Browse Source

增加查看websocket客户端信息的地方

menglu 3 years ago
parent
commit
da94f62e05

+ 20 - 0
ibms-data-sdk/src/main/java/com/persagy/ibms/data/sdk/service/rest/RestApi.java

@@ -1,5 +1,6 @@
 package com.persagy.ibms.data.sdk.service.rest;
 
+import java.text.SimpleDateFormat;
 import java.util.Date;
 import java.util.concurrent.ConcurrentHashMap;
 
@@ -22,6 +23,9 @@ import com.alibaba.fastjson.JSONArray;
 import com.alibaba.fastjson.JSONObject;
 import com.alibaba.fastjson.serializer.SerializerFeature;
 import com.persagy.ibms.core.util.CheckUtil;
+import com.persagy.ibms.data.sdk.service.websocket.server.WebSocketChannelInfo;
+import com.persagy.ibms.data.sdk.service.websocket.server.WebSocketChannelPool;
+import com.persagy.ibms.data.sdk.service.websocket.server.WebSocketUtil;
 import com.persagy.ibms.data.sdk.util.ExportUtil;
 import com.persagy.ibms.data.sdk.util.RWDAlarmUtil;
 import com.persagy.ibms.data.sdk.util.RepositoryContainer;
@@ -79,6 +83,22 @@ public class RestApi {
 		return "alarm_websocket_close";
 	}
 
+	@GetMapping(path = "/websocket_client_list")
+	public String websocket_client_list() {
+		SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
+		JSONArray result = new JSONArray();
+		for (String id : WebSocketChannelPool.id2ChannelInfo.keySet()) {
+			WebSocketChannelInfo WebSocketChannelInfo = WebSocketChannelPool.id2ChannelInfo.get(id);
+			JSONObject item = new JSONObject();
+			item.put("id", id);
+			item.put("remoteAddress", WebSocketChannelInfo.remoteAddress.toString());
+			item.put("connectTime", sdf.format(WebSocketChannelInfo.connectTime));
+			item.put("objArray", WebSocketUtil.idMap.get(id));
+			result.add(item);
+		}
+		return JSONObject.toJSONString(result, SerializerFeature.WriteMapNullValue);
+	}
+
 	@PostMapping(path = { "/check" }, produces = "application/json;charset=UTF-8")
 	public String check(@RequestBody String param) {
 		JSONObject sceneJSON_ori = JSON.parseObject(param);

+ 1 - 1
ibms-data-sdk/src/main/java/com/persagy/ibms/data/sdk/service/websocket/server/WebSocketChannelPool.java

@@ -15,7 +15,7 @@ public class WebSocketChannelPool {
 	// ChannelHandlerPool.channelGroup.writeAndFlush(new TextWebSocketFrame(message));
 
 	public static Map<String, Channel> id2Channel = new ConcurrentHashMap<String, Channel>();
-	public static Map<String, Object> id2ContentJSON = new ConcurrentHashMap<String, Object>();
+	public static Map<String, WebSocketChannelInfo> id2ChannelInfo = new ConcurrentHashMap<String, WebSocketChannelInfo>();
 
 	public static void Send(String id, Object content) {
 		try {

+ 15 - 10
ibms-data-sdk/src/main/java/com/persagy/ibms/data/sdk/service/websocket/server/WebSocketHandler.java

@@ -1,5 +1,6 @@
 package com.persagy.ibms.data.sdk.service.websocket.server;
 
+import java.util.Date;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.concurrent.atomic.AtomicInteger;
@@ -43,6 +44,7 @@ public class WebSocketHandler extends SimpleChannelInboundHandler<TextWebSocketF
 		log.info("与客户端{}建立连接,当前连接数量{}!", ctx.channel().remoteAddress(), onlineCount.incrementAndGet());
 		// 添加到channelGroup通道组
 		WebSocketChannelPool.channelGroup.add(ctx.channel());
+		this.handlerConnect(ctx.channel());
 	}
 
 	@Override
@@ -50,7 +52,7 @@ public class WebSocketHandler extends SimpleChannelInboundHandler<TextWebSocketF
 		log.info("与客户端{}断开连接,通道关闭!当前连接数量{}!", ctx.channel().remoteAddress(), onlineCount.decrementAndGet());
 		// 添加到channelGroup 通道组
 		WebSocketChannelPool.channelGroup.remove(ctx.channel());
-		handlerRemove(ctx.channel());
+		handlerDisconnect(ctx.channel());
 	}
 
 	@Override
@@ -64,7 +66,7 @@ public class WebSocketHandler extends SimpleChannelInboundHandler<TextWebSocketF
 				Map<String, String> paramMap = getUrlParams(uri);
 				log.info("接收到的参数是:{}", JSON.toJSONString(paramMap));
 				Object ContentJSON = JSON.toJSON(paramMap);
-				handlerType(ctx.channel(), ContentJSON);
+				handlerReceive(ctx.channel(), ContentJSON);
 				// url需要处理
 				String newUri = uri.substring(0, uri.indexOf("?"));
 				log.info(newUri);
@@ -76,29 +78,32 @@ public class WebSocketHandler extends SimpleChannelInboundHandler<TextWebSocketF
 			String remoteAddress = ctx.channel().remoteAddress().toString();
 			log.info("收到客户端{}数据:{}", remoteAddress, frame.text());
 			Object ContentJSON = JSON.parse(frame.text());
-			handlerType(ctx.channel(), ContentJSON);
+			handlerReceive(ctx.channel(), ContentJSON);
 		}
 		super.channelRead(ctx, msg);
 	}
 
 	@Override
