| | |
|---|
| | | 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.common.mw.protocol4Mqtt.status.DevOnLineInfo; |
|---|
| | | import com.dy.common.mw.protocol4Mqtt.status.DevRunInfo; |
|---|
| | | import com.dy.rtuMw.server.ServerProperties; |
|---|
| | | import org.apache.logging.log4j.LogManager; |
|---|
| | | import org.apache.logging.log4j.Logger; |
|---|
| | |
|---|
| | | throw new Exception("Mqtt连接池获得订阅连接不可用"); |
|---|
| | | } |
|---|
| | | subClients.add(clientSub) ; |
|---|
| | | |
|---|
| | | // 订阅主题 |
|---|
| | | for(int i = 0; i < this.configVo.subTopics.length; i++){ |
|---|
| | | 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); |
|---|
| | | if(this.configVo.subTopics != null && this.configVo.subTopics.length > 0){ |
|---|
| | | if(this.configVo.protocolAndDeviceIds != null || this.configVo.protocolAndDeviceIds.length > 0){ |
|---|
| | | for(int i = 0; i < this.configVo.subTopics.length; i++){ |
|---|
| | | 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 DevOnLineInfo){ |
|---|
| | | DevOnLineInfo onLineSt = (DevOnLineInfo)info; |
|---|
| | | if(onLineSt.onLine != null && onLineSt.onLine.booleanValue()){ |
|---|
| | | DevStatusDealer.onLine(devId, ((DevOnLineInfo)info).protocol); |
|---|
| | | }else{ |
|---|
| | | DevStatusDealer.offLine(devId); |
|---|
| | | } |
|---|
| | | } else if(info instanceof DevRunInfo){ |
|---|
| | | DevStatusDealer.setStatus(devId, (DevRunInfo)info); |
|---|
| | | } |
|---|
| | | } |
|---|
| | | } else if(info instanceof DevRunSt){ |
|---|
| | | DevStatusDealer.setStatus(devId, (DevRunSt)info); |
|---|
| | | } |
|---|
| | | } |
|---|
| | | } |
|---|
| | | } |
|---|
| | | }) |
|---|
| | | ); |
|---|
| | | }) |
|---|
| | | ); |
|---|
| | | } |
|---|
| | | } |
|---|
| | | } |
|---|
| | | } |
|---|
| | | } |
|---|
