From 70c390dea995dc203779466ac55f4238f6620aec Mon Sep 17 00:00:00 2001 From: imhao183 Date: Sat, 4 Jan 2020 12:12:53 +0800 Subject: [PATCH] =?UTF-8?q?=E6=B7=BB=E5=8A=A0=E8=AE=BE=E7=BD=AE/=E8=8E=B7?= =?UTF-8?q?=E5=8F=96=E5=A4=87=E6=B3=A8api=EF=BC=8C=E6=B7=BB=E5=8A=A0?= =?UTF-8?q?=E7=94=A8=E6=88=B7=E5=9C=A8=E7=BA=BF=E7=8A=B6=E6=80=81=E5=9B=9E?= =?UTF-8?q?=E8=B0=83=E5=8A=9F=E8=83=BD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- broker/config/wildfirechat.conf | 5 ++ .../loServer/action/admin/AliasGetAction.java | 53 +++++++++++++++++++ .../loServer/action/admin/AliasPutAction.java | 53 +++++++++++++++++++ .../java/io/moquette/BrokerConstants.java | 2 + .../persistence/MemoryMessagesStore.java | 6 +++ .../moquette/spi/impl/ProtocolProcessor.java | 33 +++++++++++- .../java/cn/wildfirechat/common/APIPath.java | 2 + 7 files changed, 153 insertions(+), 1 deletion(-) create mode 100755 broker/src/main/java/com/xiaoleilu/loServer/action/admin/AliasGetAction.java create mode 100755 broker/src/main/java/com/xiaoleilu/loServer/action/admin/AliasPutAction.java diff --git a/broker/config/wildfirechat.conf b/broker/config/wildfirechat.conf index ceeb9d77..64b7708c 100755 --- a/broker/config/wildfirechat.conf +++ b/broker/config/wildfirechat.conf @@ -129,6 +129,11 @@ push.android.server.address http://localhost:8085/android/push ##苹果推送服务器地址 push.ios.server.address http://localhost:8085/ios/push +#********************************************************************* +# User online status event callback +#********************************************************************* +##用户在线状态事件回调地址 +user.online_status_callback http://localhost:8085/user/online_event #********************************************************************* # MongoDB configuration diff --git a/broker/src/main/java/com/xiaoleilu/loServer/action/admin/AliasGetAction.java b/broker/src/main/java/com/xiaoleilu/loServer/action/admin/AliasGetAction.java new file mode 100755 index 00000000..7db7e564 --- /dev/null +++ b/broker/src/main/java/com/xiaoleilu/loServer/action/admin/AliasGetAction.java @@ -0,0 +1,53 @@ +/* + * 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.pojos.InputGetFriendList; +import cn.wildfirechat.pojos.OutputStringList; +import com.google.gson.Gson; +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 com.xiaoleilu.loServer.model.FriendData; +import io.netty.handler.codec.http.FullHttpRequest; +import io.netty.handler.codec.http.HttpResponseStatus; + +import java.util.ArrayList; +import java.util.List; + +@Route(APIPath.Friend_Get_Alias) +@HttpMethod("POST") +public class AliasGetAction extends AdminAction { + + @Override + public boolean isTransactionAction() { + return true; + } + + @Override + public boolean action(Request request, Response response) { + if (request.getNettyRequest() instanceof FullHttpRequest) { + InputGetFriendList inputGetFriendList = getRequestBody(request.getNettyRequest(), InputGetFriendList.class); + List dataList = messagesStore.getFriendList(inputGetFriendList.getUserId(), null, 0); + List list = new ArrayList<>(); + for (FriendData data : dataList) { + if (data.getState() == inputGetFriendList.getStatus()) { + list.add(data.getFriendUid()); + } + } + response.setStatus(HttpResponseStatus.OK); + RestResult result = RestResult.ok(new OutputStringList(list)); + response.setContent(new Gson().toJson(result)); + } + return true; + } +} diff --git a/broker/src/main/java/com/xiaoleilu/loServer/action/admin/AliasPutAction.java b/broker/src/main/java/com/xiaoleilu/loServer/action/admin/AliasPutAction.java new file mode 100755 index 00000000..4400e487 --- /dev/null +++ b/broker/src/main/java/com/xiaoleilu/loServer/action/admin/AliasPutAction.java @@ -0,0 +1,53 @@ +/* + * 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.pojos.InputGetFriendList; +import cn.wildfirechat.pojos.OutputStringList; +import com.google.gson.Gson; +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 com.xiaoleilu.loServer.model.FriendData; +import io.netty.handler.codec.http.FullHttpRequest; +import io.netty.handler.codec.http.HttpResponseStatus; + +import java.util.ArrayList; +import java.util.List; + +@Route(APIPath.Friend_Set_Alias) +@HttpMethod("POST") +public class AliasPutAction extends AdminAction { + + @Override + public boolean isTransactionAction() { + return true; + } + + @Override + public boolean action(Request request, Response response) { + if (request.getNettyRequest() instanceof FullHttpRequest) { + InputGetFriendList inputGetFriendList = getRequestBody(request.getNettyRequest(), InputGetFriendList.class); + List dataList = messagesStore.getFriendList(inputGetFriendList.getUserId(), null, 0); + List list = new ArrayList<>(); + for (FriendData data : dataList) { + if (data.getState() == inputGetFriendList.getStatus()) { + list.add(data.getFriendUid()); + } + } + response.setStatus(HttpResponseStatus.OK); + RestResult result = RestResult.ok(new OutputStringList(list)); + response.setContent(new Gson().toJson(result)); + } + return true; + } +} diff --git a/broker/src/main/java/io/moquette/BrokerConstants.java b/broker/src/main/java/io/moquette/BrokerConstants.java index 8efa606c..52d910b8 100755 --- a/broker/src/main/java/io/moquette/BrokerConstants.java +++ b/broker/src/main/java/io/moquette/BrokerConstants.java @@ -88,6 +88,8 @@ public final class BrokerConstants { public static final String PUSH_ANDROID_SERVER_ADDRESS = "push.android.server.address"; public static final String PUSH_IOS_SERVER_ADDRESS = "push.ios.server.address"; + public static final String USER_ONLINE_STATUS_CALLBACK = "user.online_status_callback"; + public static final String HZ_Cluster_Node_External_IP = "node_external_ip"; public static final String HZ_Cluster_Node_External_Long_Port = "node_external_long_port"; public static final String HZ_Cluster_Node_External_Short_Port = "node_external_short_port"; diff --git a/broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java b/broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java index a1a04172..13841806 100755 --- a/broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java +++ b/broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java @@ -386,6 +386,12 @@ public class MemoryMessagesStore implements IMessagesStore { continue; } + if (bundle.getMessage().getContent().getExpireDuration() > 0) { + if (System.currentTimeMillis() < bundle.getMessage().getServerTimestamp() + bundle.getMessage().getContent().getExpireDuration()) { + continue; + } + } + size += bundle.getMessage().getSerializedSize(); if (size >= 512 * 1024) { //3M if(builder.getMessageCount() == 0){ 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 c3b0b32d..f71858eb 100755 --- a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java @@ -17,8 +17,13 @@ package io.moquette.spi.impl; import cn.wildfirechat.common.ErrorCode; +import cn.wildfirechat.pojos.SendMessageData; import cn.wildfirechat.proto.ProtoConstants; import cn.wildfirechat.proto.WFCMessage; +import com.google.gson.Gson; +import com.hazelcast.core.Member; +import com.hazelcast.util.StringUtil; +import io.moquette.BrokerConstants; import io.moquette.persistence.RPCCenter; import io.moquette.interception.InterceptHandler; import io.moquette.interception.messages.InterceptAcknowledgedMessage; @@ -40,9 +45,12 @@ import io.netty.handler.codec.mqtt.*; import io.netty.handler.timeout.IdleStateHandler; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import win.liyufan.im.HttpUtils; import win.liyufan.im.Utility; import java.util.*; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; import static io.moquette.spi.impl.InternalRepublisher.createPublishForQos; import static io.moquette.spi.impl.Utils.readBytesAndRewind; @@ -59,7 +67,7 @@ import static io.moquette.server.ConnectionDescriptor.ConnectionState.*; */ public class ProtocolProcessor { - + private static ExecutorService executorCallback = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors()); public void kickoffSession(final MemorySessionStore.Session session) { mServer.getImBusinessScheduler().execute(()->{ @@ -154,8 +162,15 @@ public class ProtocolProcessor { this.connectionDescriptors, this.messagesPublisher, sessionsStore, server.getImBusinessScheduler(), server); mServer = server; + + String onlineStatusCallback = server.getConfig().getProperty(BrokerConstants.USER_ONLINE_STATUS_CALLBACK); + if (!com.hazelcast.util.StringUtil.isNullOrEmpty(onlineStatusCallback)) { + mUserOnlineStatusCallback = onlineStatusCallback; + } } + private String mUserOnlineStatusCallback; + public void processConnect(Channel channel, MqttConnectMessage msg) { MqttConnectPayload payload = msg.payload(); String clientId = payload.clientIdentifier(); @@ -238,6 +253,22 @@ public class ProtocolProcessor { LOG.info("The CONNECT message has been processed. CId={}, username={}", clientId, payload.userName()); } + private static class UserOnlineStatus { + public String userId; + public int status; //0 offline; 1 online; -1 logout + + public UserOnlineStatus(String userId, int status) { + this.userId = userId; + this.status = status; + } + } + + public void forwardOnlineStatusEvent(String userId, int status) { + if (!StringUtil.isNullOrEmpty(mUserOnlineStatusCallback)) { + executorCallback.execute(() -> HttpUtils.httpJsonPost(mUserOnlineStatusCallback, new Gson().toJson(new UserOnlineStatus(userId, status)))); + } + } + private MqttConnAckMessage connAck(MqttConnectReturnCode returnCode) { return connAck(returnCode, false); } diff --git a/common/src/main/java/cn/wildfirechat/common/APIPath.java b/common/src/main/java/cn/wildfirechat/common/APIPath.java index 2a155afa..346b2ee2 100644 --- a/common/src/main/java/cn/wildfirechat/common/APIPath.java +++ b/common/src/main/java/cn/wildfirechat/common/APIPath.java @@ -23,6 +23,8 @@ public interface APIPath { String Friend_Update_Status = "/admin/friend/status"; String Friend_Get_List = "/admin/friend/list"; + String Friend_Get_Alias = "/admin/friend/get_alias"; + String Friend_Set_Alias = "/admin/friend/set_alias"; String Msg_Send = "/admin/message/send"; String Msg_Recall = "/admin/message/recall";