MqttClient的主要配置来至于MqttConnectOptions
12512
@IocBean(name = "mqttConnectOptions")
    public MqttConnectOptions createMqttConnectOptions() {
        MqttConnectOptions options = new MqttConnectOptions();
        options.setAutomaticReconnect(conf.getBoolean(PROP_OPTIONS_AUTOMATIC_RECONNECT, true));
        options.setKeepAliveInterval(conf.getInt(PROP_OPTIONS_KEEP_ALIVE_INTERVAL, 60));
        if (!Strings.isBlank(conf.get(PROP_OPTIONS_WILL_TOPIC))) {
            options.setWill(conf.get(PROP_OPTIONS_WILL_TOPIC),
                            conf.get(PROP_OPTIONS_WILL_PAYLOAD).getBytes(),
                            conf.getInt(PROP_OPTIONS_WILL_QOS, 2),
                            conf.getBoolean(PROP_OPTIONS_WILL_RETAINED, true));
        }
        // 用户信息不一定存在,也不一定需要,所以需要判断一下是否真的要设置
        if (!Strings.isBlank(conf.get(PROP_OPTIONS_USERNAME))) {
            options.setUserName(conf.get(PROP_OPTIONS_USERNAME));
        }
        if (!Strings.isBlank(conf.get(PROP_OPTIONS_PASSWORD))) {
            options.setPassword(conf.get(PROP_OPTIONS_PASSWORD).toCharArray());
        }
        options.setCleanSession(conf.getBoolean(PROP_OPTIONS_CLEAN_SESSION, MqttConnectOptions.CLEAN_SESSION_DEFAULT));
        options.setConnectionTimeout(conf.getInt(PROP_OPTIONS_CONNECTION_TIMEOUT, MqttConnectOptions.CONNECTION_TIMEOUT_DEFAULT));
        // 事实上, urls的优先级是高于url的
        if (!Strings.isBlank(conf.get(PROP_OPTIONS_URLS))) {
            options.setServerURIs(Strings.splitIgnoreBlank(conf.get(PROP_OPTIONS_URLS), " "));
        }
        // TODO 完成SSL相关的配置
        return options;
    }
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;
  }
565210
function() {
        try {
          var server = {
            "host": params.host,
            "port": params.port,
            "timeout": params.timeout || 20000,
            "connectHeaders": {
              "host": "/",
              "login": params.username,
              "passcode": params.password,
              "heart-beat": "5000, 5000"
            }
          };

          var servers = [server];

          var reconnectOptions = {
            'maxReconnects': 20
          };

          var manager = new Stompit.ConnectFailover(servers, reconnectOptions);

          manager.connect(function(error, stompClient, reconnect) {
            if (error) {
              fireEvent("error", {
                errorCode: error.code,
                errorMessage: error.message,
                errorEvent: error
              });

              return;
            }

            client = stompClient;

            client.on('error', function(error) {
              fireEvent("error", {
                errorCode: 999,
                errorMessage: "ActiveMQ client has had an error",
                errorEvent: error
              });
              reconnect();
            });

            fireEvent("init");
          });

        } catch (error) {
          console.error(error);
        }
      }
