update version to 0.25

broker/config/wildfirechat.conf

@@ -42,6 +42,9 @@ client.proto.secret_key 0x00,0x11,0x22,0x33,0x44,0x55,0x66,0x77,0x78,0x79,0x7A,0
 ##用来生产im token的私钥，只在服务器使用，客户端不用。正式使用时为了安全一定要修改这个值，切记切记
 token.key testim
 
+##首次登录，是否接收之前的历史消息。0 不接收；1接收。
+message.roaming 1
+
 #*********************************************************************
 # hazelcase configuration
 #*********************************************************************

broker/pom.xml

@@ -5,7 +5,7 @@
         <relativePath>../</relativePath>
         <artifactId>wildfirechat-parent</artifactId>
         <groupId>cn.wildfirechat</groupId>
-        <version>0.24</version>
+        <version>0.25</version>
     </parent>
 
     <artifactId>moquette-broker</artifactId>
@@ -115,23 +115,6 @@
              -->
         </dependency>
 
-        <dependency>
-            <groupId>io.dropwizard.metrics</groupId>
-            <artifactId>metrics-core</artifactId>
-            <version>3.2.2</version>
-        </dependency>
-        <dependency>
-            <groupId>io.dropwizard.metrics</groupId>
-            <artifactId>metrics-jvm</artifactId>
-            <version>3.2.2</version>
-        </dependency>
-
-        <dependency>
-            <groupId>com.librato.metrics</groupId>
-            <artifactId>metrics-librato</artifactId>
-            <version>5.1.0</version>
-        </dependency>
-
         <dependency>
             <groupId>commons-codec</groupId>
             <artifactId>commons-codec</artifactId>

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

