连接mqtt server,并返回一个客户端对象,如果连接失败,那么返回null
351626
public MqttAsyncClient connectBroker() {
        LOG.info(String.format("mqtt=======客户端%s与rabbitMQ server: %s 准备建立连接,userName = %s", getMqttClientId(), JSON.toJSONString(serverURIs), userName));
        try {
            sampleClient = new MqttAsyncClient("tcp://overriddenByMqttConnectOptions.setServerURIs:1883", getMqttClientId(), persistence);
            connOpts = new MqttConnectOptions();
            connOpts.setAutomaticReconnect(true);
            connOpts.setServerURIs(serverURIs);
            connOpts.setUserName(userName);
            connOpts.setPassword(getPwd());
            connOpts.setCleanSession(cleanSession);
            connOpts.setMaxInflight(1000 /**默认的值是10,对于我们来说这个值太小!*/);
            connOpts.setKeepAliveInterval(keepAliveInterval);
            sampleClient.setCallback(getCallback(this));
            sampleClient.connect(connOpts).waitForCompletion(60 * 1000);
            LOG.info(String.format("mqtt=======客户端%s与rabbitMQ server: %s 建立连接完成,userName = %s", getMqttClientId(), JSON.toJSONString(serverURIs), userName));
            return sampleClient;
        } catch (MqttException me) {
            throw new RuntimeException(String.format("mqtt=======客户端%s与rabbitMQ server: %s 连接失败!!! userName = %s", getMqttClientId(), JSON.toJSONString(serverURIs), userName), me);
        }
    }
755877
function ( arg, qstring ) {
		var obj    = {},
		    result = qstring !== undefined ? ( qstring.indexOf( "?" ) > -1 ? qstring.replace( /.*\?/, "" ) : null) : ( server || string.isEmpty( location.search ) ? null : location.search.replace( "?", "" ) );

		if ( result !== null && !string.isEmpty( result ) ) {
			result = result.split( "&" );
			array.each( result, function (prop ) {
				var item = prop.split( "=" );

				if ( string.isEmpty( item[0] ) ) {
					return;
				}

				if ( item[1] === undefined ) {
					item[1] = "";
				}
				else {
					item[1] = utility.coerce( decodeURIComponent( item[1] ) );
				}

				if ( obj[item[0]] === undefined ) {
					obj[item[0]] = item[1];
				}
				else if ( !(obj[item[0]] instanceof Array) ) {
					obj[item[0]] = [obj[item[0]]];
					obj[item[0]].push( item[1] );
				}
				else {
					obj[item[0]].push( item[1] );
				}
			});
		}

		if ( arg !== null && arg !== undefined ) {
			obj = obj[arg];
		}

		return obj;
	}
672639
function(config) {
  
    //console.log('RtcConnector: createMqttClient: server:  '+ config.server);

     var client = mqtt.connect(config.server + ':' + config.port);
     client.on('message', function(topic, message) {
       /*
        * This is called when a topic matching a valid subsription is received
        *   We have a valid subscription somewhere... 
        *   TODO:  This must call the filter FOR the topic and ensure it hasn't already
        *   been called.  
        */
       l('MESSAGE') && console.log('RtcConnector.mqttClient - Received Message on topic: '+topic);
        try {
          doFilter(topic,message);
        } catch(e) {
          console.error('MqttClient Message callback chain failure:',e);
          console.log(e.stack);
        }
      });
    client.on('connect', function(error) {
      console.log('RtcConnector: mqtt is CONNECTED');
      this.connected = true;
      this.emit('connected');
    }.bind(this));
    client.on('error', function(error) {
      console.log('RtcConnector: DISCONNECTED - error:'+ error);
      this.connected = false;
      this.emit('disconnected', error);
    }.bind(this));
    client.on('close', function(error) {
      console.log('RtcConnector: DISCONNECTED - close: '+error);
      this.connected = false;
      this.emit('disconnected',error);
    }.bind(this));
    return client;
  }
