From d0671c16822cebb4fb31f35475a68df19abafad0 Mon Sep 17 00:00:00 2001 From: "heavyrain.lee" Date: Fri, 31 May 2019 07:51:54 +0800 Subject: [PATCH] add broad cast message --- .../action/admin/BroadcastMessageAction.java | 90 +++++++++++++++++++ .../imhandler/BroadcastMessageHandler.java | 49 ++++++++++ .../java/io/moquette/imhandler/IMHandler.java | 10 ++- .../moquette/persistence/DatabaseStore.java | 27 ++++++ .../persistence/MemoryMessagesStore.java | 4 + .../java/io/moquette/spi/IMessagesStore.java | 1 + .../src/main/java/win/liyufan/im/IMTopic.java | 1 + .../BroadMessageData.java | 59 ++++++++++++ .../BroadMessageResult.java | 38 ++++++++ .../java/cn/wildfirechat/common/APIPath.java | 1 + .../main/java/cn/wildfirechat/sdk/Main.java | 8 ++ .../cn/wildfirechat/sdk/MessageAdmin.java | 9 ++ 12 files changed, 296 insertions(+), 1 deletion(-) create mode 100755 broker/src/main/java/com/xiaoleilu/loServer/action/admin/BroadcastMessageAction.java create mode 100644 broker/src/main/java/io/moquette/imhandler/BroadcastMessageHandler.java create mode 100644 common/src/main/java/cn.wildfirechat.pojos/BroadMessageData.java create mode 100644 common/src/main/java/cn.wildfirechat.pojos/BroadMessageResult.java diff --git a/broker/src/main/java/com/xiaoleilu/loServer/action/admin/BroadcastMessageAction.java b/broker/src/main/java/com/xiaoleilu/loServer/action/admin/BroadcastMessageAction.java new file mode 100755 index 00000000..5bd0331c --- /dev/null +++ b/broker/src/main/java/com/xiaoleilu/loServer/action/admin/BroadcastMessageAction.java @@ -0,0 +1,90 @@ +/* + * 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.BroadMessageData; +import cn.wildfirechat.pojos.BroadMessageResult; +import cn.wildfirechat.pojos.SendMessageData; +import cn.wildfirechat.pojos.SendMessageResult; +import com.google.gson.Gson; +import com.hazelcast.util.StringUtil; +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.RPCCenter; +import io.moquette.persistence.TargetEntry; +import io.netty.buffer.ByteBuf; +import io.netty.buffer.Unpooled; +import io.netty.handler.codec.http.FullHttpRequest; +import io.netty.handler.codec.http.HttpResponseStatus; +import win.liyufan.im.IMTopic; + +import java.util.concurrent.Executor; + +@Route(APIPath.Msg_Broadcast) +@HttpMethod("POST") +public class BroadcastMessageAction extends AdminAction { + + @Override + public boolean isTransactionAction() { + return true; + } + + @Override + public boolean action(Request request, Response response) { + if (request.getNettyRequest() instanceof FullHttpRequest) { + BroadMessageData sendMessageData = getRequestBody(request.getNettyRequest(), BroadMessageData.class); + if (BroadMessageData.isValide(sendMessageData) && !StringUtil.isNullOrEmpty(sendMessageData.getSender())) { + RPCCenter.getInstance().sendRequest(sendMessageData.getSender(), null, IMTopic.BroadcastMessageTopic, sendMessageData.toProtoMessage().toByteArray(), sendMessageData.getSender(), TargetEntry.Type.TARGET_TYPE_USER, new RPCCenter.Callback() { + @Override + public void onSuccess(byte[] result) { + ByteBuf byteBuf = Unpooled.buffer(); + byteBuf.writeBytes(result); + ErrorCode errorCode = ErrorCode.fromCode(byteBuf.readByte()); + if (errorCode == ErrorCode.ERROR_CODE_SUCCESS) { + long messageId = byteBuf.readLong(); + long count = byteBuf.readLong(); + sendResponse(response, null, new BroadMessageResult(messageId, count)); + } else { + sendResponse(response, errorCode, null); + } + } + + @Override + public void onError(ErrorCode errorCode) { + sendResponse(response, errorCode, null); + } + + @Override + public void onTimeout() { + sendResponse(response, ErrorCode.ERROR_CODE_TIMEOUT, null); + } + + @Override + public Executor getResponseExecutor() { + return command -> { + ctx.executor().execute(command); + }; + } + }, true); + return false; + } else { + response.setStatus(HttpResponseStatus.OK); + RestResult result = RestResult.resultOf(ErrorCode.INVALID_PARAMETER); + response.setContent(new Gson().toJson(result)); + } + } + return true; + } +} diff --git a/broker/src/main/java/io/moquette/imhandler/BroadcastMessageHandler.java b/broker/src/main/java/io/moquette/imhandler/BroadcastMessageHandler.java new file mode 100644 index 00000000..df98c0fe --- /dev/null +++ b/broker/src/main/java/io/moquette/imhandler/BroadcastMessageHandler.java @@ -0,0 +1,49 @@ +/* + * 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 io.moquette.imhandler; + +import cn.wildfirechat.common.ErrorCode; +import cn.wildfirechat.proto.ProtoConstants; +import cn.wildfirechat.proto.WFCMessage; +import com.hazelcast.util.StringUtil; +import io.moquette.BrokerConstants; +import io.moquette.spi.impl.Qos1PublishHandler; +import io.netty.buffer.ByteBuf; +import win.liyufan.im.IMTopic; +import win.liyufan.im.MessageShardingUtil; + +import java.util.Set; + +import static cn.wildfirechat.proto.ProtoConstants.ContentType.Text; + +@Handler(value = IMTopic.BroadcastMessageTopic) +public class BroadcastMessageHandler extends IMHandler { + @Override + public ErrorCode action(ByteBuf ackPayload, String clientID, String fromUser, boolean isAdmin, WFCMessage.Message message, Qos1PublishHandler.IMCallback callback) { + ErrorCode errorCode = ErrorCode.ERROR_CODE_SUCCESS; + if (message != null) { + if (!isAdmin) { //only admin can broadcast message + return ErrorCode.ERROR_CODE_NOT_RIGHT; + } + + long timestamp = System.currentTimeMillis(); + long messageId = MessageShardingUtil.generateId(); + message = message.toBuilder().setFromUser(fromUser).setMessageId(messageId).setServerTimestamp(timestamp).setConversation(WFCMessage.Conversation.newBuilder().setTarget(fromUser).setLine(message.getConversation().getLine()).setType(ProtoConstants.ConversationType.ConversationType_Private).build()).build(); + + long count = saveAndBroadcast(fromUser, clientID, message); + ackPayload = ackPayload.capacity(20); + ackPayload.writeLong(messageId); + ackPayload.writeLong(count); + } else { + errorCode = ErrorCode.ERROR_CODE_INVALID_MESSAGE; + } + return errorCode; + } + +} diff --git a/broker/src/main/java/io/moquette/imhandler/IMHandler.java b/broker/src/main/java/io/moquette/imhandler/IMHandler.java index 786e9643..881b4966 100755 --- a/broker/src/main/java/io/moquette/imhandler/IMHandler.java +++ b/broker/src/main/java/io/moquette/imhandler/IMHandler.java @@ -8,6 +8,7 @@ package io.moquette.imhandler; +import cn.wildfirechat.proto.ProtoConstants; import cn.wildfirechat.proto.WFCMessage; import cn.wildfirechat.server.ThreadPoolExecutorWrapper; import com.google.gson.Gson; @@ -239,6 +240,13 @@ abstract public class IMHandler { message = m_messagesStore.storeMessage(username, clientID, message); int pullType = m_messagesStore.getNotifyReceivers(username, message, notifyReceivers); this.publisher.publish2Receivers(message, notifyReceivers, clientID, pullType); - return message.getMessageId(); + return notifyReceivers.size(); + } + + protected long saveAndBroadcast(String username, String clientID, WFCMessage.Message message) { + Set notifyReceivers = m_messagesStore.getAllEnds(); + WFCMessage.Message updatedMessage = m_messagesStore.storeMessage(username, clientID, message); + mServer.getImBusinessScheduler().execute(() -> publisher.publish2Receivers(updatedMessage, notifyReceivers, clientID, ProtoConstants.PullType.Pull_Normal)); + return notifyReceivers.size(); } } diff --git a/broker/src/main/java/io/moquette/persistence/DatabaseStore.java b/broker/src/main/java/io/moquette/persistence/DatabaseStore.java index 7c79065b..be846618 100644 --- a/broker/src/main/java/io/moquette/persistence/DatabaseStore.java +++ b/broker/src/main/java/io/moquette/persistence/DatabaseStore.java @@ -1774,6 +1774,33 @@ public class DatabaseStore { return null; } + Set getAllEnds() { + Connection connection = null; + PreparedStatement statement = null; + ResultSet rs = null; + try { + connection = DBUtil.getConnection(); + String sql = "select distinct(`_uid`) from t_user_session"; + statement = connection.prepareStatement(sql); + + + + rs = statement.executeQuery(); + Set out = new HashSet<>(); + while (rs.next()) { + String uid = rs.getString(1); + out.add(uid); + } + return out; + } catch (SQLException e) { + // TODO Auto-generated catch block + e.printStackTrace(); + Utility.printExecption(LOG, e); + } finally { + DBUtil.closeDB(connection, statement, rs); + } + return null; + } List getPersistFriends(String userId) { Connection connection = null; PreparedStatement statement = null; diff --git a/broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java b/broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java index 996aceec..34ccbf06 100755 --- a/broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java +++ b/broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java @@ -245,6 +245,10 @@ public class MemoryMessagesStore implements IMessagesStore { return pullType; } + @Override + public Set getAllEnds() { + return databaseStore.getAllEnds(); + } @Override public WFCMessage.PullMessageResult fetchMessage(String user, String exceptClientId, long fromMessageId, int pullType) { WFCMessage.PullMessageResult.Builder builder = WFCMessage.PullMessageResult.newBuilder(); diff --git a/broker/src/main/java/io/moquette/spi/IMessagesStore.java b/broker/src/main/java/io/moquette/spi/IMessagesStore.java index 50aa372d..45ec3f84 100755 --- a/broker/src/main/java/io/moquette/spi/IMessagesStore.java +++ b/broker/src/main/java/io/moquette/spi/IMessagesStore.java @@ -97,6 +97,7 @@ public interface IMessagesStore { DatabaseStore getDatabaseStore(); WFCMessage.Message storeMessage(String fromUser, String fromClientId, WFCMessage.Message message); int getNotifyReceivers(String fromUser, WFCMessage.Message message, Set notifyReceivers); + Set getAllEnds(); WFCMessage.PullMessageResult fetchMessage(String user, String exceptClientId, long fromMessageId, int pullType); WFCMessage.PullMessageResult loadRemoteMessages(String user, WFCMessage.Conversation conversation, long beforeUid, int count); long insertUserMessages(String sender, int conversationType, String target, int line, int messageContentType, String userId, long messageId); diff --git a/broker/src/main/java/win/liyufan/im/IMTopic.java b/broker/src/main/java/win/liyufan/im/IMTopic.java index 29a53185..dd9bc3a9 100755 --- a/broker/src/main/java/win/liyufan/im/IMTopic.java +++ b/broker/src/main/java/win/liyufan/im/IMTopic.java @@ -14,6 +14,7 @@ public interface IMTopic { String PullMessageTopic = "MP"; String NotifyMessageTopic = "MN"; String NotifyRecallMessageTopic = "RMN"; + String BroadcastMessageTopic = "MBC"; String GetUserSettingTopic = "UG"; String PutUserSettingTopic = "UP"; diff --git a/common/src/main/java/cn.wildfirechat.pojos/BroadMessageData.java b/common/src/main/java/cn.wildfirechat.pojos/BroadMessageData.java new file mode 100644 index 00000000..5f93ab5d --- /dev/null +++ b/common/src/main/java/cn.wildfirechat.pojos/BroadMessageData.java @@ -0,0 +1,59 @@ +/* + * 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 cn.wildfirechat.pojos; + +import cn.wildfirechat.proto.ProtoConstants; +import cn.wildfirechat.proto.WFCMessage; +import io.netty.util.internal.StringUtil; + +public class BroadMessageData { + private String sender; + private int line; + private MessagePayload payload; + + public String getSender() { + return sender; + } + + public void setSender(String sender) { + this.sender = sender; + } + + public int getLine() { + return line; + } + + public void setLine(int line) { + this.line = line; + } + + public MessagePayload getPayload() { + return payload; + } + + public void setPayload(MessagePayload payload) { + this.payload = payload; + } + + public static boolean isValide(BroadMessageData sendMessageData) { + if(sendMessageData == null || + StringUtil.isNullOrEmpty(sendMessageData.getSender()) || + sendMessageData.getPayload() == null) { + return false; + } + return true; + } + + public WFCMessage.Message toProtoMessage() { + return WFCMessage.Message.newBuilder().setFromUser(sender) + .setConversation(WFCMessage.Conversation.newBuilder().setType(ProtoConstants.ConversationType.ConversationType_Private).setTarget(sender).setLine(line)) + .setContent(payload.toProtoMessageContent()) + .build(); + } +} diff --git a/common/src/main/java/cn.wildfirechat.pojos/BroadMessageResult.java b/common/src/main/java/cn.wildfirechat.pojos/BroadMessageResult.java new file mode 100644 index 00000000..a27205f2 --- /dev/null +++ b/common/src/main/java/cn.wildfirechat.pojos/BroadMessageResult.java @@ -0,0 +1,38 @@ +/* + * 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 cn.wildfirechat.pojos; + +public class BroadMessageResult { + private long messageUid; + private long count; + + public BroadMessageResult() { + } + + public BroadMessageResult(long messageUid, long count) { + this.messageUid = messageUid; + this.count = count; + } + + public long getMessageUid() { + return messageUid; + } + + public void setMessageUid(long messageUid) { + this.messageUid = messageUid; + } + + public long getCount() { + return count; + } + + public void setCount(long count) { + this.count = count; + } +} diff --git a/common/src/main/java/cn/wildfirechat/common/APIPath.java b/common/src/main/java/cn/wildfirechat/common/APIPath.java index ff8e5bbd..e834f950 100644 --- a/common/src/main/java/cn/wildfirechat/common/APIPath.java +++ b/common/src/main/java/cn/wildfirechat/common/APIPath.java @@ -25,6 +25,7 @@ public interface APIPath { String Msg_Send = "/admin/message/send"; String Msg_Recall = "/admin/message/recall"; + String Msg_Broadcast = "/admin/message/broadcast"; String Create_Group = "/admin/group/create"; String Group_Dismiss = "/admin/group/del"; diff --git a/sdk/src/main/java/cn/wildfirechat/sdk/Main.java b/sdk/src/main/java/cn/wildfirechat/sdk/Main.java index ccce7e1c..b4c41fc5 100644 --- a/sdk/src/main/java/cn/wildfirechat/sdk/Main.java +++ b/sdk/src/main/java/cn/wildfirechat/sdk/Main.java @@ -264,6 +264,14 @@ public class Main { } + IMResult resultBroadcastMessage = MessageAdmin.broadcastMessage("user1", 0, payload); + if (resultBroadcastMessage != null && resultBroadcastMessage.getErrorCode() == ErrorCode.ERROR_CODE_SUCCESS) { + System.out.println("broad message success, send message to " + resultBroadcastMessage.getResult().getCount() + " users"); + } else { + System.out.println("broad message failure"); + System.exit(-1); + } + //初始化机器人API RobotHttpUtils.init("http://localhost:18080", "robot1", "123456"); diff --git a/sdk/src/main/java/cn/wildfirechat/sdk/MessageAdmin.java b/sdk/src/main/java/cn/wildfirechat/sdk/MessageAdmin.java index ff297fd9..4493ee23 100644 --- a/sdk/src/main/java/cn/wildfirechat/sdk/MessageAdmin.java +++ b/sdk/src/main/java/cn/wildfirechat/sdk/MessageAdmin.java @@ -22,4 +22,13 @@ public class MessageAdmin { messageData.setMessageUid(messageUid); return AdminHttpUtils.httpJsonPost(path, messageData, Void.class); } + + public static IMResult broadcastMessage(String sender, int line, MessagePayload payload) throws Exception { + String path = APIPath.Msg_Broadcast; + BroadMessageData messageData = new BroadMessageData(); + messageData.setSender(sender); + messageData.setLine(line); + messageData.setPayload(payload); + return AdminHttpUtils.httpJsonPost(path, messageData, BroadMessageResult.class); + } }