From b774e0d423c17b3b38d22d5d779e06bdc5f13385 Mon Sep 17 00:00:00 2001 From: heavyrian2012 Date: Tue, 28 Nov 2023 11:04:12 +0800 Subject: [PATCH] =?UTF-8?q?=E6=B7=BB=E5=8A=A0=E8=8E=B7=E5=8F=96=E5=9C=A8?= =?UTF-8?q?=E7=BA=BF=E7=94=A8=E6=88=B7=E4=BF=A1=E6=81=AF=E5=92=8C=E8=8E=B7?= =?UTF-8?q?=E5=8F=96=E7=94=A8=E6=88=B7session=E4=BF=A1=E6=81=AF=E7=9A=84?= =?UTF-8?q?=E6=8E=A5=E5=8F=A3?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../action/admin/GetOnlineUserAction.java | 47 +++++++++++++++ .../admin/GetUserOnlineCountAction.java | 50 ++++++++++++++++ .../action/admin/GetUserSessionsAction.java | 57 +++++++++++++++++++ .../persistence/MemoryMessagesStore.java | 29 ++++++++++ .../persistence/MemorySessionStore.java | 5 ++ .../java/io/moquette/spi/IMessagesStore.java | 13 ++--- .../java/io/moquette/spi/ISessionsStore.java | 2 + .../moquette/spi/impl/ProtocolProcessor.java | 4 ++ .../java/cn/wildfirechat/common/APIPath.java | 1 + .../pojos/GetUserSessionResult.java | 35 ++++++++++++ .../main/java/cn/wildfirechat/sdk/Main.java | 8 +++ .../java/cn/wildfirechat/sdk/UserAdmin.java | 5 ++ 12 files changed, 248 insertions(+), 8 deletions(-) create mode 100755 broker/src/main/java/com/xiaoleilu/loServer/action/admin/GetOnlineUserAction.java create mode 100755 broker/src/main/java/com/xiaoleilu/loServer/action/admin/GetUserOnlineCountAction.java create mode 100755 broker/src/main/java/com/xiaoleilu/loServer/action/admin/GetUserSessionsAction.java create mode 100644 common/src/main/java/cn/wildfirechat/pojos/GetUserSessionResult.java diff --git a/broker/src/main/java/com/xiaoleilu/loServer/action/admin/GetOnlineUserAction.java b/broker/src/main/java/com/xiaoleilu/loServer/action/admin/GetOnlineUserAction.java new file mode 100755 index 00000000..1bfec750 --- /dev/null +++ b/broker/src/main/java/com/xiaoleilu/loServer/action/admin/GetOnlineUserAction.java @@ -0,0 +1,47 @@ +/* + * This file is part of the Wildfire Chat package. + * (c) Heavyrain2012 + * + * For the full copyright and license information, please view the LICENSE + * file that was distributed with this source code. + */ + +package com.xiaoleilu.loServer.action.admin; + +import cn.wildfirechat.common.APIPath; +import cn.wildfirechat.common.ErrorCode; +import cn.wildfirechat.pojos.*; +import com.xiaoleilu.loServer.RestResult; +import com.xiaoleilu.loServer.annotation.HttpMethod; +import com.xiaoleilu.loServer.annotation.Route; +import com.xiaoleilu.loServer.handler.Request; +import com.xiaoleilu.loServer.handler.Response; +import io.moquette.persistence.MemorySessionStore; +import io.netty.handler.codec.http.FullHttpRequest; +import io.netty.util.internal.StringUtil; + +import java.util.Collection; + +@Route(APIPath.User_Online_List) +@HttpMethod("POST") +public class GetOnlineUserAction extends AdminAction { + + @Override + public boolean isTransactionAction() { + return true; + } + + @Override + public boolean action(Request request, Response response) { + if (request.getNettyRequest() instanceof FullHttpRequest) { + GetOnlineUserRequest getOnlineUserRequest = getRequestBody(request.getNettyRequest(), GetOnlineUserRequest.class); + if (getOnlineUserRequest.nodeId == 1) { + GetOnlineUserResult result = messagesStore.getOnlineUsers(getOnlineUserRequest.offset, getOnlineUserRequest.count); + setResponseContent(RestResult.ok(result), response); + } else { + setResponseContent(RestResult.resultOf(ErrorCode.INVALID_PARAMETER), response); + } + } + return true; + } +} diff --git a/broker/src/main/java/com/xiaoleilu/loServer/action/admin/GetUserOnlineCountAction.java b/broker/src/main/java/com/xiaoleilu/loServer/action/admin/GetUserOnlineCountAction.java new file mode 100755 index 00000000..3a076538 --- /dev/null +++ b/broker/src/main/java/com/xiaoleilu/loServer/action/admin/GetUserOnlineCountAction.java @@ -0,0 +1,50 @@ +/* + * This file is part of the Wildfire Chat package. + * (c) Heavyrain2012 + * + * For the full copyright and license information, please view the LICENSE + * file that was distributed with this source code. + */ + +package com.xiaoleilu.loServer.action.admin; + +import cn.wildfirechat.common.APIPath; +import cn.wildfirechat.common.ErrorCode; +import cn.wildfirechat.pojos.GetOnlineUserCountResult; +import cn.wildfirechat.pojos.GetUserSessionResult; +import cn.wildfirechat.pojos.InputUserId; +import com.xiaoleilu.loServer.RestResult; +import com.xiaoleilu.loServer.annotation.HttpMethod; +import com.xiaoleilu.loServer.annotation.Route; +import com.xiaoleilu.loServer.handler.Request; +import com.xiaoleilu.loServer.handler.Response; +import io.moquette.persistence.MemorySessionStore; +import io.netty.handler.codec.http.FullHttpRequest; +import io.netty.util.internal.StringUtil; + +import java.util.Collection; + +@Route(APIPath.User_Online_Count) +@HttpMethod("POST") +public class GetUserOnlineCountAction extends AdminAction { + + @Override + public boolean isTransactionAction() { + return true; + } + + @Override + public boolean action(Request request, Response response) { + if (request.getNettyRequest() instanceof FullHttpRequest) { + int count = messagesStore.getOnlineUserCount(); + GetOnlineUserCountResult result = new GetOnlineUserCountResult(); + GetOnlineUserCountResult.Node node = new GetOnlineUserCountResult.Node(); + node.node = 1; + node.count = count; + result.add(node); + + setResponseContent(RestResult.ok(result), response); + } + return true; + } +} diff --git a/broker/src/main/java/com/xiaoleilu/loServer/action/admin/GetUserSessionsAction.java b/broker/src/main/java/com/xiaoleilu/loServer/action/admin/GetUserSessionsAction.java new file mode 100755 index 00000000..3dae392f --- /dev/null +++ b/broker/src/main/java/com/xiaoleilu/loServer/action/admin/GetUserSessionsAction.java @@ -0,0 +1,57 @@ +/* + * This file is part of the Wildfire Chat package. + * (c) Heavyrain2012 + * + * For the full copyright and license information, please view the LICENSE + * file that was distributed with this source code. + */ + +package com.xiaoleilu.loServer.action.admin; + +import cn.wildfirechat.common.APIPath; +import cn.wildfirechat.common.ErrorCode; +import cn.wildfirechat.pojos.GetUserSessionResult; +import cn.wildfirechat.pojos.InputGetChatroomInfo; +import cn.wildfirechat.pojos.InputUserId; +import cn.wildfirechat.pojos.OutputStringList; +import com.xiaoleilu.loServer.RestResult; +import com.xiaoleilu.loServer.annotation.HttpMethod; +import com.xiaoleilu.loServer.annotation.Route; +import com.xiaoleilu.loServer.handler.Request; +import com.xiaoleilu.loServer.handler.Response; +import io.moquette.persistence.MemorySessionStore; +import io.netty.handler.codec.http.FullHttpRequest; +import io.netty.util.internal.StringUtil; + +import java.util.Collection; + +@Route(APIPath.User_Session_List) +@HttpMethod("POST") +public class GetUserSessionsAction extends AdminAction { + + @Override + public boolean isTransactionAction() { + return true; + } + + @Override + public boolean action(Request request, Response response) { + if (request.getNettyRequest() instanceof FullHttpRequest) { + InputUserId inputUserId = getRequestBody(request.getNettyRequest(), InputUserId.class); + String userId = inputUserId.getUserId(); + if (!StringUtil.isNullOrEmpty(userId)) { + Collection sessions = sessionsStore.sessionForUser(userId); + GetUserSessionResult result = new GetUserSessionResult(); + for (MemorySessionStore.Session session : sessions) { + boolean isOnline = sessionsStore.isClientOnline(session.getClientID()); + result.userSessions.add(new GetUserSessionResult.UserSession(session.getUsername(), session.getClientID(), session.getPlatform(), session.getPushType(), session.getDeviceToken(), session.getVoipDeviceToken(), isOnline)); + } + + setResponseContent(RestResult.ok(result), response); + } else { + setResponseContent(RestResult.resultOf(ErrorCode.INVALID_PARAMETER), response); + } + } + return true; + } +} diff --git a/broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java b/broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java index dd37b24a..251581ee 100755 --- a/broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java +++ b/broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java @@ -4367,6 +4367,35 @@ public class MemoryMessagesStore implements IMessagesStore { return getChannelListener(channelId); } + @Override + public int getOnlineUserCount() { + return m_Server.getProcessor().getConnectionDescriptors().getActiveConnectionsNo(); + } + + @Override + public GetOnlineUserResult getOnlineUsers(int offset, int count) { + List onlineUserIds = new ArrayList<>(m_Server.getProcessor().getConnectionDescriptors().getConnectedClientIds()); + GetOnlineUserResult result = new GetOnlineUserResult(); + result.userClients = new ArrayList<>(); + result.totalCount = onlineUserIds.size(); + result.offset = offset; + + if(onlineUserIds.size() > offset) { + List clientIds = onlineUserIds.subList(offset, Math.min(offset+count, onlineUserIds.size())); + for (String clientId : clientIds) { + MemorySessionStore.Session session = m_Server.getStore().sessionsStore().getSession(clientId); + if(session != null) { + GetOnlineUserResult.UserClient userClient = new GetOnlineUserResult.UserClient(); + userClient.userId = session.getUsername(); + userClient.clientId = session.getClientID(); + userClient.platform = session.getPlatform(); + result.userClients.add(userClient); + } + } + } + return result; + } + @Override public Set handleSensitiveWord(String message) { updateSensitiveWord(); diff --git a/broker/src/main/java/io/moquette/persistence/MemorySessionStore.java b/broker/src/main/java/io/moquette/persistence/MemorySessionStore.java index 16bfd784..5361652a 100755 --- a/broker/src/main/java/io/moquette/persistence/MemorySessionStore.java +++ b/broker/src/main/java/io/moquette/persistence/MemorySessionStore.java @@ -615,6 +615,11 @@ public class MemorySessionStore implements ISessionsStore { return result; } + @Override + public boolean isClientOnline(String clientId) { + return mServer.getProcessor().getConnectionDescriptors().isConnected(clientId); + } + @Override public StoredMessage inFlightAck(String clientID, int messageID) { return getSession(clientID).outboundFlightMessages.remove(messageID); diff --git a/broker/src/main/java/io/moquette/spi/IMessagesStore.java b/broker/src/main/java/io/moquette/spi/IMessagesStore.java index f9aeee61..b6dcded9 100755 --- a/broker/src/main/java/io/moquette/spi/IMessagesStore.java +++ b/broker/src/main/java/io/moquette/spi/IMessagesStore.java @@ -16,13 +16,10 @@ package io.moquette.spi; -import cn.wildfirechat.pojos.InputOutputUserInfo; -import cn.wildfirechat.pojos.OutputUserChatroom; -import cn.wildfirechat.pojos.SystemSettingPojo; +import cn.wildfirechat.pojos.*; import cn.wildfirechat.proto.ProtoConstants; import cn.wildfirechat.proto.WFCMessage; import com.xiaoleilu.loServer.model.FriendData; -import cn.wildfirechat.pojos.InputOutputUserBlockStatus; import io.moquette.persistence.DatabaseStore; import io.moquette.persistence.MemorySessionStore; import io.moquette.persistence.UserClientEntry; @@ -32,10 +29,7 @@ import io.netty.buffer.Unpooled; import io.netty.handler.codec.mqtt.MqttQoS; import cn.wildfirechat.common.ErrorCode; -import java.util.Collection; -import java.util.Date; -import java.util.List; -import java.util.Set; +import java.util.*; /** * Defines the SPI to be implemented by a StorageService that handle persistence of messages @@ -228,6 +222,9 @@ public interface IMessagesStore { boolean checkUserInChannel(String user, String channelId); Collection getChannelSubscriber(String channelId); + int getOnlineUserCount(); + GetOnlineUserResult getOnlineUsers(int offset, int count); + Set handleSensitiveWord(String message); boolean addSensitiveWords(List words); boolean removeSensitiveWords(List words); diff --git a/broker/src/main/java/io/moquette/spi/ISessionsStore.java b/broker/src/main/java/io/moquette/spi/ISessionsStore.java index 11afe1d0..4e6cdf7b 100755 --- a/broker/src/main/java/io/moquette/spi/ISessionsStore.java +++ b/broker/src/main/java/io/moquette/spi/ISessionsStore.java @@ -74,6 +74,8 @@ public interface ISessionsStore { */ Collection getAllSessions(); + boolean isClientOnline(String clientId); + StoredMessage inFlightAck(String clientID, int messageID); /** diff --git a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java index 0bcdf9f4..175b6390 100755 --- a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java @@ -731,6 +731,10 @@ public class ProtocolProcessor { channel.flush(); } + public ConnectionDescriptorStore getConnectionDescriptors() { + return connectionDescriptors; + } + public void addInterceptHandler(InterceptHandler interceptHandler) { this.m_interceptor.addInterceptHandler(interceptHandler); } diff --git a/common/src/main/java/cn/wildfirechat/common/APIPath.java b/common/src/main/java/cn/wildfirechat/common/APIPath.java index 32f343f6..413e182a 100644 --- a/common/src/main/java/cn/wildfirechat/common/APIPath.java +++ b/common/src/main/java/cn/wildfirechat/common/APIPath.java @@ -37,6 +37,7 @@ public interface APIPath { String User_Kickoff_Client = "/admin/user/kickoff_client"; String User_Online_Count = "/admin/user/online_count"; String User_Online_List = "/admin/user/online_list"; + String User_Session_List = "/admin/user/session_list"; String User_Application_Get_UserInfo = "/admin/user/app_get_user_info"; String Friend_Update_Status = "/admin/friend/status"; diff --git a/common/src/main/java/cn/wildfirechat/pojos/GetUserSessionResult.java b/common/src/main/java/cn/wildfirechat/pojos/GetUserSessionResult.java new file mode 100644 index 00000000..8f60365c --- /dev/null +++ b/common/src/main/java/cn/wildfirechat/pojos/GetUserSessionResult.java @@ -0,0 +1,35 @@ +package cn.wildfirechat.pojos; + +import java.util.ArrayList; +import java.util.List; + +public class GetUserSessionResult { + public static class UserSession { + public String userId; + public String clientId; + public int platform; + public int pushType; + public String deviceToken; + public String deviceVoipToken; + boolean isOnline; + + public UserSession() { + } + + public UserSession(String userId, String clientId, int platform, int pushType, String deviceToken, String deviceVoipToken, boolean isOnline) { + this.userId = userId; + this.clientId = clientId; + this.platform = platform; + this.pushType = pushType; + this.deviceToken = deviceToken; + this.deviceVoipToken = deviceVoipToken; + this.isOnline = isOnline; + } + } + + public List userSessions; + + public GetUserSessionResult() { + userSessions = new ArrayList<>(); + } +} diff --git a/sdk/src/main/java/cn/wildfirechat/sdk/Main.java b/sdk/src/main/java/cn/wildfirechat/sdk/Main.java index 24d3918a..ebe95f84 100644 --- a/sdk/src/main/java/cn/wildfirechat/sdk/Main.java +++ b/sdk/src/main/java/cn/wildfirechat/sdk/Main.java @@ -324,6 +324,14 @@ public class Main { System.out.println("get user online failure"); System.exit(-1); } + + IMResult getUserSessionResultIMResult = UserAdmin.getUserSession("hygqmws2k"); + if (getUserSessionResultIMResult != null && getUserSessionResultIMResult.getErrorCode() == ErrorCode.ERROR_CODE_SUCCESS) { + System.out.println("get user session success"); + } else { + System.out.println("get user session failure"); + System.exit(-1); + } } } diff --git a/sdk/src/main/java/cn/wildfirechat/sdk/UserAdmin.java b/sdk/src/main/java/cn/wildfirechat/sdk/UserAdmin.java index cac4470e..34233103 100644 --- a/sdk/src/main/java/cn/wildfirechat/sdk/UserAdmin.java +++ b/sdk/src/main/java/cn/wildfirechat/sdk/UserAdmin.java @@ -151,4 +151,9 @@ public class UserAdmin { input.setAuthCode(authCode); return AdminHttpUtils.httpJsonPost(path, input, OutputApplicationUserInfo.class); } + + public static IMResult getUserSession(String userId) throws Exception { + InputUserId inputUserId = new InputUserId(userId); + return AdminHttpUtils.httpJsonPost(APIPath.User_Session_List, inputUserId, GetUserSessionResult.class); + } }