351743
@Override
    public void onConnect(int rst, HuaweiApiClient client) {

        HMSAgentLog.d("onConnect:" + rst);

        if (client == null || !ApiClientMgr.INST.isConnect(client)) {
            HMSAgentLog.e("client not connted");
            onPMSPayEnd(rst, null);
            return;
        }

        // 调用HMS-SDK pay 接口
        PendingResult<PayResult> payResult = HuaweiPay.HuaweiPayApi.productPay(client, payReq);
        payResult.setResultCallback(new ResultCallback<PayResult>() {
            @Override
            public void onResult(PayResult result) {

                if (result == null) {
                    HMSAgentLog.e("result is null");
                    onPMSPayEnd(HMSAgent.AgentResultCode.RESULT_IS_NULL, null);
                    return;
                }

                Status status = result.getStatus();
                if (status == null) {
                    HMSAgentLog.e("status is null");
                    onPMSPayEnd(HMSAgent.AgentResultCode.STATUS_IS_NULL, null);
                    return;
                }

                int rstCode = status.getStatusCode();
                HMSAgentLog.d("status=" + status);
                // 需要重试的错误码，并且可以重试
                if ((rstCode == CommonCode.ErrorCode.SESSION_INVALID
                        || rstCode == CommonCode.ErrorCode.CLIENT_API_INVALID) && retryTimes > 0) {
                    retryTimes--;
                    connect();
                } else if (rstCode == PayStatusCodes.PAY_STATE_SUCCESS) {
                    // 支付校验完成，取当前activity进行后续支付操作
                    Activity curActivity = ActivityMgr.INST.getLastActivity();
                    if (curActivity == null) {
                        HMSAgentLog.e("activity is null");
                        onPMSPayEnd(HMSAgent.AgentResultCode.NO_ACTIVITY_FOR_USE, null);
                        return;
                    }

                    if (statusForPay != null) {
                        HMSAgentLog.e("has already a pay to dispose");
                        onPMSPayEnd(HMSAgent.AgentResultCode.REQUEST_REPEATED, null);
                        return;
                    }

                    //启动支付流程
                    try {
                        statusForPay = status;
                        Intent intent = new Intent(curActivity, HMSPMSPayAgentActivity.class);
                        intent.putExtra(BaseAgentActivity.EXTRA_IS_FULLSCREEN, UIUtils.isActivityFullscreen(curActivity));
                        curActivity.startActivity(intent);
                    } catch (Exception e) {
                        HMSAgentLog.e("start HMSPayAgentActivity error:" + e.getMessage());
                        onPMSPayEnd(HMSAgent.AgentResultCode.START_ACTIVITY_ERROR, null);
                    }
                }else {
                    onPMSPayEnd(rstCode, null);
                }
            }
        });
    }
55688
protected SofaResponse doSendMsg(ProviderInfo providerInfo, ClientTransport transport,
                                     SofaRequest request) throws SofaRpcException {
        RpcInternalContext context = RpcInternalContext.getContext();
        // 添加调用的服务端远程地址
        RpcInternalContext.getContext().setRemoteAddress(providerInfo.getHost(), providerInfo.getPort());
        try {
            checkProviderVersion(providerInfo, request); // 根据服务端版本特殊处理
            String invokeType = request.getInvokeType();
            int timeout = resolveTimeout(request, consumerConfig, providerInfo);

            SofaResponse response = null;
            // 同步调用
            if (RpcConstants.INVOKER_TYPE_SYNC.equals(invokeType)) {
                long start = RpcRuntimeContext.now();
                try {
                    response = transport.syncSend(request, timeout);
                } finally {
                    if (RpcInternalContext.isAttachmentEnable()) {
                        long elapsed = RpcRuntimeContext.now() - start;
                        context.setAttachment(RpcConstants.INTERNAL_KEY_CLIENT_ELAPSE, elapsed);
                    }
                }
            }
            // 单向调用
            else if (RpcConstants.INVOKER_TYPE_ONEWAY.equals(invokeType)) {
                long start = RpcRuntimeContext.now();
                try {
                    transport.oneWaySend(request, timeout);
                    response = buildEmptyResponse(request);
                } finally {
                    if (RpcInternalContext.isAttachmentEnable()) {
                        long elapsed = RpcRuntimeContext.now() - start;
                        context.setAttachment(RpcConstants.INTERNAL_KEY_CLIENT_ELAPSE, elapsed);
                    }
                }
            }
            // Callback调用
            else if (RpcConstants.INVOKER_TYPE_CALLBACK.equals(invokeType)) {
                // 调用级别回调监听器
                SofaResponseCallback sofaResponseCallback = request.getSofaResponseCallback();
                if (sofaResponseCallback == null) {
                    SofaResponseCallback methodResponseCallback = consumerConfig
                        .getMethodOnreturn(request.getMethodName());
                    if (methodResponseCallback != null) { // 方法的Callback
                        request.setSofaResponseCallback(methodResponseCallback);
                    }
                }
                // 记录发送开始时间
                context.setAttachment(RpcConstants.INTERNAL_KEY_CLIENT_SEND_TIME, RpcRuntimeContext.now());
                // 开始调用
                transport.asyncSend(request, timeout);
                response = buildEmptyResponse(request);
            }
            // Future调用
            else if (RpcConstants.INVOKER_TYPE_FUTURE.equals(invokeType)) {
                // 记录发送开始时间
                context.setAttachment(RpcConstants.INTERNAL_KEY_CLIENT_SEND_TIME, RpcRuntimeContext.now());
                // 开始调用
                ResponseFuture future = transport.asyncSend(request, timeout);
                // 放入线程上下文
                RpcInternalContext.getContext().setFuture(future);
                response = buildEmptyResponse(request);
            } else {
                throw new SofaRpcException(RpcErrorType.CLIENT_UNDECLARED_ERROR, "Unknown invoke type:" + invokeType);
            }
            return response;
        } catch (SofaRpcException e) {
            throw e;
        } catch (Throwable e) { // 客户端其它异常
            throw new SofaRpcException(RpcErrorType.CLIENT_UNDECLARED_ERROR, e);
        }
    }
