PublishClient.java 8.9 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279
  1. package com.ozs.service.utils;
  2. import com.alibaba.fastjson2.JSON;
  3. import com.alibaba.fastjson2.JSONObject;
  4. import com.ozs.common.utils.sign.Md5Utils;
  5. import com.ozs.common.utils.stateSecrets.SM4Utils;
  6. import com.ozs.service.entity.vo.BaseCameraVersionVo;
  7. import com.ozs.service.entity.vo.Codec;
  8. import com.ozs.service.entity.vo.Data;
  9. import com.ozs.service.entity.vo.Heartbeat;
  10. import com.ozs.service.entity.vo.Param;
  11. import com.ozs.service.entity.vo.Svp;
  12. import com.ozs.service.entity.vo.UpdateDeviceSn;
  13. import com.ozs.service.entity.vo.Venc0;
  14. import org.eclipse.paho.client.mqttv3.MqttClient;
  15. import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
  16. import org.eclipse.paho.client.mqttv3.MqttDeliveryToken;
  17. import org.eclipse.paho.client.mqttv3.MqttException;
  18. import org.eclipse.paho.client.mqttv3.MqttMessage;
  19. import org.eclipse.paho.client.mqttv3.MqttPersistenceException;
  20. import org.eclipse.paho.client.mqttv3.MqttTopic;
  21. import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
  22. import org.springframework.beans.factory.annotation.Value;
  23. import java.util.ArrayList;
  24. import java.util.HashMap;
  25. import java.util.Map;
  26. import java.util.UUID;
  27. /**
  28. * 发布客户端
  29. *
  30. * @author Administrator
  31. */
  32. public class PublishClient {
  33. /**
  34. * mqtt服务器地址
  35. */
  36. public static final String HOST = "tcp://124.70.58.209:1883";
  37. /**
  38. * 主题
  39. */
  40. private final String topic;
  41. /**
  42. * 客户端实例
  43. */
  44. private final MqttClient client;
  45. /**
  46. * 主题实例
  47. */
  48. private MqttTopic topic11;
  49. private static final String USER_NAME = "guest";
  50. private static final String PASS_WORD = "guest";
  51. private MqttMessage message;
  52. /**
  53. * 初始化客户端实例
  54. *
  55. * @param topic
  56. * @param clientId
  57. * @throws MqttException
  58. */
  59. public PublishClient(String topic, String clientId) throws MqttException {
  60. /**
  61. * mqtt 客户机ID
  62. * heart_DEVICESN
  63. */
  64. this.topic = topic;
  65. /**
  66. * MemoryPersistence设置clientId的保存形式,默认为以内存保存
  67. */
  68. client = new MqttClient(HOST, clientId, new MemoryPersistence());
  69. connect();
  70. }
  71. /**
  72. * 连接服务器
  73. */
  74. private void connect() {
  75. /**
  76. * 连接配置
  77. */
  78. MqttConnectOptions options = new MqttConnectOptions();
  79. /**
  80. * 不保存,每次重启新client
  81. */
  82. options.setCleanSession(true);
  83. options.setUserName(USER_NAME);
  84. options.setPassword(PASS_WORD.toCharArray());
  85. /**
  86. * 设置超时时间
  87. */
  88. options.setConnectionTimeout(10);
  89. /**
  90. * 设置会话心跳时间
  91. */
  92. options.setKeepAliveInterval(20);
  93. try {
  94. /**
  95. * 设置发布回调
  96. */
  97. client.setCallback(new PublishCallback());
  98. client.connect(options);
  99. topic11 = client.getTopic(topic);
  100. } catch (Exception e) {
  101. e.printStackTrace();
  102. }
  103. }
  104. /**
  105. * 发布
  106. *
  107. * @param topic
  108. * @param message
  109. * @throws MqttPersistenceException
  110. * @throws MqttException
  111. */
  112. public void publish(MqttTopic topic, MqttMessage message) throws MqttPersistenceException,
  113. MqttException {
  114. MqttDeliveryToken token = topic.publish(message);
  115. token.waitForCompletion();
  116. System.out.println("message is published completely! " + token.isComplete());
  117. }
  118. public static void pull(String sign, String deviceSn) throws MqttException, InterruptedException {
  119. /**
  120. * 发布客户端
  121. */
  122. PublishClient server = new PublishClient("heart_" + deviceSn, "heart_" + deviceSn);
  123. Heartbeat test = new Heartbeat();
  124. test.setName("HeartResponse");
  125. Data data = new Data();
  126. data.setAlarm(0);
  127. data.setStream(0);
  128. test.setData(data);
  129. test.setCode(200);
  130. test.setSign(sign);
  131. String s = JSON.toJSONString(test);
  132. server.message = new MqttMessage();
  133. /**
  134. * 保证消息能到达一次
  135. */
  136. server.message.setQos(1);
  137. /**
  138. * 消息保留
  139. */
  140. // server.message.setRetained(false);
  141. /**
  142. * 消息内容
  143. */
  144. server.message.setPayload(s.getBytes());
  145. /**
  146. * 发布
  147. */
  148. server.publish(server.topic11, server.message);
  149. }
  150. public static void updateDeviceSn(BaseCameraVersionVo baseCameraVersionVo) {
  151. /**
  152. * 发布客户端
  153. */
  154. for (String code : baseCameraVersionVo.getCameraCodeList()) {
  155. try {
  156. PublishClient server = new PublishClient("update_" + code, "update_" + code);
  157. UpdateDeviceSn updateDeviceSn = new UpdateDeviceSn();
  158. updateDeviceSn.setName("UpdateRequest");
  159. updateDeviceSn.setType(Integer.valueOf(baseCameraVersionVo.getUpgradeType()));
  160. updateDeviceSn.setMd5(baseCameraVersionVo.getMd5());
  161. updateDeviceSn.setSign(UUID.randomUUID().toString());
  162. updateDeviceSn.setUrl(baseCameraVersionVo.getVersionAddress());
  163. String s = JSON.toJSONString(updateDeviceSn);
  164. server.message = new MqttMessage();
  165. /**
  166. * 保证消息能到达一次
  167. */
  168. server.message.setQos(1);
  169. /**
  170. * 消息保留
  171. */
  172. // server.message.setRetained(false);
  173. /**
  174. * 消息内容
  175. */
  176. server.message.setPayload(s.getBytes());
  177. /**
  178. * 发布
  179. */
  180. server.publish(server.topic11, server.message);
  181. } catch (MqttException e) {
  182. System.out.println(e.getMessage());
  183. }
  184. }
  185. }
  186. public static void confidenceCoefficient(BaseCameraVersionVo baseCameraVersionVo,String value) {
  187. /**
  188. * 发布客户端
  189. */
  190. for (String code : baseCameraVersionVo.getCameraCodeList()) {
  191. try {
  192. PublishClient server = new PublishClient("config_" + code, "config_" + code);
  193. JSONObject res = new JSONObject();
  194. res.put("Name", "ConfigRequest");
  195. Svp svp = new Svp();
  196. Param param = new Param();
  197. param.setThreshold(Double.valueOf(value));
  198. svp.setParam(param);
  199. Map<String, Object> map = new HashMap<>();
  200. map.put("svp", svp);
  201. res.put("data", map);
  202. res.put("sign", UUID.randomUUID().toString());
  203. String s = JSONObject.toJSONString(res);
  204. server.message = new MqttMessage();
  205. /**
  206. * 保证消息能到达一次
  207. */
  208. server.message.setQos(1);
  209. /**
  210. * 消息保留
  211. */
  212. // server.message.setRetained(false);
  213. /**
  214. * 消息内容
  215. */
  216. server.message.setPayload(s.getBytes());
  217. /**
  218. * 发布
  219. */
  220. server.publish(server.topic11, server.message);
  221. } catch (MqttException e) {
  222. System.out.println(e.getMessage());
  223. }
  224. }
  225. }
  226. public static void configFrameRate(BaseCameraVersionVo baseCameraVersionVo,String value) {
  227. /**
  228. * 发布客户端
  229. */
  230. for (String code : baseCameraVersionVo.getCameraCodeList()) {
  231. try {
  232. PublishClient server = new PublishClient("config_" + code, "config_" + code);
  233. JSONObject res = new JSONObject();
  234. res.put("Name", "ConfigRequest");
  235. Codec codec = new Codec();
  236. Venc0 venc0 = new Venc0();
  237. venc0.setFps(Double.valueOf(value));
  238. codec.setVenc0(venc0);
  239. Map<String, Object> map = new HashMap<>();
  240. map.put("codec", codec);
  241. res.put("data", map);
  242. res.put("sign", UUID.randomUUID().toString());
  243. String s = JSONObject.toJSONString(res);
  244. server.message = new MqttMessage();
  245. /**
  246. * 保证消息能到达一次
  247. */
  248. server.message.setQos(1);
  249. /**
  250. * 消息保留
  251. */
  252. // server.message.setRetained(false);
  253. /**
  254. * 消息内容
  255. */
  256. server.message.setPayload(s.getBytes());
  257. /**
  258. * 发布
  259. */
  260. server.publish(server.topic11, server.message);
  261. } catch (MqttException e) {
  262. System.out.println(e.getMessage());
  263. }
  264. }
  265. }
  266. }