@@ -60,10 +60,6 @@ public final class BrokerConstants {
     public static final String NETTY_SO_KEEPALIVE_PROPERTY_NAME = "netty.so_keepalive";
     public static final String NETTY_CHANNEL_TIMEOUT_SECONDS_PROPERTY_NAME = "netty.channel_timeout.seconds";
     public static final String NETTY_EPOLL_PROPERTY_NAME = "netty.epoll";
-    public static final String METRICS_ENABLE_PROPERTY_NAME = "use_metrics";
-    public static final String METRICS_LIBRATO_EMAIL_PROPERTY_NAME = "metrics.librato.email";
-    public static final String METRICS_LIBRATO_TOKEN_PROPERTY_NAME = "metrics.librato.token";
-    public static final String METRICS_LIBRATO_SOURCE_PROPERTY_NAME = "metrics.librato.source";
 
     public static final String STORAGE_CLASS_NAME = "storage_class";
 
@@ -100,7 +96,7 @@ public final class BrokerConstants {
     public static final String HTTP_SERVER_API_NO_CHECK_TIME = "http.admin.no_check_time";
 
     public static final String CLIENT_PROTO_SECRET_KEY = "client.proto.secret_key";
-    public static final String TOKEN_SECRET_KEY = "token.key ";
+    public static final String TOKEN_SECRET_KEY = "token.key";
 
     public static final String EMBED_DB_PROPERTY_NAME = "embed.db";
 
@@ -110,6 +106,7 @@ public final class BrokerConstants {
 
     public static final String MESSAGE_Forward_Url = "message.forward.url";
 
+    public static final String MESSAGE_ROAMING = "message.roaming";
 
 
     private BrokerConstants() {

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

@@ -210,23 +210,27 @@ public class DatabaseStore {
         return 0;
     }
 
-    void reloadGroupMemberFromDB(HazelcastInstance hzInstance) {
+    synchronized Collection<WFCMessage.GroupMember>  reloadGroupMemberFromDB(HazelcastInstance hzInstance, String groupId) {
         MultiMap<String, WFCMessage.GroupMember> groupMembers = hzInstance.getMultiMap(MemoryMessagesStore.GROUP_MEMBERS);
+        if (groupMembers.get(groupId).size() > 0) {
+            return groupMembers.get(groupId);
+        }
 
         Connection connection = null;
         PreparedStatement statement = null;
         ResultSet rs = null;
         try {
             connection = DBUtil.getConnection();
-            String sql = "select `_gid`" +
-                ", `_mid`" +
+            String sql = "select `_mid`" +
                 ", `_alias`" +
                 ", `_type`" +
-                ", `_dt` from t_group_member";
+                ", `_dt` from t_group_member where _gid = ?";
             statement = connection.prepareStatement(sql);
 
+            statement.setString(1, groupId);
             int index;
 
+
             rs = statement.executeQuery();
             while (rs.next()) {
                 WFCMessage.GroupMember.Builder builder = WFCMessage.GroupMember.newBuilder();
@@ -234,10 +238,6 @@ public class DatabaseStore {
 
                 String value = rs.getString(index++);
                 value = (value == null ? "" : value);
-                String groupId = value;
-
-                value = rs.getString(index++);
-                value = (value == null ? "" : value);
                 builder.setMemberId(value);
 
                 value = rs.getString(index++);
@@ -254,6 +254,7 @@ public class DatabaseStore {
                 WFCMessage.GroupMember member = builder.build();
                 groupMembers.put(groupId, member);
             }
+            return groupMembers.get(groupId);
         } catch (SQLException e) {
             // TODO Auto-generated catch block
             e.printStackTrace();
@@ -261,6 +262,7 @@ public class DatabaseStore {
         } finally {
             DBUtil.closeDB(connection, statement, rs);
         }
+        return new ArrayList<>();
     }
 
     void reloadFriendsFromDB(HazelcastInstance hzInstance) {
@@ -688,6 +690,43 @@ public class DatabaseStore {
         });
     }
 
+    void removeFavGroup(final String groupId, final List<String> memberIds) {
+        Connection connection = null;
+        PreparedStatement statement = null;
+        try {
+            connection = DBUtil.getConnection();
+
+            StringBuilder sb = new StringBuilder("update t_user_setting set _value = ?, _dt = ? where _uid in (");
+            for (int i = 0; i < memberIds.size(); i++) {
+                sb.append("?");
+                if (i != memberIds.size() - 1) {
+                    sb.append(",");
+                }
+            }
+            sb.append(")");
+
+            sb.append(" and _scope = 6 and _key = ?");
+
+            statement = connection.prepareStatement(sb.toString());
+            int index = 1;
+            statement.setString(index++, "0");
+            statement.setLong(index++, System.currentTimeMillis());
+            for (int i = 0; i < memberIds.size(); i++) {
+                statement.setString(index++, memberIds.get(i));
+            }
+            statement.setString(index++, groupId);
+
+            int count = statement.executeUpdate();
+            LOG.info("Update rows {}", count);
+        } catch (SQLException e) {
+            // TODO Auto-generated catch block
+            e.printStackTrace();
+            Utility.printExecption(LOG, e);
+        } finally {
+            DBUtil.closeDB(connection, statement);
+        }
+    }
+
     void persistUserSetting(final String userId, WFCMessage.UserSettingEntry entry) {
         mScheduler.execute(()->{
             Connection connection = null;

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

@@ -61,7 +61,6 @@ import static win.liyufan.im.MyInfoType.*;
 public class MemoryMessagesStore implements IMessagesStore {
     private static final String MESSAGES_MAP = "messages_map";
     private static final String GROUPS_MAP = "groups_map";
-    private static final String GROUP_ID_COUNTER = "group_id_counter";
     static int dumy = 0;
     static final String GROUP_MEMBERS = "group_members";
 
@@ -83,6 +82,8 @@ public class MemoryMessagesStore implements IMessagesStore {
 
     private static final String CHANNEL_LISTENERS = "channel_listeners";
 
+    private static boolean IS_MESSAGE_ROAMING = true;
+
     static final String USER_ROBOTS = "user_robots";
     static final String USER_THINGS = "user_things";
 
@@ -116,22 +117,19 @@ public class MemoryMessagesStore implements IMessagesStore {
     MemoryMessagesStore(Server server, DatabaseStore databaseStore) {
         m_Server = server;
         this.databaseStore = databaseStore;
+        IS_MESSAGE_ROAMING = "1".equals(m_Server.getConfig().getProperty(MESSAGE_ROAMING));
     }
 
     @Override
     public void initStore() {
-        //TODO reload data from mysql
-        HazelcastInstance hzInstance = m_Server.getHazelcastInstance();
-        MultiMap<String, WFCMessage.GroupMember> groupMembers = hzInstance.getMultiMap(MemoryMessagesStore.GROUP_MEMBERS);
-        if (groupMembers.size() == 0) {
-            databaseStore.reloadGroupMemberFromDB(hzInstance);
-            databaseStore.reloadFriendsFromDB(hzInstance);
-            databaseStore.reloadFriendRequestsFromDB(hzInstance);
-        }
-
         updateSensitiveWord();
     }
 
+    private Collection<WFCMessage.GroupMember> loadGroupMemberFromDB(HazelcastInstance hzInstance, String groupId) {
+        Collection<WFCMessage.GroupMember> members = databaseStore.reloadGroupMemberFromDB(hzInstance, groupId);
+        return members;
+    }
+
     private void updateSensitiveWord() {
         long now = System.currentTimeMillis();
         if (now - lastUpdateSensitiveTime > 2 * 60 * 60 * 1000) {
@@ -195,6 +193,9 @@ public class MemoryMessagesStore implements IMessagesStore {
             } else {
                 MultiMap<String, WFCMessage.GroupMember> groupMembers = hzInstance.getMultiMap(GROUP_MEMBERS);
                 Collection<WFCMessage.GroupMember> members = groupMembers.get(message.getConversation().getTarget());
+                if (members == null || members.size() == 0) {
+                    members = loadGroupMemberFromDB(hzInstance, message.getConversation().getTarget());
+                }
                 for (WFCMessage.GroupMember member : members) {
                     if (member.getType() != GroupMemberType_Removed) {
                         notifyReceivers.add(member.getMemberId());
@@ -316,6 +317,10 @@ public class MemoryMessagesStore implements IMessagesStore {
                 });
             }
 
+            boolean noRoaming = false;
+            if (pullType == ProtoConstants.PullType.Pull_Normal && fromMessageId == 0 && !IS_MESSAGE_ROAMING) {
+                noRoaming = true;
+            }
             while (true) {
                 Map.Entry<Long, Long> entry = maps.higherEntry(current);
                 if (entry == null) {
@@ -335,6 +340,10 @@ public class MemoryMessagesStore implements IMessagesStore {
                             }
                         }
 
+                        if (noRoaming && (System.currentTimeMillis() - bundle.getMessage().getServerTimestamp() > 5 * 60 * 1000)) {
+                            continue;
+                        }
+
                         size += bundle.getMessage().getSerializedSize();
                         if (size >= 1 * 1024 * 1024) { //3M
                             break;
@@ -630,20 +639,21 @@ public class MemoryMessagesStore implements IMessagesStore {
         }
 
 
-        MultiMap<String, WFCMessage.GroupMember> groupMembers = hzInstance.getMultiMap(GROUP_MEMBERS);
-
         if (!isAdmin) {
             boolean isMember = false;
             boolean isManager = false;
             WFCMessage.GroupMember gm = getGroupMember(groupId, operator);
             if (gm != null) {
-                if (gm.getType() == GroupMemberType_Removed) {
+                if (gm.getType() == GroupMemberType_Removed && !(memberList.size() == 1 && operator.equals(memberList.get(0).getMemberId()))) {
                     return ErrorCode.ERROR_CODE_NOT_IN_GROUP;
                 } else if (gm.getType() == GroupMemberType_Silent) {
                     return ErrorCode.ERROR_CODE_NOT_RIGHT;
                 }
                 isManager = (gm.getType() == GroupMemberType_Manager || gm.getType() == GroupMemberType_Owner);
-                isMember = true;
+
+                if (gm.getType() != GroupMemberType_Removed) {
+                    isMember = true;
+                }
             }
 
             if (groupInfo.getType() == ProtoConstants.GroupType.GroupType_Restricted && ((groupInfo.getOwner() == null || !groupInfo.getOwner().equals(operator)) && !isManager)) {
@@ -655,7 +665,7 @@ public class MemoryMessagesStore implements IMessagesStore {
                 }
             }
 
-            if (!isMember) {
+            if (!isMember && !(memberList.size() == 1 && operator.equals(memberList.get(0).getMemberId()))) {
                 return ErrorCode.ERROR_CODE_NOT_IN_GROUP;
             }
         }
@@ -671,7 +681,13 @@ public class MemoryMessagesStore implements IMessagesStore {
         }
         memberList = tmp;
 
-        for (WFCMessage.GroupMember member : groupMembers.get(groupId)) {
+        MultiMap<String, WFCMessage.GroupMember> groupMembers = hzInstance.getMultiMap(GROUP_MEMBERS);
+        Collection<WFCMessage.GroupMember> members = groupMembers.get(groupId);
+        if (members == null || members.size() == 0) {
+            members = loadGroupMemberFromDB(hzInstance, groupId);
+        }
+
+        for (WFCMessage.GroupMember member : members) {
             if (newInviteUsers.contains(member.getMemberId())) {
                 groupMembers.remove(groupId, member);
             }
@@ -713,7 +729,14 @@ public class MemoryMessagesStore implements IMessagesStore {
         int removeCount = 0;
         long updateDt = System.currentTimeMillis();
         ArrayList<WFCMessage.GroupMember> list = new ArrayList<>();
-        List<WFCMessage.GroupMember> allMembers = new ArrayList<>(groupMembers.get(groupId));
+        Collection<WFCMessage.GroupMember> members = groupMembers.get(groupId);
+        if (members == null || members.size() == 0) {
+            members = loadGroupMemberFromDB(hzInstance, groupId);
+        }
+
+        List<WFCMessage.GroupMember> allMembers = new ArrayList<>(members);
+
+        List<String> removedIds = new ArrayList<>();
         for (WFCMessage.GroupMember member : allMembers) {
             if (memberList.contains(member.getMemberId())) {
                 boolean removed = groupMembers.remove(groupId, member);
@@ -722,6 +745,7 @@ public class MemoryMessagesStore implements IMessagesStore {
                     member = member.toBuilder().setType(GroupMemberType_Removed).setUpdateDt(updateDt).build();
                     groupMembers.put(groupId, member);
                     list.add(member);
+                    removedIds.add(member.getMemberId());
                 }
             }
         }
@@ -732,9 +756,23 @@ public class MemoryMessagesStore implements IMessagesStore {
             mIMap.put(groupId, groupInfo.toBuilder().setMemberUpdateDt(updateDt).setUpdateDt(updateDt).setMemberCount(groupInfo.getMemberCount() - removeCount).build());
         }
 
+        removeFavGroup(groupId, removedIds);
+
         return ErrorCode.ERROR_CODE_SUCCESS;
     }
 
+    void removeFavGroup(String groupId, List<String> memberIds) {
+        HazelcastInstance hzInstance = m_Server.getHazelcastInstance();
+        MultiMap<String, WFCMessage.UserSettingEntry> userSettingMap = hzInstance.getMultiMap(USER_SETTING);
+
+        databaseStore.removeFavGroup(groupId, memberIds);
+
+        for (String member : memberIds) {
+            userSettingMap.remove(member);
+        }
+
+    }
+
     @Override
     public ErrorCode quitGroup(String operator, String groupId) {
         HazelcastInstance hzInstance = m_Server.getHazelcastInstance();
@@ -743,8 +781,11 @@ public class MemoryMessagesStore implements IMessagesStore {
         WFCMessage.GroupInfo groupInfo = mIMap.get(groupId);
         if (groupInfo == null) {
             MultiMap<String, WFCMessage.GroupMember> groupMembers = hzInstance.getMultiMap(GROUP_MEMBERS);
-            for (WFCMessage.GroupMember member : groupMembers.get(groupId)
-                ) {
+            Collection<WFCMessage.GroupMember> members = groupMembers.get(groupId);
+            if (members == null || members.size() == 0) {
+                members = loadGroupMemberFromDB(hzInstance, groupId);
+            }
+            for (WFCMessage.GroupMember member :members) {
                 if (member.getMemberId().equals(operator)) {
                     groupMembers.remove(groupId, member);
                     break;
@@ -760,8 +801,11 @@ public class MemoryMessagesStore implements IMessagesStore {
 
         long updateDt = System.currentTimeMillis();
         boolean removed = false;
-        for (WFCMessage.GroupMember member : groupMembers.get(groupId)
-            ) {
+        Collection<WFCMessage.GroupMember> members = groupMembers.get(groupId);
+        if (members == null || members.size() == 0) {
+            members = loadGroupMemberFromDB(hzInstance, groupId);
+        }
+        for (WFCMessage.GroupMember member : members) {
             if (member.getMemberId().equals(operator)) {
                 removed = groupMembers.remove(groupId, member);
                 if (removed) {
@@ -780,6 +824,8 @@ public class MemoryMessagesStore implements IMessagesStore {
         if (removed) {
             mIMap.put(groupId, groupInfo.toBuilder().setMemberUpdateDt(updateDt).setUpdateDt(updateDt).setMemberCount(groupInfo.getMemberCount() - 1).build());
         }
+
+        removeFavGroup(groupId, Arrays.asList(operator));
         return ErrorCode.ERROR_CODE_SUCCESS;
     }
 
@@ -805,11 +851,24 @@ public class MemoryMessagesStore implements IMessagesStore {
 
         MultiMap<String, WFCMessage.GroupMember> groupMembers = hzInstance.getMultiMap(GROUP_MEMBERS);
 
+        Collection<WFCMessage.GroupMember> members = groupMembers.get(groupId);
+        if (members == null || members.size() == 0) {
+            members = loadGroupMemberFromDB(hzInstance, groupId);
+        }
+
         groupMembers.remove(groupId);
         mIMap.remove(groupId);
 
         databaseStore.removeGroupMemberFromDB(groupId);
 
+        ArrayList<String> ids = new ArrayList<>();
+        if (members != null) {
+            for (WFCMessage.GroupMember member : members) {
+                ids.add(member.getMemberId());
+            }
+        }
+        removeFavGroup(groupId, ids);
+
         return ErrorCode.ERROR_CODE_SUCCESS;
     }
 
@@ -898,8 +957,10 @@ public class MemoryMessagesStore implements IMessagesStore {
         long updateDt = System.currentTimeMillis();
         MultiMap<String, WFCMessage.GroupMember> groupMembers = hzInstance.getMultiMap(GROUP_MEMBERS);
         Collection<WFCMessage.GroupMember> members = groupMembers.get(groupId);
-        for (WFCMessage.GroupMember member : members
-            ) {
+        if (members == null || members.size() == 0) {
+            members = loadGroupMemberFromDB(hzInstance, groupId);
+        }
+        for (WFCMessage.GroupMember member : members) {
             if (member.getMemberId().equals(operator)) {
                 groupMembers.remove(groupId, member);
                 member = member.toBuilder().setAlias(alias).setUpdateDt(updateDt).build();
@@ -965,8 +1026,16 @@ public class MemoryMessagesStore implements IMessagesStore {
         }
 
         MultiMap<String, WFCMessage.GroupMember> groupMembers = hzInstance.getMultiMap(GROUP_MEMBERS);
+        Collection<WFCMessage.GroupMember> memberCollection = groupMembers.get(groupId);
+        if (memberCollection == null || memberCollection.size() == 0) {
+            memberCollection = loadGroupMemberFromDB(hzInstance, groupId);
+        }
+        for (WFCMessage.GroupMember member:memberCollection) {
+            if (member.getUpdateDt() > maxDt) {
+                members.add(member);
+            }
+        }
 
-        members.addAll(groupMembers.get(groupId));
         return ErrorCode.ERROR_CODE_SUCCESS;
     }
 
@@ -977,6 +1046,10 @@ public class MemoryMessagesStore implements IMessagesStore {
         MultiMap<String, WFCMessage.GroupMember> groupMembers = hzInstance.getMultiMap(GROUP_MEMBERS);
 
         Collection<WFCMessage.GroupMember> members = groupMembers.get(groupId);
+        if (members == null || members.size() == 0) {
+            members = loadGroupMemberFromDB(hzInstance, groupId);
+        }
+
         if (members != null) {
             for (WFCMessage.GroupMember gm : members) {
                 if (gm.getMemberId().equals(memberId)) {
@@ -1007,8 +1080,8 @@ public class MemoryMessagesStore implements IMessagesStore {
         }
 
         //check the new owner is in member list? is that necessary?
-
-        groupInfo = groupInfo.toBuilder().setOwner(newOwner).build();
+        long updateDt = System.currentTimeMillis();
+        groupInfo = groupInfo.toBuilder().setOwner(newOwner).setUpdateDt(updateDt).build();
 
         mIMap.set(groupId, groupInfo);
 
@@ -1035,10 +1108,13 @@ public class MemoryMessagesStore implements IMessagesStore {
         long updateDt = System.currentTimeMillis();
         MultiMap<String, WFCMessage.GroupMember> groupMembers = hzInstance.getMultiMap(GROUP_MEMBERS);
         Collection<WFCMessage.GroupMember> members = groupMembers.get(groupId);
+        if (members == null || members.size() == 0) {
+            members = loadGroupMemberFromDB(hzInstance, groupId);
+        }
         for (WFCMessage.GroupMember member : members) {
             if (userList.contains(member.getMemberId())) {
                 groupMembers.remove(groupId, member);
-                member = member.toBuilder().setType(ProtoConstants.GroupMemberType.GroupMemberType_Manager).setUpdateDt(updateDt).build();
+                member = member.toBuilder().setType(type == 0 ? ProtoConstants.GroupMemberType.GroupMemberType_Normal : ProtoConstants.GroupMemberType.GroupMemberType_Manager).setUpdateDt(updateDt).build();
                 databaseStore.persistGroupMember(groupId, Arrays.asList(member));
                 groupMembers.put(groupId, member);
 //                userList.remove(member.getMemberId());
@@ -1055,7 +1131,9 @@ public class MemoryMessagesStore implements IMessagesStore {
         MultiMap<String, WFCMessage.GroupMember> groupMembers = hzInstance.getMultiMap(GROUP_MEMBERS);
 
         Collection<WFCMessage.GroupMember> members = groupMembers.get(groupId);
-
+        if (members == null || members.size() == 0) {
+            members = loadGroupMemberFromDB(hzInstance, groupId);
+        }
         for (WFCMessage.GroupMember member : members
             ) {
             if (member.getMemberId().equals(memberId) && member.getType() != GroupMemberType_Removed)
@@ -1080,7 +1158,9 @@ public class MemoryMessagesStore implements IMessagesStore {
         }
 
         Collection<WFCMessage.GroupMember> members = groupMembers.get(groupId);
-
+        if (members == null || members.size() == 0) {
+            members = loadGroupMemberFromDB(hzInstance, groupId);
+        }
         for (WFCMessage.GroupMember member : members
             ) {
             if (member.getMemberId().equals(memberId)) {
@@ -1129,6 +1209,9 @@ public class MemoryMessagesStore implements IMessagesStore {
 
                 MultiMap<String, WFCMessage.GroupMember> groupMembers = hzInstance.getMultiMap(GROUP_MEMBERS);
                 Collection<WFCMessage.GroupMember> members = groupMembers.get(message.getConversation().getTarget());
+                if (members == null || members.size() == 0) {
+                    members = loadGroupMemberFromDB(hzInstance, message.getConversation().getTarget());
+                }
                 for (WFCMessage.GroupMember member : members) {
                     if (member.getMemberId().equals(operatorId)) {
                         if (member.getType() == GroupMemberType_Manager || member.getType() == ProtoConstants.GroupMemberType.GroupMemberType_Owner) {

broker/src/main/java/io/moquette/server/netty/NettyAcceptor.java

@@ -139,19 +139,11 @@ public class NettyAcceptor implements ServerAcceptor {
 
         final NettyMQTTHandler mqttHandler = new NettyMQTTHandler(processor);
 
-        final boolean useFineMetrics = Boolean.parseBoolean(props.getProperty(METRICS_ENABLE_PROPERTY_NAME, "false"));
-        if (useFineMetrics) {
-            DropWizardMetricsHandler metricsHandler = new DropWizardMetricsHandler();
-            metricsHandler.init(props);
-            this.metrics = Optional.of(metricsHandler);
-        } else {
-            this.metrics = Optional.empty();
-        }
+        this.metrics = Optional.empty();
 
         this.errorsCather = Optional.empty();
 
         initializePlainTCPTransport(mqttHandler, props);
-        initializeWebSocketTransport(mqttHandler, props);
         String sslTcpPortProp = props.getProperty(BrokerConstants.SSL_PORT_PROPERTY_NAME);
         String wssPortProp = props.getProperty(BrokerConstants.WSS_PORT_PROPERTY_NAME);
         if (sslTcpPortProp != null || wssPortProp != null) {
@@ -233,42 +225,6 @@ public class NettyAcceptor implements ServerAcceptor {
         });
     }
 
-    private void initializeWebSocketTransport(final NettyMQTTHandler handler, IConfig props) throws IOException {
-        LOG.info("Configuring Websocket MQTT transport");
-        String webSocketPortProp = props.getProperty(WEB_SOCKET_PORT_PROPERTY_NAME, DISABLED_PORT_BIND);
-        if (DISABLED_PORT_BIND.equals(webSocketPortProp)) {
-            // Do nothing no WebSocket configured
-            LOG.info("Property {} has been setted to {}. Websocket MQTT will be disabled",
-                BrokerConstants.WEB_SOCKET_PORT_PROPERTY_NAME, DISABLED_PORT_BIND);
-            return;
-        }
-        int port = Integer.parseInt(webSocketPortProp);
-
-        final MoquetteIdleTimeoutHandler timeoutHandler = new MoquetteIdleTimeoutHandler();
-
-        String host = props.getProperty(BrokerConstants.HOST_PROPERTY_NAME);
-        initFactory(host, port, "Websocket MQTT", new PipelineInitializer() {
-
-            @Override
-            void init(ChannelPipeline pipeline) {
-                pipeline.addLast(new HttpServerCodec());
-                pipeline.addLast("aggregator", new HttpObjectAggregator(65536));
-                pipeline.addLast("webSocketHandler",
-                        new WebSocketServerProtocolHandler("/", MQTT_SUBPROTOCOL_CSV_LIST));
-                pipeline.addLast("ws2bytebufDecoder", new WebSocketFrameToByteBufDecoder());
-                pipeline.addLast("bytebuf2wsEncoder", new ByteBufToWebSocketFrameEncoder());
-                pipeline.addFirst("idleStateHandler", new IdleStateHandler(nettyChannelTimeoutSeconds, 0, 0));
-                pipeline.addAfter("idleStateHandler", "idleEventHandler", timeoutHandler);
-                pipeline.addFirst("bytemetrics", new BytesMetricsHandler(m_bytesMetricsCollector));
-                pipeline.addLast("decoder", new MqttDecoder());
-                pipeline.addLast("encoder", MqttEncoder.INSTANCE);
-                pipeline.addLast("metrics", new MessageMetricsHandler(m_metricsCollector));
-                pipeline.addLast("messageLogger", new MQTTMessageLogger());
-                pipeline.addLast("handler", handler);
-            }
-        });
-    }
-
     private void initializeSSLTCPTransport(final NettyMQTTHandler handler, IConfig props, final SSLContext sslContext)
             throws IOException {
         LOG.info("Configuring SSL MQTT transport");

broker/src/main/java/io/moquette/server/netty/metrics/BytesMetricsHandler.java

@@ -16,6 +16,7 @@
 
 package io.moquette.server.netty.metrics;
 
+import io.moquette.server.netty.NettyUtils;
 import io.netty.buffer.ByteBuf;
 import io.netty.channel.Channel;
 import io.netty.channel.ChannelDuplexHandler;
@@ -23,10 +24,15 @@ import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.ChannelPromise;
 import io.netty.util.Attribute;
 import io.netty.util.AttributeKey;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import static io.moquette.server.netty.NettyUtils.ATTR_USERNAME;
 
 public class BytesMetricsHandler extends ChannelDuplexHandler {
-
+    private static final Logger LOG = LoggerFactory.getLogger(BytesMetricsHandler.class);
     private static final AttributeKey<BytesMetrics> ATTR_KEY_METRICS = AttributeKey.valueOf("BytesMetrics");
+    private static final AttributeKey<String> ATTR_KEY_USERNAME = AttributeKey.valueOf(ATTR_USERNAME);
 
     private BytesMetricsCollector m_collector;
 
@@ -59,6 +65,12 @@ public class BytesMetricsHandler extends ChannelDuplexHandler {
     @Override
     public void close(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception {
         BytesMetrics metrics = ctx.channel().attr(ATTR_KEY_METRICS).get();
+        String userId = ctx.channel().attr(ATTR_KEY_USERNAME).get();
+        if (userId == null) {
+            userId = "";
+        }
+
+        LOG.info("channel<{}> closing after read {} bytes and wrote {} bytes", userId,  metrics.readBytes(), metrics.wroteBytes());
         m_collector.sumReadBytes(metrics.readBytes());
         m_collector.sumWroteBytes(metrics.wroteBytes());
         super.close(ctx, promise);

broker/src/main/java/io/moquette/server/netty/metrics/DropWizardMetricsHandler.java

@@ -1,95 +0,0 @@
-/*
- * Copyright (c) 2012-2017 The original author or authors
- * ------------------------------------------------------
- * All rights reserved. This program and the accompanying materials
- * are made available under the terms of the Eclipse Public License v1.0
- * and Apache License v2.0 which accompanies this distribution.
- *
- * The Eclipse Public License is available at
- * http://www.eclipse.org/legal/epl-v10.html
- *
- * The Apache License v2.0 is available at
- * http://www.opensource.org/licenses/apache2.0.php
- *
- * You may elect to redistribute this code under either of these licenses.
- */
-package io.moquette.server.netty.metrics;
-
-import com.codahale.metrics.Counter;
-import com.codahale.metrics.Meter;
-import com.codahale.metrics.MetricRegistry;
-import com.librato.metrics.reporter.Librato;
-import io.moquette.server.config.IConfig;
-import io.moquette.server.netty.NettyUtils;
-import io.netty.channel.ChannelHandlerContext;
-import io.netty.channel.ChannelInboundHandlerAdapter;
-import io.netty.handler.codec.mqtt.MqttMessage;
-import io.netty.handler.codec.mqtt.MqttMessageType;
-
-import java.util.concurrent.TimeUnit;
-
-import static io.moquette.BrokerConstants.*;
-import static io.netty.channel.ChannelHandler.Sharable;
-
-/**
- * Pipeline handler use to track some MQTT metrics.
- */
-@Sharable
-public final class DropWizardMetricsHandler extends ChannelInboundHandlerAdapter {
-    private MetricRegistry metrics;
-    private Meter publishesMetrics;
-    private Meter subscribeMetrics;
-    private Counter connectedClientsMetrics;
-
-    public void init(IConfig props) {
-        this.metrics = new MetricRegistry();
-        this.publishesMetrics = metrics.meter("publish.requests");
-        this.subscribeMetrics = metrics.meter("subscribe.requests");
-        this.connectedClientsMetrics = metrics.counter("connect.num_clients");
-//        ConsoleReporter reporter = ConsoleReporter.forRegistry(metrics)
-//            .convertRatesTo(TimeUnit.SECONDS)
-//            .convertDurationsTo(TimeUnit.MILLISECONDS)
-//            .build();
-//        reporter.start(1, TimeUnit.MINUTES);
-        final String email = props.getProperty(METRICS_LIBRATO_EMAIL_PROPERTY_NAME);
-        final String token = props.getProperty(METRICS_LIBRATO_TOKEN_PROPERTY_NAME);
-        final String source = props.getProperty(METRICS_LIBRATO_SOURCE_PROPERTY_NAME);
-
-        Librato.reporter(this.metrics, email, token)
-            .setSource(source)
-            .start(10, TimeUnit.SECONDS);
-    }
-
-    @Override
-    public void channelRead(ChannelHandlerContext ctx, Object message) {
-        MqttMessage msg = (MqttMessage) message;
-        MqttMessageType messageType = msg.fixedHeader().messageType();
-        switch (messageType) {
-            case PUBLISH:
-                this.publishesMetrics.mark();
-                break;
-            case SUBSCRIBE:
-                this.subscribeMetrics.mark();
-                break;
-            case CONNECT:
-                this.connectedClientsMetrics.inc();
-                break;
-            case DISCONNECT:
-                this.connectedClientsMetrics.dec();
-                break;
-            default:
-                break;
-        }
-        ctx.fireChannelRead(message);
-    }
-
-    @Override
-    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
-        String clientID = NettyUtils.clientID(ctx.channel());
-        if (clientID != null && !clientID.isEmpty()) {
-            this.connectedClientsMetrics.dec();
-        }
-        ctx.fireChannelInactive();
-    }
-
-}

broker/src/main/java/io/moquette/server/netty/metrics/MessageMetricsHandler.java

@@ -22,10 +22,16 @@ import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.ChannelPromise;
 import io.netty.util.Attribute;
 import io.netty.util.AttributeKey;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import static io.moquette.server.netty.NettyUtils.ATTR_USERNAME;
 
 public class MessageMetricsHandler extends ChannelDuplexHandler {
-
+    private static final Logger LOG = LoggerFactory.getLogger(MessageMetricsHandler.class);
     private static final AttributeKey<MessageMetrics> ATTR_KEY_METRICS = AttributeKey.valueOf("MessageMetrics");
+    private static final AttributeKey<String> ATTR_KEY_USERNAME = AttributeKey.valueOf(ATTR_USERNAME);
+
 
     private MessageMetricsCollector m_collector;
 
@@ -58,6 +64,12 @@ public class MessageMetricsHandler extends ChannelDuplexHandler {
     @Override
     public void close(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception {
         MessageMetrics metrics = ctx.channel().attr(ATTR_KEY_METRICS).get();
+        String userId = ctx.channel().attr(ATTR_KEY_USERNAME).get();
+        if (userId == null) {
+            userId = "";
+        }
+
+        LOG.info("channel<{}> closing after read {} messages and wrote {} messages", userId,  metrics.messagesRead(), metrics.messagesWrote());
         m_collector.sumReadMessages(metrics.messagesRead());
         m_collector.sumWroteMessages(metrics.messagesWrote());
         super.close(ctx, promise);

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

@@ -623,7 +623,8 @@ public class ProtocolProcessor {
 
     public void onRpcMsg(String fromUser, String clientId, byte[] message, int messageId, String from, String request, boolean isAdmin) {
         if(request.equals(RPCCenter.KICKOFF_USER_REQUEST)) {
-            mServer.getImBusinessScheduler().execute(()->handleTargetRemovedFromCurrentNode(new TargetEntry(TargetEntry.Type.TARGET_TYPE_USER, from)));
+            String userId = new String(message);
+            mServer.getImBusinessScheduler().execute(()->handleTargetRemovedFromCurrentNode(new TargetEntry(TargetEntry.Type.TARGET_TYPE_USER, userId)));
             return;
         }
         qos1PublishHandler.onRpcMsg(fromUser, clientId, message, messageId, from, request, isAdmin);

client/pom.xml

@@ -6,7 +6,7 @@
         <relativePath>../</relativePath>
         <artifactId>wildfirechat-parent</artifactId>
         <groupId>cn.wildfirechat</groupId>
-        <version>0.24</version>
+        <version>0.25</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 

common/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <artifactId>wildfirechat-parent</artifactId>
         <groupId>cn.wildfirechat</groupId>
-        <version>0.24</version>
+        <version>0.25</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 

distribution/pom.xml

@@ -6,7 +6,7 @@
         <!--<relativePath>../pom.xml</relativePath>-->
         <artifactId>wildfirechat-parent</artifactId>
         <groupId>cn.wildfirechat</groupId>
-        <version>0.24</version>
+        <version>0.25</version>
     </parent>
 
     <artifactId>distribution</artifactId>

distribution/src/main/resources/wildfirechat.conf

@@ -42,6 +42,9 @@ client.proto.secret_key 0x00,0x11,0x22,0x33,0x44,0x55,0x66,0x77,0x78,0x79,0x7A,0
 ##用来生产im token的私钥，只在服务器使用，客户端不用。正式使用时为了安全一定要修改这个值，切记切记
 token.key testim
 
+##首次登录，是否接收之前的历史消息。0 不接收；1接收。
+message.roaming 1
+
 #*********************************************************************
 # hazelcase configuration
 #*********************************************************************

pom.xml

@@ -12,7 +12,7 @@
     <artifactId>wildfirechat-parent</artifactId>
 
     <packaging>pom</packaging>
-    <version>0.24</version>
+    <version>0.25</version>
     <name>Wildfire Chat</name>
     <description>Wildfire Chat Server</description>
     <inceptionYear>2019</inceptionYear>

sdk/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <artifactId>wildfirechat-parent</artifactId>
         <groupId>cn.wildfirechat</groupId>
-        <version>0.24</version>
+        <version>0.25</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 

sdk/src/main/java/cn/wildfirechat/sdk/Main.java

@@ -103,7 +103,7 @@ public class Main {
             System.exit(-1);
         }
 
-        IMResult<Void> resultVoid =UserAdmin.updateUserBlockStatus(userInfo.getUserId(), 1);
+        IMResult<Void> resultVoid =UserAdmin.updateUserBlockStatus(userInfo.getUserId(), 2);
         if (resultVoid != null && resultVoid.getErrorCode() == ErrorCode.ERROR_CODE_SUCCESS) {
             System.out.println("block user done");
         } else {
@@ -113,7 +113,7 @@ public class Main {
 
         IMResult<OutputUserStatus> resultCheckUserStatus = UserAdmin.checkUserBlockStatus(userInfo.getUserId());
         if (resultCheckUserStatus != null && resultCheckUserStatus.getErrorCode() == ErrorCode.ERROR_CODE_SUCCESS) {
-            if (resultCheckUserStatus.getResult().getStatus() == 1) {
+            if (resultCheckUserStatus.getResult().getStatus() == 2) {
                 System.out.println("check user status success");
             } else {
                 System.out.println("user status not correct");
@@ -128,7 +128,7 @@ public class Main {
         if (resultBlockStatusList != null && resultBlockStatusList.getErrorCode() == ErrorCode.ERROR_CODE_SUCCESS) {
             boolean success = false;
             for (InputOutputUserBlockStatus blockStatus : resultBlockStatusList.getResult().getStatusList()) {
-                if (blockStatus.getUserId().equals(userInfo.getUserId()) && blockStatus.getStatus() == 1) {
+                if (blockStatus.getUserId().equals(userInfo.getUserId()) && blockStatus.getStatus() == 2) {
                     System.out.println("get block list done");
                     success = true;
                     break;
@@ -294,7 +294,7 @@ public class Main {
 
 
         //初始化机器人API
-        RobotHttpUtils.init("http://localhost:18080", "robot1", "123456");
+        RobotHttpUtils.init("http://localhost", "robot1", "123456");
         //***********************************************
         //****  机器人API
         //***********************************************
@@ -313,7 +313,7 @@ public class Main {
             System.exit(-1);
         }
 
-        IMResult<InputOutputUserInfo> resultRobotGetUserInfo = RobotService.getUserInfo("user1");
+        IMResult<InputOutputUserInfo> resultRobotGetUserInfo = RobotService.getUserInfo("userId1");
         if (resultRobotGetUserInfo != null && resultRobotGetUserInfo.getErrorCode() == ErrorCode.ERROR_CODE_SUCCESS) {
             System.out.println("robot get user info success");
         } else {

wildfire_jmeter/pom.xml

@@ -5,7 +5,7 @@
     <relativePath>../</relativePath>
     <artifactId>wildfirechat-parent</artifactId>
     <groupId>cn.wildfirechat</groupId>
-    <version>0.24</version>
+    <version>0.25</version>
   </parent>
 
   <groupId>net.xmeter</groupId>