1256621
@Override
  public InetSocketAddress ipRemote()
  {
    SocketChannel s = _channel;
    
    if (s != null) {
      try {
        return (InetSocketAddress) s.getRemoteAddress();
      } catch (IOException e) {
        return null;
        //throw new RuntimeException(e);
      }
    }
    else {
      return null;
    }
  }
522010
def shift_string(buffer)
      len = shift_short(buffer)
      str = shift_data(buffer, len)
      # Strings in MQTT v3.1 are all UTF-8
      str.force_encoding('UTF-8')
    end
916772
public void publish(String subtopic, Object data) {
    try {
      MqttConnection.publish(settingsController.getMqttSettings(), subtopic, data);
    } catch (JsonProcessingException | MqttException e) {
      logger.error("Failed to publish MQTT message", e);
    }
  }
533691
def _parse_message_to_mqtt(self, data):
        
        msg = Message(data, self)
        payload = str(msg.payload)
        msg.payload = ''
        # prefix/node/child/type/ack/subtype : payload
        return ('{}/{}'.format(self._out_prefix, msg.encode('/'))[:-2],
                payload, msg.ack)
371257
function send() {
  var message = arguments.length <= 0 || arguments[0] === undefined ? 'hello mqtt-controls' : arguments[0];
  var topic = arguments.length <= 1 || arguments[1] === undefined ? null : arguments[1];

  var t = undefined;
  if (topic === null) {
    t = topics.publish;
  } else {
    t = topic;
  }
  client.publish(t, message);
}
1275241
function (data) {
        var that = this;
        if (!data) {
            if (this._socketTimeout) {
                clearTimeout(this._socketTimeout);  //just in case
            }

            //waiting on data.
            //logger.log("server: waiting on hello");
            this._socketTimeout = setTimeout(function () {
                that.handshakeFail("get_hello timed out");
            }, 30 * 1000);

            return;
        }
        clearTimeout(this._socketTimeout);

        var env = this.client.parseMessage(data);
        var msg = (env && env.hello) ? env.hello : env;
        if (!msg) {
            this.handshakeFail("failed to parse hello");
            return;
        }
        this.client.recvCounter = msg.getId();
        //logger.log("server: got a good hello! Counter was: " + msg.getId());

        try {
            if (msg.getPayload) {
                var payload = msg.getPayload();
                if (payload.length > 0) {
                    var r = new buffers.BufferReader(payload);
                    this.client.spark_product_id = r.shiftUInt16();
                    this.client.product_firmware_version = r.shiftUInt16();
                    //logger.log('version of core firmware is ', this.client.spark_product_id, this.client.product_firmware_version);
                }
            }
            else {
                logger.log('msg object had no getPayload fn');
            }
        }
        catch (ex) {
            logger.log('error while parsing hello payload ', ex);
        }

//        //remind ourselves later that this key worked.
//        if (that.corePublicKeyWasUncertain) {
//            process.nextTick(function () {
//                try {
//					//set preferred key for device
//					//that.coreFullPublicKeyObject
//                }
//                catch (ex) {
//                    logger.error("error marking key as valid " + ex);
//                }
//            });
//        }


        this.stage++;
        this.nextStep();
    }
