package com.hdl.sdk.link.core.utils.mqtt;
|
|
import android.os.SystemClock;
|
import android.text.TextUtils;
|
|
import com.alibaba.fastjson.JSON;
|
import com.hdl.sdk.link.common.utils.LogUtils;
|
import com.hdl.sdk.link.common.utils.ThreadToolUtils;
|
import com.hdl.sdk.link.core.bean.LinkPacket;
|
import com.hdl.sdk.link.core.bean.eventbus.BaseEventBus;
|
import com.hdl.sdk.link.core.bean.eventbus.EventBindMiniRemoteSuccessInfo;
|
import com.hdl.sdk.link.core.bean.eventbus.EventNotifyRefreshGatewayAesKeyInfo;
|
import com.hdl.sdk.link.core.bean.gateway.GatewayBean;
|
import com.hdl.sdk.link.core.config.HDLLinkConfig;
|
import com.hdl.sdk.link.core.connect.HDLConnectHelper;
|
import com.hdl.sdk.link.core.protocol.LinkMessageDecoder;
|
import com.hdl.sdk.link.core.utils.AesUtil;
|
import com.hdl.sdk.link.core.utils.ByteUtils;
|
import com.hdl.sdk.link.core.utils.JsonUtils;
|
import com.hdl.sdk.link.core.utils.QueueUtils;
|
import com.hdl.sdk.link.gateway.HDLLinkLocalGateway;
|
import com.hdl.sdk.link.socket.bean.Packet;
|
|
import org.eclipse.paho.client.mqttv3.IMqttActionListener;
|
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
|
import org.eclipse.paho.client.mqttv3.IMqttToken;
|
import org.eclipse.paho.client.mqttv3.MqttAsyncClient;
|
import org.eclipse.paho.client.mqttv3.MqttCallbackExtended;
|
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
|
import org.eclipse.paho.client.mqttv3.MqttException;
|
import org.eclipse.paho.client.mqttv3.MqttMessage;
|
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
|
import org.greenrobot.eventbus.EventBus;
|
|
import java.util.ArrayList;
|
import java.util.List;
|
import java.util.concurrent.ExecutorService;
|
|
/**
|
* Created by Zoro on 2018/8/1.
|
* desc:2019/7/24 连接成功以后再订阅 不需要发送心跳 不需要自己处理重连
|
*/
|
|
public class MqttRecvClient {
|
private final String TAG = "MqttRecvClient";
|
private static volatile MqttRecvClient mqttRecvClient;
|
private MqttAsyncClient sampleClient;
|
private String mBroker;
|
private String mClientId;
|
private String mUserName;
|
private String mPassWord;
|
private final String[] ignoreTopics = new String[]{"/thing/topo/found", "/ota/device/progress/up"};
|
/**
|
* 上次的主题需要记录 更改主题的时候需要取消订阅
|
*/
|
private List<String> lastTopicFilters = new ArrayList<>();
|
private final int connectionTimeout = 8;//秒
|
private final int keepAliveInterval = 10;
|
|
/**
|
* 设置远程连接参数
|
*
|
* @param broker 地址
|
* @param clientId 客户端Id
|
* @param userName 用户名
|
* @param pwd 密码
|
*/
|
public void setConnectParam(String broker, String clientId, String userName, String pwd) {
|
mBroker = broker;
|
mClientId = clientId;
|
mUserName = userName;
|
mPassWord = pwd;
|
}
|
|
/**
|
* 发送数据
|
*
|
* @param topic
|
* @param bytes
|
*/
|
public void send(String topic, byte[] bytes) {
|
try {
|
if (TextUtils.isEmpty(topic) || bytes == null) {
|
LogUtils.e(TAG, "数据不发送,Topic:" + topic + " Bytes:" + bytes);
|
return;
|
}
|
checkAndsubscribeAllTopics(topic);
|
publish(topic, bytes);
|
} catch (MqttException e) {
|
e.printStackTrace();
|
}
|
}
|
|
/**
|
* 发布
|
*
|
* @param topic 主题
|
* @param bytes 内容
|
* @throws MqttException
|
*/
|
public void publish(String topic, byte[] bytes) throws MqttException {
|
if (sampleClient == null) {
|
LogUtils.i(TAG, "Mqtt未初始化");
|
return;
|
}
|
//如果还没有连接成功，可以试着休眠等一下。这种情况可能是刚开始连接，接着马上发送数据
|
if (!sampleClient.isConnected()) {
|
reConnect();
|
SystemClock.sleep(500);
|
}
|
//回复时，mqtt主题中的方向要变化，要做方向替换
|
/**
|
* 从网关会出现mqtt发出去失败 所以改成模式2
|
*/
|
int qos = 0;
|
if(topic.contains("/custom/native/zigbee/down/slaveoid/")){
|
qos = 2;
|
}
|
sampleClient.publish(topic, bytes, qos, false, null, new IMqttActionListener() {
|
@Override
|
public void onSuccess(IMqttToken asyncActionToken) {
|
LogUtils.d(TAG, "mqtt发送成功,Topic:" + topic);
|
}
|
|
@Override
|
public void onFailure(IMqttToken asyncActionToken, Throwable exception) {
|
LogUtils.d(TAG, "mqtt发送失败,Topic:" + topic);
|
}
|
});
|
}
|
|
/**
|
* 使用的时候需要判断非空
|
*/
|
public static MqttRecvClient getInstance() {
|
if (null == mqttRecvClient) {
|
synchronized (MqttRecvClient.class) {
|
if (null == mqttRecvClient) {
|
mqttRecvClient = new MqttRecvClient();
|
}
|
}
|
}
|
return mqttRecvClient;
|
}
|
|
/**
|
* 重新连接
|
*/
|
public void reConnect() {
|
try {
|
if (sampleClient == null) {
|
return;
|
}
|
LogUtils.i(TAG, "mqtt重新连接");
|
//重新连接
|
sampleClient.reconnect();
|
} catch (Exception e) {
|
}
|
}
|
|
/**
|
* 连接mqtt服务器
|
*/
|
public void connect() {
|
try {
|
//先关闭之前的连接
|
disConnect();
|
|
if (TextUtils.isEmpty(mBroker) || TextUtils.isEmpty(mClientId) || TextUtils.isEmpty(mUserName) || TextUtils.isEmpty(mPassWord)) {
|
LogUtils.i(TAG, "连接参数为空 mBroker:" + mBroker + " mClientId:" + mClientId + " mUserName:" + mUserName + " mPassWord:" + mPassWord);
|
return;
|
}
|
sampleClient = new MqttAsyncClient(mBroker, mClientId, new MemoryPersistence());
|
MqttConnectOptions connOpts = new MqttConnectOptions();
|
connOpts.setUserName(mUserName);
|
// connOpts.setServerURIs(new String[]{mBroker});
|
connOpts.setPassword(mPassWord.toCharArray());
|
connOpts.setCleanSession(true);
|
connOpts.setKeepAliveInterval(keepAliveInterval);
|
connOpts.setAutomaticReconnect(true);
|
connOpts.setConnectionTimeout(connectionTimeout);
|
connOpts.setMqttVersion(MqttConnectOptions.MQTT_VERSION_3_1_1);
|
connOpts.setMaxInflight(1000);
|
sampleClient.setCallback(new MqttCallbackExtended() {
|
public void connectComplete(boolean reconnect, String serverURI) {
|
LogUtils.d(TAG, "连接成功");
|
checkAndsubscribeAllTopics("");
|
}
|
|
public void connectionLost(Throwable throwable) {
|
LogUtils.d(TAG, "连接断开");
|
lastTopicFilters.clear();
|
}
|
|
public void messageArrived(String topic, MqttMessage mqttMessage) throws Exception {
|
managerMqttMsg(topic, mqttMessage);
|
}
|
|
public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
|
}
|
});
|
LogUtils.d(TAG, "开始连接，连接参数 mBroker:" + mBroker + " mClientId:" + mClientId + " mUserName:" + mUserName + " mPassWord:" + mPassWord);
|
sampleClient.connect(connOpts);
|
} catch (Exception me) {
|
me.printStackTrace();
|
}
|
}
|
|
/**
|
* 处理接收的mqtt数据
|
*
|
* @param topic 接收主题
|
* @param mqttMessage 接收数据
|
* @throws Exception
|
*/
|
private void managerMqttMsg(String topic, MqttMessage mqttMessage) throws Exception {
|
LogUtils.d(TAG, "远程接收主题，Topic：" + topic);
|
if (HDLConnectHelper.isLocal()) {
|
boolean needReturn = false;
|
//如果是本地模式，云端下来的网关数据部分数据不接收(如：ota升级反馈进度)
|
for (String ignoreTopic : ignoreTopics) {
|
if (topic.endsWith(ignoreTopic)) {
|
needReturn = true;
|
break;
|
}
|
}
|
if (needReturn) {
|
return;
|
}
|
}
|
if (topic.contains("/custom/mqtt/secret/change")) {
|
/**
|
* 网关重连mqtt 需要更换aesKey 不然网关无法解密 通知刷新网关列表并且更新主网关的aesKey
|
*/
|
String[] topics = topic.split("/");
|
//非当前住宅网关的数据返回
|
if (topics.length < 3) {
|
return;
|
}
|
LogUtils.d(TAG, "网关重连mqtt秘钥更新通知" + topic);
|
|
BaseEventBus baseEventBus = new BaseEventBus();
|
baseEventBus.setTopic(topic);
|
EventBus.getDefault().post(baseEventBus);
|
|
EventNotifyRefreshGatewayAesKeyInfo eventNotifyRefreshGatewayAesKeyInfo = new EventNotifyRefreshGatewayAesKeyInfo();
|
eventNotifyRefreshGatewayAesKeyInfo.setGatewayId(topics[2]);
|
EventBus.getDefault().post(eventNotifyRefreshGatewayAesKeyInfo);
|
return;
|
}
|
|
String[] topics = topic.split("/");
|
//非当前住宅网关的数据返回
|
if (topics.length < 3) {
|
return;
|
}
|
String aes = HDLLinkConfig.getInstance().getAesKey();//默认用主网关的密钥
|
String cloudsGatewayId = topics[2];//云端上GatewayId
|
GatewayBean gatewayBean = HDLLinkLocalGateway.getInstance().getGatewayByOidOrGatewayId(cloudsGatewayId);
|
//云端转换网关的数据，非网关直接过来的数据，这个用的是住宅密钥
|
if (cloudsGatewayId.equals(HDLLinkConfig.getInstance().getHomeId())) {
|
aes = getHomeAES();
|
} else {
|
//网关的数据
|
if (gatewayBean != null) {
|
//毫米波mqtt专用秘钥、逆变器mqtt专用秘钥
|
aes = gatewayBean.getAesKey();
|
}
|
}
|
|
if (TextUtils.isEmpty(aes)) {
|
LogUtils.e("找不到远程解密的密钥，这个问题要排期解决");
|
return;
|
}
|
|
byte[] bytes = AesUtil.aesDecrypt(mqttMessage.getPayload(), aes);
|
if (null == bytes) {
|
LogUtils.i(TAG, "远程回复数据 密钥解密失败");
|
return;
|
}
|
//zigbee数据比较特殊，是json，但前面有其它数据
|
if (JsonUtils.isJson(bytes) || topic.contains("/custom/native/zigbee/up")) {
|
LogUtils.i(TAG, "远程接收数据，Payload：" + new String(bytes));
|
} else {
|
LogUtils.i(TAG, "远程接收数据，Payload：" + ByteUtils.encodeHexString(bytes));
|
}
|
/**
|
* 红外宝设备通过/thing/topo/found主题 上报红外宝设备已经入网了 然后直接return 不需要再下行了
|
*/
|
if (topic.endsWith("/thing/topo/found")) {
|
/**
|
* {"id":"0000016E","time_stamp":"366574","objects":[{"sid":"010105D370451908110100000000",
|
* "name":"Mini智能遥控器","spk":"ir.module","oid":"010105D370451908","omodel":"MIR01R-LK.10",
|
* "online":"true","attributes":[],"status":[],"from":"010105D370451908","src":"010105D370451908"}]}
|
*/
|
if (new String(bytes).contains("ir.module")) {
|
EventBus.getDefault().post(new EventBindMiniRemoteSuccessInfo());
|
return;
|
}
|
}
|
//Link从网关透传特殊处理
|
if (topic.contains("/native/a/") && topic.contains("/slaveoid/")) {
|
LinkMessageDecoder.getInstance().read(new Packet(bytes));
|
} else {
|
byte[] topBytes = new byte[3];
|
if (bytes.length > 2) {
|
topBytes[0] = bytes[0];
|
topBytes[1] = bytes[1];
|
topBytes[2] = bytes[2];
|
}
|
if (new String(topBytes).equals("hex")) {//link原生数据，判断是否是文件处理通知
|
LinkMessageDecoder.getInstance().read(new Packet(bytes));
|
} else {
|
QueueUtils.getInstance().add(new LinkPacket(topic, bytes, true));
|
}
|
}
|
}
|
|
/**
|
* 检查主题是否已订阅，没有订阅就订阅
|
*
|
* @param sendTopic 请求主题
|
*/
|
public synchronized void checkAndsubscribeAllTopics(String sendTopic) {
|
// LogUtils.d(TAG, "收到订阅主题\r\n" + sendTopic);
|
if (null == sampleClient) {
|
return;
|
}
|
if (sampleClient.isConnected() == false) {
|
return;
|
}
|
try {
|
String topics[] = nativeAndLinkTopic(sendTopic);
|
for (String topic : topics) {
|
if (lastTopicFilters.contains(topic)) {
|
continue;
|
}
|
LogUtils.d(TAG, "订阅主题：" + topic);
|
sampleClient.subscribe(topic, 0, null, new IMqttActionListener() {
|
@Override
|
public void onSuccess(IMqttToken asyncActionToken) {
|
if (!lastTopicFilters.contains(topic)) {
|
lastTopicFilters.add(topic);
|
}
|
}
|
|
@Override
|
public void onFailure(IMqttToken asyncActionToken, Throwable exception) {
|
|
}
|
});
|
}
|
|
} catch (MqttException e) {
|
e.printStackTrace();
|
}
|
}
|
|
/**
|
* 订阅平台下来的主题
|
*
|
* @return string数组
|
*/
|
private String[] nativeAndLinkTopic(String sendTopic) {
|
String[] topicArray = sendTopic.split("/");
|
//非当前住宅网关的数据返回,默认先订阅当前
|
if (topicArray.length < 3) {
|
if (TextUtils.isEmpty(HDLLinkConfig.getInstance().getGatewayId())) {
|
return new String[]{};
|
}
|
//默认订阅主网关的信息
|
return new String[]{
|
String.format("/user/%s/#", HDLLinkConfig.getInstance().getGatewayId()),
|
String.format("/base/%s/#", HDLLinkConfig.getInstance().getGatewayId())};
|
}
|
|
String gatewayId = topicArray[2];//云端上GatewayId
|
if (gatewayId.equals(HDLLinkConfig.getInstance().getGatewayId())) {
|
return new String[]{
|
String.format("/user/%s/#", gatewayId),
|
String.format("/base/%s/#", gatewayId)
|
};
|
} else {
|
return new String[]{
|
String.format("/user/%s/#", gatewayId),
|
String.format("/base/%s/#", gatewayId),
|
String.format("/user/%s/#", HDLLinkConfig.getInstance().getGatewayId()),
|
String.format("/base/%s/#", HDLLinkConfig.getInstance().getGatewayId())
|
};
|
}
|
}
|
|
|
/**
|
* APP订阅云端主题解密密文的秘钥(这个密钥只能用于订阅云端主题，订阅网关主题另外一个密钥)
|
*
|
* @return -返回解密密文的秘钥
|
*/
|
private String getHomeAES() {
|
String homeId = HDLLinkConfig.getInstance().getHomeId();
|
if (TextUtils.isEmpty(homeId)) {
|
return null;
|
}
|
//解密密钥规则：已现有的住宅ID为基准,从右边一一获取值,最后如果不够16位,则往右补零
|
StringBuilder aesKey = new StringBuilder();
|
for (int i = homeId.length() - 1; i >= 0; i--) {
|
aesKey.append(homeId.charAt(i));
|
if (aesKey.length() == 16) {
|
break;
|
}
|
}
|
return this.PadRight(aesKey.toString(), 16, "0");
|
}
|
|
/**
|
* 从右边添加空格或其它字符
|
*
|
* @param currentValueStr 当前值
|
* @param count 总长数
|
* @param others 其它字符(自定义)
|
* @return 总长度
|
*/
|
private String PadRight(String currentValueStr, int count, String others) {
|
if (count > currentValueStr.length()) {
|
StringBuilder stringBuilder = new StringBuilder();
|
int subLen = count - currentValueStr.length();
|
for (int i = 0; i < subLen; i++) {
|
stringBuilder.append(others);
|
}
|
currentValueStr = currentValueStr + stringBuilder;
|
}
|
return currentValueStr;
|
|
}
|
|
/**
|
* 切换住宅的时候订阅要全部取消
|
*/
|
public void removeAllTopic() {
|
if (null == sampleClient) {
|
return;
|
}
|
if (sampleClient.isConnected() == false) {
|
return;
|
}
|
try {
|
if (lastTopicFilters.size() == 0) {
|
return;
|
}
|
LogUtils.d(TAG, "移除主题\r\n" + JSON.toJSONString(lastTopicFilters));
|
sampleClient.unsubscribe(lastTopicFilters.toArray(new String[lastTopicFilters.size()]));
|
lastTopicFilters.clear();
|
} catch (MqttException e) {
|
e.printStackTrace();
|
}
|
}
|
|
/**
|
* 关闭当前连接，一般不要用，只创建一个连接就好了
|
*/
|
public void disConnect() {
|
try {
|
if (sampleClient == null) {
|
return;
|
}
|
LogUtils.i(TAG, "断开mqtt连接");
|
sampleClient.disconnect();
|
sampleClient = null;//不是置空mqttRecvClient，而是mqttRecvClient.sampleClient
|
lastTopicFilters.clear();
|
} catch (MqttException e) {
|
e.printStackTrace();
|
}
|
}
|
|
public String getmBroker() {
|
return mBroker;
|
}
|
|
public void setmBroker(String mBroker) {
|
this.mBroker = mBroker;
|
}
|
|
public String getmClientId() {
|
return mClientId;
|
}
|
|
public void setmClientId(String mClientId) {
|
this.mClientId = mClientId;
|
}
|
|
public String getmUserName() {
|
return mUserName;
|
}
|
|
public void setmUserName(String mUserName) {
|
this.mUserName = mUserName;
|
}
|
|
public String getmPassWord() {
|
return mPassWord;
|
}
|
|
public void setmPassWord(String mPassWord) {
|
this.mPassWord = mPassWord;
|
}
|
|
/**
|
* 是否已经初始化及调用连接
|
*
|
* @return
|
*/
|
public boolean isInit() {
|
return sampleClient != null;
|
}
|
|
public boolean isConnected() {
|
if (sampleClient == null) {
|
return false;
|
}
|
return sampleClient.isConnected();
|
}
|
}
|
