diff --git a/broker/src/main/java/io/moquette/imhandler/ChannelListenedListHandler.java b/broker/src/main/java/io/moquette/imhandler/ChannelListenedListHandler.java new file mode 100644 index 00000000..1ac08038 --- /dev/null +++ b/broker/src/main/java/io/moquette/imhandler/ChannelListenedListHandler.java @@ -0,0 +1,23 @@ +package io.moquette.imhandler; + +import cn.wildfirechat.common.ErrorCode; +import cn.wildfirechat.proto.ProtoConstants; +import cn.wildfirechat.proto.WFCMessage; +import io.moquette.spi.impl.Qos1PublishHandler; +import io.netty.buffer.ByteBuf; +import win.liyufan.im.IMTopic; + +import java.util.List; + +@Handler(IMTopic.ListenedChannelListTopic) +public class ChannelListenedListHandler extends IMHandler { + @Override + public ErrorCode action(ByteBuf ackPayload, String clientID, String fromUser, ProtoConstants.RequestSourceType requestSourceType, Void request, Qos1PublishHandler.IMCallback callback) { + List channelInfoList = m_messagesStore.getListenedChannels(fromUser); + WFCMessage.IDListBuf.Builder builder = WFCMessage.IDListBuf.newBuilder(); + builder.addAllId(channelInfoList); + byte[] data = builder.build().toByteArray(); + ackPayload.ensureWritable(data.length).writeBytes(data); + return ErrorCode.ERROR_CODE_SUCCESS; + } +} diff --git a/broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java b/broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java index 029def62..f6774444 100755 --- a/broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java +++ b/broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java @@ -4139,6 +4139,11 @@ public class MemoryMessagesStore implements IMessagesStore { return databaseStore.searchChannelFromDB(keyword, buzzy, page); } + + @Override + public List getListenedChannels(String userId) { + return databaseStore.getUserChannels(userId); + } @Override public ErrorCode listenChannel(String operator, String channelId, boolean listen) { HazelcastInstance hzInstance = m_Server.getHazelcastInstance(); diff --git a/broker/src/main/java/io/moquette/spi/IMessagesStore.java b/broker/src/main/java/io/moquette/spi/IMessagesStore.java index ea2d8ada..c25c315a 100755 --- a/broker/src/main/java/io/moquette/spi/IMessagesStore.java +++ b/broker/src/main/java/io/moquette/spi/IMessagesStore.java @@ -218,6 +218,7 @@ public interface IMessagesStore { ErrorCode transferChannel(String operator, String channelId, String newOwner); ErrorCode destroyChannel(String operator, String channelId, boolean isAdmin); List searchChannel(String keyword, boolean buzzy, int page); + List getListenedChannels(String userId); ErrorCode listenChannel(String operator, String channelId, boolean listen); WFCMessage.ChannelInfo getChannelInfo(String channelId); boolean canSendMessageInChannel(String user, String channelId); diff --git a/broker/src/main/java/io/moquette/spi/impl/Qos1PublishHandler.java b/broker/src/main/java/io/moquette/spi/impl/Qos1PublishHandler.java index 7065bbdd..e0159799 100755 --- a/broker/src/main/java/io/moquette/spi/impl/Qos1PublishHandler.java +++ b/broker/src/main/java/io/moquette/spi/impl/Qos1PublishHandler.java @@ -243,20 +243,15 @@ public class Qos1PublishHandler extends QosPublishHandler { String imtopic = topic.getTopic(); ByteBuf payload = msg.payload(); byte[] payloadContent = readBytesAndRewind(payload); - if(payloadContent.length == 0) { - ByteBuf ackPayload = Unpooled.buffer(); - ackPayload.ensureWritable(1).writeByte(ERROR_CODE_INVALID_DATA.getCode()); - sendPubAck(clientID, messageID, ackPayload, ERROR_CODE_INVALID_DATA); - return; - } - - MemorySessionStore.Session session = m_sessionStore.getSession(clientID); - payloadContent = AES.AESDecrypt(payloadContent, session.getSecret(), true); - if(payloadContent == null) { - ByteBuf ackPayload = Unpooled.buffer(); - ackPayload.ensureWritable(1).writeByte(ERROR_CODE_INVALID_DATA.getCode()); - sendPubAck(clientID, messageID, ackPayload, ERROR_CODE_INVALID_DATA); - return; + if(payloadContent.length > 0) { + MemorySessionStore.Session session = m_sessionStore.getSession(clientID); + payloadContent = AES.AESDecrypt(payloadContent, session.getSecret(), true); + if(payloadContent == null) { + ByteBuf ackPayload = Unpooled.buffer(); + ackPayload.ensureWritable(1).writeByte(ERROR_CODE_INVALID_DATA.getCode()); + sendPubAck(clientID, messageID, ackPayload, ERROR_CODE_INVALID_DATA); + return; + } } imHandler(clientID, username, imtopic, payloadContent, (errorCode, ackPayload) -> sendPubAck(clientID, messageID, ackPayload, errorCode), ProtoConstants.RequestSourceType.Request_From_User); diff --git a/broker/src/main/java/win/liyufan/im/IMTopic.java b/broker/src/main/java/win/liyufan/im/IMTopic.java index c4c854a3..4ff22e17 100755 --- a/broker/src/main/java/win/liyufan/im/IMTopic.java +++ b/broker/src/main/java/win/liyufan/im/IMTopic.java @@ -77,6 +77,7 @@ public interface IMTopic { String ChannelSearchTopic = "CHS"; String ChannelListenTopic = "CHL"; String ChannelPullTopic = "CHP"; + String ListenedChannelListTopic = "CHLL"; String GetTokenTopic = "GETTOKEN"; String DestroyUserTopic = "DESTROYUSER";