update version to 0.34

Fix race condition during SensitiveFilter update and potential resource leak of GZIPOutputStream on exception

README.md

@@ -2,17 +2,18 @@
 
 野火IM是一套跨平台、核心功能开源的即时通讯解决方案，主要包含以下内容。
 
-| 仓库                                                         | 说明                                                    | 备注                                           |
-| ------------------------------------------------------------ | ------------------------------------------------------- | ---------------------------------------------- |
-| [android-chat](https://github.com/wildfirechat/android-chat) | 野火IM Android SDK源码和App源码                                      | 可以很方便地进行二次开发，或集成到现有应用当中 |
-| [ios-chat](https://github.com/wildfirechat/ios-chat)         | 野火IM iOS SDK源码和App源码                                          | 可以很方便地进行二次开发，或集成到现有应用当中 |
-| [pc-chat](https://github.com/wildfirechat/pc-chat)           | 基于[Electron](https://electronjs.org/)开发的PC平台应用 |                                                |
-| [proto](https://github.com/wildfirechat/proto)               | 野火IM的协议栈实现                                      |                                                |
-| [server](https://github.com/wildfirechat/server)             | IM server                                               |                                                |
-| [app server](https://github.com/wildfirechat/app_server)     | 应用服务端                                              |                                                |
-| [robot_server](https://github.com/wildfirechat/robot_server) | 机器人服务端                                            |                                                |
-| [push_server](https://github.com/wildfirechat/push_server)   | 推送服务器                                              |                                                |
-| [docs](https://github.com/wildfirechat/docs)                 | 野火IM相关文档，包含设计、概念、开发、使用说明          |                                                |
+| 仓库                                                         | 说明                                                    | 备注 |
+| ------------------------------------------------------------ | ------------------------------------------------------- | ---- |
+| [android-chat](https://github.com/wildfirechat/android-chat) | 野火IM Android SDK源码和App源码                       |可以很方便地进行二次开发，或集成到现有应用当中      |
+| [ios-chat](https://github.com/wildfirechat/ios-chat)         | 野火IM iOS SDK源码和App源码                            |可以很方便地进行二次开发，或集成到现有应用当中      |
+| [pc-chat](https://github.com/wildfirechat/pc-chat)           | 基于[Electron](https://electronjs.org/)开发的PC平台应用 |      |
+| [web-chat](https://github.com/wildfirechat/web-chat)          | Web平台的Demo, [体验地址](http://web.wildfirechat.cn)   |      |
+| [wx-chat](https://github.com/wildfirechat/wx-chat)           | 微信小程序平台的Demo   |      |
+| [server](https://github.com/wildfirechat/server)             | IM server                                               |      |
+| [app server](https://github.com/wildfirechat/app_server)     | 应用服务端                                          |      |
+| [robot_server](https://github.com/wildfirechat/robot_server) | 机器人服务端                                        |      |
+| [push_server](https://github.com/wildfirechat/push_server)   | 推送服务器                                              |      |
+| [docs](https://github.com/wildfirechat/docs)                 | 野火IM相关文档，包含设计、概念、开发、使用说明          |      | |                                            |
 
 # server
 本工程为野火IM 社区版IM服务软件。野火IM作为一个通用的即时通讯解决方案，可以集成到各种应用中。请阅读[docs](http://docs.wildfirechat.cn)或下载服务器[发布版本](https://github.com/wildfirechat/server/releases)
@@ -28,13 +29,10 @@
 2. imndx  邮箱: imndx@wildfirechat.cn  微信：wfchat
 
 ### 问题交流
-1. 微信交流群（请先加我们好友，我们再邀请您进群）
 
-  ![微信二维码](http://static.wildfirechat.cn/wx_qr_wildfirechat.jpeg)   ![微信二维码](http://static.wildfirechat.cn/wx_qr_wfc.jpeg)
-  
-2. 如果大家发现bug，请在GitHub提issue
-3. 其他问题，请到[野火IM论坛](http://bbs.wildfirechat.cn/)进行交流学习
-4. 微信公众号
+1. 如果大家发现bug，请在GitHub提issue
+2. 其他问题，请到[野火IM论坛](http://bbs.wildfirechat.cn/)进行交流学习
+3. 微信公众号
 
 <img src="http://static.wildfirechat.cn/wx_wfc_qrcode.jpg" width = 50% height = 50% />
 
@@ -45,6 +43,11 @@
 
 ![野火IM](http://static.wildfirechat.cn/download_qrcode.png)
 
+#### 加入野火官方体验交流群
+使用野火IM移动客户端扫码下面二维码（不是使用微信扫码），和野火IM的爱好者一起畅谈野火IM的问题和经验：
+
+![野火IM官方交流群](http://static.wildfirechat.cn/wildfirechat_official_group.jpeg)
+
 #### 应用截图
 ![ios-demo](http://static.wildfirechat.cn/ios-demo.gif)
 

broker/config/hazelcast.xml

@@ -1,3 +1,7 @@
+
+<!--    # 本配置文件为debug使用，修改这里不会在Release包中生效。Release中包含的配置文件在 ${Porject_Path}/distribution/src/main/resources目录下-->
+
+
 <hazelcast
     xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
     xsi:schemaLocation="http://www.hazelcast.com/schema/config

broker/config/wildfirechat.conf

@@ -1,8 +1,17 @@
+#
+#
+# 本配置文件为debug使用，修改这里不会在Release包中生效。Release中包含的配置文件在 ${Porject_Path}/distribution/src/main/resources目录下
+#
+#
+
+
+
 #*********************************************************************
 # Server configuration
 #*********************************************************************
 #服务器的接入IP。给客户端提供是${server.ip}和${http_port}。
-#客户端会从${http_port}端口获取到长链接端口。这个地址一定要改成该服务的公网IP。
+#客户端会从${http_port}端口获取到长链接端口。这个地址一定要改成客户端可以访问到的IP地址
+# （如果您部署云服务器上或者具有独立公网出口的服务器上，请改为对应的公网IP；如果您部署在在内网环境下，在内网使用，这个地方改成内网地址）
 server.ip 0.0.0.0
 
 ##原生客户端长链接端口
@@ -49,6 +58,9 @@ message.roaming 0
 ##是否开启拉取远程历史消息
 message.remote_history_message 0
 
+##服务器为每个用户缓存的消息数量。这个值改得太大，拉取消息时间变长，另外会占用大量内存。
+message.max_queue 1024
+
 #*********************************************************************
 # hazelcase configuration
 #*********************************************************************
@@ -93,6 +105,9 @@ qiniu.bucket_favorite_domain http://cdn2.wildfirechat.cn
 # 本地媒体服务器配置。
 local.media.storage.root media
 
+# 是否支持多端登陆
+server.multi_endpoint false
+
 #*********************************************************************
 # Push server configuration
 #*********************************************************************

broker/migrate/h2/V16__add_friend_blocked.sql

@@ -0,0 +1,2 @@
+alter table t_friend add column `_blacked` tinyint DEFAULT 0;
+update t_friend set `_blacked` = 1 where `_state` = 2;

broker/migrate/h2/V17__add_user_session_token_index.sql

@@ -0,0 +1 @@
+alter table `t_user_session` ADD INDEX `session_token_index` ( `_token` );

broker/migrate/h2/V18__add_friend_request_index.sql

@@ -0,0 +1 @@
+alter table t_friend_request add index `friend_request_uid_index` (`_friend_uid`);

broker/migrate/h2/V19__add_user_session_uid_index.sql

@@ -0,0 +1 @@
+alter table `t_user_session` ADD INDEX `session_uid_index` ( `_uid` );

broker/migrate/h2/V20__user_session_add_deleted.sql

@@ -0,0 +1 @@
+alter table `t_user_session` add column `_deleted` tinyint DEFAULT 0;

broker/migrate/h2/V21__user_add_deleted.sql

@@ -0,0 +1 @@
+alter table `t_user` add column `_deleted` tinyint DEFAULT 0;

broker/migrate/mysql/V21__add_friend_blocked.sql

@@ -0,0 +1,2 @@
+alter table t_friend add column `_blacked` tinyint DEFAULT 0 COMMENT "0, normal; 1, blacked";
+update t_friend set `_blacked` = 1 where `_state` = 2;

broker/migrate/mysql/V22__add_user_session_token_index.sql

@@ -0,0 +1 @@
+alter table `t_user_session` ADD INDEX `session_token_index` ( `_token` );

broker/migrate/mysql/V23__add_friend_request_index.sql

@@ -0,0 +1 @@
+alter table t_friend_request add index `friend_request_uid_index` (`_friend_uid`);

broker/migrate/mysql/V24__add_user_session_uid_index.sql

@@ -0,0 +1 @@
+alter table `t_user_session` ADD INDEX `session_uid_index` ( `_uid` );

broker/migrate/mysql/V25__user_session_add_deleted.sql

@@ -0,0 +1 @@
+alter table `t_user_session` add column `_deleted` tinyint DEFAULT 0;

broker/migrate/mysql/V26__user_add_deleted.sql

@@ -0,0 +1 @@
+alter table `t_user` add column `_deleted` tinyint DEFAULT 0;

broker/pom.xml

@@ -5,7 +5,7 @@
         <relativePath>../</relativePath>
         <artifactId>wildfirechat-parent</artifactId>
         <groupId>cn.wildfirechat</groupId>
-        <version>0.28</version>
+        <version>0.34</version>
     </parent>
 
     <artifactId>moquette-broker</artifactId>
@@ -161,7 +161,7 @@
         <dependency>
             <groupId>org.eclipse.paho</groupId>
             <artifactId>org.eclipse.paho.client.mqttv3</artifactId>
-            <version>1.0.2</version>
+            <version>1.2.1</version>
         </dependency>
 
         <dependency>

broker/src/main/java/cn/wildfirechat/push/PushServer.java

@@ -57,7 +57,7 @@ public class PushServer {
     }
 
     public void pushMessage(PushMessage pushMessage, String deviceId, String pushContent) {
-        LOG.info("try to delivery push diviceId = {}, pushContent", deviceId, pushContent);
+        LOG.info("try to delivery push diviceId = {}, pushContent = {}", deviceId, pushContent);
         executorService.execute(() ->{
                 try {
                     pushMessageInternel(pushMessage, deviceId, pushContent);

broker/src/main/java/com/xiaoleilu/loServer/action/RouteAction.java

@@ -82,7 +82,7 @@ public class RouteAction extends Action {
 
             MemorySessionStore.Session session = sessionsStore.sessionForClientAndUser(uid, cid);
             if (session == null) {
-                ErrorCode errorCode = sessionsStore.createNewSession(uid, cid, true, true);
+                ErrorCode errorCode = sessionsStore.loadActiveSession(uid, cid);
                 if (errorCode != ErrorCode.ERROR_CODE_SUCCESS) {
                     sendResponse(response, errorCode, null, base64Response);
                     return true;

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

@@ -0,0 +1,97 @@
+/*
+ * 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.InputDestroyUser;
+import cn.wildfirechat.pojos.InputOutputUserInfo;
+import cn.wildfirechat.pojos.OutputCreateUser;
+import cn.wildfirechat.pojos.OutputGetIMTokenData;
+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.handler.codec.http.FullHttpRequest;
+import io.netty.handler.codec.http.HttpResponseStatus;
+import io.netty.util.internal.StringUtil;
+import win.liyufan.im.IMTopic;
+import win.liyufan.im.UUIDGenerator;
+
+import java.util.Base64;
+import java.util.concurrent.Executor;
+
+@Route(APIPath.Destroy_User)
+@HttpMethod("POST")
+public class DestroyUserAction extends AdminAction {
+
+    @Override
+    public boolean isTransactionAction() {
+        return true;
+    }
+
+    @Override
+    public boolean action(Request request, Response response) {
+        if (request.getNettyRequest() instanceof FullHttpRequest) {
+            InputDestroyUser inputDestroyUser = getRequestBody(request.getNettyRequest(), InputDestroyUser.class);
+            if (inputDestroyUser != null
+                && !StringUtil.isNullOrEmpty(inputDestroyUser.getUserId())) {
+
+                WFCMessage.IDBuf idBuf = WFCMessage.IDBuf.newBuilder().setId(inputDestroyUser.getUserId()).build();
+                RPCCenter.getInstance().sendRequest(inputDestroyUser.getUserId(), null, IMTopic.DestroyUserTopic, idBuf.toByteArray(), inputDestroyUser.getUserId(), TargetEntry.Type.TARGET_TYPE_USER, new RPCCenter.Callback() {
+                    @Override
+                    public void onSuccess(byte[] result) {
+                        ErrorCode errorCode1 = ErrorCode.fromCode(result[0]);
+                        if (errorCode1 == ErrorCode.ERROR_CODE_SUCCESS) {
+                            //ba errorcode qudiao
+                            byte[] data = new byte[result.length -1];
+                            for (int i = 0; i < data.length; i++) {
+                                data[i] = result[i+1];
+                            }
+                            String token = Base64.getEncoder().encodeToString(data);
+
+                            sendResponse(response, null, null);
+                        } else {
+                            sendResponse(response, errorCode1, 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/com/xiaoleilu/loServer/action/admin/FriendRelationGetAction.java

@@ -36,7 +36,7 @@ public class FriendRelationGetAction extends AdminAction {
     public boolean action(Request request, Response response) {
         if (request.getNettyRequest() instanceof FullHttpRequest) {
             InputGetFriendList inputGetFriendList = getRequestBody(request.getNettyRequest(), InputGetFriendList.class);
-            List<FriendData> dataList = messagesStore.getFriendList(inputGetFriendList.getUserId(), 0);
+            List<FriendData> dataList = messagesStore.getFriendList(inputGetFriendList.getUserId(), null, 0);
             List<String> list = new ArrayList<>();
             for (FriendData data : dataList) {
                 if (data.getState() == inputGetFriendList.getStatus()) {

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

@@ -41,7 +41,7 @@ public class GetIMTokenAction extends AdminAction {
             String userId = input.getUserId();
 
 
-            WFCMessage.GetTokenRequest getTokenRequest = WFCMessage.GetTokenRequest.newBuilder().setUserId(userId).setClientId(input.getClientId()).build();
+            WFCMessage.GetTokenRequest getTokenRequest = WFCMessage.GetTokenRequest.newBuilder().setUserId(userId).setClientId(input.getClientId()).setPlatform(input.getPlatform() == null ? 0 : input.getPlatform()).build();
             RPCCenter.getInstance().sendRequest(userId, input.getClientId(), IMTopic.GetTokenTopic, getTokenRequest.toByteArray(), userId, TargetEntry.Type.TARGET_TYPE_USER, new RPCCenter.Callback() {
                 @Override
                 public void onSuccess(byte[] result) {

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

@@ -0,0 +1,63 @@
+/*
+ * 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.InputGetGroup;
+import cn.wildfirechat.pojos.InputUserId;
+import cn.wildfirechat.pojos.OutputGroupIds;
+import cn.wildfirechat.pojos.PojoGroupInfo;
+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.netty.handler.codec.http.FullHttpRequest;
+import io.netty.handler.codec.http.HttpResponseStatus;
+import io.netty.util.internal.StringUtil;
+
+import java.util.ArrayList;
+import java.util.Set;
+
+@Route(APIPath.Get_User_Groups)
+@HttpMethod("POST")
+public class GetUserGroupsAction extends AdminAction {
+
+    @Override
+    public boolean isTransactionAction() {
+        return true;
+    }
+
+    @Override
+    public boolean action(Request request, Response response) {
+        if (request.getNettyRequest() instanceof FullHttpRequest) {
+            InputUserId inputUserId = getRequestBody(request.getNettyRequest(), InputUserId.class);
+            if (inputUserId != null
+                && (!StringUtil.isNullOrEmpty(inputUserId.getUserId()))) {
+
+                Set<String> groupIds = messagesStore.getUserGroupIds(inputUserId.getUserId());
+                response.setStatus(HttpResponseStatus.OK);
+                OutputGroupIds outputGroupIds = new OutputGroupIds();
+                outputGroupIds.setGroupIds(new ArrayList<>(groupIds));
+                RestResult result = RestResult.ok(outputGroupIds);
+
+                response.setContent(new Gson().toJson(result));
+            } else {
+                response.setStatus(HttpResponseStatus.OK);
+                RestResult result = RestResult.resultOf(ErrorCode.INVALID_PARAMETER);
+                response.setContent(new Gson().toJson(result));
+            }
+
+        }
+        return true;
+    }
+}

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

@@ -0,0 +1,85 @@
+/*
+ * 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.InputAddGroupMember;
+import cn.wildfirechat.pojos.InputSetGroupManager;
+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 win.liyufan.im.IMTopic;
+
+import java.util.concurrent.Executor;
+
+@Route(APIPath.Group_Set_Manager)
+@HttpMethod("POST")
+public class SetGroupManagerAction extends AdminAction {
+
+    @Override
+    public boolean isTransactionAction() {
+        return true;
+    }
+
+    @Override
+    public boolean action(Request request, Response response) {
+        if (request.getNettyRequest() instanceof FullHttpRequest) {
+            InputSetGroupManager inputAddGroupMember = getRequestBody(request.getNettyRequest(), InputSetGroupManager.class);
+            if (inputAddGroupMember.isValide()) {
+                RPCCenter.getInstance().sendRequest(inputAddGroupMember.getOperator(), null, IMTopic.SetGroupManagerTopic, inputAddGroupMember.toProtoGroupRequest().toByteArray(), inputAddGroupMember.getOperator(), 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) {
+                            sendResponse(response, null, null);
+                        } 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/com/xiaoleilu/loServer/model/FriendData.java

@@ -15,14 +15,16 @@ public class FriendData implements Serializable {
     private String friendUid;
     private String alias;
     private int state;
+    private int blacked;
     private long timestamp;
 
 
-    public FriendData(String userId, String friendUid, String alias, int state, long timestamp) {
+    public FriendData(String userId, String friendUid, String alias, int state, int blacked, long timestamp) {
         this.userId = userId;
         this.friendUid = friendUid;
         this.alias = alias;
         this.state = state;
+        this.blacked = blacked;
         this.timestamp = timestamp;
     }
 
@@ -61,6 +63,14 @@ public class FriendData implements Serializable {
         this.state = state;
     }
 
+    public int getBlacked() {
+        return blacked;
+    }
+
+    public void setBlacked(int blacked) {
+        this.blacked = blacked;
+    }
+
     public long getTimestamp() {
         return timestamp;
     }

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

@@ -106,12 +106,17 @@ public final class BrokerConstants {
 
     public static final String MESSAGE_Forward_Url = "message.forward.url";
 
+    public static final String SERVER_MULTI_ENDPOINT = "server.multi_endpoint";
+
     public static final String MONGODB_Client_URI = "mongodb.client_uri";
     public static final String MONGODB_Database = "mongodb.database";
     public static final String MONGODB_Data_Expire_Days = "mongodb.data_expire_days";
 
     public static final String MESSAGE_ROAMING = "message.roaming";
-    public static final String MESSAGE_Remote_History_Message = "message.reomte_history_message";
+    public static final String MESSAGE_Remote_History_Message = "message.remote_history_message";
+
+    public static final String MESSAGE_Max_Queue = "message.max_queue";
+
 
 
     private BrokerConstants() {

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

@@ -0,0 +1,38 @@
+/*
+ * 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.WFCMessage;
+import io.moquette.persistence.MemorySessionStore;
+import io.moquette.spi.impl.Qos1PublishHandler;
+import io.moquette.spi.impl.security.TokenAuthenticator;
+import io.netty.buffer.ByteBuf;
+import win.liyufan.im.IMTopic;
+
+@Handler(IMTopic.DestroyUserTopic)
+public class DestroyUserHandler extends IMHandler<WFCMessage.IDBuf> {
+    @Override
+    public ErrorCode action(ByteBuf ackPayload, String clientID, String fromUser, boolean isAdmin, WFCMessage.IDBuf request, Qos1PublishHandler.IMCallback callback) {
+        if (isAdmin) {
+            mServer.getImBusinessScheduler().execute(()-> {
+                m_sessionsStore.clearUserSession(fromUser);
+                m_messagesStore.clearUserMessages(fromUser);
+                m_messagesStore.clearUserSettings(fromUser);
+                m_messagesStore.clearUserFriend(fromUser);
+                m_messagesStore.clearUserGroups(fromUser);
+                m_messagesStore.clearUserChannels(fromUser);
+                m_messagesStore.destoryUser(fromUser);
+            });
+            return ErrorCode.ERROR_CODE_SUCCESS;
+        } else {
+            return ErrorCode.ERROR_CODE_NOT_RIGHT;
+        }
+    }
+}

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

@@ -22,11 +22,11 @@ import java.util.List;
 public class FriendPullHandler extends IMHandler<WFCMessage.Version> {
     @Override
     public ErrorCode action(ByteBuf ackPayload, String clientID, String fromUser, boolean isAdmin, WFCMessage.Version request, Qos1PublishHandler.IMCallback callback) {
-        List<FriendData> friendDatas = m_messagesStore.getFriendList(fromUser, request.getVersion());
+        List<FriendData> friendDatas = m_messagesStore.getFriendList(fromUser, clientID, request.getVersion());
         WFCMessage.GetFriendsResult.Builder builder = WFCMessage.GetFriendsResult.newBuilder();
         for (FriendData data : friendDatas
             ) {
-            WFCMessage.Friend.Builder builder1 = WFCMessage.Friend.newBuilder().setState(data.getState()).setUid(data.getFriendUid()).setUpdateDt(data.getTimestamp());
+            WFCMessage.Friend.Builder builder1 = WFCMessage.Friend.newBuilder().setState(data.getState()).setBlacked(data.getBlacked()).setUid(data.getFriendUid()).setUpdateDt(data.getTimestamp());
             if (!StringUtil.isNullOrEmpty(data.getAlias())) {
                 builder1.setAlias(data.getAlias());
             }

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

@@ -20,7 +20,7 @@ import win.liyufan.im.IMTopic;
 public class GetTokenHandler extends IMHandler<WFCMessage.GetTokenRequest> {
     @Override
     public ErrorCode action(ByteBuf ackPayload, String clientID, String fromUser, boolean isAdmin, WFCMessage.GetTokenRequest request, Qos1PublishHandler.IMCallback callback) {
-        MemorySessionStore.Session session = m_sessionsStore.createUserSession(fromUser, clientID);
+        MemorySessionStore.Session session = m_sessionsStore.createUserSession(fromUser, clientID, request.getPlatform());
         TokenAuthenticator authenticator = new TokenAuthenticator();
         String strToken = authenticator.generateToken(fromUser);
         String result = strToken + "|" + session.getSecret() + "|" + session.getDbSecret();

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

@@ -83,11 +83,10 @@ abstract public class IMHandler<T> {
                 }
             }
 
-
             for (Method method : getClass().getDeclaredMethods()
                  ) {
 
-                if (method.getName() == actionName && method.getParameterCount() == 6) {
+                if (method.getName() == actionName && method.getParameterCount() == 6 && !method.getParameterTypes()[4].equals(Object.class)) {
                     dataCls = method.getParameterTypes()[4];
                     break;
                 }
@@ -240,7 +239,7 @@ abstract public class IMHandler<T> {
 
         WFCMessage.Message.Builder messageBuilder = message.toBuilder();
         int pullType = m_messagesStore.getNotifyReceivers(username, messageBuilder, notifyReceivers);
-        this.publisher.publish2Receivers(messageBuilder.build(), notifyReceivers, clientID, pullType);
+        mServer.getImBusinessScheduler().execute(() -> this.publisher.publish2Receivers(messageBuilder.build(), notifyReceivers, clientID, pullType));
         return notifyReceivers.size();
     }
 
@@ -250,7 +249,7 @@ abstract public class IMHandler<T> {
         message = m_messagesStore.storeMessage(username, clientID, message);
         WFCMessage.Message.Builder messageBuilder = message.toBuilder();
         int pullType = m_messagesStore.getNotifyReceivers(username, messageBuilder, notifyReceivers);
-        this.publisher.publish2Receivers(messageBuilder.build(), notifyReceivers, clientID, pullType);
+        mServer.getImBusinessScheduler().execute(() -> this.publisher.publish2Receivers(messageBuilder.build(), notifyReceivers, clientID, pullType));
         return notifyReceivers.size();
     }
 

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

@@ -32,7 +32,7 @@ public class RouteHandler extends IMHandler<WFCMessage.RouteRequest> {
         ClientSession clientSession = m_sessionsStore.sessionForClient(clientID);
         boolean isSessionAlreadyStored = clientSession != null;
         if (!isSessionAlreadyStored) {
-            m_sessionsStore.createNewSession(fromUser, clientID, true, true);
+            m_sessionsStore.loadActiveSession(fromUser, clientID);
         } else {
             m_sessionsStore.updateExistSession(fromUser, clientID, request, true);
         }

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

@@ -26,12 +26,12 @@ public class UploadDeviceTokenHandler extends IMHandler<WFCMessage.UploadDeviceT
             session.setAppName(request.getAppName());
             if (request.getPlatform() == ProtoConstants.Platform.Platform_iOS && request.getPushType() == 2) {
                 session.setVoipDeviceToken(request.getDeviceToken());
-//                session.setPushType(0);//如果有voip token，肯定是生产环境的推送
                 m_sessionsStore.updateSessionToken(session, true);
             } else {
                 session.setDeviceToken(request.getDeviceToken());
                 session.setPushType(request.getPushType());
                 m_sessionsStore.updateSessionToken(session, false);
+                m_sessionsStore.cleanDuplatedToken(session.getClientID(), session.getPushType(), session.getDeviceToken(), false);
             }
 
             return ErrorCode.ERROR_CODE_SUCCESS;

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

@@ -16,10 +16,10 @@ import com.hazelcast.core.HazelcastInstance;
 import com.hazelcast.core.MultiMap;
 import com.hazelcast.util.StringUtil;
 import com.xiaoleilu.loServer.model.FriendData;
+import io.moquette.server.Server;
 import io.moquette.spi.ClientSession;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-import sun.misc.BASE64Encoder;
 import win.liyufan.im.DBUtil;
 import win.liyufan.im.MessageBundle;
 import win.liyufan.im.MessageShardingUtil;
@@ -31,6 +31,7 @@ import java.sql.*;
 import java.util.*;
 import java.util.function.Function;
 
+
 import static cn.wildfirechat.proto.ProtoConstants.PersistFlag.Transparent;
 import static io.moquette.persistence.MemoryMessagesStore.USER_STATUS;
 import static io.moquette.server.Constants.MAX_MESSAGE_QUEUE;
@@ -57,8 +58,6 @@ public class DatabaseStore {
 
             statement.setString(1, userId);
 
-            statement.executeQuery();
-
             rs = statement.executeQuery();
             while (rs.next()) {
                 int index = 1;
@@ -173,13 +172,17 @@ public class DatabaseStore {
             if (buzzy) {
                 sql += " where (`_display_name` like ? or `_name` = ? or `_mobile` = ?) ";
             } else {
-                sql += " where (`_display_name` = ? or `_name` = ? or `_mobile` = ?) ";
+                sql += " where (`_name` = ? or `_mobile` = ?) ";
             }
 
             sql += " and _type <> 2"; //can search normal user(0) and robot(1), can not search things
 
-            sql += " limit 20";
-
+            if (buzzy) {
+                sql += " limit 20";
+            } else {
+                sql += " limit 1";
+            }
+            
             if (page > 0) {
                 sql += "offset = '" + page * 20 + "'";
             }
@@ -189,9 +192,8 @@ public class DatabaseStore {
             int index = 1;
             if (buzzy) {
                 statement.setString(index++, "%" + keyword + "%");
-            } else {
-                statement.setString(index++, keyword);
             }
+
             statement.setString(index++, keyword);
             statement.setString(index++, keyword);
 
@@ -350,7 +352,7 @@ public class DatabaseStore {
 
         try {
             connection = DBUtil.getConnection();
-            String sql = "select `_uid`, `_friend_uid`, `_alias`, `_state`, `_dt` from t_friend";
+            String sql = "select `_uid`, `_friend_uid`, `_alias`, `_state`, `_blacked`, `_dt` from t_friend";
             statement = connection.prepareStatement(sql);
 
             int index;
@@ -375,6 +377,9 @@ public class DatabaseStore {
                 int intvalue = rs.getInt(index++);
                 builder.setState(intvalue);
 
+                intvalue = rs.getInt(index++);
+                builder.setBlacked(intvalue);
+
                 long longvalue = rs.getLong(index++);
                 builder.setTimestamp(longvalue);
 
@@ -635,11 +640,12 @@ public class DatabaseStore {
         String sql = "select  `_from`, `_type`, `_target`, `_line`, `_data`, `_dt` from " + MessageShardingUtil.getMessageTable(messageId) +" where _mid = ? limit 1";
         Connection connection = null;
         PreparedStatement statement = null;
+        ResultSet resultSet = null;
         try {
             connection = DBUtil.getConnection();
             statement = connection.prepareStatement(sql);
             statement.setLong(1, messageId);
-            ResultSet resultSet = statement.executeQuery();
+            resultSet = statement.executeQuery();
             if (resultSet.next()) {
                 WFCMessage.Message.Builder builder = WFCMessage.Message.newBuilder();
                 builder.setMessageId(messageId);
@@ -658,12 +664,11 @@ public class DatabaseStore {
                 WFCMessage.Message message = builder.build();
                 return new MessageBundle(messageId, message.getFromUser(), null, message);
             }
-            resultSet.close();
         } catch (SQLException | IOException e) {
             e.printStackTrace();
             Utility.printExecption(LOG, e);
         } finally {
-            DBUtil.closeDB(connection, statement);
+            DBUtil.closeDB(connection, statement, resultSet);
         }
         return null;
     }
@@ -783,6 +788,40 @@ public class DatabaseStore {
         });
     }
 
+    void clearUserMessage(final String userId) {
+        mScheduler.execute(()->{
+            String tableName = getUserMessageTable(userId);
+            Connection connection = null;
+            PreparedStatement statement = null;
+            try {
+                connection = DBUtil.getConnection();
+
+                String sql = "delete from " + tableName + " where _uid = ?";
+                statement = connection.prepareStatement(sql);
+                statement.setString(1, userId);
+                int count = statement.executeUpdate();
+                LOG.info("Update rows {}", count);
+
+                try {
+                    if (statement!=null) {
+                        statement.close();
+                    }
+                    statement = null;
+                } catch (SQLException e) {
+                    e.printStackTrace();
+                    Utility.printExecption(LOG, e);
+                }
+            } catch (SQLException e) {
+                // TODO Auto-generated catch block
+                e.printStackTrace();
+                Utility.printExecption(LOG, e);
+            } finally {
+                DBUtil.closeDB(connection, statement);
+            }
+        });
+
+    }
+
     void removeFavGroup(final String groupId, final List<String> memberIds) {
         Connection connection = null;
         PreparedStatement statement = null;
@@ -858,6 +897,31 @@ public class DatabaseStore {
         });
     }
 
+    void clearUserSetting(final String userId) {
+        mScheduler.execute(()->{
+            Connection connection = null;
+            PreparedStatement statement = null;
+            try {
+                connection = DBUtil.getConnection();
+
+                String sql = "delete from t_user_setting where `_uid` = ?";
+
+                statement = connection.prepareStatement(sql);
+                int index = 1;
+                statement.setString(index++, userId);
+
+                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);
+            }
+        });
+    }
+
     List<WFCMessage.UserSettingEntry> getPersistUserSetting(final String userId) {
         Connection connection = null;
         PreparedStatement statement = null;
@@ -1021,16 +1085,82 @@ public class DatabaseStore {
         });
     }
 
-    MemorySessionStore.Session getSession(String uid, String clientId, ClientSession clientSession) {
-        String sql = "select  `_package_name`,`_token`,`_voip_token`,`_secret`,`_db_secret`,`_platform`,`_push_type`,`_device_name`,`_device_version`,`_phone_name`,`_language`,`_carrier_name`, `_dt` from t_user_session where `_uid` = ? and `_cid` = ? limit 1";
+    List<MemorySessionStore.Session> getUserActivedSessions(String uid) {
+        String sql = "select  `_cid`, `_package_name`,`_token`,`_voip_token`,`_secret`,`_db_secret`,`_platform`,`_push_type`,`_device_name`,`_device_version`,`_phone_name`,`_language`,`_carrier_name`, `_dt` from t_user_session where `_uid` = ? and `_deleted` = 0";
         Connection connection = null;
         PreparedStatement statement = null;
+        List<MemorySessionStore.Session> result = new ArrayList<>();
+        ResultSet resultSet = null;
+        try {
+            connection = DBUtil.getConnection();
+            statement = connection.prepareStatement(sql);
+            statement.setString(1, uid);
+
+            resultSet = statement.executeQuery();
+            while (resultSet.next()) {
+                int index = 1;
+                String cid = resultSet.getString(index++);
+
+                ClientSession clientSession = new ClientSession(cid, Server.getServer().getStore().sessionsStore());
+                MemorySessionStore.Session session = new MemorySessionStore.Session(uid, cid, clientSession);
+
+                session.setAppName(resultSet.getString(index++));
+                session.setDeviceToken(resultSet.getString(index++));
+                session.setVoipDeviceToken(resultSet.getString(index++));
+                session.setSecret(resultSet.getString(index++));
+                session.setDbSecret(resultSet.getString(index++));
+                session.setPlatform(resultSet.getInt(index++));
+                session.setPushType(resultSet.getInt(index++));
+
+                session.setDeviceName(resultSet.getString(index++));
+                session.setDeviceVersion(resultSet.getString(index++));
+                session.setPhoneName(resultSet.getString(index++));
+                session.setLanguage(resultSet.getString(index++));
+                session.setCarrierName(resultSet.getString(index++));
+                session.setUpdateDt(resultSet.getLong(index++));
+                result.add(session);
+            }
+        } catch (SQLException e) {
+            e.printStackTrace();
+            Utility.printExecption(LOG, e);
+        } finally {
+            DBUtil.closeDB(connection, statement, resultSet);
+        }
+        return result;
+    }
+
+    void clearUserSessions(String uid) {
+        Connection connection = null;
+        PreparedStatement statement = null;
+        try {
+            connection = DBUtil.getConnection();
+            String sql = "delete from t_user_session where `_uid`=?";
+            statement = connection.prepareStatement(sql);
+            int index = 1;
+            statement.setString(index++, uid);
+
+            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);
+        }
+    }
+
+    MemorySessionStore.Session getSession(String uid, String clientId, ClientSession clientSession) {
+        String sql = "select  `_package_name`,`_token`,`_voip_token`,`_secret`,`_db_secret`,`_platform`,`_push_type`,`_device_name`,`_device_version`,`_phone_name`,`_language`,`_carrier_name`, `_dt`, `_deleted` from t_user_session where `_uid` = ? and `_cid` = ? limit 1";
+        Connection connection = null;
+        PreparedStatement statement = null;
+        ResultSet resultSet = null;
         try {
             connection = DBUtil.getConnection();
             statement = connection.prepareStatement(sql);
             statement.setString(1, uid);
             statement.setString(2, clientId);
-            ResultSet resultSet = statement.executeQuery();
+            resultSet = statement.executeQuery();
             if (resultSet.next()) {
                 MemorySessionStore.Session session = new MemorySessionStore.Session(uid, clientId, clientSession);
 
@@ -1048,23 +1178,58 @@ public class DatabaseStore {
                 session.setPhoneName(resultSet.getString(index++));
                 session.setLanguage(resultSet.getString(index++));
                 session.setCarrierName(resultSet.getString(index++));
-                session.setUpdateDt(resultSet.getLong(index));
+                session.setUpdateDt(resultSet.getLong(index++));
+                session.setDeleted(resultSet.getInt(index));
                 return session;
             }
-            resultSet.close();
         } catch (SQLException e) {
             e.printStackTrace();
             Utility.printExecption(LOG, e);
+        } finally {
+            DBUtil.closeDB(connection, statement, resultSet);
+        }
+        return null;
+    }
+
+    void clearOtherSessionToken(String cid, String token, int pushType, boolean voipToken) {
+        if (voipToken) {
+            return;
+        }
+
+        Connection connection = null;
+        PreparedStatement statement = null;
+        try {
+            connection = DBUtil.getConnection();
+
+            String sql = "update t_user_session set `_token` = ?, `_dt` = ? where `_token` = ? and `_push_type` = ? and `_cid` <> ?";
+
+            statement = connection.prepareStatement(sql);
+
+            int index = 1;
+            statement.setString(index++, "");
+            statement.setLong(index++, System.currentTimeMillis());
+            statement.setString(index++, token);
+            statement.setInt(index++, pushType);
+            statement.setString(index++, cid);
+
+            int c = statement.executeUpdate();
+            LOG.info("Update rows {}", c);
+        } catch (SQLException e) {
+            // TODO Auto-generated catch block
+            e.printStackTrace();
+            Utility.printExecption(LOG, e);
         } finally {
             DBUtil.closeDB(connection, statement);
         }
-        return null;
     }
 
-    void updateSessionToken(String uid, String cid, String token, boolean voipToken) {
+
+    void updateSessionToken(String uid, String cid, String token, int pushType, boolean voipToken) {
         mScheduler.execute(()->{
             Connection connection = null;
             PreparedStatement statement = null;
+
+            clearOtherSessionToken(cid, token, pushType, voipToken);
             try {
                 connection = DBUtil.getConnection();
 
@@ -1073,12 +1238,15 @@ public class DatabaseStore {
                 if (voipToken) {
                     sql = "update t_user_session set `_voip_token` = ?, `_dt` = ? where `_uid` = ? and `_cid` = ?";
                 } else {
-                    sql = "update t_user_session set `_token` = ?, `_dt` = ? where `_uid` = ? and `_cid` = ?";
+                    sql = "update t_user_session set `_token` = ?, `_push_type` = ?, `_dt` = ? where `_uid` = ? and `_cid` = ?";
                 }
                 statement = connection.prepareStatement(sql);
 
                 int index = 1;
                 statement.setString(index++, token);
+                if (!voipToken) {
+                    statement.setInt(index++, pushType);
+                }
                 statement.setLong(index++, System.currentTimeMillis());
                 statement.setString(index++, uid);
                 statement.setString(index++, cid);
@@ -1095,6 +1263,64 @@ public class DatabaseStore {
         });
     }
 
+    void updateSessionDeleted(String uid, String cid, int deleted) {
+        mScheduler.execute(()->{
+            Connection connection = null;
+            PreparedStatement statement = null;
+
+            try {
+                connection = DBUtil.getConnection();
+
+                String sql = "update t_user_session set `_deleted` = ? where `_uid` = ? and `_cid` = ?";
+
+                statement = connection.prepareStatement(sql);
+
+                int index = 1;
+                statement.setInt(index++, deleted);
+                statement.setString(index++, uid);
+                statement.setString(index++, cid);
+
+                int c = statement.executeUpdate();
+                LOG.info("Update rows {}", c);
+            } catch (SQLException e) {
+                // TODO Auto-generated catch block
+                e.printStackTrace();
+                Utility.printExecption(LOG, e);
+            } finally {
+                DBUtil.closeDB(connection, statement);
+            }
+        });
+    }
+
+    void updateSessionPlatform(String uid, String cid, int platform) {
+        mScheduler.execute(()->{
+            Connection connection = null;
+            PreparedStatement statement = null;
+
+            try {
+                connection = DBUtil.getConnection();
+
+                String sql = "update t_user_session set `_platform` = ? where `_uid` = ? and `_cid` = ?";
+
+                statement = connection.prepareStatement(sql);
+
+                int index = 1;
+                statement.setInt(index++, platform);
+                statement.setString(index++, uid);
+                statement.setString(index++, cid);
+
+                int c = statement.executeUpdate();
+                LOG.info("Update rows {}", c);
+            } catch (SQLException e) {
+                // TODO Auto-generated catch block
+                e.printStackTrace();
+                Utility.printExecption(LOG, e);
+            } finally {
+                DBUtil.closeDB(connection, statement);
+            }
+        });
+    }
+
     private boolean strEqual(String left, String right) {
         if (left == right)
             return true;
@@ -1187,23 +1413,71 @@ public class DatabaseStore {
         });
     }
 
-    MemorySessionStore.Session createSession(String uid, String clientId, ClientSession clientSession) {
+    void clearMultiEndpoint(String uid, String clientId, int platform) {
+        LOG.info("clearMultiEndpoint {}, {}", uid, clientId);
+        Connection connection = null;
+        PreparedStatement statement = null;
+        try {
+            connection = DBUtil.getConnection();
+            String sql;
+            if (platform == ProtoConstants.Platform.Platform_Windows || platform == ProtoConstants.Platform.Platform_OSX
+                    || platform == ProtoConstants.Platform.Platform_iOS || platform == ProtoConstants.Platform.Platform_Android) {
+                sql = "update t_user_session set `_deleted` = ?, `_token` = ?, `_voip_token` = ?, `_dt` = ?  where `_uid`=? and (`_platform` = ? or `_platform` = ?)  and `_cid` <> ? and `_deleted` = 0";
+            } else {
+                sql = "update t_user_session set `_deleted` = ?, `_token` = ?, `_voip_token` = ?, `_dt` = ?  where `_uid`=? and `_platform` = ? and `_cid` <> ? and `_deleted` = 0";
+            }
+
+            statement = connection.prepareStatement(sql);
+            int index = 1;
+
+            statement.setInt(index++, 1);
+            statement.setString(index++, "");
+            statement.setString(index++, "");
+            statement.setLong(index++, System.currentTimeMillis());
+
+            statement.setString(index++, uid);
+
+            if (platform == ProtoConstants.Platform.Platform_Windows || platform == ProtoConstants.Platform.Platform_OSX) {
+                statement.setInt(index++, ProtoConstants.Platform.Platform_Windows);
+                statement.setInt(index++, ProtoConstants.Platform.Platform_OSX);
+            } else if(platform == ProtoConstants.Platform.Platform_iOS || platform == ProtoConstants.Platform.Platform_Android) {
+                statement.setInt(index++, ProtoConstants.Platform.Platform_iOS);
+                statement.setInt(index++, ProtoConstants.Platform.Platform_Android);
+            } else {
+                statement.setInt(index++, platform);
+            }
+
+            statement.setString(index++, clientId);
+
+            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);
+        }
+    }
+
+    MemorySessionStore.Session createSession(String uid, String clientId, ClientSession clientSession, int platform) {
         Connection connection = null;
         PreparedStatement statement = null;
         LOG.info("Database create session {},{}", uid, clientId);
         try {
             connection = DBUtil.getConnection();
-            String sql = "insert into t_user_session  (`_uid`,`_cid`,`_secret`,`_db_secret`, `_dt`) values (?,?,?,?,?)";
+            String sql = "insert into t_user_session  (`_uid`,`_cid`,`_platform`,`_secret`,`_db_secret`, `_dt`) values (?,?,?,?,?,?)";
 
             statement = connection.prepareStatement(sql);
 
             int index = 1;
 
             MemorySessionStore.Session session = new MemorySessionStore.Session(uid, clientId, clientSession);
-
+            session.setPlatform(platform);
 
             statement.setString(index++, uid);
             statement.setString(index++, clientId);
+            statement.setInt(index++, platform);
 
 
             session.setSecret(UUID.randomUUID().toString());
@@ -1228,6 +1502,35 @@ public class DatabaseStore {
         return null;
     }
 
+    Set<String> getUserGroupIds(String userId) {
+        Connection connection = null;
+        PreparedStatement statement = null;
+        ResultSet rs = null;
+        try {
+            connection = DBUtil.getConnection();
+            String sql = "select `_gid` from t_group_member where `_mid` = ? and `_type` <> ?";
+            statement = connection.prepareStatement(sql);
+
+            statement.setString(1, userId);
+            statement.setInt(2, ProtoConstants.GroupMemberType.GroupMemberType_Removed);
+
+
+            rs = statement.executeQuery();
+            Set<String> out = new HashSet<>();
+            while (rs.next()) {
+                String uid = rs.getString(1);
+                out.add(uid);
+            }
+            return out;
+        } catch (SQLException e) {
+            // TODO Auto-generated catch block
+            e.printStackTrace();
+            Utility.printExecption(LOG, e);
+        } finally {
+            DBUtil.closeDB(connection, statement, rs);
+        }
+        return null;
+    }
     void updateGroupMemberCountDt(final String groupId, final int count, final long dt) {
         mScheduler.execute(()->{
             Connection connection = null;
@@ -1628,8 +1931,7 @@ public class DatabaseStore {
 
                 try {
                     MessageDigest md5 = MessageDigest.getInstance("MD5");
-                    BASE64Encoder base64en = new BASE64Encoder();
-                    String passwdMd5 = base64en.encode(md5.digest(password.getBytes("utf-8")));
+                    String passwdMd5 = Base64.getEncoder().encodeToString(md5.digest(password.getBytes("utf-8")));
                     statement.setString(index, passwdMd5);
                 } catch (Exception e) {
                     statement.setString(index, "");
@@ -1833,6 +2135,7 @@ public class DatabaseStore {
                     ", `_social`=?" +
                     ", `_extra`=?" +
                     ", `_type`=?" +
+                    ", `_deleted`=?" +
                     ", `_dt`=?";
 
                 statement = connection.prepareStatement(sql);
@@ -1866,6 +2169,7 @@ public class DatabaseStore {
 
                 statement.setString(index++, user.getExtra());
                 statement.setInt(index++, user.getType());
+                statement.setInt(index++, user.getDeleted());
                 statement.setLong(index++, user.getUpdateDt() == 0 ? System.currentTimeMillis() : user.getUpdateDt());
 
                 int count = statement.executeUpdate();
@@ -1992,6 +2296,7 @@ public class DatabaseStore {
                 ", `_social`" +
                 ", `_extra`" +
                 ", `_type`" +
+                ", `_deleted`" +
                 ", `_dt` from t_user where `_uid` = ?";
             statement = connection.prepareStatement(sql);
 
@@ -2046,6 +2351,9 @@ public class DatabaseStore {
                 int type = rs.getInt(index++);
                 builder.setType(type);
 
+                int deleted = rs.getInt(index++);
+                builder.setDeleted(deleted);
+
                 long longValue = rs.getLong(index++);
                 builder.setUpdateDt(longValue);
 
@@ -2113,7 +2421,7 @@ public class DatabaseStore {
         ResultSet rs = null;
         try {
             connection = DBUtil.getConnection();
-            String sql = "select distinct(`_uid`) from t_user_session";
+            String sql = "select distinct(`_uid`) from t_user_session where `_deleted` = 0";
             statement = connection.prepareStatement(sql);
 
 
@@ -2140,7 +2448,7 @@ public class DatabaseStore {
         ResultSet rs = null;
         try {
             connection = DBUtil.getConnection();
-            String sql = "select `_friend_uid`, `_alias`, `_state`, `_dt` from t_friend where `_uid` = ?";
+            String sql = "select `_friend_uid`, `_alias`, `_state`, `_blacked`, `_dt` from t_friend where `_uid` = ?";
             statement = connection.prepareStatement(sql);
 
 
@@ -2154,9 +2462,10 @@ public class DatabaseStore {
                 String uid = rs.getString(1);
                 String alias = rs.getString(2);
                 int state = rs.getInt(3);
-                long timestamp = rs.getLong(4);
+                int blacked = rs.getInt(4);
+                long timestamp = rs.getLong(5);
 
-                FriendData data = new FriendData(userId, uid, alias, state, timestamp);
+                FriendData data = new FriendData(userId, uid, alias, state, blacked, timestamp);
                 out.add(data);
             }
             return out;
@@ -2170,19 +2479,72 @@ public class DatabaseStore {
         return null;
     }
 
+
+    void removeUserFriend(String userId) {
+        Connection connection = null;
+        PreparedStatement statement = null;
+        try {
+            connection = DBUtil.getConnection();
+            String sql = "update t_friend  set `_alias` = '', `_state` = 1, `_blacked` = 0, `_dt` = ? where `_uid` = ? or `_friend_uid` = ?";
+
+            statement = connection.prepareStatement(sql);
+            int index = 1;
+            statement.setLong(index++, System.currentTimeMillis());
+            statement.setString(index++, userId);
+            statement.setString(index++, userId);
+            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 removeUserFriendRequest(String userId) {
+        Connection connection = null;
+        PreparedStatement statement = null;
+        try {
+            connection = DBUtil.getConnection();
+            String sql = "delete from t_friend_request  where _uid = ? or _friend_uid = ?";
+
+            statement = connection.prepareStatement(sql);
+            int index = 1;
+            statement.setString(index++, userId);
+            statement.setString(index++, userId);
+            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);
+        }
+    }
+
     List<WFCMessage.FriendRequest> getPersistFriendRequests(String userId) {
         Connection connection = null;
         PreparedStatement statement = null;
         ResultSet rs = null;
         try {
             connection = DBUtil.getConnection();
-            String sql = "select   `_uid`" +
+            String sql = "select  `_uid`" +
                 ", `_friend_uid`" +
                 ", `_reason`" +
                 ", `_status`" +
                 ", `_dt`" +
                 ", `_from_read_status`" +
-                ", `_to_read_status` from t_friend_request where `_uid` = ? or `_friend_uid` = ?";
+                ", `_to_read_status` from t_friend_request where `_uid` = ? UNION ALL " +
+                "select   `_uid`" +
+                ", `_friend_uid`" +
+                ", `_reason`" +
+                ", `_status`" +
+                ", `_dt`" +
+                ", `_from_read_status`" +
+                ", `_to_read_status` from t_friend_request where `_friend_uid` = ?";
             statement = connection.prepareStatement(sql);
 
             int index = 1;
@@ -2306,10 +2668,11 @@ public class DatabaseStore {
             PreparedStatement statement = null;
             try {
                 connection = DBUtil.getConnection();
-                String sql = "insert into t_friend (`_uid`, `_friend_uid`, `_alias`, `_state`, `_dt`) values(?, ?, ?, ?, ?)" +
+                String sql = "insert into t_friend (`_uid`, `_friend_uid`, `_alias`, `_state`, `_blacked`, `_dt`) values(?, ?, ?, ?, ?, ?)" +
                     " ON DUPLICATE KEY UPDATE " +
                     "`_alias` = ?," +
                     "`_state` = ?," +
+                    "`_blacked` = ?," +
                     "`_dt` = ?";
 
 
@@ -2319,9 +2682,11 @@ public class DatabaseStore {
                 statement.setString(index++, request.getFriendUid());
                 statement.setString(index++, request.getAlias());
                 statement.setInt(index++, request.getState());
+                statement.setInt(index++, request.getBlacked());
                 statement.setLong(index++, request.getTimestamp());
                 statement.setString(index++, request.getAlias());
                 statement.setInt(index++, request.getState());
+                statement.setInt(index++, request.getBlacked());
                 statement.setLong(index++, request.getTimestamp());
                 int count = statement.executeUpdate();
                 LOG.info("Update rows {}", count);

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

@@ -25,14 +25,15 @@ import com.hazelcast.util.StringUtil;
 import com.xiaoleilu.loServer.model.FriendData;
 import cn.wildfirechat.pojos.InputOutputUserBlockStatus;
 import cn.wildfirechat.common.ErrorCode;
+import io.moquette.server.Constants;
 import io.moquette.server.Server;
 import io.moquette.spi.IMatchingCondition;
 import io.moquette.spi.IMessagesStore;
 import io.moquette.spi.security.Tokenor;
 import io.moquette.spi.impl.subscriptions.Topic;
+import io.netty.handler.codec.mqtt.MqttVersion;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-import sun.misc.BASE64Encoder;
 import win.liyufan.im.*;
 
 import java.security.MessageDigest;
@@ -41,6 +42,7 @@ import java.sql.PreparedStatement;
 import java.sql.ResultSet;
 import java.sql.SQLException;
 import java.util.*;
+import java.util.concurrent.atomic.AtomicReference;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.locks.Lock;
@@ -110,7 +112,7 @@ public class MemoryMessagesStore implements IMessagesStore {
     private final DatabaseStore databaseStore;
     private ConcurrentHashMap<String, Long> userMaxPullSeq = new ConcurrentHashMap<>();
 
-    private SensitiveFilter mSensitiveFilter;
+    private AtomicReference<SensitiveFilter> mSensitiveFilter = new AtomicReference<>();
     private volatile long lastUpdateSensitiveTime = 0;
 
     private ConcurrentHashMap<String, Boolean> userGlobalSlientMap = new ConcurrentHashMap<>();
@@ -122,6 +124,7 @@ public class MemoryMessagesStore implements IMessagesStore {
         this.databaseStore = databaseStore;
         IS_MESSAGE_ROAMING = "1".equals(m_Server.getConfig().getProperty(MESSAGE_ROAMING));
         IS_MESSAGE_REMOTE_HISTORY_MESSAGE = "1".equals(m_Server.getConfig().getProperty(MESSAGE_Remote_History_Message));
+        Constants.MAX_MESSAGE_QUEUE = Integer.parseInt(m_Server.getConfig().getProperty(MESSAGE_Max_Queue));
     }
 
     @Override
@@ -145,7 +148,7 @@ public class MemoryMessagesStore implements IMessagesStore {
                 }
             }
             Set<String> sensitiveWords = databaseStore.getSensitiveWord();
-            mSensitiveFilter = new SensitiveFilter(sensitiveWords);
+            mSensitiveFilter.set(new SensitiveFilter(sensitiveWords));
         }
     }
 
@@ -349,7 +352,7 @@ public class MemoryMessagesStore implements IMessagesStore {
                         }
 
                         size += bundle.getMessage().getSerializedSize();
-                        if (size >= 1 * 1024 * 1024) { //3M
+                        if (size >= 512 * 1024) { //3M
                             break;
                         }
                         builder.addMessage(bundle.getMessage());
@@ -772,7 +775,31 @@ public class MemoryMessagesStore implements IMessagesStore {
         }
 
         if (groupInfo.getType() != ProtoConstants.GroupType.GroupType_Free && groupInfo.getOwner() != null && groupInfo.getOwner().equals(operator)) {
-            return ErrorCode.ERROR_CODE_NOT_RIGHT;
+            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);
+            }
+
+            WFCMessage.GroupMember newOwner = null;
+            for (WFCMessage.GroupMember member :members) {
+                if (!member.getMemberId().equals(operator) && member.getType() != GroupMemberType_Removed) {
+                    newOwner = member;
+                    break;
+                }
+            }
+
+            if (newOwner == null) {
+                 return dismissGroup(operator, groupId, false);
+            }
+
+            transferGroup(operator, groupId, newOwner.getMemberId(), false);
+
+            try {
+                Thread.sleep(500);
+            } catch (InterruptedException e) {
+                e.printStackTrace();
+            }
         }
 
         removeGroupMember(groupId, Arrays.asList(operator));
@@ -781,6 +808,18 @@ public class MemoryMessagesStore implements IMessagesStore {
         return ErrorCode.ERROR_CODE_SUCCESS;
     }
 
+    @Override
+    public void clearUserGroups(String userId) {
+        Set<String> groupIds = databaseStore.getUserGroupIds(userId);
+        int[] removedCount = new int[1];
+        for (String groupId : groupIds) {
+            removedCount[0] = 0;
+            ErrorCode errorCode = quitGroup(userId, groupId);
+            LOG.info("clear user {} group {} result {}", userId, groupId, errorCode);
+        }
+    }
+
+
     private void removeGroupUserSettings(String groupId, List<String> users) {
         HazelcastInstance hzInstance = m_Server.getHazelcastInstance();
         MultiMap<String, WFCMessage.UserSettingEntry> userSettingMap = hzInstance.getMultiMap(USER_SETTING);
@@ -971,6 +1010,11 @@ public class MemoryMessagesStore implements IMessagesStore {
         return groupInfo;
     }
 
+    @Override
+    public Set<String> getUserGroupIds(String userId) {
+        return databaseStore.getUserGroupIds(userId);
+    }
+
     @Override
     public ErrorCode getGroupMembers(String groupId, long maxDt, List<WFCMessage.GroupMember> members) {
         HazelcastInstance hzInstance = m_Server.getHazelcastInstance();
@@ -1039,25 +1083,31 @@ public class MemoryMessagesStore implements IMessagesStore {
 
         //check the new owner is in member list? is that necessary?
         long updateDt = System.currentTimeMillis();
-        groupInfo = groupInfo.toBuilder().setOwner(newOwner).setUpdateDt(updateDt).build();
-
+        groupInfo = groupInfo.toBuilder().setOwner(newOwner).setUpdateDt(updateDt).setMemberUpdateDt(updateDt).build();
         mIMap.set(groupId, groupInfo);
+
         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);
         }
+        int modifyMemeberCount = 0;
         for (WFCMessage.GroupMember member : members) {
+            if (modifyMemeberCount == 2) {
+                break;
+            }
             if (newOwner.equals(member.getMemberId())) {
                 groupMembers.remove(groupId, member);
                 member = member.toBuilder().setType(GroupMemberType_Owner).setUpdateDt(updateDt).build();
                 databaseStore.persistGroupMember(groupId, Arrays.asList(member));
                 groupMembers.put(groupId, member);
+                modifyMemeberCount++;
             } else if(member.getType() == GroupMemberType_Owner) {
                 groupMembers.remove(groupId, member);
                 member = member.toBuilder().setType(GroupMemberType_Normal).setUpdateDt(updateDt).build();
                 databaseStore.persistGroupMember(groupId, Arrays.asList(member));
                 groupMembers.put(groupId, member);
+                modifyMemeberCount++;
             }
         }
 
@@ -1224,6 +1274,18 @@ public class MemoryMessagesStore implements IMessagesStore {
         }
     }
 
+    @Override
+    public void clearUserMessages(String userId) {
+        mWriteLock.lock();
+        try {
+            userMessages.remove(userId);
+        } finally {
+            mWriteLock.unlock();
+        }
+
+        databaseStore.clearUserMessage(userId);
+    }
+
     @Override
     public WFCMessage.Robot getRobot(String robotId) {
         HazelcastInstance hzInstance = m_Server.getHazelcastInstance();
@@ -1394,6 +1456,18 @@ public class MemoryMessagesStore implements IMessagesStore {
         databaseStore.updateUserPassword(user.getUid(), password);
     }
 
+    @Override
+    public void destoryUser(String userId) {
+        HazelcastInstance hzInstance = m_Server.getHazelcastInstance();
+        IMap<String, WFCMessage.User> mUserMap = hzInstance.getMap(USERS);
+        WFCMessage.User us = mUserMap.get(userId);
+        if (us != null) {
+            us = WFCMessage.User.newBuilder().setUid(userId).setUpdateDt(System.currentTimeMillis()).setDisplayName("已删除用户").setName(userId).setDeleted(1).build();
+            mUserMap.put(userId, us);
+        }
+
+    }
+
     @Override
     public WFCMessage.User getUserInfo(String userId) {
         if (StringUtil.isNullOrEmpty(userId)) {
@@ -1556,7 +1630,7 @@ public class MemoryMessagesStore implements IMessagesStore {
 
         for (FriendData friendData : friendDatas) {
             if (friendData.getFriendUid().equals(userId)) {
-                if (friendData.getState() == 2) {
+                if (friendData.getBlacked() == 1) {
                     return true;
                 } else {
                     return false;
@@ -1592,8 +1666,7 @@ public class MemoryMessagesStore implements IMessagesStore {
                 String pwd_md5 = rs.getString(2);
                 try {
                     MessageDigest md5 = MessageDigest.getInstance("MD5");
-                    BASE64Encoder base64en = new BASE64Encoder();
-                    String passwdMd5 = base64en.encode(md5.digest(password.getBytes("utf-8")));
+                    String passwdMd5 = Base64.getEncoder().encodeToString(md5.digest(password.getBytes("utf-8")));
                     if (passwdMd5.equals(pwd_md5)) {
                         LOG.info("login success userName={}, userId={}", name, uid);
                         userIdRet.add(uid);
@@ -1648,7 +1721,7 @@ public class MemoryMessagesStore implements IMessagesStore {
     }
 
     @Override
-    public List<FriendData> getFriendList(String userId, long version) {
+    public List<FriendData> getFriendList(String userId, String clientId, long version) {
         List<FriendData> out = new ArrayList<FriendData>();
 
         HazelcastInstance hzInstance = m_Server.getHazelcastInstance();
@@ -1658,15 +1731,57 @@ public class MemoryMessagesStore implements IMessagesStore {
             friends = loadFriend(friendsMap, userId);
         }
 
+        boolean needFriendMigrate = false;
+
+        if (!StringUtil.isNullOrEmpty(clientId)) {
+            MemorySessionStore.Session session = m_Server.getStore().sessionsStore().getSession(clientId);
+            if(session != null && session.getMqttVersion() != null && session.getMqttVersion().protocolLevel() < MqttVersion.Wildfire_2.protocolLevel()) {
+                needFriendMigrate = true;
+            }
+        }
+
         for (FriendData friend : friends) {
             if (friend.getTimestamp() > version) {
-                out.add(friend);
+                if (needFriendMigrate) {
+                    if (friend.getBlacked() > 0) {
+                        friend.setState(2);
+                    }
+                    out.add(friend);
+                } else {
+                    out.add(friend);
+                }
+
             }
         }
 
         return out;
     }
 
+    @Override
+    public void clearUserFriend(String userId) {
+        HazelcastInstance hzInstance = m_Server.getHazelcastInstance();
+        MultiMap<String, FriendData> friendsMap = hzInstance.getMultiMap(USER_FRIENDS);
+
+        Collection<FriendData> friendDatas = friendsMap.get(userId);
+        if (friendDatas == null || friendDatas.size() == 0) {
+            friendDatas = loadFriend(friendsMap, userId);
+        }
+
+        databaseStore.removeUserFriend(userId);
+        databaseStore.removeUserFriendRequest(userId);
+
+        MultiMap<String, WFCMessage.FriendRequest> requestMap = hzInstance.getMultiMap(USER_FRIENDS_REQUEST);
+        requestMap.remove(userId);
+
+
+        if (friendDatas != null) {
+            for (FriendData fd : friendDatas) {
+                friendsMap.remove(fd.getFriendUid());
+            }
+            friendsMap.remove(userId);
+        }
+    }
+
     @Override
     public List<WFCMessage.FriendRequest> getFriendRequestList(String userId, long version) {
         List<WFCMessage.FriendRequest> out = new ArrayList<>();
@@ -1715,6 +1830,24 @@ public class MemoryMessagesStore implements IMessagesStore {
                 return ErrorCode.ERROR_CODE_FRIEND_ALREADY_REQUEST;
             }
         }
+
+        MultiMap<String, FriendData> friendsMap = hzInstance.getMultiMap(USER_FRIENDS);
+
+        FriendData friendData1 = null;
+        Collection<FriendData> friendDatas = friendsMap.get(userId);
+        if (friendDatas == null || friendDatas.size() == 0) {
+            friendDatas = loadFriend(friendsMap, userId);
+        }
+        for (FriendData fd : friendDatas) {
+            if (fd.getFriendUid().equals(request.getTargetUid())) {
+                friendData1 = fd;
+                break;
+            }
+        }
+        if (friendData1 != null && friendData1.getBlacked() > 0) {
+            return ErrorCode.ERROR_CODE_IN_BLACK_LIST;
+        }
+
         WFCMessage.FriendRequest newRequest = WFCMessage.FriendRequest
             .newBuilder()
             .setFromUid(userId)
@@ -1766,7 +1899,7 @@ public class MemoryMessagesStore implements IMessagesStore {
                 }
             }
             if (friendData1 == null) {
-                friendData1 = new FriendData(userId, request.getTargetUid(), "", request.getStatus(), System.currentTimeMillis());
+                friendData1 = new FriendData(userId, request.getTargetUid(), "", request.getStatus(), 0, System.currentTimeMillis());
             } else {
                 friendData1.setState(request.getStatus());
                 friendData1.setTimestamp(System.currentTimeMillis());
@@ -1788,7 +1921,7 @@ public class MemoryMessagesStore implements IMessagesStore {
                     }
                 }
                 if (friendData2 == null) {
-                    friendData2 = new FriendData(request.getTargetUid(), userId, "", request.getStatus(), friendData1.getTimestamp());
+                    friendData2 = new FriendData(request.getTargetUid(), userId, "", request.getStatus(), 0, friendData1.getTimestamp());
                 } else {
                     friendsMap.remove(request.getTargetUid(), friendData2);
                     friendData2.setState(request.getStatus());
@@ -1829,10 +1962,10 @@ public class MemoryMessagesStore implements IMessagesStore {
                 databaseStore.persistOrUpdateFriendRequest(existRequest);
                 MultiMap<String, FriendData> friendsMap = hzInstance.getMultiMap(USER_FRIENDS);
 
-                FriendData friendData1 = new FriendData(userId, request.getTargetUid(), "", 0, System.currentTimeMillis());
+                FriendData friendData1 = new FriendData(userId, request.getTargetUid(), "", 0, 0, System.currentTimeMillis());
                 databaseStore.persistOrUpdateFriendData(friendData1);
 
-                FriendData friendData2 = new FriendData(request.getTargetUid(), userId, "", 0, friendData1.getTimestamp());
+                FriendData friendData2 = new FriendData(request.getTargetUid(), userId, "", 0, 0, friendData1.getTimestamp());
                 databaseStore.persistOrUpdateFriendData(friendData2);
 
                 requestMap.remove(userId);
@@ -1854,6 +1987,12 @@ public class MemoryMessagesStore implements IMessagesStore {
 
     @Override
     public ErrorCode blackUserRequest(String fromUser, String targetUserId, int state, long[] heads) {
+        if (state == 2) {
+            state = 1;
+        } else {
+            state = 0;
+        }
+        
         HazelcastInstance hzInstance = m_Server.getHazelcastInstance();
         MultiMap<String, FriendData> friendsMap = hzInstance.getMultiMap(USER_FRIENDS);
 
@@ -1871,9 +2010,9 @@ public class MemoryMessagesStore implements IMessagesStore {
         }
 
         if (friendData == null) {
-            friendData = new FriendData(fromUser, targetUserId, "", state, System.currentTimeMillis());
+            friendData = new FriendData(fromUser, targetUserId, "", 1, state, System.currentTimeMillis());
         }
-        friendData.setState(state);
+        friendData.setBlacked(state);
         friendData.setTimestamp(System.currentTimeMillis());
 
 
@@ -1994,6 +2133,8 @@ public class MemoryMessagesStore implements IMessagesStore {
     public ErrorCode getUserSettings(String userId, long version, WFCMessage.GetUserSettingResult.Builder builder) {
         HazelcastInstance hzInstance = m_Server.getHazelcastInstance();
         MultiMap<String, WFCMessage.UserSettingEntry> userSettingMap = hzInstance.getMultiMap(USER_SETTING);
+        IMap<String, WFCMessage.GroupInfo> mGroups = hzInstance.getMap(GROUPS_MAP);
+        IMap<String, WFCMessage.ChannelInfo> mChannels = hzInstance.getMap(CHANNELS);
 
         Collection<WFCMessage.UserSettingEntry> entries = userSettingMap.get(userId);
         if (entries == null || entries.size() == 0) {
@@ -2001,12 +2142,51 @@ public class MemoryMessagesStore implements IMessagesStore {
         }
 
         ErrorCode ec = ErrorCode.ERROR_CODE_NOT_MODIFIED;
+        Calendar c = Calendar.getInstance();
+        c.setTime(new Date());
+        c.add(Calendar.MONTH, -1);
+        long monthAgo = c.getTimeInMillis();
+
         if (entries != null) {
             for (WFCMessage.UserSettingEntry entry : entries
             ) {
                 if (entry.getUpdateDt() > version) {
                     ec = ErrorCode.ERROR_CODE_SUCCESS;
-                    builder.addEntry(entry);
+                    boolean skip = false;
+                    if (version == 0 && entry.getUpdateDt() > monthAgo) {
+                        if (entry.getScope() == UserSettingScope.kUserSettingConversationSync) {
+                            try {
+                                String key = entry.getKey();
+                                int pos1 = key.indexOf('-');
+                                String typeStr = key.substring(0, pos1);
+                                int type = Integer.parseInt(typeStr);
+                                if (type == ProtoConstants.ConversationType.ConversationType_Group
+                                    || type == ProtoConstants.ConversationType.ConversationType_Channel) {
+                                    int pos2 = key.indexOf('-', pos1 + 1);
+                                    String target = key.substring(pos2 + 1);
+
+                                    if (type == ProtoConstants.ConversationType.ConversationType_Group) {
+                                        WFCMessage.GroupInfo groupInfo = mGroups.get(target);
+                                        if (groupInfo == null) {
+                                            skip = true;
+                                        }
+                                    } else {
+                                        WFCMessage.ChannelInfo channelInfo = mChannels.get(target);
+                                        if (channelInfo == null) {
+                                            skip = true;
+                                        }
+                                    }
+                                }
+                            } catch (Exception e) {
+                                e.printStackTrace();
+                            }
+
+                        }
+                    }
+
+                    if (!skip) {
+                        builder.addEntry(entry);
+                    }
                 }
             }
         }
@@ -2102,6 +2282,14 @@ public class MemoryMessagesStore implements IMessagesStore {
         return updateDt;
     }
 
+    @Override
+    public void clearUserSettings(String userId) {
+        databaseStore.clearUserSetting(userId);
+        HazelcastInstance hzInstance = m_Server.getHazelcastInstance();
+        MultiMap<String, WFCMessage.UserSettingEntry> userSettingMap = hzInstance.getMultiMap(USER_SETTING);
+        userSettingMap.remove(userId);
+    }
+
     @Override
     public boolean getUserGlobalSlient(String userId) {
         Boolean slient = userGlobalSlientMap.get(userId);
@@ -2121,7 +2309,7 @@ public class MemoryMessagesStore implements IMessagesStore {
     public boolean getUserPushHiddenDetail(String userId) {
         Boolean hidden = userPushHiddenDetail.get(userId);
         if (hidden == null) {
-            WFCMessage.UserSettingEntry entry = getUserSetting(userId, UserSettingScope.kUserSettingGlobalSilent, null);
+            WFCMessage.UserSettingEntry entry = getUserSetting(userId, UserSettingScope.kUserSettingHiddenNotificationDetail, null);
             if (entry == null || !entry.getValue().equals("1")) {
                 hidden = false;
             } else {
@@ -2159,6 +2347,12 @@ public class MemoryMessagesStore implements IMessagesStore {
         return ErrorCode.ERROR_CODE_SUCCESS;
     }
 
+    @Override
+    public void clearUserChannels(String userId) {
+
+    }
+
+
     @Override
     public ErrorCode modifyChannelInfo(String operator, String channelId, int modifyType, String value) {
         HazelcastInstance hzInstance = m_Server.getHazelcastInstance();
@@ -2301,7 +2495,7 @@ public class MemoryMessagesStore implements IMessagesStore {
     @Override
     public Set<String> handleSensitiveWord(String message) {
         updateSensitiveWord();
-        return mSensitiveFilter.getSensitiveWords(message, SensitiveFilter.MatchType.MAX_MATCH);
+        return mSensitiveFilter.get().getSensitiveWords(message, SensitiveFilter.MatchType.MAX_MATCH);
     }
 
     @Override
@@ -2320,6 +2514,7 @@ public class MemoryMessagesStore implements IMessagesStore {
             words) {
             databaseStore.deleteSensitiveWord(word);
         }
+        lastUpdateSensitiveTime = 0;
         return true;
     }
 

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

@@ -16,17 +16,19 @@
 
 package io.moquette.persistence;
 
+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.Constants;
 import io.moquette.server.Server;
 import io.moquette.spi.ClientSession;
-import io.moquette.spi.IMessagesStore.StoredMessage;
 import io.moquette.spi.ISessionsStore;
-
+import io.moquette.spi.IMessagesStore.StoredMessage;
 import io.netty.handler.codec.mqtt.MqttVersion;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-import cn.wildfirechat.common.ErrorCode;
 
 import java.util.*;
 import java.util.concurrent.*;
@@ -35,6 +37,8 @@ public class MemorySessionStore implements ISessionsStore {
     private static int dumy = 1;
     private static final Logger LOG = LoggerFactory.getLogger(MemorySessionStore.class);
 
+    private boolean supportMultiEndpoint = false;
+
     public static class Session implements Comparable<Session>{
         final String clientID;
         String username;
@@ -50,6 +54,16 @@ public class MemorySessionStore implements ISessionsStore {
 
         private volatile int unReceivedMsgs;
 
+        private int deleted;
+
+        public int getDeleted() {
+            return deleted;
+        }
+
+        public void setDeleted(int deleted) {
+            this.deleted = deleted;
+        }
+
         private MqttVersion mqttVersion = MqttVersion.MQTT_3_1_1;
 
         public long getLastActiveTime() {
@@ -130,6 +144,9 @@ public class MemorySessionStore implements ISessionsStore {
 
         public void setUpdateDt(long updateDt) {
             this.updateDt = updateDt;
+            if (this.lastActiveTime == 0) {
+                this.lastActiveTime = updateDt;
+            }
         }
 
         private int pushType;
@@ -248,8 +265,14 @@ public class MemorySessionStore implements ISessionsStore {
     private final Server mServer;
     private final DatabaseStore databaseStore;
     public MemorySessionStore(Server server, DatabaseStore databaseStore) {
-    		mServer = server;
-    		this.databaseStore = databaseStore;
+        mServer = server;
+        this.databaseStore = databaseStore;
+
+        try {
+            supportMultiEndpoint = Boolean.parseBoolean(server.getConfig().getProperty(BrokerConstants.SERVER_MULTI_ENDPOINT));
+        } catch (Exception e) {
+            e.printStackTrace();
+        }
     }
 
     @Override
@@ -261,6 +284,21 @@ public class MemorySessionStore implements ISessionsStore {
         return session;
     }
 
+    @Override
+    public void cleanDuplatedToken(String cid, int pushType, String token, boolean isVoip) {
+        if (StringUtil.isNullOrEmpty(token) || isVoip) {
+            return;
+        }
+
+        Iterator<Map.Entry<String, Session>> it = sessions.entrySet().iterator();
+        while (it.hasNext()) {
+            Session session = it.next().getValue();
+            if (!session.getClientID().equals(cid) && (session.pushType == pushType && token.equals(session.deviceToken))) {
+                session.deviceToken = null;
+            }
+        }
+    }
+
     @Override
     public void initStore() {
     }
@@ -272,18 +310,82 @@ public class MemorySessionStore implements ISessionsStore {
 
     @Override
     public void updateSessionToken(Session session, boolean voip) {
-        databaseStore.updateSessionToken(session.getUsername(), session.getClientID(), voip ? session.getVoipDeviceToken() : session.getDeviceToken(), voip);
+        databaseStore.updateSessionToken(session.getUsername(), session.getClientID(), voip ? session.getVoipDeviceToken() : session.getDeviceToken(), session.getPushType(), voip);
     }
 
     @Override
-    public Session createUserSession(String username, String clientID) {
+    public void clearUserSession(String username) {
+        LOG.info("Fooooooooo <{}>", username);
+
+        databaseStore.clearUserSessions(username);
+
+        ConcurrentSkipListSet<String> sessionSet = getUserSessionSet(username);
+        for (String clientID : sessionSet) {
+            Session s = sessions.remove(clientID);
+            mServer.getProcessor().kickoffSession(s);
+        }
+        userSessions.remove(username);
+    }
+
+    @Override
+    public Session createUserSession(String username, String clientID, int platform) {
         LOG.debug("createUserSession for client <{}>, user <{}>", clientID, username);
 
         ClientSession clientSession = new ClientSession(clientID, this);
         Session session = databaseStore.getSession(username, clientID, clientSession);
 
         if (session == null) {
-            session = databaseStore.createSession(username, clientID, clientSession);
+            session = databaseStore.createSession(username, clientID, clientSession, platform);
+        }
+
+        if (session.getDeleted() > 0) {
+            session.setDeleted(0);
+            databaseStore.updateSessionDeleted(username, clientID, 0);
+        }
+
+        if (session.getPlatform() != platform) {
+            session.setPlatform(platform);
+            databaseStore.updateSessionPlatform(username, clientID, platform);
+        }
+
+
+        if (!supportMultiEndpoint && platform > 0) {
+            databaseStore.clearMultiEndpoint(username, clientID, platform);
+            if (userSessions.get(username) != null) {
+                Iterator<String> it = userSessions.get(username).iterator();
+                while (it.hasNext()) {
+                    String c = it.next();
+                    if (!clientID.equals(c)) {
+                        Session s = sessions.get(c);
+                        if (s == null) {
+                            it.remove();
+                            continue;
+                        }
+
+                        boolean remove = false;
+                        if (platform == ProtoConstants.Platform.Platform_Android || platform == ProtoConstants.Platform.Platform_iOS) {
+                            if (s.getPlatform() == ProtoConstants.Platform.Platform_Android || s.getPlatform() == ProtoConstants.Platform.Platform_iOS) {
+                                remove = true;
+                            }
+                        } else if(platform == ProtoConstants.Platform.Platform_OSX || platform == ProtoConstants.Platform.Platform_Windows) {
+                            if (s.getPlatform() == ProtoConstants.Platform.Platform_OSX || s.getPlatform() == ProtoConstants.Platform.Platform_Windows) {
+                                remove = true;
+                            }
+                        } else {
+                            if (s.getPlatform() ==platform) {
+                                remove = true;
+                            }
+                        }
+
+                        if (remove) {
+                            sessions.remove(c);
+                            mServer.getProcessor().kickoffSession(s);
+                            it.remove();
+                        }
+                    }
+                }
+            }
+
         }
 
         return session;
@@ -291,34 +393,29 @@ public class MemorySessionStore implements ISessionsStore {
 
 
     @Override
-    public ErrorCode createNewSession(String username, String clientID, boolean cleanSession, boolean createWhenNoExist) {
+    public ErrorCode loadActiveSession(String username, String clientID) {
         LOG.debug("createNewSession for client <{}>", clientID);
 
         Session session = sessions.get(clientID);
-        if (session != null) {
+
+        if (session != null && session.getDeleted() == 0) {
             LOG.error("already exists a session for client <{}>, bad condition", clientID);
             throw new IllegalArgumentException("Can't create a session with the ID of an already existing" + clientID);
         }
 
+        if (session != null && session.getDeleted() > 0) {
+            return ErrorCode.ERROR_CODE_SECRECT_KEY_MISMATCH;
+        }
 
         ClientSession clientSession = new ClientSession(clientID, this);
         session = databaseStore.getSession(username, clientID, clientSession);
 
-        if (session == null) {
-            if (!createWhenNoExist) {
-                return ErrorCode.ERROR_CODE_NOT_EXIST;
-            }
-
-            session = databaseStore.createSession(username, clientID, clientSession);
+        if (session == null || session.getDeleted() > 0) {
+            return ErrorCode.ERROR_CODE_SECRECT_KEY_MISMATCH;
         }
 
         sessions.put(clientID, session);
-        ConcurrentSkipListSet<String> sessionSet = userSessions.get(username);
-        if (sessionSet == null) {
-			sessionSet = new ConcurrentSkipListSet<>();
-			userSessions.put(username, sessionSet);
-		}
-        sessionSet = userSessions.get(username);
+        ConcurrentSkipListSet<String> sessionSet = getUserSessionSet(username);
         sessionSet.add(clientID);
 
         return ErrorCode.ERROR_CODE_SUCCESS;
@@ -342,12 +439,7 @@ public class MemorySessionStore implements ISessionsStore {
         session.setUsername(username);
         sessions.put(clientID, session);
 
-        ConcurrentSkipListSet<String> sessionSet = userSessions.get(username);
-        if (sessionSet == null) {
-            sessionSet = new ConcurrentSkipListSet<>();
-            userSessions.put(username, sessionSet);
-        }
-        sessionSet = userSessions.get(username);
+        ConcurrentSkipListSet<String> sessionSet = getUserSessionSet(username);
         sessionSet.add(clientID);
 
         if (endpoint != null) {
@@ -393,14 +485,26 @@ public class MemorySessionStore implements ISessionsStore {
         }
     }
 
+    private ConcurrentSkipListSet<String> getUserSessionSet(String username) {
+        ConcurrentSkipListSet<String> sessionSet = userSessions.get(username);
+        if (sessionSet == null) {
+            sessionSet = new ConcurrentSkipListSet<String>();
+            List<Session> ss = databaseStore.getUserActivedSessions(username);
+            for (Session s : ss) {
+                sessionSet.add(s.getClientID());
+                sessions.put(s.getClientID(), s);
+            }
+            userSessions.put(username, sessionSet);
+        }
+
+        sessionSet = userSessions.get(username);
+        return sessionSet;
+    }
+
     @Override
     public Collection<Session> sessionForUser(String username) {
-    	ConcurrentSkipListSet<String> sessionSet = userSessions.get(username);
-        if (sessionSet == null) {
-			sessionSet = new ConcurrentSkipListSet<String>();
-			userSessions.put(username, sessionSet);
-		}
-        sessionSet = userSessions.get(username);
+    	ConcurrentSkipListSet<String> sessionSet = getUserSessionSet(username);
+
         ArrayList<Session> out = new ArrayList<>();
         for (String clientId : sessionSet
              ) {
@@ -466,7 +570,9 @@ public class MemorySessionStore implements ISessionsStore {
 
     @Override
     public void dropQueue(String clientID) {
-        sessions.get(clientID).queue.clear();
+        if (sessions.get(clientID) != null) {
+            sessions.get(clientID).queue.clear();
+        }
     }
 
     @Override
@@ -542,12 +648,7 @@ public class MemorySessionStore implements ISessionsStore {
             LOG.error("Can't find the session for client <{}>", clientID);
             return;
         }
-        ConcurrentSkipListSet<String> sessionSet = userSessions.get(session.username);
-        if (sessionSet == null) {
-			sessionSet = new ConcurrentSkipListSet<>();
-			userSessions.put(session.username, sessionSet);
-		}
-        sessionSet = userSessions.get(session.username);
+        ConcurrentSkipListSet<String> sessionSet = getUserSessionSet(session.username);
         sessionSet.remove(clientID);
 
         // remove also the messages stored of type QoS1/2

broker/src/main/java/io/moquette/server/Constants.java

@@ -24,8 +24,10 @@ public final class Constants {
     public static final String ATTR_CLIENTID = "ClientID";
     public static final String CLEAN_SESSION = "cleanSession";
     public static final String KEEP_ALIVE = "keepAlive";
-    public static final int MAX_MESSAGE_QUEUE = 1024; // number of messages
-    public static final int MAX_CHATROOM_MESSAGE_QUEUE = 256; // number of chatroom messages
+
+
+    public static int MAX_MESSAGE_QUEUE = 1024; // number of messages
+    public static int MAX_CHATROOM_MESSAGE_QUEUE = 256; // number of chatroom messages
 
     private Constants() {
     }

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

@@ -107,11 +107,13 @@ public interface IMessagesStore {
     ErrorCode addGroupMembers(String operator, boolean isAdmin, String groupId, List<WFCMessage.GroupMember> memberList);
     ErrorCode kickoffGroupMembers(String operator, boolean isAdmin, String groupId, List<String> memberList);
     ErrorCode quitGroup(String operator, String groupId);
+    void clearUserGroups(String userId);
     ErrorCode dismissGroup(String operator, String groupId, boolean isAdmin);
     ErrorCode modifyGroupInfo(String operator, String groupId, int modifyType, String value, boolean isAdmin);
     ErrorCode modifyGroupAlias(String operator, String groupId, String alias);
     List<WFCMessage.GroupInfo> getGroupInfos(List<WFCMessage.UserRequest> requests);
     WFCMessage.GroupInfo getGroupInfo(String groupId);
+    Set<String> getUserGroupIds(String userId);
     ErrorCode getGroupMembers(String groupId, long maxDt, List<WFCMessage.GroupMember> members);
     WFCMessage.GroupMember getGroupMember(String groupId, String memberId);
     ErrorCode transferGroup(String operator, String groupId, String newOwner, boolean isAdmin);
@@ -121,6 +123,8 @@ public interface IMessagesStore {
 
     ErrorCode recallMessage(long messageUid, String operatorId, boolean isAdmin);
 
+    void clearUserMessages(String userId);
+
     WFCMessage.Robot getRobot(String robotId);
     void addRobot(WFCMessage.Robot robot);
     ErrorCode getUserInfo(List<WFCMessage.UserRequest> requestList, WFCMessage.PullUserResult.Builder builder);
@@ -131,6 +135,7 @@ public interface IMessagesStore {
     List<InputOutputUserBlockStatus> getUserStatusList();
 
     void addUserInfo(WFCMessage.User user, String password);
+    void destoryUser(String userId);
     WFCMessage.User getUserInfo(String userId);
     WFCMessage.User getUserInfoByName(String name);
     WFCMessage.User getUserInfoByMobile(String mobile);
@@ -153,7 +158,8 @@ public interface IMessagesStore {
     ErrorCode verifyToken(String userId, String token, List<String> serverIPs, List<Integer> ports);
     ErrorCode login(String name, String password, List<String> userIdRet);
 
-    List<FriendData> getFriendList(String userId, long version);
+    List<FriendData> getFriendList(String userId, String clientId, long version);
+    void clearUserFriend(String userId);
     List<WFCMessage.FriendRequest> getFriendRequestList(String userId, long version);
 
     ErrorCode saveAddFriendRequest(String userId, WFCMessage.AddFriendRequest request, long[] head);
@@ -171,12 +177,14 @@ public interface IMessagesStore {
     WFCMessage.UserSettingEntry getUserSetting(String userId, int scope, String key);
     List<WFCMessage.UserSettingEntry> getUserSetting(String userId, int scope);
     long updateUserSettings(String userId, WFCMessage.ModifyUserSettingReq request);
+    void clearUserSettings(String userId);
 
     boolean getUserGlobalSlient(String userId);
     boolean getUserPushHiddenDetail(String userId);
     boolean getUserConversationSlient(String userId, WFCMessage.Conversation conversation);
 
     ErrorCode createChannel(String operator, WFCMessage.ChannelInfo channelInfo);
+    void clearUserChannels(String userId);
     ErrorCode modifyChannelInfo(String operator, String channelId, int modifyType, String value);
     ErrorCode transferChannel(String operator, String channelId, String newOwner);
     ErrorCode distoryChannel(String operator, String channelId);

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

@@ -38,15 +38,19 @@ public interface ISessionsStore {
      */
     boolean contains(String clientID);
 
-    Session createUserSession(String username, String clientID);
+    Session createUserSession(String username, String clientID, int platform);
 
-    ErrorCode createNewSession(String username, String clientID, boolean cleanSession, boolean createNoExist);
+    ErrorCode loadActiveSession(String username, String clientID);
 
     ClientSession updateExistSession(String username, String clientID, WFCMessage.RouteRequest endpoint, boolean cleanSession);
 
     Session getSession(String clientID);
 
+    void cleanDuplatedToken(String cid, int pushType, String token, boolean isVoip);
+
     void updateSessionToken(Session session, boolean voip);
+
+    void clearUserSession(String username);
     /**
      * @param clientID
      *            the client owning the session.

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

@@ -16,6 +16,7 @@
 
 package io.moquette.spi.impl;
 
+import cn.wildfirechat.common.ErrorCode;
 import cn.wildfirechat.proto.WFCMessage;
 import io.moquette.persistence.RPCCenter;
 import io.moquette.interception.InterceptHandler;
@@ -55,7 +56,24 @@ import static io.moquette.server.ConnectionDescriptor.ConnectionState.*;
  *
  * Used by the front facing class ProtocolProcessorBootstrapper.
  */
+
 public class ProtocolProcessor {
+
+
+    public void kickoffSession(final MemorySessionStore.Session session) {
+        mServer.getImBusinessScheduler().execute(()->{
+            ConnectionDescriptor descriptor = connectionDescriptors.getConnection(session.getClientID());
+            try {
+                if (descriptor != null) {
+                    processDisconnect(descriptor.getChannel(), true);
+                }
+            } catch (InterruptedException e) {
+                e.printStackTrace();
+                Utility.printExecption(LOG, e);
+            }
+        });
+    }
+
     private void handleTargetRemovedFromCurrentNode(TargetEntry target) {
         System.out.println("kickof user " + target);
         if (target.type == TargetEntry.Type.TARGET_TYPE_USER) {
@@ -142,9 +160,8 @@ public class ProtocolProcessor {
         String clientId = payload.clientIdentifier();
         LOG.info("Processing CONNECT message. CId={}, username={}", clientId, payload.userName());
 
-        if (msg.variableHeader().version() != MqttVersion.MQTT_3_1.protocolLevel()
-                && msg.variableHeader().version() != MqttVersion.MQTT_3_1_1.protocolLevel()
-                && msg.variableHeader().version() != MqttVersion.Wildfire_1.protocolLevel()) {
+        if (msg.variableHeader().version() < MqttVersion.MQTT_3_1_1.protocolLevel() ||
+                msg.variableHeader().version() >= MqttVersion.Wildfire_Max.protocolLevel()) {
             MqttConnAckMessage badProto = connAck(CONNECTION_REFUSED_UNACCEPTABLE_PROTOCOL_VERSION);
 
             LOG.error("MQTT protocol version is not valid. CId={}", clientId);
@@ -165,7 +182,11 @@ public class ProtocolProcessor {
 
         MqttVersion mqttVersion = MqttVersion.fromProtocolLevel(msg.variableHeader().version());
         if (!login(channel, msg, clientId, mqttVersion)) {
+            MqttConnAckMessage badId = connAck(CONNECTION_REFUSED_BAD_USER_NAME_OR_PASSWORD);
+
+            channel.writeAndFlush(badId);
             channel.close();
+            LOG.error("The MQTT login failure. Username={}", payload.userName());
             return;
         }
         if (!mServer.m_initialized) {
@@ -254,9 +275,16 @@ public class ProtocolProcessor {
 
                 MemorySessionStore.Session session = m_sessionsStore.getSession(clientId);
                 if (session == null) {
-                    m_sessionsStore.createNewSession(msg.payload().userName(), clientId, true, false);
+                    ErrorCode errorCode = m_sessionsStore.loadActiveSession(msg.payload().userName(), clientId);
+                    if (errorCode != ErrorCode.ERROR_CODE_SUCCESS) {
+                        return false;
+                    }
                     session = m_sessionsStore.getSession(clientId);
                 }
+
+                if (session.getDeleted() != 0) {
+                    return false;
+                }
                 
                 if (session != null && session.getUsername().equals(msg.payload().userName())) {
                     pwd = AES.AESDecrypt(pwd, session.getSecret(), true);
@@ -502,15 +530,12 @@ public class ProtocolProcessor {
             return;
         }
 
-        boolean stillPresent = this.connectionDescriptors.removeConnection(existingDescriptor);
-        if (!stillPresent) {
-            // another descriptor was inserted
-            LOG.warn("Another descriptor has been inserted. CId={}", clientID);
-            return;
-        }
+        this.connectionDescriptors.removeConnection(existingDescriptor);
 
         LOG.info("The DISCONNECT message has been processed. CId={}", clientID);
 
+        channel.closeFuture();
+
         //disconnect the session
         m_sessionsStore.sessionForClient(clientID).disconnect(clearSession);
     }

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

@@ -178,17 +178,16 @@ public class Qos1PublishHandler extends QosPublishHandler {
                         } else {
                             MemorySessionStore.Session session = m_sessionStore.getSession(clientID);
                             if (session != null && session.getUsername().equals(fromUser)) {
-                                if (data.length > 7*1024 && session.getMqttVersion() == MqttVersion.Wildfire_1) {
-                                    ByteArrayOutputStream out = new ByteArrayOutputStream();
-                                    GZIPOutputStream gzip;
-                                    try {
-                                        gzip = new GZIPOutputStream(out);
+                                if (data.length > 7*1024 && session.getMqttVersion().protocolLevel() >= MqttVersion.Wildfire_1.protocolLevel()) {
+                                    try (
+                                        ByteArrayOutputStream out = new ByteArrayOutputStream();
+                                        GZIPOutputStream gzip = new GZIPOutputStream(out)
+                                    ) {
                                         gzip.write(data);
-                                        gzip.close();
+                                        data = out.toByteArray();
                                     } catch (Exception e) {
                                         e.printStackTrace();
                                     }
-                                    data = out.toByteArray();
                                     code = (byte)ErrorCode.ERROR_CODE_SUCCESS_GZIPED.code;
                                 }
 

broker/src/main/java/io/netty/handler/codec/mqtt/MqttCodecUtil.java

@@ -43,7 +43,7 @@ final class MqttCodecUtil {
             return clientId != null && clientId.length() >= MIN_CLIENT_ID_LENGTH &&
                 clientId.length() <= MAX_CLIENT_ID_LENGTH;
         }
-        if (mqttVersion == MqttVersion.MQTT_3_1_1 || mqttVersion == MqttVersion.Wildfire_1) {
+        if (mqttVersion.protocolLevel() >= MqttVersion.MQTT_3_1_1.protocolLevel()) {
             // In 3.1.3.1 Client Identifier of MQTT 3.1.1 specification, The Server MAY allow ClientId’s
             // that contain more than 23 encoded bytes. And, The Server MAY allow zero-length ClientId.
             return clientId != null;

broker/src/main/java/io/netty/handler/codec/mqtt/MqttDecoder.java

@@ -223,7 +223,7 @@ public final class MqttDecoder extends ReplayingDecoder<DecoderState> {
         final int willQos = (b1 & 0x18) >> 3;
         final boolean willFlag = (b1 & 0x04) == 0x04;
         final boolean cleanSession = (b1 & 0x02) == 0x02;
-        if (mqttVersion == MqttVersion.MQTT_3_1_1 || mqttVersion == MqttVersion.Wildfire_1) {
+        if (mqttVersion.protocolLevel() >= MqttVersion.MQTT_3_1_1.protocolLevel()) {
             final boolean zeroReservedFlag = (b1 & 0x01) == 0x0;
             if (!zeroReservedFlag) {
                 // MQTT v3.1.1: The Server MUST validate that the reserved flag in the CONNECT Control Packet is

broker/src/main/java/io/netty/handler/codec/mqtt/MqttVersion.java

@@ -25,7 +25,9 @@ import io.netty.util.internal.ObjectUtil;
 public enum MqttVersion {
     MQTT_3_1("MQIsdp", (byte) 3),
     MQTT_3_1_1("MQTT", (byte) 4),
-    Wildfire_1("MQTT", (byte) 5);
+    Wildfire_1("MQTT", (byte) 5),
+    Wildfire_2("MQTT", (byte) 6),
+    Wildfire_Max("MQTT", (byte) 7);
 
     private final String name;
     private final byte level;

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

@@ -52,7 +52,14 @@ public class HttpUtils {
 
             @Override
             public void onResponse(Call call, Response response) throws IOException {
-                LOG.info("POST to {} success with response", url, response.body().string());
+                LOG.info("POST to {} success with response: {}", url, response.body());
+                try {
+                    if (response.body() != null) {
+                        response.body().close();
+                    }
+                } catch (Exception e) {
+                    e.printStackTrace();
+                }
             }
         });
     }

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

@@ -71,6 +71,7 @@ public interface IMTopic {
     String ChannelPullTopic = "CHP";
 
     String GetTokenTopic = "GETTOKEN";
+    String DestroyUserTopic = "DESTROYUSER";
 
     String LoadRemoteMessagesTopic = "LRM";
 }

broker/src/main/resources/c3p0-config.xml

@@ -1,4 +1,7 @@
 <?xml version ="1.0" encoding="UTF-8"?>
+
+<!--    # 本配置文件为debug使用，修改这里不会在Release包中生效。Release中包含的配置文件在 ${Porject_Path}/distribution/src/main/resources目录下-->
+
 <c3p0-config>
     <!-- This is default config! -->
     <default-config>
@@ -19,7 +22,7 @@
         <!--MySQL数据库用户名-->
         <property name="user">root</property>
         <!--MySQL数据库密码-->
-        <property name="password">lt123456</property>
+        <property name="password">123456</property>
         <!-- 初始化连接池中的连接数，取值应在minPoolSize与maxPoolSize之间，默认为3-->
         <property name="initialPoolSize">10</property>
         <!--最大空闲时间，60秒内未使用则连接被丢弃。若为0则永不丢弃。默认值: 0 -->

common/pom.xml

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

common/src/main/java/cn.wildfirechat.pojos/InputDestroyUser.java

@@ -0,0 +1,22 @@
+/*
+ * This file is part of the Wildfire Chat package.
+ * (c) Heavyrain2012 <heavyrain.lee@gmail.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+package cn.wildfirechat.pojos;
+
+
+public class InputDestroyUser {
+    private String userId;
+
+    public String getUserId() {
+        return userId;
+    }
+
+    public void setUserId(String userId) {
+        this.userId = userId;
+    }
+}

common/src/main/java/cn/wildfirechat/common/APIPath.java

@@ -11,6 +11,7 @@ public interface APIPath {
     String Sensitive_Query = "/admin/sensitive/query";
 
     String Create_User = "/admin/user/create";
+    String Destroy_User = "/admin/user/destroy";
     String Create_Robot = "/admin/robot/create";
     String User_Get_Token = "/admin/user/get_token";
     String User_Update_Block_Status = "/admin/user/update_block_status";
@@ -36,6 +37,8 @@ public interface APIPath {
     String Group_Member_List = "/admin/group/member/list";
     String Group_Member_Add = "/admin/group/member/add";
     String Group_Member_Kickoff = "/admin/group/member/del";
+    String Group_Set_Manager = "/admin/group/manager/set";
+    String Get_User_Groups = "/admin/group/of_user";
 
 
     String Create_Channel = "/admin/channel/create";

common/src/main/java/cn/wildfirechat/pojos/InputGetToken.java

@@ -3,10 +3,20 @@ package cn.wildfirechat.pojos;
 public class InputGetToken {
     private String userId;
     private String clientId;
+    private Integer platform;
 
-    public InputGetToken(String userId, String clientId) {
+    public InputGetToken(String userId, String clientId, int platform) {
         this.userId = userId;
         this.clientId = clientId;
+        this.platform = platform;
+    }
+
+    public Integer getPlatform() {
+        return platform;
+    }
+
+    public void setPlatform(Integer platform) {
+        this.platform = platform;
     }
 
     public String getUserId() {

common/src/main/java/cn/wildfirechat/pojos/InputSetGroupManager.java

@@ -0,0 +1,72 @@
+/*
+ * This file is part of the Wildfire Chat package.
+ * (c) Heavyrain2012 <heavyrain.lee@gmail.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+package cn.wildfirechat.pojos;
+
+
+import cn.wildfirechat.proto.WFCMessage;
+import io.netty.util.internal.StringUtil;
+
+import java.util.List;
+
+public class InputSetGroupManager extends InputGroupBase {
+    private String group_id;
+    private List<String> members;
+    private boolean is_manager;
+
+    public String getGroup_id() {
+        return group_id;
+    }
+
+    public void setGroup_id(String group_id) {
+        this.group_id = group_id;
+    }
+
+    public List<String> getMembers() {
+        return members;
+    }
+
+    public void setMembers(List<String> members) {
+        this.members = members;
+    }
+
+    public boolean isIs_manager() {
+        return is_manager;
+    }
+
+    public void setIs_manager(boolean is_manager) {
+        this.is_manager = is_manager;
+    }
+
+    public boolean isValide() {
+        if (StringUtil.isNullOrEmpty(operator) || StringUtil.isNullOrEmpty(group_id) || members == null || members.isEmpty()) {
+            return false;
+        }
+        return true;
+    }
+
+    public WFCMessage.SetGroupManagerRequest toProtoGroupRequest() {
+        WFCMessage.SetGroupManagerRequest.Builder addGroupBuilder = WFCMessage.SetGroupManagerRequest.newBuilder();
+        addGroupBuilder.setGroupId(group_id);
+        addGroupBuilder.addAllUserId(members);
+        addGroupBuilder.setType(is_manager ? 1 : 0);
+
+        if (to_lines != null) {
+            for (Integer line : to_lines
+            ) {
+                addGroupBuilder.addToLine(line);
+            }
+        }
+
+        if (notify_message != null) {
+            addGroupBuilder.setNotifyContent(notify_message.toProtoMessageContent());
+        }
+        return addGroupBuilder.build();
+    }
+
+}

common/src/main/java/cn/wildfirechat/pojos/InputUserId.java

@@ -0,0 +1,21 @@
+/*
+ * This file is part of the Wildfire Chat package.
+ * (c) Heavyrain2012 <heavyrain.lee@gmail.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+package cn.wildfirechat.pojos;
+
+public class InputUserId {
+    private String userId;
+
+    public String getUserId() {
+        return userId;
+    }
+
+    public void setUserId(String userId) {
+        this.userId = userId;
+    }
+}

common/src/main/java/cn/wildfirechat/pojos/OutputGroupIds.java

@@ -0,0 +1,23 @@
+/*
+ * This file is part of the Wildfire Chat package.
+ * (c) Heavyrain2012 <heavyrain.lee@gmail.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+package cn.wildfirechat.pojos;
+
+import java.util.List;
+
+public class OutputGroupIds {
+    private List<String> groupIds;
+
+    public List<String> getGroupIds() {
+        return groupIds;
+    }
+
+    public void setGroupIds(List<String> groupIds) {
+        this.groupIds = groupIds;
+    }
+}