819087
function server(options) {
  status = 'server.started'
  config = _.extend(defaultServerOptions, options)

  open(config.datafile)

  if (config.logToConsole)
    console.log("✔ log to stdout")

  // ws [
  var ServerMessageLog = {}
  var queueMessages = []
  var timer

  function flushMessages(n) {
    queueMessages = _.sortBy(queueMessages, 'time')
    var messages = queueMessages.slice(0, n)
    queueMessages = queueMessages.slice(n)
//    console.log(messages);
    _.each(messages, function (m) {
//      var filename = options.logpath+'/1.log'
//      logString(filename, m.text)
      logMessage(new Date(m.time), m.path, m.data, m.type == 'log' ? '' : m.type)
    })
  }

  console.log("✔ ws server listening on port %d", config.port)

  var wss = new WebSocketServer({port: config.port});

  wss.on('connection', function(ws) {
    ws.on('message', function(m) {
      var message = JSON.parse(m)
      queueMessages.push(message)
      if (queueMessages.length > config.queueMessagesMax)
        flushMessages(config.flushMessagesCount)

      if (timer)
        clearTimeout(timer)
      timer = setTimeout(function () {
        timer = null
        flushMessages(queueMessages.length)
      }, config.serverFlushTimeout)
    })
  })
  // ws ]
}
1125815
function getServer(client, type, server, port, cb) {
    var serverType    = type || 'chat';
    var serverAddress = server || null;
    var serverPort    = port || 443;

    if (serverAddress === null) {
        var serverTypes = ['chat', 'events', 'groups'];
        if (serverTypes.indexOf(serverType) === -1) {
            serverType = 'chat';
        }

        var serverPortsChat = [80, 443, 6667];
        var serverPorts     = [80, 443];
        if (serverType === 'chat' && serverPortsChat.indexOf(serverPort) === -1) {
            serverPort = 443;
        } else if (serverType !== 'chat' && serverPorts.indexOf(serverPort) === -1) {
            serverPort = 443;
        }

        function findServer(cb) {
            function scan() {
                var serverAddr = serverList[serverType][serverPort][Math.floor(Math.random() * (serverList[serverType][serverPort].length))];
                if (unavailable.indexOf(serverAddr) >= 0) {
                    scan();
                } else {
                    probeServer(serverAddr, serverPort, function (err, available) {
                        if (!available || err) {
                            unavailable.push(serverAddr);
                            if (unavailable.length > serverList[serverType][serverPort].length-1) {
                                unavailable = [];
                                setTimeout(function(){
                                    if (!client.connected) { scan(); }
                                }, 60000);
                            } else {
                                return scan();
                            }
                        } else {
                            unavailable = [];
                            return cb(serverAddr);
                        }
                    });
                }
            }
            scan();
        }

        client.logger.info('Searching for a Twitch server..');
        findServer(function(server) {
            if (server) {
                cb(server + ':' + serverPort);
            }
        });
    }
    return cb(serverAddress + ':' + serverPort);
}
480006
public MqttSettings getMqttSettings() {
    MqttSettings settings = new MqttSettings();
    settings.setServerUrl(getSettingValue("mqtt.serverUrl"));
    settings.setClientUrl(getSettingValue("mqtt.clientUrl"));    
    settings.setTopic(getSettingValue("mqtt.topic"));
    settings.setWildcard(getSettingValue("mqtt.wildcard"));
    return settings;
  }
