diff --git a/broker/src/main/java/com/xiaoleilu/loServer/action/admin/RecallMultiCastMessageAction.java b/broker/src/main/java/com/xiaoleilu/loServer/action/admin/RecallMultiCastMessageAction.java new file mode 100755 index 00000000..620a8f7c --- /dev/null +++ b/broker/src/main/java/com/xiaoleilu/loServer/action/admin/RecallMultiCastMessageAction.java @@ -0,0 +1,84 @@ +/* + * 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.RecallMessageData; +import cn.wildfirechat.pojos.RecallMultiCastMessageData; +import cn.wildfirechat.proto.WFCMessage; +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 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 io.netty.util.internal.StringUtil; +import win.liyufan.im.IMTopic; + +import java.util.concurrent.Executor; + +@Route(APIPath.Msg_RecallMultiCast) +@HttpMethod("POST") +public class RecallMultiCastMessageAction extends AdminAction { + + @Override + public boolean isTransactionAction() { + return true; + } + + @Override + public boolean action(Request request, Response response) { + if (request.getNettyRequest() instanceof FullHttpRequest) { + RecallMultiCastMessageData recallMessageData = getRequestBody(request.getNettyRequest(), RecallMultiCastMessageData.class); + if (recallMessageData != null && !StringUtil.isNullOrEmpty(recallMessageData.operator)) { + + WFCMessage.RecallMultiCastMessageRequest recallRequest = WFCMessage.RecallMultiCastMessageRequest.newBuilder().setMessageId(recallMessageData.messageUid).addAllReceiver(recallMessageData.receivers).build(); + RPCCenter.getInstance().sendRequest(recallMessageData.operator, null, IMTopic.RecallMultiCastMessageTopic, recallRequest.toByteArray(), recallMessageData.operator, 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()); + 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/IMHandler.java b/broker/src/main/java/io/moquette/imhandler/IMHandler.java index 15d84576..d695149f 100755 --- a/broker/src/main/java/io/moquette/imhandler/IMHandler.java +++ b/broker/src/main/java/io/moquette/imhandler/IMHandler.java @@ -268,4 +268,13 @@ abstract public class IMHandler { mServer.getImBusinessScheduler().execute(() -> publisher.publish2Receivers(updatedMessage, notifyReceivers, clientID, ProtoConstants.PullType.Pull_Normal)); return notifyReceivers.size(); } + protected long publishRecallMultiCastMsg(long messageUid, List receivers) { + WFCMessage.Message updatedMessage = m_messagesStore.getMessage(messageUid); + + Set notifyReceivers = new HashSet<>(receivers); + LOG.info("Multicast recall receiver count: {}", notifyReceivers.size()); + mServer.getImBusinessScheduler().execute(() -> publisher.publish2Receivers(updatedMessage, notifyReceivers, null, ProtoConstants.PullType.Pull_Normal, false)); + + return notifyReceivers.size(); + } } diff --git a/broker/src/main/java/io/moquette/imhandler/RecallMultiCastMessageHandler.java b/broker/src/main/java/io/moquette/imhandler/RecallMultiCastMessageHandler.java new file mode 100644 index 00000000..7b31ed2e --- /dev/null +++ b/broker/src/main/java/io/moquette/imhandler/RecallMultiCastMessageHandler.java @@ -0,0 +1,36 @@ +/* + * 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.WFCMessage; +import io.moquette.spi.impl.Qos1PublishHandler; +import io.netty.buffer.ByteBuf; +import win.liyufan.im.IMTopic; + +@Handler(value = IMTopic.RecallMultiCastMessageTopic) +public class RecallMultiCastMessageHandler extends IMHandler { + @Override + public ErrorCode action(ByteBuf ackPayload, String clientID, String fromUser, boolean isAdmin, WFCMessage.RecallMultiCastMessageRequest request, Qos1PublishHandler.IMCallback callback) { + if (!isAdmin) { + return ErrorCode.ERROR_CODE_NOT_RIGHT; + } + + ErrorCode errorCode = m_messagesStore.recallCastMessage(request.getMessageId(), fromUser); + + if(errorCode != ErrorCode.ERROR_CODE_SUCCESS) { + return errorCode; + } + + publishRecallMultiCastMsg(request.getMessageId(), request.getReceiverList()); + + return errorCode; + } + +} diff --git a/broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java b/broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java index 12af98b5..752c6bcc 100755 --- a/broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java +++ b/broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java @@ -1754,6 +1754,26 @@ public class MemoryMessagesStore implements IMessagesStore { } } + public ErrorCode recallCastMessage(long messageUid, String operatorId) { + HazelcastInstance hzInstance = m_Server.getHazelcastInstance(); + IMap mIMap = hzInstance.getMap(MESSAGES_MAP); + + MessageBundle messageBundle = mIMap.get(messageUid); + if (messageBundle != null) { + WFCMessage.Message message = messageBundle.getMessage(); + + message = message.toBuilder().setContent(WFCMessage.MessageContent.newBuilder().setContent(operatorId).setType(80).setData(ByteString.copyFrom(String.valueOf(messageUid).getBytes()))).build(); + messageBundle.setMessage(message); + messageBundle.setFromClientId(null); + + databaseStore.deleteMessage(messageUid); + mIMap.set(messageUid, messageBundle, 7, TimeUnit.DAYS); + return ErrorCode.ERROR_CODE_SUCCESS; + } else { + return ErrorCode.ERROR_CODE_NOT_EXIST; + } + } + @Override public void clearUserMessages(String userId) { mWriteLock.lock(); diff --git a/broker/src/main/java/io/moquette/spi/IMessagesStore.java b/broker/src/main/java/io/moquette/spi/IMessagesStore.java index b62d8f82..fdd228fa 100755 --- a/broker/src/main/java/io/moquette/spi/IMessagesStore.java +++ b/broker/src/main/java/io/moquette/spi/IMessagesStore.java @@ -124,6 +124,7 @@ public interface IMessagesStore { ErrorCode canSendMessageInGroup(String member, String groupId); ErrorCode recallMessage(long messageUid, String operatorId, String clientId, boolean isAdmin); + ErrorCode recallCastMessage(long messageUid, String operatorId); void clearUserMessages(String userId); diff --git a/broker/src/main/java/win/liyufan/im/IMTopic.java b/broker/src/main/java/win/liyufan/im/IMTopic.java index 8d6ded54..101993ae 100755 --- a/broker/src/main/java/win/liyufan/im/IMTopic.java +++ b/broker/src/main/java/win/liyufan/im/IMTopic.java @@ -12,6 +12,7 @@ public interface IMTopic { String SendMessageTopic = "MS"; String MultiCastMessageTopic = "MMC"; String RecallMessageTopic = "MR"; + String RecallMultiCastMessageTopic = "MRMC"; String PullMessageTopic = "MP"; String NotifyMessageTopic = "MN"; String NotifyRecallMessageTopic = "RMN";