123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279 |
- package com.ozs.service.utils;
- import com.alibaba.fastjson2.JSON;
- import com.alibaba.fastjson2.JSONObject;
- import com.ozs.common.utils.sign.Md5Utils;
- import com.ozs.common.utils.stateSecrets.SM4Utils;
- import com.ozs.service.entity.vo.BaseCameraVersionVo;
- import com.ozs.service.entity.vo.Codec;
- import com.ozs.service.entity.vo.Data;
- import com.ozs.service.entity.vo.Heartbeat;
- import com.ozs.service.entity.vo.Param;
- import com.ozs.service.entity.vo.Svp;
- import com.ozs.service.entity.vo.UpdateDeviceSn;
- import com.ozs.service.entity.vo.Venc0;
- import org.eclipse.paho.client.mqttv3.MqttClient;
- import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
- import org.eclipse.paho.client.mqttv3.MqttDeliveryToken;
- import org.eclipse.paho.client.mqttv3.MqttException;
- import org.eclipse.paho.client.mqttv3.MqttMessage;
- import org.eclipse.paho.client.mqttv3.MqttPersistenceException;
- import org.eclipse.paho.client.mqttv3.MqttTopic;
- import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
- import org.springframework.beans.factory.annotation.Value;
- import java.util.ArrayList;
- import java.util.HashMap;
- import java.util.Map;
- import java.util.UUID;
- /**
- * 发布客户端
- *
- * @author Administrator
- */
- public class PublishClient {
- /**
- * mqtt服务器地址
- */
- public static final String HOST = "tcp://124.70.58.209:1883";
- /**
- * 主题
- */
- private final String topic;
- /**
- * 客户端实例
- */
- private final MqttClient client;
- /**
- * 主题实例
- */
- private MqttTopic topic11;
- private static final String USER_NAME = "guest";
- private static final String PASS_WORD = "guest";
- private MqttMessage message;
- /**
- * 初始化客户端实例
- *
- * @param topic
- * @param clientId
- * @throws MqttException
- */
- public PublishClient(String topic, String clientId) throws MqttException {
- /**
- * mqtt 客户机ID
- * heart_DEVICESN
- */
- this.topic = topic;
- /**
- * MemoryPersistence设置clientId的保存形式,默认为以内存保存
- */
- client = new MqttClient(HOST, clientId, new MemoryPersistence());
- connect();
- }
- /**
- * 连接服务器
- */
- private void connect() {
- /**
- * 连接配置
- */
- MqttConnectOptions options = new MqttConnectOptions();
- /**
- * 不保存,每次重启新client
- */
- options.setCleanSession(true);
- options.setUserName(USER_NAME);
- options.setPassword(PASS_WORD.toCharArray());
- /**
- * 设置超时时间
- */
- options.setConnectionTimeout(10);
- /**
- * 设置会话心跳时间
- */
- options.setKeepAliveInterval(20);
- try {
- /**
- * 设置发布回调
- */
- client.setCallback(new PublishCallback());
- client.connect(options);
- topic11 = client.getTopic(topic);
- } catch (Exception e) {
- e.printStackTrace();
- }
- }
- /**
- * 发布
- *
- * @param topic
- * @param message
- * @throws MqttPersistenceException
- * @throws MqttException
- */
- public void publish(MqttTopic topic, MqttMessage message) throws MqttPersistenceException,
- MqttException {
- MqttDeliveryToken token = topic.publish(message);
- token.waitForCompletion();
- System.out.println("message is published completely! " + token.isComplete());
- }
- public static void pull(String sign, String deviceSn) throws MqttException, InterruptedException {
- /**
- * 发布客户端
- */
- PublishClient server = new PublishClient("heart_" + deviceSn, "heart_" + deviceSn);
- Heartbeat test = new Heartbeat();
- test.setName("HeartResponse");
- Data data = new Data();
- data.setAlarm(0);
- data.setStream(0);
- test.setData(data);
- test.setCode(200);
- test.setSign(sign);
- String s = JSON.toJSONString(test);
- server.message = new MqttMessage();
- /**
- * 保证消息能到达一次
- */
- server.message.setQos(1);
- /**
- * 消息保留
- */
- // server.message.setRetained(false);
- /**
- * 消息内容
- */
- server.message.setPayload(s.getBytes());
- /**
- * 发布
- */
- server.publish(server.topic11, server.message);
- }
- public static void updateDeviceSn(BaseCameraVersionVo baseCameraVersionVo) {
- /**
- * 发布客户端
- */
- for (String code : baseCameraVersionVo.getCameraCodeList()) {
- try {
- PublishClient server = new PublishClient("update_" + code, "update_" + code);
- UpdateDeviceSn updateDeviceSn = new UpdateDeviceSn();
- updateDeviceSn.setName("UpdateRequest");
- updateDeviceSn.setType(Integer.valueOf(baseCameraVersionVo.getUpgradeType()));
- updateDeviceSn.setMd5(baseCameraVersionVo.getMd5());
- updateDeviceSn.setSign(UUID.randomUUID().toString());
- updateDeviceSn.setUrl(baseCameraVersionVo.getVersionAddress());
- String s = JSON.toJSONString(updateDeviceSn);
- server.message = new MqttMessage();
- /**
- * 保证消息能到达一次
- */
- server.message.setQos(1);
- /**
- * 消息保留
- */
- // server.message.setRetained(false);
- /**
- * 消息内容
- */
- server.message.setPayload(s.getBytes());
- /**
- * 发布
- */
- server.publish(server.topic11, server.message);
- } catch (MqttException e) {
- System.out.println(e.getMessage());
- }
- }
- }
- public static void confidenceCoefficient(BaseCameraVersionVo baseCameraVersionVo,String value) {
- /**
- * 发布客户端
- */
- for (String code : baseCameraVersionVo.getCameraCodeList()) {
- try {
- PublishClient server = new PublishClient("config_" + code, "config_" + code);
- JSONObject res = new JSONObject();
- res.put("Name", "ConfigRequest");
- Svp svp = new Svp();
- Param param = new Param();
- param.setThreshold(Double.valueOf(value));
- svp.setParam(param);
- Map<String, Object> map = new HashMap<>();
- map.put("svp", svp);
- res.put("data", map);
- res.put("sign", UUID.randomUUID().toString());
- String s = JSONObject.toJSONString(res);
- server.message = new MqttMessage();
- /**
- * 保证消息能到达一次
- */
- server.message.setQos(1);
- /**
- * 消息保留
- */
- // server.message.setRetained(false);
- /**
- * 消息内容
- */
- server.message.setPayload(s.getBytes());
- /**
- * 发布
- */
- server.publish(server.topic11, server.message);
- } catch (MqttException e) {
- System.out.println(e.getMessage());
- }
- }
- }
- public static void configFrameRate(BaseCameraVersionVo baseCameraVersionVo,String value) {
- /**
- * 发布客户端
- */
- for (String code : baseCameraVersionVo.getCameraCodeList()) {
- try {
- PublishClient server = new PublishClient("config_" + code, "config_" + code);
- JSONObject res = new JSONObject();
- res.put("Name", "ConfigRequest");
- Codec codec = new Codec();
- Venc0 venc0 = new Venc0();
- venc0.setFps(Double.valueOf(value));
- codec.setVenc0(venc0);
- Map<String, Object> map = new HashMap<>();
- map.put("codec", codec);
- res.put("data", map);
- res.put("sign", UUID.randomUUID().toString());
- String s = JSONObject.toJSONString(res);
- server.message = new MqttMessage();
- /**
- * 保证消息能到达一次
- */
- server.message.setQos(1);
- /**
- * 消息保留
- */
- // server.message.setRetained(false);
- /**
- * 消息内容
- */
- server.message.setPayload(s.getBytes());
- /**
- * 发布
- */
- server.publish(server.topic11, server.message);
- } catch (MqttException e) {
- System.out.println(e.getMessage());
- }
- }
- }
- }
|