添加频道新的回调特性

broker/src/main/java/io/moquette/BrokerConstants.java

@@ -163,6 +163,8 @@ public final class BrokerConstants {
 
     public static final String ROBOT_Callback_With_Client_Info = "robot.callback_with_client_info";
     public static final String CHANNEL_Callback_With_Client_Info = "channel.callback_with_client_info";
+    public static final String CHANNEL_New_Callback_Feature = "channel.new_callback_feature";
+
 
     public static final String FRIEND_Disable_Search = "friend.disable_search";
     public static final String FRIEND_Disable_NickName_Search = "friend.disable_nick_name_search";

broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java

@@ -203,6 +203,7 @@ public class MemoryMessagesStore implements IMessagesStore {
     private boolean mForwardMessageWithClientInfo = false;
     private boolean mRobotCallbackWithClientInfo = false;
     private boolean mChannelCallbackWithClientInfo = false;
+    private boolean mChannelNewCallbackFeature = true;
 
     private Set<Integer> mUserHideProperties = new HashSet<>();
 
@@ -602,8 +603,12 @@ public class MemoryMessagesStore implements IMessagesStore {
         try {
             mChannelCallbackWithClientInfo = Boolean.parseBoolean(server.getConfig().getProperty(BrokerConstants.CHANNEL_Callback_With_Client_Info, "false"));
         } catch (Exception e) {
-
         }
+        try {
+            mChannelNewCallbackFeature = Boolean.parseBoolean(server.getConfig().getProperty(BrokerConstants.CHANNEL_New_Callback_Feature, "true"));
+        } catch (Exception e) {
+        }
+
         try {
             mRecallForwardUrl = server.getConfig().getProperty(BrokerConstants.MESSAGE_RecallMsg_Forward_Url);
         } catch (Exception e) {
@@ -732,7 +737,7 @@ public class MemoryMessagesStore implements IMessagesStore {
             if (channelInfo != null) {
                 if (channelInfo.getOwner().equals(fromUser)) {
                     if(requestSourceType == ProtoConstants.RequestSourceType.Request_From_Channel
-                        && channelInfo.getAutomatic() != 0) {
+                        && channelInfo.getAutomatic() != 0 && !isChannelNewCallbackFeature()) {
                         LOG.info("Channel api message not send to the owner when automatic");
                     } else {
                         notifyReceivers.add(fromUser);
@@ -762,7 +767,7 @@ public class MemoryMessagesStore implements IMessagesStore {
                     }
                 } else {
                     notifyReceivers.add(fromUser);
-                    if (channelInfo.getAutomatic() == 0) {
+                    if (channelInfo.getAutomatic() == 0 || isChannelNewCallbackFeature()) {
                         notifyReceivers.add(channelInfo.getOwner());
                     } else {
                         LOG.info("Channel api message not send to the owner when automatic");
@@ -4645,6 +4650,11 @@ public class MemoryMessagesStore implements IMessagesStore {
         return mChannelCallbackWithClientInfo;
     }
 
+    @Override
+    public boolean isChannelNewCallbackFeature() {
+        return mChannelNewCallbackFeature;
+    }
+
     @Override
     public boolean isRobotAutoAcceptFriendRequest() {
         return mFriendRobotAutoAccept;

broker/src/main/java/io/moquette/spi/IMessagesStore.java

@@ -243,6 +243,7 @@ public interface IMessagesStore {
     boolean isForwardMessageWithClientInfo();
     boolean isRobotCallbackWithClientInfo();
     boolean isChannelCallbackWithClientInfo();
+    boolean isChannelNewCallbackFeature();
 
     boolean isRobotAutoAcceptFriendRequest();
 

broker/src/main/java/io/moquette/spi/impl/MessagesPublisher.java

@@ -668,7 +668,15 @@ public class MessagesPublisher {
     public void publish2Receivers(WFCMessage.Message message, Set<String> receivers, String exceptClientId, int pullType) {
         if (message.getConversation().getType() == ProtoConstants.ConversationType.ConversationType_Channel) {
             WFCMessage.ChannelInfo channelInfo = m_messagesStore.getChannelInfo(message.getConversation().getTarget());
-            if (channelInfo != null && !StringUtil.isNullOrEmpty(channelInfo.getCallback()) && channelInfo.getAutomatic() == 1 && !message.getFromUser().equals(channelInfo.getOwner())) {
+
+            boolean forwardMsg;
+            if(m_messagesStore.isChannelNewCallbackFeature()) {
+                forwardMsg = channelInfo != null && !StringUtil.isNullOrEmpty(channelInfo.getCallback()) && !StringUtil.isNullOrEmpty(exceptClientId);
+            } else {
+                forwardMsg = channelInfo != null && !StringUtil.isNullOrEmpty(channelInfo.getCallback()) && channelInfo.getAutomatic() == 1 && !message.getFromUser().equals(channelInfo.getOwner());
+            }
+
+            if (forwardMsg) {
                 OutputClient outputClient = null;
                 if(m_messagesStore.isChannelCallbackWithClientInfo() && !StringUtil.isNullOrEmpty(exceptClientId)) {
                     Session session = m_sessionsStore.getSession(exceptClientId);
@@ -767,7 +775,10 @@ public class MessagesPublisher {
     }
 
     public void notifyChannelListenStatusChanged(WFCMessage.ChannelInfo channelInfo, String user, boolean listen) {
-        if (channelInfo == null || StringUtil.isNullOrEmpty(channelInfo.getCallback()) || channelInfo.getAutomatic() == 0) {
+        if (channelInfo == null || StringUtil.isNullOrEmpty(channelInfo.getCallback())) {
+            return;
+        }
+        if(!m_messagesStore.isChannelNewCallbackFeature() && channelInfo.getAutomatic() == 0) {
             return;
         }