添加组播消息

broker/src/main/java/com/xiaoleilu/loServer/action/admin/MulticastMessageAction.java

@@ -0,0 +1,90 @@
+/*
+ * This file is part of the Wildfire Chat package.
+ * (c) Heavyrain2012 <[email protected]>
+ *
+ * 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.MultiMessageResult;
+import cn.wildfirechat.pojos.MulticastMessageData;
+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_Multicast)
+@HttpMethod("POST")
+public class MulticastMessageAction extends AdminAction {
+
+    @Override
+    public boolean isTransactionAction() {
+        return true;
+    }
+
+    @Override
+    public boolean action(Request request, Response response) {
+        if (request.getNettyRequest() instanceof FullHttpRequest) {
+            MulticastMessageData sendMessageData = getRequestBody(request.getNettyRequest(), MulticastMessageData.class);
+            if (MulticastMessageData.isValide(sendMessageData)) {
+                RPCCenter.getInstance().sendRequest(sendMessageData.getSender(), null, IMTopic.MultiCastMessageTopic, 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 timestamp = byteBuf.readLong();
+                            sendResponse(response, null, new MultiMessageResult(messageId, timestamp));
+                        } 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;
+    }
+}

broker/src/main/java/io/moquette/imhandler/IMHandler.java

@@ -32,6 +32,8 @@ import java.lang.annotation.Retention;
 import java.lang.annotation.RetentionPolicy;
 import java.lang.reflect.InvocationTargetException;
 import java.lang.reflect.Method;
+import java.util.Collection;
+import java.util.HashSet;
 import java.util.LinkedHashSet;
 import java.util.Set;
 
@@ -250,4 +252,12 @@ abstract public class IMHandler<T> {
         mServer.getImBusinessScheduler().execute(() -> publisher.publish2Receivers(updatedMessage, notifyReceivers, clientID, ProtoConstants.PullType.Pull_Normal));
         return notifyReceivers.size();
     }
+
+    protected long saveAndMulticast(String username, String clientID, WFCMessage.Message message, Collection<String> targets) {
+        Set<String> notifyReceivers = new HashSet<>();
+        notifyReceivers.addAll(targets);
+        WFCMessage.Message updatedMessage = m_messagesStore.storeMessage(username, clientID, message);
+        mServer.getImBusinessScheduler().execute(() -> publisher.publish2Receivers(updatedMessage, notifyReceivers, clientID, ProtoConstants.PullType.Pull_Normal));
+        return notifyReceivers.size();
+    }
 }

broker/src/main/java/io/moquette/imhandler/MultiCastMessageHandler.java

@@ -0,0 +1,57 @@
+/*
+ * This file is part of the Wildfire Chat package.
+ * (c) Heavyrain2012 <[email protected]>
+ *
+ * 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.server.Server;
+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;
+import static cn.wildfirechat.proto.ProtoConstants.ConversationType.ConversationType_Private;
+
+@Handler(value = IMTopic.MultiCastMessageTopic)
+public class MultiCastMessageHandler extends IMHandler<WFCMessage.MultiCastMessage> {
+
+
+    @Override
+    public ErrorCode action(ByteBuf ackPayload, String clientID, String fromUser, boolean isAdmin, WFCMessage.MultiCastMessage multiCastMessage, Qos1PublishHandler.IMCallback callback) {
+        ErrorCode errorCode = ErrorCode.ERROR_CODE_SUCCESS;
+        if (!isAdmin) {
+            return ErrorCode.ERROR_CODE_NOT_RIGHT;
+        }
+
+        long timestamp = System.currentTimeMillis();
+        long messageId = MessageShardingUtil.generateId();
+        WFCMessage.Message message = WFCMessage.Message.newBuilder()
+            .setContent(multiCastMessage.getContent())
+            .setConversation(WFCMessage.Conversation.newBuilder().setTarget(fromUser).setType(ConversationType_Private).setLine(multiCastMessage.getLine()).build())
+            .setFromUser(fromUser)
+            .setMessageId(messageId)
+            .setServerTimestamp(timestamp)
+            .build();
+
+        saveAndMulticast(fromUser, clientID, message, multiCastMessage.getToList());
+
+        ackPayload = ackPayload.capacity(20);
+        ackPayload.writeLong(messageId);
+        ackPayload.writeLong(timestamp);
+
+        return errorCode;
+    }
+
+}

broker/src/main/java/win/liyufan/im/IMTopic.java

@@ -10,6 +10,7 @@ package win.liyufan.im;
 
 public interface IMTopic {
 	String SendMessageTopic = "MS";
+    String MultiCastMessageTopic = "MMC";
     String RecallMessageTopic = "MR";
 	String PullMessageTopic = "MP";
 	String NotifyMessageTopic = "MN";