1333724
AmqpClient startOkConnection(AmqpArguments clientProperties, String mechanism, String response, String locale) {
        this.startOkConnection(clientProperties, mechanism, response, locale, null, null);
        return this;
    }
678279
@Override
    public void onConnect(int rst, HuaweiApiClient client) {
        if (client == null || !ApiClientMgr.INST.isConnect(client)) {
            HMSAgentLog.e("client not connted");
            onSignOutResult(rst, null);
            return;
        }

        PendingResult<SignOutResult> signOutResult = HuaweiId.HuaweiIdApi.signOut(client);
        signOutResult.setResultCallback(new ResultCallback<SignOutResult>() {
            @Override
            public void onResult(SignOutResult result) {
                if (result == null) {
                    HMSAgentLog.e("result is null");
                    onSignOutResult(HMSAgent.AgentResultCode.RESULT_IS_NULL, null);
                    return;
                }

                Status status = result.getStatus();
                if (status == null) {
                    HMSAgentLog.e("status is null");
                    onSignOutResult(HMSAgent.AgentResultCode.STATUS_IS_NULL, null);
                    return;
                }

                int rstCode = status.getStatusCode();
                HMSAgentLog.d("status=" + status);
                // 需要重试的错误码，并且可以重试
                if ((rstCode == CommonCode.ErrorCode.SESSION_INVALID
                        || rstCode == CommonCode.ErrorCode.CLIENT_API_INVALID) && retryTimes > 0) {
                    retryTimes--;
                    connect();
                } else {
                    onSignOutResult(rstCode, result);
                }
            }
        });
    }
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);
}
1051571
function(args) {
    var mqttClient = this.mqttClient;
    return promised(function(cb) {
      mqttClient.subscribe(args.channel, _getPahoCallbacks(cb));
    });
  }
902528
@Override
	protected ThriftConnection<T> getConnectionInternal() throws ThriftConnectionPoolException {
		ThriftConnection<T> result = pollConnection();
		// 如果立即获取连接失败 则换一个分区继续获取
		// TODO 设置当连接获取超时返回null？
		if (result == null) {
			if (this.pool.getThriftServerCount() == 0) {
				throw new ThriftConnectionPoolException("当前没有可用的服务器  无法获取连接");
			}
			int partition = (int) (Thread.currentThread().getId() % this.pool.thriftServerCount);

			ThriftConnectionPartition<T> thriftConnectionPartition = this.pool.partitions.get(partition);

			try {
				result = thriftConnectionPartition.poolFreeConnection(this.pool.connectionTimeoutInMs,
						TimeUnit.MILLISECONDS);
				if (result == null) {

					throw new ThriftConnectionPoolException("Timed out waiting for a free available connection.");
				}
			} catch (InterruptedException e) {
				throw new ThriftConnectionPoolException(e);
			}
		}
		return result;
	}
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;
        }
    }
