MQ类，建立channel
@constructor
968688
function (config, options) {
    this.config = config;
    this.options = options;
    this.channel = null;
    this._status = CONNECT_STATUS.unconnected;
}
1313934
function execute(req, res) {
  // must coerce channel to string
  var len = this.state.pubsub.publish(
    req.conn, '' + req.args[0], req.args[1]);
  res.send(null, len);
}
617252
function(arr, req, res, indx, cb) {
    if (arr.length === indx) {
      cb();
    } else {
      arr[indx](req, res, function() {
        middleware(arr, req, res, ++indx, cb);
      });
    }
  }
1003452
function CommunicationAbstract(object, layerID, assignID) {
		var self = this;

		this.assignID = assignID;
		this.layerID = layerID;
		this.online = false;
		this.object = object;
		this.handlers = {};
		this.handle = null;

		this.on('message', function (message) {
			this.handle.handleMessage(message);
		});

		this.on('ready', function () {
			self.online = true;
		});

		this.on('close', function () {
			self.online = false;
		});
	}
1161486
function initiateSubscribe(clientObj) {
  this.__channels = {};
  var self = this;
  clientObj.subscribe('internalPing');
  clientObj.on('message', function(channelName, message) {
    if (channelName === 'internalPing') return;  // this is our internal channel for ping
    if (typeof self.__channels[channelName] === 'undefined') return; // we have no channel.
    try {
      var data = JSON.parse(message);
    } catch (e) {
      data = message;
    }
    for (var i = 0; i < self.__channels[channelName].length; i++) {
      self.__channels[channelName][i](data);
    }
  });
}
507954
private synchronized void createChannel(String channel) {
        if (!channelExists(channel)) {
            Queue queue = new Queue(channel, true, false, true);
            admin.declareQueue(queue);
            Binding binding = new Binding(channel, DestinationType.QUEUE, exchange.getName(), channel + ".#", null);
            admin.declareBinding(binding);
        }
    }
986052
function(callback){
        if(source['brokerId']!='null'){
            db.db(dbName).collection(collectionName).find({brokerId:source['brokerId']}).toArray(function(err,result){
                mqttHost = result[0]['ipAddress'];
                mqttPort = result[0]['port'];
                console.log(result);
            });
        }else{
            mqttHost = mqttPort = null;
        }
    }
832209
function RpcChannel(connection) {
    if (!connection) {
        var msg = 'The connection is mandatory!';
        logger.error(msg);
        throw new TypeError(msg);
    }
    this._connection = connection;
    this._open = true;
    this._queue = [];
    var self = this;
    setInterval(function () {
        executor.call(self);
    }, 100).unref();
}
971893
function(channel) {
			return function() {
				var args = Array.prototype.slice.call(arguments);
				args.unshift(__class+':');
				console[channel].apply(console, args);
			};
		}
468656
function AMQPConnection(transport, id, receive, connReady) {
  this.transport = transport;
  this.id = id;
  this.channel = null;

  var me = this;
  this.ready = new Promise(function (resolve, reject) {
    connReady.then(function () {
      me.transport.connection.createChannel(function (err, ch) {
        if (err != null) {
          console.error(err);
          reject();
        } else {
          me.channel = ch;
          ch.assertQueue(me.id);
          ch.consume(me.id, function (message) {
              if (message !== null && message.content && message.content.toString() != "") {
                var body = JSON.parse(message.content.toString());
                if (body.to != me.id) {
                  console.warn("Received message not meant for me?", body);
                } else {
                  receive(body.from, body.message);
                }
                ch.ack(message);
              }
            }
          );
          resolve();
        }
      });
    });
  })
}
328232
function(channel){
    
        var tokenInfo   = this._tokenData,
            reader      = this._reader,
            value,
            i           =0,
            len         = tokenInfo.length,
            found       = false,
            token,
            info;
            
        //check the lookahead buffer first
        if (this._lt.length && this._ltIndex >= 0 && this._ltIndex < this._lt.length){  
                           
            i++;
            this._token = this._lt[this._ltIndex++];
            info = tokenInfo[this._token.type];
            
            //obey channels logic
            while((info.channel !== undefined && channel !== info.channel) &&
                    this._ltIndex < this._lt.length){
                this._token = this._lt[this._ltIndex++];
                info = tokenInfo[this._token.type];
                i++;
            }
            
            //here be dragons
            if ((info.channel === undefined || channel === info.channel) &&
                    this._ltIndex <= this._lt.length){
                this._ltIndexCache.push(i);
                return this._token.type;
            }
        }
        
        //call token retriever method
        token = this._getToken();

        //if it should be hidden, don't save a token
        if (token.type > -1 && !tokenInfo[token.type].hide){
                     
            //apply token channel
            token.channel = tokenInfo[token.type].channel;
         
            //save for later
            this._token = token;
            this._lt.push(token);

            //save space that will be moved (must be done before array is truncated)
            this._ltIndexCache.push(this._lt.length - this._ltIndex + i);  
        
            //keep the buffer under 5 items
            if (this._lt.length > 5){
                this._lt.shift();                
            }
            
            //also keep the shift buffer under 5 items
            if (this._ltIndexCache.length > 5){
                this._ltIndexCache.shift();
            }
                
            //update lookahead index
            this._ltIndex = this._lt.length;
        }
            
        /*
         * Skip to the next token if:
         * 1. The token type is marked as hidden.
         * 2. The token type has a channel specified and it isn't the current channel.
         */
        info = tokenInfo[token.type];
        if (info && 
                (info.hide || 
                (info.channel !== undefined && channel !== info.channel))){
            return this.get(channel);
        } else {
            //return just the type
            return token.type;
        }
    }