-	protected void channelRead0(ChannelHandlerContext channelHandlerContext, TextWebSocketFrame textWebSocketFrame)
-			throws Exception {
+	protected void channelRead0(ChannelHandlerContext channelHandlerContext, TextWebSocketFrame textWebSocketFrame) throws Exception {
 
 	}
 
-	private synchronized void handlerType(Channel channel, Object ContentJSON) {
+	private synchronized void handlerConnect(Channel channel) {
 		String id = channel.id().asLongText();
 		WebSocketChannelPool.id2Channel.put(id, channel);
-		WebSocketChannelPool.id2ContentJSON.put(id, ContentJSON);
-		WebSocketUtil.ProcessConnected(id, ContentJSON);
+		WebSocketChannelPool.id2ChannelInfo.put(id, new WebSocketChannelInfo(channel.remoteAddress(), new Date()));
 	}
 
-	private synchronized void handlerRemove(Channel channel) {
+	private synchronized void handlerReceive(Channel channel, Object ContentJSON) {
+		String id = channel.id().asLongText();
+		WebSocketUtil.ProcessReceive(id, ContentJSON);
+	}
+
+	private synchronized void handlerDisconnect(Channel channel) {
 		try {
 			String id = channel.id().asLongText();
 			WebSocketChannelPool.id2Channel.remove(id);
-			WebSocketChannelPool.id2ContentJSON.remove(id);
+			WebSocketChannelPool.id2ChannelInfo.remove(id);
 			WebSocketUtil.ProcessDisconnected(id);
 		} catch (Exception e) {
 			log.error(e.getMessage(), e);

+ 18 - 2
ibms-data-sdk/src/main/java/com/persagy/ibms/data/sdk/service/websocket/server/WebSocketUtil.java

@@ -33,18 +33,25 @@ public class WebSocketUtil {
 	public static Map<String, Map<String, Boolean>> objInfoId2idList = new ConcurrentHashMap<String, Map<String, Boolean>>();
 	public static Map<String, Map<String, Boolean>> id2objInfoIdList = new ConcurrentHashMap<String, Map<String, Boolean>>();
 
-	public static synchronized void ProcessConnected(String id, Object ContentJSON) {
+	public static Map<String, Map<String, Object>> id2objId2Tag = new ConcurrentHashMap<String, Map<String, Object>>();
+
+	public static synchronized void ProcessReceive(String id, Object ContentJSON) {
 		ProcessDisconnected(id);
-		
+
 		idMap.put(id, ContentJSON);
 		Map<String, Boolean> objIdList = new ConcurrentHashMap<String, Boolean>();
 		id2objIdList.put(id, objIdList);
 		Map<String, Boolean> objInfoIdList = new ConcurrentHashMap<String, Boolean>();
 		id2objInfoIdList.put(id, objInfoIdList);
+		Map<String, Object> objId2Tag = new ConcurrentHashMap<String, Object>();
+		id2objId2Tag.put(id, objId2Tag);
 		JSONArray objArray = (JSONArray) ContentJSON;
 		for (int i = 0; i < objArray.size(); i++) {
 			JSONObject objJSON = objArray.getJSONObject(i);
 			String objId = (String) objJSON.get("objId");
+			if (objJSON.containsKey("websocket_tag")) {
+				objId2Tag.put(objId, objJSON.get("websocket_tag"));
+			}
 			if (objJSON.containsKey("infoCodeArray")) {
 				JSONArray infoCodeArray = (JSONArray) objJSON.get("infoCodeArray");
 				for (int ii = 0; ii < infoCodeArray.size(); ii++) {
@@ -91,6 +98,7 @@ public class WebSocketUtil {
 		}
 		id2objInfoIdList.remove(id);
 
+		id2objId2Tag.remove(id);
 		idMap.remove(id);
 	}
 
@@ -100,6 +108,7 @@ public class WebSocketUtil {
 				@Override
 				public void run() {
 					RepositoryImpl Repository = RepositoryContainer.instance;
+					Map<String, Object> objId2Tag = id2objId2Tag.get(id);
 
 					JSONArray sendArray = new JSONArray();
 					Map<String, JSONObject> sendObj = new HashMap<String, JSONObject>();
@@ -148,6 +157,9 @@ public class WebSocketUtil {
 					for (String objId : sendObj.keySet()) {
 						JSONObject sendItem = sendObj.get(objId);
 						sendItem.put("id", objId);
+						if (objId2Tag.containsKey(objId)) {
+							sendItem.put("websocket_tag", objId2Tag.get(objId));
+						}
 						sendArray.add(sendItem);
 					}
 					if (sendArray.size() > 0) {
@@ -172,6 +184,7 @@ public class WebSocketUtil {
 						String message = json.getString("data");
 						String[] splits = message.split(";");
 						for (String id : idMap.keySet()) {
+							Map<String, Object> objId2Tag = id2objId2Tag.get(id);
 							try {
 								Map<String, Boolean> objIdList = id2objIdList.get(id);
 								Map<String, Boolean> objInfoIdList = id2objInfoIdList.get(id);
@@ -201,6 +214,9 @@ public class WebSocketUtil {
 								for (String objId : sendObj.keySet()) {
 									JSONObject sendItem = sendObj.get(objId);
 									sendItem.put("id", objId);
+									if (objId2Tag.containsKey(objId)) {
+										sendItem.put("websocket_tag", objId2Tag.get(objId));
+									}
 									sendArray.add(sendItem);
 								}
 								if (sendArray.size() > 0) {

+ 1 - 1
ibms-data-sdk/src/main/resources/application.properties

@@ -1,5 +1,5 @@
 server.port=8808
-websocket.port=8818
+websocket.port=9908
 spring.application.name=ibms-data-sdk
 
 autolog.app-name: ${spring.application.name}