| | |
|---|
| | | package com.dy.rtuMw.server.mqtt; |
|---|
| | | |
|---|
| | | import com.dy.common.mw.channel.mqtt.MqttClientPool; |
|---|
| | | import com.dy.common.mw.protocol4Mqtt.MqttNotify; |
|---|
| | | import com.dy.common.mw.protocol4Mqtt.MqttNotifyInfo; |
|---|
| | | import com.dy.common.mw.protocol4Mqtt.status.DevOnLineSt; |
|---|
| | | import com.dy.common.mw.protocol4Mqtt.status.DevRunSt; |
|---|
| | | import com.dy.rtuMw.server.ServerProperties; |
|---|
| | | import org.apache.logging.log4j.LogManager; |
|---|
| | | import org.apache.logging.log4j.Logger; |
|---|
| | | import org.eclipse.paho.client.mqttv3.MqttClient; |
|---|
| | |
|---|
| | | this.configVo = configVo; |
|---|
| | | } |
|---|
| | | |
|---|
| | | /** |
|---|
| | | * 创建连接池 + 订阅主题 |
|---|
| | | * @throws Exception |
|---|
| | | */ |
|---|
| | | public void start()throws Exception{ |
|---|
| | | String URL = "tcp://" + this.configVo.svIp + ":" + this.configVo.svPort; |
|---|
| | | this.pool = new MqttClientPool(URL, this.configVo.svUserName, this.configVo.svUserPassword, this.configVo.poolMaxSize); |
|---|
| | | if(this.pool.isClose()){ |
|---|
| | | throw new Exception("Mqtt连接池初始化失败"); |
|---|
| | | } |
|---|
| | | MqttClient clientSub = null ; |
|---|
| | | MqttClient clientSub ; |
|---|
| | | try { |
|---|
| | | clientSub = pool.popClient();//新创建一个Client时，此Client实际去连接MQTT服务器，如果连接不上，就会抛出异常 |
|---|
| | | }catch (Exception e){ |
|---|
| | |
|---|
| | | if(clientSub == null || !clientSub.isConnected()){ |
|---|
| | | throw new Exception("Mqtt连接池获得订阅连接不可用"); |
|---|
| | | } |
|---|
| | | // 订阅主题 |
|---|
| | | for(int i = 0; i < this.configVo.subTopics.length; i++){ |
|---|
| | | clientSub.subscribe(this.configVo.subTopics[i], this.configVo.topicsQos[i], new MqttMessageListener()); |
|---|
| | | for(int j = 0 ; j < this.configVo.protocolAndDeviceIds.length; j++){ |
|---|
| | | clientSub.subscribe(ServerProperties.orgTag + "/" |
|---|
| | | + this.configVo.protocolAndDeviceIds[j] + "/" |
|---|
| | | + this.configVo.subTopics[i], |
|---|
| | | this.configVo.subTopicsQos[i], |
|---|
| | | //每一个订阅主题都有一个MqttMessageListener实例 |
|---|
| | | new MqttMessageListener(new MqttNotify(){ |
|---|
| | | @Override |
|---|
| | | public void notify(String devId, MqttNotifyInfo... infos) { |
|---|
| | | if(devId != null && infos != null && infos.length > 0){ |
|---|
| | | for(MqttNotifyInfo info : infos){ |
|---|
| | | if(info instanceof DevOnLineSt){ |
|---|
| | | DevOnLineSt onLineSt = (DevOnLineSt)info; |
|---|
| | | if(onLineSt.onLine != null && onLineSt.onLine.booleanValue()){ |
|---|
| | | DevStatusDealer.onLine(devId, ((DevOnLineSt)info).protocol); |
|---|
| | | }else{ |
|---|
| | | DevStatusDealer.offLine(devId); |
|---|
| | | } |
|---|
| | | } else if(info instanceof DevRunSt){ |
|---|
| | | DevStatusDealer.setStatus(devId, (DevRunSt)info); |
|---|
| | | } |
|---|
| | | } |
|---|
| | | } |
|---|
| | | } |
|---|
| | | }) |
|---|
| | | ); |
|---|
| | | } |
|---|
| | | } |
|---|
| | | } |
|---|
| | | |
|---|
| | |
|---|
| | | } |
|---|
| | | |
|---|
| | | public void publishMsg(MqttClient client, String topic, byte[] msg) throws Exception{ |
|---|
| | | client.publish(topic, msg, this.configVo.publishQos, false); |
|---|
| | | client.publish(topic, msg, this.configVo.pubTopicQos, false); |
|---|
| | | } |
|---|
| | | |
|---|
| | | public void publishMsg(MqttClient client, String topic, String msg) throws Exception{ |
|---|
| | | byte[] bs = msg.getBytes("UTF-8") ; |
|---|
| | | client.publish(topic, bs, this.configVo.publishQos, false); |
|---|
| | | client.publish(topic, bs, this.configVo.pubTopicQos, false); |
|---|
| | | } |
|---|
| | | |
|---|
| | | public boolean poolIsClose(){ |
|---|