954670
AmqpClient tuneOkConnection(int channelMax, int frameMax, int heartbeat) {
        this.tuneOkConnection(channelMax, frameMax, heartbeat, null, null);
        return this;
    }
547279
function NodeMqttTransport(options) {
  Transport.call(this, options);

  this._options = options;
  this._client = null;
  this._sendTimer = null;
  this._buf = [];

  this._status = '';

  this._connHandler = onConnect.bind(this);
  this._messageHandler = onMessage.bind(this);
  this._sendOutHandler = sendOut.bind(this);
  this._disconnHandler = onDisconnect.bind(this);
  this._errorHandler = onError.bind(this);

  init(this);
}
1177274
private void startConnect() {

        // 触发一次连接将重试次数减1
        curLeftResolveTimes--;

        HMSAgentLog.d("start thread to connect");
        ThreadUtil.INST.excute(new Runnable() {
            @Override
            public void run() {
                HuaweiApiClient client =  getApiClient();

                if (client != null) {
                    HMSAgentLog.d("connect");
                    Activity curActivity = ActivityMgr.INST.getLastActivity();

                    // 考虑到有cp后台需要调用接口，HMSSDK去掉了activity不能为空的判断。这里只是取当前activity，可能为空
                    timeoutHandler.sendEmptyMessageDelayed(APICLIENT_TIMEOUT_HANDLE_MSG, APICLIENT_CONNECT_TIMEOUT);
                    client.connect(curActivity);
                } else {
                    HMSAgentLog.d("client is generate error");
                    onConnectEnd(HMSAgent.AgentResultCode.RESULT_IS_NULL);
                }
            }
        });
    }
167405
@Override
    public void onConnect(int rst, HuaweiApiClient client) {

        HMSAgentLog.d("onConnect:" + rst);

        if (client == null || !ApiClientMgr.INST.isConnect(client)) {
            HMSAgentLog.e("client not connted");
            onProductDetailResult(rst, null);
            return;
        }

        // 调用HMS-SDK getOrderDetail 接口
        PendingResult<ProductDetailResult> checkPayResult = HuaweiPay.HuaweiPayApi.getProductDetails(client, productDetailReq);
        checkPayResult.setResultCallback(new ResultCallback<ProductDetailResult>() {
            @Override
            public void onResult(ProductDetailResult result) {

                if (result == null) {
                    HMSAgentLog.e("result is null");
                    onProductDetailResult(HMSAgent.AgentResultCode.RESULT_IS_NULL, null);
                    return;
                }

                Status status = result.getStatus();
                if (status == null) {
                    HMSAgentLog.e("status is null");
                    onProductDetailResult(HMSAgent.AgentResultCode.STATUS_IS_NULL, null);
                    return;
                }

                int rstCode = status.getStatusCode();
                HMSAgentLog.d("status=" + status);
                // 需要重试的错误码，并且可以重试
                if ((rstCode == CommonCode.ErrorCode.SESSION_INVALID
                        || rstCode == CommonCode.ErrorCode.CLIENT_API_INVALID) && retryTimes > 0) {
                    retryTimes--;
                    connect();
                } else {
                    onProductDetailResult(rstCode, result);
                }
            }
        });
    }
888083
function (client, backlog, channel, message) {
    if ( addToBacklog(client, backlog, publishBrowser, [client, backlog, channel, message]) ) return;
    message = new mqtt_lib.Message(message);
    message.destinationName = channel;
    client.send(message);
}
1220447
function (config, options) {
    //获取Mq的配置
    Mq.call(this, config, options);
    this.queue = options.queueName;
    this.consumers = [];
    this.isAlive = false;
    this.options = options || {};
}