From 2096fcf9d948dbdcea2aae6f29d508086da9793b Mon Sep 17 00:00:00 2001
From: 于紫祥_1901 <email@yuzixiang_1910>
Date: Thu, 22 Oct 2020 09:09:05 +0800
Subject: [PATCH] updata

---
 src/main/java/com/moral/webSocketServer/WebSocketServer.java |   76 ++++++++++++++++++++++++++++++--------
 1 files changed, 60 insertions(+), 16 deletions(-)

diff --git a/src/main/java/com/moral/webSocketServer/WebSocketServer.java b/src/main/java/com/moral/webSocketServer/WebSocketServer.java
index 38ef281..f886cd2 100644
--- a/src/main/java/com/moral/webSocketServer/WebSocketServer.java
+++ b/src/main/java/com/moral/webSocketServer/WebSocketServer.java
@@ -1,6 +1,9 @@
 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 javax.websocket.OnClose;
@@ -11,13 +14,21 @@
 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.moral.util.RabbitMQUtils;
+import com.rabbitmq.client.AMQP;
 import com.rabbitmq.client.Channel;
 import com.rabbitmq.client.Connection;
 import com.rabbitmq.client.ConnectionFactory;
+import com.rabbitmq.client.Consumer;
+import com.rabbitmq.client.DefaultConsumer;
+import com.rabbitmq.client.Envelope;
 import com.rabbitmq.client.QueueingConsumer;
 
 import lombok.extern.slf4j.Slf4j;
+
 import org.springframework.stereotype.Component;
 
 /**
@@ -29,14 +40,20 @@
 @Component
 public class WebSocketServer {
 
-    public static MonitorPointService monitorPointService;
+    public static DeviceService deviceService;
 
     // ������������������������������������������������������������������������������
     private Session session;
 
     private String orgId;
 
+    private String accountId;
+
     private String regionCode;
+
+    private Connection connection;
+
+    private Channel channel;
 
     // ������session���������������������������
     private static CopyOnWriteArraySet<WebSocketServer> webSocketSet = new CopyOnWriteArraySet<WebSocketServer>();
@@ -51,27 +68,42 @@
         this.session = session;
         //������������������������������������������������
         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();
+            /*ConnectionFactory factory = new ConnectionFactory();
             //������MabbitMQ������������ip���������������
-            factory.setHost("172.16.206.8");
+            factory.setHost("47.96.15.25");
             factory.setPort(5672);
             factory.setUsername("guest");
-            factory.setPassword("guest_pass");
-            String routingKey = orgId+".*";
-            Connection connection = factory.newConnection();
-            Channel channel = connection.createChannel();
+            factory.setPassword("guest_pass");*/
+            //Connection connection = RabbitMQUtils.getConnection();
+            String routingKey;
+            this.connection = RabbitMQUtils.getConnection();
+            channel = this.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);
+            /*java������������*/
+            /*QueueingConsumer consumer = new QueueingConsumer(channel);
             //������������������
             channel.basicConsume(QUEUE_NAME, true, consumer);
 
@@ -80,8 +112,16 @@
                 QueueingConsumer.Delivery delivery = consumer.nextDelivery();
                 String message = new String(delivery.getBody());
                 sendMessage(message);
-            }
-        }catch (Exception e){
+            }*/
+            channel.basicConsume(QUEUE_NAME, false, new DefaultConsumer(channel){
+                @Override
+                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
+                    String msg = new String(body, "utf-8");
+                    sendMessage(msg);
+                    channel.basicAck(envelope.getDeliveryTag(), false);
+                }
+            });
+        } catch (Exception e) {
             log.error(e.getMessage());
         }
     }
@@ -93,11 +133,15 @@
     public void onClose() {
         /**���������Set��� ������������������������*/
         webSocketSet.remove(this);
+        try {
+            connection.close();
+        } 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