1268683
function assetQueue(addr, queue, options) {
  return createChannel(addr).then(data => {
    let ch = data[1];

    return ch.assertQueue(queue, options).then(q => [ch, q]);
  });
}
663832
private void handle(SelectionKey key) {
		// 有客户端接入此服务端
		if (key.isAcceptable()) {
			// 获取通道 转化为要处理的类型
			final ServerSocketChannel server = (ServerSocketChannel) key.channel();
			SocketChannel socketChannel;
			try {
				// 获取连接到此服务器的客户端通道
				socketChannel = server.accept();
			} catch (IOException e) {
				throw new IORuntimeException(e);
			}

			// SocketChannel通道的可读事件注册到Selector中
			registerChannel(selector, socketChannel, Operation.READ);
		}

		// 读事件就绪
		if (key.isReadable()) {
			final SocketChannel socketChannel = (SocketChannel) key.channel();
			read(socketChannel);

			// SocketChannel通道的可写事件注册到Selector中
			registerChannel(selector, socketChannel, Operation.WRITE);
		}

		// 写事件就绪
		if (key.isWritable()) {
			final SocketChannel socketChannel = (SocketChannel) key.channel();
			write(socketChannel);
			// SocketChannel通道的可读事件注册到Selector中
			registerChannel(selector, socketChannel, Operation.READ);
		}
	}
128544
function processChannel(channel) {
    if (0 == bestAlpha) {
      return bottom[channel]
    } else {
      return bottom[channel] + (top[channel] - bottom[channel]) / bestAlpha
    }
  }
845740
function Channel(session, wsConn) {
  DEBUG("new Channel()")
  this.session     = session
  this.wsConn      = wsConn
  this.agents      = new SMap()
  this.nextAgentID = session.nextAgentID

  this.session.addChannel(this)

  wsConn.addListener("message", onMessage)
  wsConn.addListener("close",   onClose)
  wsConn.addListener("error",   onError)

  var channel = this

  //-----------------------------------
  function onMessage(message) {
    if (message.type != "utf8") return

    DEBUGMSG("recv: " + message.utf8Data)
    message = JSON.parse(message.utf8Data)

    if      (message.type == "request")  session.onRequest(channel, message)
    else if (message.type == "response") session.onResponse(channel, message)
    else if (message.type == "event")    session.onEvent(channel, message)
  }

  //-----------------------------------
  function onClose() {
    DEBUG("channel.onClose()")
    channel.eachAgent(function(agent){
      agent.shutDown()
    })

    channel.session.delChannel(this)
  }

  //-----------------------------------
  function onError(err) {
    DEBUG("channel.onErr():" +  err)

    if (err.errno == "ECONNRESET") return

    console.log("websocket error: " + err)
    console.log(err.stack)
  }
}
239165
function (ch) {
  if (ch.hasInterval('M3')) {
    return ch.hasInterval('m7') ? 'x' : 'M';
  }

  if (ch.hasInterval('m3')) {
    if (ch.hasInterval('dim5')) {
      return ch.hasInterval('m7') ? 'ø' : 'o';
    }

    return 'm';
  }

  if (ch.hasInterval('P4')) {
    return 's';
  }

  return null;
}
407877
function call(channel, args, onError) {
        
        var result;
        var nextCalls = 0;
        var version = channel.split("@")[1] || "1.x";
        var channelName = normalizeChannelName(channel);
        var subscribers = prioritize(channelName in channels ? channels[channelName] : {}, version);
        
        var decorations = prioritize(
            merge(channelName in decorators ? decorators[channelName] : {}, decorators["*"] || {})
        );
        
        args = args || [];
        assert(Array.isArray(args), "Parameter `args` must be an array.");
        
        if (!subscribers.length) {
            throw new Error("No implementation connected for channel `" + channel + "`!");
        }
        
        subscribers.reverse().some(function (subscriber) {
            
            var newValue, lastValue, decorated;
            var decos = decorations.slice();
            
            decorated = decos.filter(versionMatches).map(function (decorator) {
                return {
                    fn: decorator.fn(next, {
                        method: "call",
                        channel: channel
                    }),
                    config: decorator.config
                };
            });
            
            decorated.push(subscriber);
            
            function versionMatches(decorator) {
                
                if (decorator.channel === "*") {
                    return true;
                }
                
                return semver.satisfies(subscriber.version, decorator.version);
            }
            
            function next() {
                
                var current, lastCallCount;
                var value = lastValue;
                
                nextCalls += 1;
                
                while (decorated.length) {
                    
                    current = decorated.shift();
                    
                    if (!decorated.length) {
                        
                        try {
                            value = current.fn.apply(null, arguments);
                        }
                        catch (error) {
                            error.mustReThrow = true;
                            throw error;
                        }
                        
                        break;
                    }
                    
                    try {
                        
                        lastCallCount = nextCalls;
                        value = current.fn.apply(null, arguments);
                        lastValue = value;
                        
                        if (nextCalls === lastCallCount) {
                            break;
                        }
                    }
                    catch (error) {
                        if (error && typeof error === "object" && error.mustReThrow) {
                            throw error;
                        }
                        else {
                            
                            if (current.config.onError) {
                                current.config.onError(error);
                            }
                            
                            bus.emit("decoratorError", current);
                        }
                    }
                    
                }
                
                return value;
            }
            
            try {
                
                newValue = next.apply(null, args);
                result = newValue;
                
                return true;
            }
            catch (error) {
                
                if (subscriber.config.onError) {
                    subscriber.config.onError(error);
                }
                
                if (onError) {
                    onError(error);
                }
                
                bus.emit("subscriberError", subscriber);
            }
            
            return false;
        });
        
        return result;
    }