1104006
def _parse_mqtt_to_message(self, topic, payload, qos):
        
        topic_levels = topic.split('/')
        topic_levels = not_prefix = topic_levels[-5:]
        prefix_end_idx = topic.find('/'.join(not_prefix)) - 1
        prefix = topic[:prefix_end_idx]
        if prefix != self._in_prefix:
            return None
        if qos and qos > 0:
            ack = '1'
        else:
            ack = '0'
        topic_levels[3] = ack
        topic_levels.append(str(payload))
        return ';'.join(topic_levels)
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();
        }
      });
    });
  })
}
1206267
public function connect()
    {
        /*
         The Server MUST process a second CONNECT Packet sent from a Client as a protocol violation
         and disconnect the Client [MQTT-3.1.0-2]

         So CONNECT SHOULD be a blocking connection.
         */
        $this->socket->connect(function ($cli) {
            Debug::Log(Debug::INFO, 'connect(): Connection established.');
            /**
             * @var Message\CONNECT $connectobj
             */
            $connectobj = $this->getMessageObject(Message::CONNECT);

            if (!$this->connect_clean && empty($this->clientid)) {
                throw new Exception('Client id must be provided if Clean Session flag is set false.');
            }

            # default client id
            if (empty($this->clientid)) {
                $this->clientid = 'mqtt' . substr(md5(uniqid('mqtt', true)), 8, 16);
            } else {
                $connectobj->setClientId($this->clientid);
            }
            Debug::Log(Debug::DEBUG, 'connect(): clientid=' . $this->clientid);

            $connectobj->setKeepalive($this->keepalive);
            Debug::Log(Debug::DEBUG, 'connect(): keepalive=' . $this->keepalive);

            $connectobj->setAuth($this->username, $this->password);
            Debug::Log(Debug::DEBUG, 'connect(): username=' . $this->username . ' password=' . $this->password);

            $connectobj->setClean($this->connect_clean);

            if ($this->connect_will instanceof Will) {
                $connectobj->setWill($this->connect_will);
            }

            $length = 0;

            $bytes_written = $this->message_write($connectobj, $length);
            Debug::Log(Debug::DEBUG, 'connect(): bytes written=' . $bytes_written);

            $this->keepalive_time = swoole_timer_tick($this->keepalive * 500, [$this, 'keepalive']);
        }, [$this, 'onReceive'],function (){
            secho("MQTT", "连接mqtt服务器失败,正在重试");
            swoole_timer_after(1000,[$this,'connect']);
        },function (){
            if ($this->keepalive_time!=null) {
                swoole_timer_clear($this->keepalive_time);
                $this->keepalive_time = null;
            }
            $this->connect();
        });
    }
956966
function connect () {
  // connection
  if (!alreadyConnectedOnce) {
    if (connectedPromise === null) connection = createConnection()
    var url = AppConfig.apiUrl + 'websocket/chat'
    if (accessToken) {
      token = accessToken
      url += '?access_token=' + accessToken
    }
    socket = new SockJS(url)
    stompClient = webstomp.over(socket)
    // stompClient.debug = null
    console.tron.log(`Connecting to ${url}`)
    var headers = {}
    stompClient.connect(headers, onConnect, onError)
    alreadyConnectedOnce = true
  }
}
103432
def connect(clientid = nil)
      @client_id = clientid unless clientid.nil?

      if @client_id.nil? || @client_id.empty?
        raise 'Must provide a client_id if clean_session is set to false' unless @clean_session

        # Empty client id is not allowed for version 3.1.0
        @client_id = MQTT::Client.generate_client_id if @version == '3.1.0'
      end

      raise 'No MQTT server host set when attempting to connect' if @host.nil?

      unless connected?
        # Create network socket
        tcp_socket = TCPSocket.new(@host, @port)

        if @ssl
          # Set the protocol version
          ssl_context.ssl_version = @ssl if @ssl.is_a?(Symbol)

          @socket = OpenSSL::SSL::SSLSocket.new(tcp_socket, ssl_context)
          @socket.sync_close = true

          # Set hostname on secure socket for Server Name Indication (SNI)
          @socket.hostname = @host if @socket.respond_to?(:hostname=)

          @socket.connect
        else
          @socket = tcp_socket
        end

        # Construct a connect packet
        packet = MQTT::Packet::Connect.new(
          :version => @version,
          :clean_session => @clean_session,
          :keep_alive => @keep_alive,
          :client_id => @client_id,
          :username => @username,
          :password => @password,
          :will_topic => @will_topic,
          :will_payload => @will_payload,
          :will_qos => @will_qos,
          :will_retain => @will_retain
        )

        # Send packet
        send_packet(packet)

        # Receive response
        receive_connack

        # Start packet reading thread
        @read_thread = Thread.new(Thread.current) do |parent|
          Thread.current[:parent] = parent
          receive_packet while connected?
        end
      end

      return unless block_given?

      # If a block is given, then yield and disconnect
      begin
        yield(self)
      ensure
        disconnect
      end
    end