From fe6dd5be9c34c948093db62484d3abf637267baf Mon Sep 17 00:00:00 2001 From: jinpengyong <jpy123456> Date: Mon, 21 Sep 2020 15:41:39 +0800 Subject: [PATCH] update --- src/main/java/com/moral/webSocketServer/WebSocketServer.java | 38 ++++++++++++++++++++++++++------------ 1 files changed, 26 insertions(+), 12 deletions(-) diff --git a/src/main/java/com/moral/webSocketServer/WebSocketServer.java b/src/main/java/com/moral/webSocketServer/WebSocketServer.java index 59a9783..c6531e2 100644 --- a/src/main/java/com/moral/webSocketServer/WebSocketServer.java +++ b/src/main/java/com/moral/webSocketServer/WebSocketServer.java @@ -1,8 +1,10 @@ package com.moral.webSocketServer; import java.io.IOException; +import java.util.HashMap; +import java.util.List; +import java.util.Map; import java.util.concurrent.CopyOnWriteArraySet; -import java.util.concurrent.TimeoutException; import javax.websocket.OnClose; import javax.websocket.OnError; @@ -12,7 +14,9 @@ import javax.websocket.server.PathParam; import javax.websocket.server.ServerEndpoint; -import com.moral.service.MonitorPointService; +import com.moral.common.util.ParameterUtils; +import com.moral.entity.Device; +import com.moral.service.DeviceService; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory; @@ -30,12 +34,14 @@ @Component public class WebSocketServer { - public static MonitorPointService monitorPointService; + public static DeviceService deviceService; // ������������������������������������������������������������������������������ private Session session; private String orgId; + + private String accountId; private String regionCode; @@ -57,9 +63,16 @@ //������������������������������������������������ webSocketSet.add(this); int flag = param.indexOf("&"); + int regionCodeIndex = param.indexOf("_"); orgId = param.substring(0, flag); - regionCode = param.substring(flag + 1); - String QUEUE_NAME = "deviceInfo"; + accountId = param.substring(flag + 1,regionCodeIndex); + regionCode = param.substring(regionCodeIndex + 1); + String QUEUE_NAME = "deviceInfo_" + accountId; + Map<String, Object> paramMap = new HashMap<String, Object>(); + paramMap.put("orgId", orgId); + paramMap.put("regionCode", regionCode); + ParameterUtils.getRegionType4RegionCode(paramMap); + List<Device> deviceList = deviceService.queryDevice(paramMap); try { //������������������������������������������������ ConnectionFactory factory = new ConnectionFactory(); @@ -68,13 +81,15 @@ factory.setPort(5672); factory.setUsername("guest"); factory.setPassword("guest_pass"); - String routingKey = orgId+".*"; + String routingKey; connection = factory.newConnection(); channel = connection.createChannel(); //��������������������������������������������������������������������������������������������������������� channel.queueDeclare(QUEUE_NAME, false, false, true, null); - channel.queueBind(QUEUE_NAME,"screens_data",routingKey); - + for (Device d : deviceList) { + routingKey = orgId + "." + d.getMac(); + channel.queueBind(QUEUE_NAME, "screens_data", routingKey); + } //��������������������� QueueingConsumer consumer = new QueueingConsumer(channel); //������������������ @@ -86,7 +101,7 @@ String message = new String(delivery.getBody()); sendMessage(message); } - }catch (Exception e){ + } catch (Exception e) { log.error(e.getMessage()); } } @@ -99,16 +114,15 @@ /**���������Set��� ������������������������*/ webSocketSet.remove(this); try { - channel.close(); + channel.queueDelete("deviceInfo_" + accountId); connection.close(); - }catch (IOException | TimeoutException e){ + } catch (IOException e) { log.error(e.getMessage()); } } @OnMessage public void onMessage(String message) { - System.out.println(message); for (WebSocketServer webSocketServer : webSocketSet) { webSocketServer.sendMessage(message); } -- Gitblit v1.8.0