Compare commits

...
85 Commits
Author SHA1 Message Date
heavyrain.lee 6bfd175781 update 2020-02-19 14:04:44 +08:00
heavyrain.lee 1fc093a2ed 遗漏的配置 2020-02-19 13:58:09 +08:00
imhao183 969d865061 添加检查token接口 2020-02-18 09:57:42 +08:00
imhao183 6a6c59bdf2 修正发送消息给指定频道目标错误问题 2020-02-17 16:06:19 +08:00
imhao183 7adadd21a6 update 2020-02-13 15:09:52 +08:00
imhao183 21394985b3 增加日志的大小 2020-02-12 19:20:46 +08:00
heavyrain.lee b55694624d update log4j version 2020-02-03 12:24:07 +08:00
imhao183 cca9cafb7e 修改群头像时,通知消息里带上头像内容 2020-02-02 16:09:15 +08:00
imhao183 950334aad3 add user type parameter when create user 2020-01-22 21:08:22 +08:00
imhao183 ce578234f4 聊天室相关测试代码 2020-01-16 21:16:58 +08:00
imhao183 5daa4b1dc8 修正黑名单策略为1时发送消息仍失败的错误 2020-01-15 17:06:21 +08:00
imhao183 22a23a3f33 添加好友请求过期时间配置 2020-01-14 10:39:36 +08:00
heavyrain.lee 5d9196939f 修正聊天室错误 2020-01-11 11:06:17 +08:00
heavyrain.lee 2e4cf566c7 add exception log 2020-01-10 12:34:31 +08:00
heavyrain.lee a856b96b59 update 2020-01-10 11:37:07 +08:00
heavyrain.lee 9eb5fdd805 更新用户信息时,先写入数据库再存入缓存中,防止异步操作写入数据库失败造成数据不一致 2020-01-10 10:25:34 +08:00
heavyrain.lee bace7ec70c 加强对群组权限的检查 2020-01-08 23:20:20 +08:00
heavyrain.lee 0bcf2fea7b update version to 0.37 2020-01-08 14:56:36 +08:00
heavyrain.lee 60730c57cb update sdk test case 2020-01-08 14:49:21 +08:00
heavyrain.lee 3b61f8f6dd 添加频道获取自己信息的接口 2020-01-08 11:40:59 +08:00
heavyrain.lee e6ca9f5c78 更新channel接口,订阅/取消订阅只能逐个操作 2020-01-06 12:02:54 +08:00
heavyrain.lee 95beca8e05 fix build error 2020-01-05 23:47:57 +08:00
heavyrain.lee 7c4fe831b3 优化聊天室逻辑,添加聊天室空闲退出时间和发消息重新加入配置 2020-01-05 12:36:36 +08:00
heavyrain.lee e4da3780f5 添加频道sdk 2020-01-04 17:04:43 +08:00
heavyrain.lee 998f01a584 默认关掉设备在线状态回调 2020-01-04 14:51:46 +08:00
heavyrain.lee d71644c209 添加设置/获取备注api,添加用户在线状态回调功能 2020-01-04 14:47:27 +08:00
imhao183 70c390dea9 添加设置/获取备注api,添加用户在线状态回调功能 2020-01-04 14:45:49 +08:00
imhao183 41d326d40f 禁止私聊添加对机器人/管理员/物联网设备的例外 2020-01-04 14:35:34 +08:00
x86 1dbf431e4d 处理好友请求间隔 2020-01-02 10:15:25 +08:00
imhao183 0c0d712869 update 2019-12-30 16:00:14 +08:00
imhao183 73015c7a17 更新黑名单逻辑 2019-12-30 15:20:59 +08:00
imhao183 9baa1f9b23 修正编译错误,优化部分逻辑 2019-12-30 08:54:46 +08:00
imhao183 190da8934f 修改配置,增加如下功能
1,添加黑名单处理策略,发送失败或者忽略掉对方。
2,添加好友再次请求的时间间隔配置和好友请求被拒绝后再次请求的时间间隔配置
2019-12-30 08:48:17 +08:00
heavyrain.lee 9d1031594e 清除重复devicetoken时对包名进行判断,如果不同包名的client不清除重复推送token 2019-12-28 18:18:40 +08:00
imhao183 1d987c1569 remove useless files 2019-12-27 09:43:02 +08:00
imhao183 ac225b1541 添加退出群组api接口 2019-12-27 09:41:44 +08:00
imhao183 11375133ba 禁止发送消息给被封禁用户 2019-12-26 21:18:34 +08:00
imhao183 91b209d0fa 添加是否允许陌生人聊天的配置 2019-12-26 19:35:15 +08:00
imhao183 216860c573 update 2019-12-24 12:32:24 +08:00
imhao183 c7579877a1 修正sdk好友和频道接口错误 2019-12-24 11:43:47 +08:00
imhao183 6c5354d17b 优化session处理 2019-12-18 11:37:39 +08:00
imhao183 c98bd1fb9b update version to 0.36 2019-12-18 10:54:08 +08:00
imhao183 fe525e9483 优化用户session逻辑 2019-12-18 09:02:59 +08:00
imhao183 6acea9b140 创建群时同步操作 2019-12-17 14:39:49 +08:00
imhao183 4beb573052 add ignore 2019-12-17 14:36:49 +08:00
imhao183 220e010193 update version to 0.35 2019-12-13 13:02:27 +08:00
imhao183 865888824c Revert "Fix race condition during SensitiveFilter update"
This reverts commit ab7b20871a.
2019-12-13 12:58:07 +08:00
imhao183 583df110fd Revert "Fix potential GZIPOutputStream resource leak on exception"
This reverts commit 6c7d2ab3ad.
2019-12-13 12:57:59 +08:00
imhao183 f204bd693e Revert "fix build error"
This reverts commit 5cd2e7affe.
2019-12-13 12:57:46 +08:00
x86 b6be1ca045 fix pull message issue 2019-12-10 23:34:15 +08:00
imhao183 beb403a9f7 添加搜索类型 2019-12-06 10:36:26 +08:00
imhao183 00f1187200 添加群主进群时,修改群主的的状态问题 2019-12-05 15:42:20 +08:00
heavyrain.lee 7ee490cc1b update version to 0.34 2019-12-03 13:54:17 +08:00
imhao183 3dc0ae7a47 update 2019-12-03 08:24:15 +08:00
imhao183 994a8b5c4a 添加服务器为客户缓存消息数量的配置 2019-12-03 08:10:38 +08:00
imhao183 5cd2e7affe fix build error 2019-12-02 15:21:04 +08:00
heavyrain.lee f29ffe4cd2 Merge pull request #124 from k81/wildfirechat
Fix race condition during SensitiveFilter update and potential resource leak of GZIPOutputStream on exception
2019-12-02 14:43:55 +08:00
gitlab-runner 6c7d2ab3ad Fix potential GZIPOutputStream resource leak on exception 2019-11-30 23:05:28 +08:00
gitlab-runner ab7b20871a Fix race condition during SensitiveFilter update 2019-11-30 17:27:56 +08:00
imhao183 6883418e22 解决okhttp内存泄漏问题 2019-11-30 14:25:41 +08:00
fengyun 32f7266a5f add deleted property 2019-11-30 13:11:25 +08:00
fengyun 35db76e9cd set use deleted 2019-11-30 13:11:18 +08:00
heavyrain.lee 4f6aa87b9f update proto 2019-11-30 13:10:53 +08:00
fengyun f5816e20dd add get user group list and destroy user function 2019-11-30 11:46:52 +08:00
heavyrain2012 e72ae59e49 update version to 0.33 2019-11-28 17:33:35 +08:00
heavyrain2012 f0c2fd82ca 调整部分文件包名 2019-11-28 14:20:34 +08:00
heavyrain2012 5a588c8268 update 2019-11-27 12:57:07 +08:00
heavyrain2012 2adc030d6f 修改日志缺陷 2019-11-27 11:09:07 +08:00
heavyrain2012 b0ca8ed2b7 update 2019-11-27 10:39:32 +08:00
heavyrain2012 f9d5c861ce 优化用户session 2019-11-27 10:30:39 +08:00
heavyrain2012 9aaaa71ce9 清除掉重复token 2019-11-26 14:44:25 +08:00
heavyrain2012 73a971bde3 同步已读设置时,如果是排除掉已经解散的群组和销毁的频道 2019-11-25 18:50:00 +08:00
heavyrain2012 5942eb7b15 update version to 0.32 2019-11-25 12:12:06 +08:00
heavyrain2012 b97fea1841 添加索引 2019-11-25 11:22:36 +08:00
heavyrain2012 33637229d6 限制多端登陆 2019-11-25 11:19:53 +08:00
heavyrain2012 af366fcdb2 update proto 2019-11-25 11:19:44 +08:00
heavyrain2012 c9089c5d51 性能优化 2019-11-25 11:18:44 +08:00
imndx 3afc6a2da9 update readme 2019-11-24 20:37:45 +08:00
heavyrain2012 12d22e78bf 根据devicetoken进行排重 2019-11-21 20:51:52 +08:00
heavyrain2012 bcf969e56f 兼容旧客户端 2019-11-21 12:20:53 +08:00
heavyrain2012 b3a95c0310 update 2019-11-21 10:24:16 +08:00
heavyrain2012 ae863ab6e4 update version 2019-11-21 10:12:19 +08:00
heavyrain2012 688ca4e31d 黑名单功能 2019-11-21 09:36:41 +08:00
heavyrain2012 724b6dddd6 消息推送显示详情配置错误 2019-11-21 08:08:20 +08:00
heavyrain.lee bf3b30eb2c remove invalid message 2019-11-18 21:51:33 +08:00
151 changed files with 3242 additions and 478 deletions
+3
View File
@@ -56,3 +56,6 @@ broker/media
common/target
server/target
sdk/target
moments/target
monitor
wildfire_jmeter
+12 -10
View File
@@ -2,16 +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平台应用 | |
| [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)
+34 -1
View File
@@ -36,7 +36,7 @@ host 0.0.0.0
local_port 80
##是否使用内置DB。0使用mysql;1使用h2db;2使用mysql+mongodb。社区版只支持0和1,专业版支持2
embed.db 0
embed.db 1
##服务器管理接口密钥
http.admin.secret_key 123456
@@ -58,6 +58,31 @@ message.roaming 0
##是否开启拉取远程历史消息
message.remote_history_message 0
##服务器为每个用户缓存的消息数量。这个值改得太大,拉取消息时间变长,另外会占用大量内存。
message.max_queue 1024
##是否禁止陌生人聊天
message.disable_stranger_chat false
##黑名单处理策略,0 发送失败,返回被拉黑的错误码;1 发送成功但消息被服务器直接丢弃;
message.blacklist.strategy 0
##好友请求重复发送的间隔,单位是毫秒,默认是7天,0为不限制
friend.repeat_request_duration 604800000
##好友请求被拒绝后再次发送的间隔,单位是毫秒,默认是30天,0为不限制
friend.reject_request_duration 2592000000
##好友请求过期时间,单位是毫秒,默认是7天,0为不限制
friend.request_expiration_duration 604800000
## 聊天室观众空闲退出时间,单位为毫秒,默认为15分钟,0为永远不退出
chatroom.participant_idle_time 900000
## 用户向聊天室发送消息,自动加入聊天室
chatroom.rejoin_when_active true
#*********************************************************************
# hazelcase configuration
#*********************************************************************
@@ -102,6 +127,9 @@ qiniu.bucket_favorite_domain http://cdn2.wildfirechat.cn
# 本地媒体服务器配置。
local.media.storage.root media
# 是否支持多端登陆
server.multi_endpoint false
#*********************************************************************
# Push server configuration
#*********************************************************************
@@ -110,6 +138,11 @@ push.android.server.address http://localhost:8085/android/push
##苹果推送服务器地址
push.ios.server.address http://localhost:8085/ios/push
#*********************************************************************
# User online status event callback
#*********************************************************************
##用户在线状态事件回调地址
#user.online_status_callback http://localhost:8888/user/online_event
#*********************************************************************
# MongoDB configuration
+1
View File
@@ -0,0 +1 @@
alter table `t_user_session` ADD INDEX `session_token_index` ( `_token` );
+1
View File
@@ -0,0 +1 @@
alter table t_friend_request add index `friend_request_uid_index` (`_friend_uid`);
+1
View File
@@ -0,0 +1 @@
alter table `t_user_session` ADD INDEX `session_uid_index` ( `_uid` );
+1
View File
@@ -0,0 +1 @@
alter table `t_user_session` add column `_deleted` tinyint DEFAULT 0;
+1
View File
@@ -0,0 +1 @@
alter table `t_user` add column `_deleted` tinyint DEFAULT 0;
@@ -0,0 +1 @@
alter table `t_user_session` ADD INDEX `session_token_index` ( `_token` );
+1
View File
@@ -0,0 +1 @@
alter table t_friend_request add index `friend_request_uid_index` (`_friend_uid`);
+1
View File
@@ -0,0 +1 @@
alter table `t_user_session` ADD INDEX `session_uid_index` ( `_uid` );
+1
View File
@@ -0,0 +1 @@
alter table `t_user_session` add column `_deleted` tinyint DEFAULT 0;
+1
View File
@@ -0,0 +1 @@
alter table `t_user` add column `_deleted` tinyint DEFAULT 0;
+26 -5
View File
@@ -9,19 +9,40 @@ server {
root html;
index index.html index.htm index.php;
## send request back to im server ##
location /im {
location /route {
proxy_set_header Host $host;
proxy_set_header X-real-ip $remote_addr;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
proxy_pass http://imserver;
}
location /route {
#社区版需要启用下面一段,专业版不能用
location /im {
proxy_set_header Host $host;
proxy_set_header X-real-ip $remote_addr;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
proxy_pass http://imserver;
}
location /api {
proxy_pass http://imserver;
}
location / {
proxy_pass https://wildfirechat.github.io/docs/;
location /fs {
proxy_pass http://imserver;
}
location /robot {
proxy_set_header Host $host;
proxy_set_header X-real-ip $remote_addr;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
proxy_pass http://imserver;
}
location /channel {
proxy_set_header Host $host;
proxy_set_header X-real-ip $remote_addr;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
proxy_pass http://imserver;
}
}
+2 -2
View File
@@ -5,7 +5,7 @@
<relativePath>../</relativePath>
<artifactId>wildfirechat-parent</artifactId>
<groupId>cn.wildfirechat</groupId>
<version>0.30</version>
<version>0.37</version>
</parent>
<artifactId>moquette-broker</artifactId>
@@ -41,7 +41,7 @@
<dependency>
<groupId>log4j</groupId>
<artifactId>log4j</artifactId>
<version>1.2.14</version>
<version>1.2.17</version>
</dependency>
<!-- https://mvnrepository.com/artifact/mysql/mysql-connector-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);
@@ -21,6 +21,7 @@ import io.netty.handler.codec.http.HttpResponseEncoder;
import io.netty.handler.stream.ChunkedWriteHandler;
import com.xiaoleilu.loServer.action.ClassUtil;
import org.slf4j.LoggerFactory;
import win.liyufan.im.Utility;
import java.io.IOException;
@@ -141,6 +142,7 @@ public class LoServer {
}
} catch (IOException e) {
e.printStackTrace();
Utility.printExecption(Logger, e);
} catch (ClassNotFoundException e) {
e.printStackTrace();
}
@@ -0,0 +1,75 @@
/*
* 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;
import com.google.gson.Gson;
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.MemorySessionStore;
import io.moquette.spi.security.Tokenor;
import io.netty.handler.codec.http.FullHttpRequest;
import io.netty.handler.codec.http.HttpResponseStatus;
import io.netty.util.internal.StringUtil;
import win.liyufan.im.GitRepositoryState;
import win.liyufan.im.RateLimiter;
import win.liyufan.im.Utility;
import java.io.IOException;
@Route("/api/verify_token")
@HttpMethod("GET")
public class CheckTokenAction extends Action {
private final RateLimiter mLimitCounter = new RateLimiter(10, 1);
@Override
public boolean action(Request request, Response response) {
if (request.getNettyRequest() instanceof FullHttpRequest) {
response.setStatus(HttpResponseStatus.OK);
String userId = request.getParam("userId");
String clientId = request.getParam("clientId");
String token = request.getParam("token");
String result = "这是个检查token有效性的接口,如果客户端无法连接成功,可以使用这个接口检查token是否正确。\n";
result += "使用方法是在浏览器中输入http://imserverip/api/verify_token?userId=${userId}&clientId=${clientId}&token=${token}。\n";
result += "例如:http://localhost/api/verify_token?userId=123&clientId=456&token=789。\n";
result += "特别注意的是:必须使用正确的clientId,clientId必须是在测试手机上调用im接口获取。不用手机上获取到的clientId也都是不同的,一定不能用错!!!\n\n\n";
if (StringUtil.isNullOrEmpty(userId)) {
result += "错误,userId为空";
} else if (StringUtil.isNullOrEmpty(clientId)) {
result += "错误,clientId为空";
} else if (StringUtil.isNullOrEmpty(token)) {
result += "错误,token为空";
} else if(!mLimitCounter.isGranted("verify_token")) {
result += "接口请求超频!接口限制每10秒只能验证一次!";
} else {
MemorySessionStore.Session session = sessionsStore.getSession(clientId);
if (session == null) {
result += "错误,session不存在。请确认token是从本服务通过getToken接口获取的,另外请确认clientId是否正确";
} else if (session.getDeleted() == 1) {
result += "错误,session已经被清除,请确认当前客户没有多端登录或者主动退出";
} else if(!session.getUsername().equals(userId)) {
result += "错误,当前客户端的登录用户不是" + userId + "。这一般发生在当前clientId又为其它用户获取过token";
} else {
String id = Tokenor.getUserId(token.getBytes());
if (id == null) {
result += "错误,无效的token";
} else if(!id.equals(userId)) {
result += "错误,改token是用户" + id + "的";
} else {
result += "恭喜,您的信息是正确的";
}
}
}
response.setContent(result);
}
return true;
}
}
@@ -57,7 +57,7 @@ public class RouteAction extends Action {
try {
bytes = Base64.getDecoder().decode(str);
} catch (IllegalArgumentException e) {
sendResponse(response, ErrorCode.ERROR_CODE_SECRECT_KEY_MISMATCH, null, base64Response);
sendResponse(response, ErrorCode.ERROR_CODE_INVALID_DATA, null, base64Response);
return true;
}
@@ -65,7 +65,7 @@ public class RouteAction extends Action {
byte[] cbytes = Base64.getDecoder().decode(cid);
cbytes = AES.AESDecrypt(cbytes, "", true);
if (cbytes == null) {
sendResponse(response, ErrorCode.ERROR_CODE_SECRECT_KEY_MISMATCH, null, base64Response);
sendResponse(response, ErrorCode.ERROR_CODE_INVALID_DATA, null, base64Response);
return true;
}
cid = new String(cbytes);
@@ -74,7 +74,7 @@ public class RouteAction extends Action {
byte[] ubytes = Base64.getDecoder().decode(uid);
ubytes = AES.AESDecrypt(ubytes, "", true);
if (ubytes == null) {
sendResponse(response, ErrorCode.ERROR_CODE_SECRECT_KEY_MISMATCH, null, base64Response);
sendResponse(response, ErrorCode.ERROR_CODE_INVALID_DATA, null, base64Response);
return true;
}
uid = new String(ubytes);
@@ -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;
@@ -28,7 +28,7 @@ public class UnknownErrorAction extends Action{
@Override
public boolean action(Request request, Response response) {
response.setStatus(HttpResponseStatus.NOT_FOUND);
response.setStatus(HttpResponseStatus.INTERNAL_SERVER_ERROR);
return true;
}
}
@@ -9,6 +9,7 @@
package com.xiaoleilu.loServer.action.admin;
import com.google.gson.Gson;
import com.xiaoleilu.loServer.LoServer;
import com.xiaoleilu.loServer.RestResult;
import com.xiaoleilu.loServer.action.Action;
import com.xiaoleilu.loServer.handler.Request;
@@ -17,9 +18,12 @@ import io.netty.handler.codec.http.HttpResponseStatus;
import io.netty.util.internal.StringUtil;
import org.apache.commons.codec.digest.DigestUtils;
import cn.wildfirechat.common.ErrorCode;
import org.slf4j.LoggerFactory;
import win.liyufan.im.RateLimiter;
import win.liyufan.im.Utility;
abstract public class AdminAction extends Action {
private static final org.slf4j.Logger LOG = LoggerFactory.getLogger(AdminAction.class);
private static String SECRET_KEY = "123456";
private static boolean NO_CHECK_TIME = false;
private final RateLimiter mLimitCounter = new RateLimiter(10, 500);
@@ -50,8 +54,9 @@ abstract public class AdminAction extends Action {
Long ts;
try {
ts = Long.parseLong(timestamp);
} catch (NumberFormatException e) {
} catch (Exception e) {
e.printStackTrace();
Utility.printExecption(LOG, e);
return ErrorCode.INVALID_PARAMETER;
}
@@ -0,0 +1,58 @@
/*
* 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.pojos.InputGetAlias;
import cn.wildfirechat.pojos.InputGetFriendList;
import cn.wildfirechat.pojos.OutputGetAlias;
import cn.wildfirechat.pojos.OutputStringList;
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 com.xiaoleilu.loServer.model.FriendData;
import io.netty.handler.codec.http.FullHttpRequest;
import io.netty.handler.codec.http.HttpResponseStatus;
import java.util.ArrayList;
import java.util.List;
@Route(APIPath.Friend_Get_Alias)
@HttpMethod("POST")
public class AliasGetAction extends AdminAction {
@Override
public boolean isTransactionAction() {
return true;
}
@Override
public boolean action(Request request, Response response) {
if (request.getNettyRequest() instanceof FullHttpRequest) {
InputGetAlias input = getRequestBody(request.getNettyRequest(), InputGetAlias.class);
List<FriendData> dataList = messagesStore.getFriendList(input.getOperator(), null, 0);
List<String> list = new ArrayList<>();
OutputGetAlias out = new OutputGetAlias(input.getOperator(), input.getTargetId());
for (FriendData data : dataList) {
if (data.getFriendUid().equals(input.getTargetId())) {
out.setAlias(data.getAlias());
break;
}
}
response.setStatus(HttpResponseStatus.OK);
RestResult result = RestResult.ok(out);
response.setContent(new Gson().toJson(result));
}
return true;
}
}
@@ -0,0 +1,49 @@
/*
* 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.InputGetFriendList;
import cn.wildfirechat.pojos.InputUpdateAlias;
import cn.wildfirechat.pojos.OutputStringList;
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 com.xiaoleilu.loServer.model.FriendData;
import io.netty.handler.codec.http.FullHttpRequest;
import io.netty.handler.codec.http.HttpResponseStatus;
import java.util.ArrayList;
import java.util.List;
@Route(APIPath.Friend_Set_Alias)
@HttpMethod("POST")
public class AliasPutAction extends AdminAction {
@Override
public boolean isTransactionAction() {
return true;
}
@Override
public boolean action(Request request, Response response) {
if (request.getNettyRequest() instanceof FullHttpRequest) {
InputUpdateAlias input = getRequestBody(request.getNettyRequest(), InputUpdateAlias.class);
ErrorCode errorCode = messagesStore.setFriendAliasRequest(input.getOperator(), input.getTargetId(), input.getAlias(), new long[1]);
response.setStatus(HttpResponseStatus.OK);
RestResult result = RestResult.resultOf(errorCode);
response.setContent(new Gson().toJson(result));
}
return true;
}
}
@@ -44,7 +44,8 @@ public class CreateChannelAction extends AdminAction {
if (request.getNettyRequest() instanceof FullHttpRequest) {
InputCreateChannel inputCreateChannel = getRequestBody(request.getNettyRequest(), InputCreateChannel.class);
if (inputCreateChannel != null
&& !StringUtil.isNullOrEmpty(inputCreateChannel.getName())) {
&& !StringUtil.isNullOrEmpty(inputCreateChannel.getName())
&& !StringUtil.isNullOrEmpty(inputCreateChannel.getOwner())) {
if(StringUtil.isNullOrEmpty(inputCreateChannel.getTargetId())) {
@@ -58,7 +59,10 @@ public class CreateChannelAction extends AdminAction {
byteBuf.writeBytes(result);
ErrorCode errorCode = ErrorCode.fromCode(byteBuf.readByte());
if (errorCode == ErrorCode.ERROR_CODE_SUCCESS) {
sendResponse(response, null, new OutputCreateChannel(inputCreateChannel.getTargetId()));
byte[] data = new byte[byteBuf.readableBytes()];
byteBuf.readBytes(data);
String secret = new String(data);
sendResponse(response, null, new OutputCreateChannel(inputCreateChannel.getTargetId(), secret));
} else {
sendResponse(response, errorCode, null);
}
@@ -22,11 +22,15 @@ import io.netty.handler.codec.http.FullHttpRequest;
import io.netty.handler.codec.http.HttpResponseStatus;
import io.netty.util.internal.StringUtil;
import cn.wildfirechat.common.ErrorCode;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import win.liyufan.im.UUIDGenerator;
import win.liyufan.im.Utility;
@Route(APIPath.Create_Robot)
@HttpMethod("POST")
public class CreateRobotAction extends AdminAction {
private static final Logger LOG = LoggerFactory.getLogger(CreateRobotAction.class);
@Override
public boolean isTransactionAction() {
@@ -55,7 +59,16 @@ public class CreateRobotAction extends AdminAction {
WFCMessage.User newUser = inputCreateRobot.toUser();
messagesStore.addUserInfo(newUser, inputCreateRobot.getPassword());
try {
messagesStore.addUserInfo(newUser, inputCreateRobot.getPassword());
} catch (Exception e) {
e.printStackTrace();
Utility.printExecption(LOG, e);
response.setStatus(HttpResponseStatus.OK);
RestResult result = RestResult.resultOf(ErrorCode.ERROR_CODE_SERVER_ERROR, e.getMessage());
response.setContent(new Gson().toJson(result));
return true;
}
if (StringUtil.isNullOrEmpty(inputCreateRobot.getOwner())) {
inputCreateRobot.setOwner(inputCreateRobot.getUserId());
@@ -9,6 +9,7 @@
package com.xiaoleilu.loServer.action.admin;
import cn.wildfirechat.common.APIPath;
import cn.wildfirechat.proto.ProtoConstants;
import cn.wildfirechat.proto.WFCMessage;
import com.google.gson.Gson;
import com.xiaoleilu.loServer.RestResult;
@@ -22,11 +23,16 @@ import io.netty.handler.codec.http.FullHttpRequest;
import io.netty.handler.codec.http.HttpResponseStatus;
import io.netty.util.internal.StringUtil;
import cn.wildfirechat.common.ErrorCode;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import win.liyufan.im.DBUtil;
import win.liyufan.im.UUIDGenerator;
import win.liyufan.im.Utility;
@Route(APIPath.Create_User)
@HttpMethod("POST")
public class CreateUserAction extends AdminAction {
private static final Logger LOG = LoggerFactory.getLogger(CreateUserAction.class);
@Override
public boolean isTransactionAction() {
@@ -38,7 +44,8 @@ public class CreateUserAction extends AdminAction {
if (request.getNettyRequest() instanceof FullHttpRequest) {
InputOutputUserInfo inputCreateUser = getRequestBody(request.getNettyRequest(), InputOutputUserInfo.class);
if (inputCreateUser != null
&& !StringUtil.isNullOrEmpty(inputCreateUser.getName())) {
&& !StringUtil.isNullOrEmpty(inputCreateUser.getName())
&& (inputCreateUser.getType() == ProtoConstants.UserType.UserType_Normal || inputCreateUser.getType() == ProtoConstants.UserType.UserType_Admin || inputCreateUser.getType() == ProtoConstants.UserType.UserType_Super_Admin)) {
if(StringUtil.isNullOrEmpty(inputCreateUser.getPassword())) {
inputCreateUser.setPassword(UUIDGenerator.getUUID());
@@ -73,9 +80,20 @@ public class CreateUserAction extends AdminAction {
if (inputCreateUser.getExtra() != null)
newUserBuilder.setExtra(inputCreateUser.getExtra());
newUserBuilder.setType(inputCreateUser.getType());
newUserBuilder.setUpdateDt(System.currentTimeMillis());
messagesStore.addUserInfo(newUserBuilder.build(), inputCreateUser.getPassword());
try {
messagesStore.addUserInfo(newUserBuilder.build(), inputCreateUser.getPassword());
} catch (Exception e) {
e.printStackTrace();
Utility.printExecption(LOG, e);
response.setStatus(HttpResponseStatus.OK);
RestResult result = RestResult.resultOf(ErrorCode.ERROR_CODE_SERVER_ERROR, e.getMessage());
response.setContent(new Gson().toJson(result));
return true;
}
response.setStatus(HttpResponseStatus.OK);
RestResult result = RestResult.ok(new OutputCreateUser(inputCreateUser.getUserId(), inputCreateUser.getName()));
@@ -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;
}
}
@@ -9,6 +9,7 @@
package com.xiaoleilu.loServer.action.admin;
import cn.wildfirechat.common.APIPath;
import cn.wildfirechat.pojos.OutputStringList;
import com.google.gson.Gson;
import com.xiaoleilu.loServer.RestResult;
import com.xiaoleilu.loServer.annotation.HttpMethod;
@@ -44,7 +45,7 @@ public class FriendRelationGetAction extends AdminAction {
}
}
response.setStatus(HttpResponseStatus.OK);
RestResult result = RestResult.ok(list);
RestResult result = RestResult.ok(new OutputStringList(list));
response.setContent(new Gson().toJson(result));
}
return true;
@@ -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) {
@@ -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;
}
}
@@ -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.InputKickoffGroupMember;
import cn.wildfirechat.pojos.InputQuitGroup;
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_Member_Quit)
@HttpMethod("POST")
public class QuitGroupMemberAction extends AdminAction {
@Override
public boolean isTransactionAction() {
return true;
}
@Override
public boolean action(Request request, Response response) {
if (request.getNettyRequest() instanceof FullHttpRequest) {
InputQuitGroup inputQuitGroup = getRequestBody(request.getNettyRequest(), InputQuitGroup.class);
if (inputQuitGroup.isValide()) {
RPCCenter.getInstance().sendRequest(inputQuitGroup.getOperator(), null, IMTopic.QuitGroupTopic, inputQuitGroup.toProtoGroupRequest().toByteArray(), inputQuitGroup.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;
}
}
@@ -12,6 +12,7 @@ import cn.wildfirechat.proto.WFCMessage;
import com.google.gson.Gson;
import com.xiaoleilu.loServer.RestResult;
import com.xiaoleilu.loServer.action.Action;
import com.xiaoleilu.loServer.action.admin.AdminAction;
import com.xiaoleilu.loServer.handler.Request;
import com.xiaoleilu.loServer.handler.Response;
import io.moquette.spi.impl.Utils;
@@ -21,9 +22,12 @@ import io.netty.handler.codec.http.HttpResponseStatus;
import io.netty.util.internal.StringUtil;
import org.apache.commons.codec.digest.DigestUtils;
import cn.wildfirechat.common.ErrorCode;
import org.slf4j.LoggerFactory;
import win.liyufan.im.RateLimiter;
import win.liyufan.im.Utility;
abstract public class ChannelAction extends Action {
private static final org.slf4j.Logger LOG = LoggerFactory.getLogger(ChannelAction.class);
private final RateLimiter mLimitCounter = new RateLimiter(10, 200);
protected WFCMessage.ChannelInfo channelInfo;
@@ -45,8 +49,9 @@ abstract public class ChannelAction extends Action {
Long ts;
try {
ts = Long.parseLong(timestamp);
} catch (NumberFormatException e) {
} catch (Exception e) {
e.printStackTrace();
Utility.printExecption(LOG, e);
return ErrorCode.INVALID_PARAMETER;
}
@@ -10,6 +10,7 @@ package com.xiaoleilu.loServer.action.channel;
import cn.wildfirechat.common.APIPath;
import cn.wildfirechat.pojos.SendChannelMessageData;
import com.google.gson.Gson;
import com.xiaoleilu.loServer.RestResult;
import com.xiaoleilu.loServer.annotation.HttpMethod;
@@ -44,11 +45,15 @@ public class SendMessageAction extends ChannelAction {
@Override
public boolean action(Request request, Response response) {
if (request.getNettyRequest() instanceof FullHttpRequest) {
SendMessageData sendMessageData = getRequestBody(request.getNettyRequest(), SendMessageData.class);
SendChannelMessageData sendChannelMessageData = getRequestBody(request.getNettyRequest(), SendChannelMessageData.class);
SendMessageData sendMessageData = new SendMessageData();
sendMessageData.setConv(new Conversation());
sendMessageData.getConv().setType(ConversationType_Channel);
sendMessageData.getConv().setTarget(channelInfo.getTargetId());
sendMessageData.getConv().setLine(sendChannelMessageData.getLine());
sendMessageData.setSender(channelInfo.getOwner());
sendMessageData.setPayload(sendChannelMessageData.getPayload());
sendMessageData.setToUsers(sendChannelMessageData.getTargets());
if (SendMessageData.isValide(sendMessageData)) {
RPCCenter.getInstance().sendRequest(sendMessageData.getSender(), null, IMTopic.SendMessageTopic, sendMessageData.toProtoMessage().toByteArray(), sendMessageData.getSender(), TargetEntry.Type.TARGET_TYPE_USER, new RPCCenter.Callback() {
@Override
@@ -14,6 +14,7 @@ import com.xiaoleilu.loServer.RestResult;
import com.xiaoleilu.loServer.action.Action;
import com.xiaoleilu.loServer.handler.Request;
import com.xiaoleilu.loServer.handler.Response;
import io.moquette.imhandler.IMHandler;
import io.moquette.spi.impl.Utils;
import io.netty.handler.codec.http.FullHttpRequest;
import io.netty.handler.codec.http.HttpRequest;
@@ -21,11 +22,15 @@ import io.netty.handler.codec.http.HttpResponseStatus;
import io.netty.util.internal.StringUtil;
import org.apache.commons.codec.digest.DigestUtils;
import cn.wildfirechat.common.ErrorCode;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import win.liyufan.im.RateLimiter;
import win.liyufan.im.Utility;
import java.io.UnsupportedEncodingException;
abstract public class RobotAction extends Action {
protected static final Logger LOG = LoggerFactory.getLogger(RobotAction.class);
private final RateLimiter mLimitCounter = new RateLimiter(10, 1000);
protected WFCMessage.Robot robot;
@@ -47,8 +52,9 @@ abstract public class RobotAction extends Action {
Long ts;
try {
ts = Long.parseLong(timestamp);
} catch (NumberFormatException e) {
} catch (Exception e) {
e.printStackTrace();
Utility.printExecption(LOG, e);
return ErrorCode.INVALID_PARAMETER;
}
@@ -89,6 +95,7 @@ abstract public class RobotAction extends Action {
return t;
} catch (UnsupportedEncodingException e) {
e.printStackTrace();
Utility.printExecption(LOG, e);
}
}
return null;
@@ -21,6 +21,7 @@ import io.netty.handler.codec.http.FullHttpRequest;
import io.netty.handler.codec.http.HttpResponseStatus;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import win.liyufan.im.Utility;
/**
* Action处理单元
@@ -51,6 +52,8 @@ abstract public class ActionHandler extends SimpleChannelInboundHandler<FullHttp
this.doAction(ctx, request, response);
}
} catch (Exception e) {
e.printStackTrace();
Utility.printExecption(Logger, e);
Action errorAction = ServerSetting.getErrorAction(ServerSetting.MAPPING_ERROR);
request.putParam(UnknownErrorAction.ERROR_PARAM_NAME, e);
response.setContent(e.toString());
@@ -20,6 +20,7 @@ import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.http.FullHttpRequest;
import io.netty.util.internal.StringUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -37,7 +38,7 @@ public class IMActionHandler extends ActionHandler {
@Override
boolean isValidePath(String path) {
if (!path.startsWith("/admin")) {
if (!StringUtil.isNullOrEmpty(path) && !path.startsWith("/admin")) {
return true;
}
return false;
@@ -88,6 +88,8 @@ public final class BrokerConstants {
public static final String PUSH_ANDROID_SERVER_ADDRESS = "push.android.server.address";
public static final String PUSH_IOS_SERVER_ADDRESS = "push.ios.server.address";
public static final String USER_ONLINE_STATUS_CALLBACK = "user.online_status_callback";
public static final String HZ_Cluster_Node_External_IP = "node_external_ip";
public static final String HZ_Cluster_Node_External_Long_Port = "node_external_long_port";
public static final String HZ_Cluster_Node_External_Short_Port = "node_external_short_port";
@@ -106,6 +108,8 @@ 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";
@@ -113,6 +117,18 @@ public final class BrokerConstants {
public static final String MESSAGE_ROAMING = "message.roaming";
public static final String MESSAGE_Remote_History_Message = "message.remote_history_message";
public static final String MESSAGE_Max_Queue = "message.max_queue";
public static final String MESSAGE_Disable_Stranger_Chat = "message.disable_stranger_chat";
public static final String MESSAGE_Blacklist_Strategy = "message.blacklist.strategy";
public static final String FRIEND_Repeat_Request_Duration = "friend.repeat_request_duration";
public static final String FRIEND_Reject_Request_Duration = "friend.reject_request_duration";
public static final String FRIEND_Request_Expiration_Duration = "friend.request_expiration_duration";
public static final String CHATROOM_Participant_Idle_Time = "chatroom.participant_idle_time";
public static final String CHATROOM_Rejoin_When_Active = "chatroom.rejoin_when_active";
private BrokerConstants() {
}
@@ -41,7 +41,7 @@ public class CreateChannelHandler extends GroupHandler<WFCMessage.ChannelInfo> {
if (errorCode == ErrorCode.ERROR_CODE_SUCCESS) {
WFCMessage.ModifyUserSettingReq modifyUserSettingReq = WFCMessage.ModifyUserSettingReq.newBuilder().setScope(kUserSettingMyChannels).setKey(request.getTargetId()).setValue("1").build();
mServer.internalRpcMsg(fromUser, null, modifyUserSettingReq.toByteArray(), 0, fromUser, PutUserSettingTopic, false);
byte[] data = request.getTargetId().getBytes();
byte[] data = request.getSecret().getBytes();
ackPayload.ensureWritable(data.length).writeBytes(data);
return ErrorCode.ERROR_CODE_SUCCESS;
} else {
@@ -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;
}
}
}
@@ -26,7 +26,7 @@ public class FriendPullHandler extends IMHandler<WFCMessage.Version> {
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());
}
@@ -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.updateOrCreateUserSession(fromUser, clientID, request.getPlatform());
TokenAuthenticator authenticator = new TokenAuthenticator();
String strToken = authenticator.generateToken(fromUser);
String result = strToken + "|" + session.getSecret() + "|" + session.getDbSecret();
@@ -32,7 +32,7 @@ abstract public class GroupHandler<T> extends IMHandler<T> {
builder.setFromUser(fromUser);
long messageId = MessageShardingUtil.generateId();
builder.setMessageId(messageId);
saveAndPublish(fromUser, null, builder.build());
saveAndPublish(fromUser, null, builder.build(), false);
}
}
@@ -32,7 +32,7 @@ public class HandleFriendRequestHandler extends IMHandler<WFCMessage.HandleFrien
long timestamp = System.currentTimeMillis();
builder.setMessageId(messageId);
builder.setServerTimestamp(timestamp);
saveAndPublish(request.getTargetUid(), null, builder.build());
saveAndPublish(request.getTargetUid(), null, builder.build(), false);
WFCMessage.MessageContent.Builder contentBuilder = WFCMessage.MessageContent.newBuilder().setType(90).setContent("以上是打招呼信息");
builder = WFCMessage.Message.newBuilder();
@@ -44,7 +44,7 @@ public class HandleFriendRequestHandler extends IMHandler<WFCMessage.HandleFrien
messageId = MessageShardingUtil.generateId();
builder.setMessageId(messageId);
saveAndPublish(request.getTargetUid(), null, builder.build());
saveAndPublish(request.getTargetUid(), null, builder.build(), false);
contentBuilder.setContent("你们已经成为好友了,现在可以开始聊天了");
builder.setContent(contentBuilder);
@@ -52,7 +52,7 @@ public class HandleFriendRequestHandler extends IMHandler<WFCMessage.HandleFrien
builder.setMessageId(messageId);
timestamp = System.currentTimeMillis();
builder.setServerTimestamp(timestamp);
saveAndPublish(request.getTargetUid(), null, builder.build());
saveAndPublish(request.getTargetUid(), null, builder.build(), false);
publisher.publishNotification(IMTopic.NotifyFriendTopic, request.getTargetUid(), heads[0]);
publisher.publishNotification(IMTopic.NotifyFriendTopic, fromUser, heads[1]);
@@ -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;
}
@@ -239,18 +238,18 @@ abstract public class IMHandler<T> {
Set<String> notifyReceivers = new LinkedHashSet<>();
WFCMessage.Message.Builder messageBuilder = message.toBuilder();
int pullType = m_messagesStore.getNotifyReceivers(username, messageBuilder, notifyReceivers);
this.publisher.publish2Receivers(messageBuilder.build(), notifyReceivers, clientID, pullType);
int pullType = m_messagesStore.getNotifyReceivers(username, messageBuilder, notifyReceivers, false);
mServer.getImBusinessScheduler().execute(() -> this.publisher.publish2Receivers(messageBuilder.build(), notifyReceivers, clientID, pullType));
return notifyReceivers.size();
}
protected long saveAndPublish(String username, String clientID, WFCMessage.Message message) {
protected long saveAndPublish(String username, String clientID, WFCMessage.Message message, boolean ignoreMsg) {
Set<String> notifyReceivers = new LinkedHashSet<>();
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);
int pullType = m_messagesStore.getNotifyReceivers(username, messageBuilder, notifyReceivers, ignoreMsg);
mServer.getImBusinessScheduler().execute(() -> this.publisher.publish2Receivers(messageBuilder.build(), notifyReceivers, clientID, pullType));
return notifyReceivers.size();
}
@@ -8,6 +8,7 @@
package io.moquette.imhandler;
import cn.wildfirechat.proto.ProtoConstants;
import cn.wildfirechat.proto.WFCMessage;
import io.moquette.spi.impl.Qos1PublishHandler;
import io.netty.buffer.ByteBuf;
@@ -24,6 +25,13 @@ public class LoadRemoteMessagesHandler extends IMHandler<WFCMessage.LoadRemoteMe
if (beforeUid == 0) {
beforeUid = Long.MAX_VALUE;
}
if (request.getConversation().getType() == ProtoConstants.ConversationType.ConversationType_Group) {
if (!m_messagesStore.isMemberInGroup(fromUser, request.getConversation().getTarget())) {
return ErrorCode.ERROR_CODE_NOT_IN_GROUP;
}
}
WFCMessage.PullMessageResult result = m_messagesStore.loadRemoteMessages(fromUser, request.getConversation(), beforeUid, request.getCount());
byte[] data = result.toByteArray();
LOG.info("User {} load message with count({}), payload size({})", fromUser, result.getMessageCount(), data.length);
@@ -31,7 +31,7 @@ public class ModifyGroupInfoHandler extends GroupHandler<WFCMessage.ModifyGroupI
if (request.getType() == Modify_Group_Name) {
content = new GroupNotificationBinaryContent(request.getGroupId(), fromUser, request.getValue(), "").getChangeGroupNameNotifyContent();
} else if(request.getType() == Modify_Group_Portrait) {
content = new GroupNotificationBinaryContent(request.getGroupId(), fromUser, null, "").getChangeGroupPortraitNotifyContent();
content = new GroupNotificationBinaryContent(request.getGroupId(), fromUser, request.getValue(), "").getChangeGroupPortraitNotifyContent();
} else if(request.getType() == Modify_Group_Mute) {
content = new GroupNotificationBinaryContent(request.getGroupId(), fromUser, request.getValue(), "").getChangeGroupMuteNotifyContent();
} else if(request.getType() == Modify_Group_JoinType) {
@@ -13,11 +13,18 @@ import io.moquette.spi.impl.Qos1PublishHandler;
import io.netty.buffer.ByteBuf;
import cn.wildfirechat.common.ErrorCode;
import win.liyufan.im.IMTopic;
import win.liyufan.im.Utility;
@Handler(IMTopic.ModifyMyInfoTopic)
public class ModifyMyInfoHandler extends IMHandler<WFCMessage.ModifyMyInfoRequest> {
@Override
public ErrorCode action(ByteBuf ackPayload, String clientID, String fromUser, boolean isAdmin, WFCMessage.ModifyMyInfoRequest request, Qos1PublishHandler.IMCallback callback) {
try {
return m_messagesStore.modifyUserInfo(fromUser, request);
} catch (Exception e) {
e.printStackTrace();
Utility.printExecption(LOG, e);
return ErrorCode.ERROR_CODE_SERVER_ERROR;
}
}
}
@@ -37,7 +37,7 @@ public class RecallMessageHandler extends IMHandler<WFCMessage.INT64Buf> {
//等待客户端实现根据撤回消息更新内容,之后可以删掉这段代码
m_messagesStore.getNotifyReceivers(fromUser, message.toBuilder(), notifyReceivers);
m_messagesStore.getNotifyReceivers(fromUser, message.toBuilder(), notifyReceivers, false);
this.publisher.publishRecall2Receivers(int64Buf.getId(), fromUser, notifyReceivers, clientID);
return errorCode;
@@ -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);
}
@@ -17,6 +17,7 @@ import io.netty.buffer.ByteBuf;
import cn.wildfirechat.common.ErrorCode;
import win.liyufan.im.IMTopic;
import win.liyufan.im.MessageShardingUtil;
import win.liyufan.im.Utility;
import java.util.Set;
@@ -26,6 +27,7 @@ import static cn.wildfirechat.proto.ProtoConstants.ContentType.Text;
public class SendMessageHandler extends IMHandler<WFCMessage.Message> {
private int mSensitiveType = 0; //命中敏感词时,0 失败,1 吞掉, 2 敏感词替换成*。
private String mForwardUrl = null;
private int mBlacklistStrategy = 0; //黑名单中时,0失败,1吞掉。
public SendMessageHandler() {
super();
@@ -38,6 +40,14 @@ public class SendMessageHandler extends IMHandler<WFCMessage.Message> {
mSensitiveType = Integer.parseInt(mServer.getConfig().getProperty(BrokerConstants.SENSITIVE_Filter_Type));
} catch (Exception e) {
e.printStackTrace();
Utility.printExecption(LOG, e);
}
try {
mBlacklistStrategy = Integer.parseInt(mServer.getConfig().getProperty(BrokerConstants.MESSAGE_Blacklist_Strategy));
} catch (Exception e) {
e.printStackTrace();
Utility.printExecption(LOG, e);
}
}
@@ -45,23 +55,34 @@ public class SendMessageHandler extends IMHandler<WFCMessage.Message> {
public ErrorCode action(ByteBuf ackPayload, String clientID, String fromUser, boolean isAdmin, WFCMessage.Message message, Qos1PublishHandler.IMCallback callback) {
ErrorCode errorCode = ErrorCode.ERROR_CODE_SUCCESS;
if (message != null) {
boolean ignoreMsg = false;
if (!isAdmin) { //admin do not check the right
// 不能在端上直接发送撤回和群通知
if (message.getContent().getType() == 80 || (message.getContent().getType() >= 100 && message.getContent().getType() < 200)) {
return ErrorCode.INVALID_PARAMETER;
}
int userStatus = m_messagesStore.getUserStatus(fromUser);
if (userStatus == 1 || userStatus == 2) {
if (userStatus == ProtoConstants.UserStatus.Muted || userStatus == ProtoConstants.UserStatus.Forbidden) {
return ErrorCode.ERROR_CODE_FORBIDDEN_SEND_MSG;
}
if (message.getConversation().getType() == ProtoConstants.ConversationType.ConversationType_Private) {
if (m_messagesStore.isBlacked(message.getConversation().getTarget(), fromUser)) {
return ErrorCode.ERROR_CODE_IN_BLACK_LIST;
errorCode = m_messagesStore.isAllowUserMessage(message.getConversation().getTarget(), fromUser);
if (errorCode != ErrorCode.ERROR_CODE_SUCCESS) {
if (errorCode == ErrorCode.ERROR_CODE_IN_BLACK_LIST && mBlacklistStrategy != ProtoConstants.BlacklistStrategy.Message_Reject) {
ignoreMsg = true;
errorCode = ErrorCode.ERROR_CODE_SUCCESS;
} else {
return errorCode;
}
}
userStatus = m_messagesStore.getUserStatus(message.getConversation().getTarget());
if (userStatus == ProtoConstants.UserStatus.Forbidden) {
return ErrorCode.ERROR_CODE_USER_FORBIDDEN;
}
}
if (message.getConversation().getType() == ProtoConstants.ConversationType.ConversationType_Group ) {
errorCode = m_messagesStore.canSendMessageInGroup(fromUser, message.getConversation().getTarget());
if (errorCode != ErrorCode.ERROR_CODE_SUCCESS) {
@@ -86,7 +107,7 @@ public class SendMessageHandler extends IMHandler<WFCMessage.Message> {
publisher.forwardMessage(message, mForwardUrl);
}
boolean ignoreMsg = false;
if (!isAdmin && message.getContent().getType() == Text) {
Set<String> matched = m_messagesStore.handleSensitiveWord(message.getContent().getSearchableContent());
if (matched != null && !matched.isEmpty()) {
@@ -109,9 +130,7 @@ public class SendMessageHandler extends IMHandler<WFCMessage.Message> {
}
if (errorCode == ErrorCode.ERROR_CODE_SUCCESS) {
if (!ignoreMsg) {
saveAndPublish(fromUser, clientID, message);
}
saveAndPublish(fromUser, clientID, message, ignoreMsg);
ackPayload = ackPayload.capacity(20);
ackPayload.writeLong(messageId);
ackPayload.writeLong(timestamp);
@@ -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, session.getAppName());
}
return ErrorCode.ERROR_CODE_SUCCESS;
@@ -20,7 +20,7 @@ import java.util.List;
public class UserSearchHandler extends IMHandler<WFCMessage.SearchUserRequest> {
@Override
public ErrorCode action(ByteBuf ackPayload, String clientID, String fromUser, boolean isAdmin, WFCMessage.SearchUserRequest request, Qos1PublishHandler.IMCallback callback) {
List<WFCMessage.User> users = m_messagesStore.searchUser(request.getKeyword(), request.getFuzzy() > 0, request.getPage());
List<WFCMessage.User> users = m_messagesStore.searchUser(request.getKeyword(), request.getFuzzy(), request.getPage());
WFCMessage.SearchUserResult.Builder builder = WFCMessage.SearchUserResult.newBuilder();
builder.addAllEntry(users);
byte[] data = builder.build().toByteArray();
@@ -16,6 +16,7 @@ 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;
@@ -30,9 +31,11 @@ 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;
import static cn.wildfirechat.proto.ProtoConstants.SearchUserType.*;
public class DatabaseStore {
private static final Logger LOG = LoggerFactory.getLogger(DatabaseStore.class);
@@ -146,8 +149,12 @@ public class DatabaseStore {
}
return null;
}
List<WFCMessage.User> searchUserFromDB(String keyword, boolean buzzy, int page) {
//searchType
// 0 模糊匹配displayName, 精确匹配账户名和电话号码
// 1 精确匹配账户名和电话号码
// 2 精确匹配账户名
// 3 精确匹配电话号码
List<WFCMessage.User> searchUserFromDB(String keyword, int searchType, int page) {
Connection connection = null;
PreparedStatement statement = null;
ResultSet rs = null;
@@ -167,18 +174,29 @@ public class DatabaseStore {
", `_social`" +
", `_extra`" +
", `_dt` from t_user";
if (buzzy) {
sql += " where (`_display_name` like ? or `_name` = ? or `_mobile` = ?) ";
} else {
sql += " where (`_name` = ? or `_mobile` = ?) ";
switch (searchType) {
case SearchUserType_Name_Mobile:
sql += " where (`_name` = ? or `_mobile` = ?) ";
break;
case SearchUserType_Name:
sql += " where `_name` = ? ";
break;
case SearchUserType_Mobile:
sql += " where `_mobile` = ? ";
break;
case SearchUserType_General:
default:
sql += " where (`_display_name` like ? or `_name` = ? or `_mobile` = ?) ";
break;
}
sql += " and _type <> 2"; //can search normal user(0) and robot(1), can not search things
if (buzzy) {
sql += " limit 20";
} else {
sql += " and _type <> 2"; //can search normal user(0) and robot(1) and admin(100), can not search things
if (searchType == SearchUserType_Name_Mobile || searchType == SearchUserType_Name || searchType == SearchUserType_Mobile) {
sql += " limit 1";
} else {
sql += " limit 20";
}
if (page > 0) {
@@ -188,12 +206,24 @@ public class DatabaseStore {
statement = connection.prepareStatement(sql);
int index = 1;
if (buzzy) {
statement.setString(index++, "%" + keyword + "%");
switch (searchType) {
case SearchUserType_Name_Mobile:
statement.setString(index++, keyword);
statement.setString(index++, keyword);
break;
case SearchUserType_Name:
case SearchUserType_Mobile:
statement.setString(index++, keyword);
break;
case SearchUserType_General:
default:
statement.setString(index++, "%" + keyword + "%");
statement.setString(index++, keyword);
statement.setString(index++, keyword);
break;
}
statement.setString(index++, keyword);
statement.setString(index++, keyword);
rs = statement.executeQuery();
while (rs.next()) {
@@ -638,11 +668,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);
@@ -661,12 +692,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;
}
@@ -786,6 +816,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;
@@ -861,6 +925,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;
@@ -950,90 +1039,154 @@ public class DatabaseStore {
}
void persistGroupInfo(final WFCMessage.GroupInfo groupInfo) {
mScheduler.execute(()->{
Connection connection = null;
PreparedStatement statement = null;
try {
connection = DBUtil.getConnection();
String sql = "insert into t_group (`_gid`" +
", `_name`" +
", `_portrait`" +
", `_owner`" +
", `_type`" +
", `_extra`" +
", `_dt`" +
", `_member_count`" +
", `_mute`" +
", `_join_type`" +
", `_private_chat`" +
", `_searchable`" +
", `_member_dt`) values(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)" +
" ON DUPLICATE KEY UPDATE " +
"`_name` = ?," +
"`_portrait` = ?," +
"`_owner` = ?," +
"`_type` = ?," +
"`_extra` = ?," +
"`_dt` = ?," +
"`_member_count` = ?," +
"`_mute` = ?" +
", `_join_type` = ?" +
", `_private_chat` = ?" +
", `_searchable` = ?, " +
"`_member_dt` = ?";
Connection connection = null;
PreparedStatement statement = null;
try {
connection = DBUtil.getConnection();
String sql = "insert into t_group (`_gid`" +
", `_name`" +
", `_portrait`" +
", `_owner`" +
", `_type`" +
", `_extra`" +
", `_dt`" +
", `_member_count`" +
", `_mute`" +
", `_join_type`" +
", `_private_chat`" +
", `_searchable`" +
", `_member_dt`) values(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)" +
" ON DUPLICATE KEY UPDATE " +
"`_name` = ?," +
"`_portrait` = ?," +
"`_owner` = ?," +
"`_type` = ?," +
"`_extra` = ?," +
"`_dt` = ?," +
"`_member_count` = ?," +
"`_mute` = ?" +
", `_join_type` = ?" +
", `_private_chat` = ?" +
", `_searchable` = ?, " +
"`_member_dt` = ?";
statement = connection.prepareStatement(sql);
statement = connection.prepareStatement(sql);
int index = 1;
statement.setString(index++, groupInfo.getTargetId());
statement.setString(index++, groupInfo.getName());
statement.setString(index++, groupInfo.getPortrait());
statement.setString(index++, groupInfo.getOwner());
statement.setInt(index++, groupInfo.getType());
statement.setString(index++, groupInfo.getExtra());
statement.setLong(index++, groupInfo.getUpdateDt() == 0 ? System.currentTimeMillis() : groupInfo.getUpdateDt());
statement.setInt(index++, groupInfo.getMemberCount());
statement.setInt(index++, groupInfo.getMute());
statement.setInt(index++, groupInfo.getJoinType());
statement.setInt(index++, groupInfo.getPrivateChat());
statement.setInt(index++, groupInfo.getSearchable());
statement.setLong(index++, groupInfo.getMemberUpdateDt() == 0 ? System.currentTimeMillis() : groupInfo.getUpdateDt());
statement.setString(index++, groupInfo.getName());
statement.setString(index++, groupInfo.getPortrait());
statement.setString(index++, groupInfo.getOwner());
statement.setInt(index++, groupInfo.getType());
statement.setString(index++, groupInfo.getExtra());
statement.setLong(index++, groupInfo.getUpdateDt() == 0 ? System.currentTimeMillis() : groupInfo.getUpdateDt());
statement.setInt(index++, groupInfo.getMemberCount());
statement.setInt(index++, groupInfo.getMute());
statement.setInt(index++, groupInfo.getJoinType());
statement.setInt(index++, groupInfo.getPrivateChat());
statement.setInt(index++, groupInfo.getSearchable());
statement.setLong(index++, groupInfo.getMemberUpdateDt() == 0 ? System.currentTimeMillis() : groupInfo.getUpdateDt());
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<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;
statement.setString(index++, groupInfo.getTargetId());
statement.setString(index++, groupInfo.getName());
statement.setString(index++, groupInfo.getPortrait());
statement.setString(index++, groupInfo.getOwner());
statement.setInt(index++, groupInfo.getType());
statement.setString(index++, groupInfo.getExtra());
statement.setLong(index++, groupInfo.getUpdateDt() == 0 ? System.currentTimeMillis() : groupInfo.getUpdateDt());
statement.setInt(index++, groupInfo.getMemberCount());
statement.setInt(index++, groupInfo.getMute());
statement.setInt(index++, groupInfo.getJoinType());
statement.setInt(index++, groupInfo.getPrivateChat());
statement.setInt(index++, groupInfo.getSearchable());
statement.setLong(index++, groupInfo.getMemberUpdateDt() == 0 ? System.currentTimeMillis() : groupInfo.getUpdateDt());
String cid = resultSet.getString(index++);
statement.setString(index++, groupInfo.getName());
statement.setString(index++, groupInfo.getPortrait());
statement.setString(index++, groupInfo.getOwner());
statement.setInt(index++, groupInfo.getType());
statement.setString(index++, groupInfo.getExtra());
statement.setLong(index++, groupInfo.getUpdateDt() == 0 ? System.currentTimeMillis() : groupInfo.getUpdateDt());
statement.setInt(index++, groupInfo.getMemberCount());
statement.setInt(index++, groupInfo.getMute());
statement.setInt(index++, groupInfo.getJoinType());
statement.setInt(index++, groupInfo.getPrivateChat());
statement.setInt(index++, groupInfo.getSearchable());
statement.setLong(index++, groupInfo.getMemberUpdateDt() == 0 ? System.currentTimeMillis() : groupInfo.getUpdateDt());
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);
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` from t_user_session where `_uid` = ? and `_cid` = ? limit 1";
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);
@@ -1051,23 +1204,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();
@@ -1076,12 +1264,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);
@@ -1098,6 +1289,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;
@@ -1190,23 +1439,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());
@@ -1231,40 +1528,67 @@ public class DatabaseStore {
return null;
}
void updateGroupMemberCountDt(final String groupId, final int count, final long dt) {
mScheduler.execute(()->{
Connection connection = null;
PreparedStatement statement = null;
try {
connection = DBUtil.getConnection();
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);
String sql ;
statement.setString(1, userId);
statement.setInt(2, ProtoConstants.GroupMemberType.GroupMemberType_Removed);
if (count >= 0) {
sql = "update t_group set `_member_count` = ?, `_member_dt` = ? , `_dt` = ? where `_gid` = ?";
} else {
sql = "update t_group set `_member_dt` = ?, `_dt` = ? where `_gid` = ?";
}
statement = connection.prepareStatement(sql);
int index = 1;
if (count >=0) {
statement.setInt(index++, count);
}
statement.setLong(index++, dt);
statement.setLong(index++, dt);
statement.setString(index++, groupId);
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);
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) {
Connection connection = null;
PreparedStatement statement = null;
try {
connection = DBUtil.getConnection();
String sql ;
if (count >= 0) {
sql = "update t_group set `_member_count` = ?, `_member_dt` = ? , `_dt` = ? where `_gid` = ?";
} else {
sql = "update t_group set `_member_dt` = ?, `_dt` = ? where `_gid` = ?";
}
statement = connection.prepareStatement(sql);
int index = 1;
if (count >=0) {
statement.setInt(index++, count);
}
statement.setLong(index++, dt);
statement.setLong(index++, dt);
statement.setString(index++, groupId);
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 persistGroupMember(final String groupId, final List<WFCMessage.GroupMember> memberList) {
@@ -1804,7 +2128,7 @@ public class DatabaseStore {
});
}
void updateUser(final WFCMessage.User user) {
void updateUser(final WFCMessage.User user) throws Exception {
LOG.info("Database update user info {} {}", user.getUid(), user.getUpdateDt());
Connection connection = null;
PreparedStatement statement = null;
@@ -1835,6 +2159,7 @@ public class DatabaseStore {
", `_social`=?" +
", `_extra`=?" +
", `_type`=?" +
", `_deleted`=?" +
", `_dt`=?";
statement = connection.prepareStatement(sql);
@@ -1868,6 +2193,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();
@@ -1876,6 +2202,7 @@ public class DatabaseStore {
// TODO Auto-generated catch block
e.printStackTrace();
Utility.printExecption(LOG, e);
throw new Exception(e.getMessage());
} finally {
DBUtil.closeDB(connection, statement);
}
@@ -1994,6 +2321,7 @@ public class DatabaseStore {
", `_social`" +
", `_extra`" +
", `_type`" +
", `_deleted`" +
", `_dt` from t_user where `_uid` = ?";
statement = connection.prepareStatement(sql);
@@ -2048,6 +2376,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);
@@ -2115,7 +2446,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);
@@ -2173,19 +2504,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;
@@ -25,6 +25,8 @@ import com.hazelcast.util.StringUtil;
import com.xiaoleilu.loServer.model.FriendData;
import cn.wildfirechat.pojos.InputOutputUserBlockStatus;
import cn.wildfirechat.common.ErrorCode;
import io.moquette.BrokerConstants;
import io.moquette.server.Constants;
import io.moquette.server.Server;
import io.moquette.spi.IMatchingCondition;
import io.moquette.spi.IMessagesStore;
@@ -117,11 +119,63 @@ public class MemoryMessagesStore implements IMessagesStore {
private ConcurrentHashMap<String, Boolean> userPushHiddenDetail = new ConcurrentHashMap<>();
private ConcurrentHashMap<String, Boolean> userConvSlientMap = new ConcurrentHashMap<>();
private long mFriendRequestDuration = 7 * 24 * 60 * 60 * 1000;
private long mFriendRejectDuration = 30 * 24 * 60 * 60 * 1000;
private long mFriendRequestExpiration = 7 * 24 * 60 * 60 * 1000;
private boolean mDisableStrangerChat = false;
private long mChatroomParticipantIdleTime = 900000;
private boolean mChatroomRejoinWhenActive = true;
MemoryMessagesStore(Server server, DatabaseStore databaseStore) {
m_Server = server;
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));
try {
mDisableStrangerChat = Boolean.parseBoolean(m_Server.getConfig().getProperty(BrokerConstants.MESSAGE_Disable_Stranger_Chat));
} catch (Exception e) {
e.printStackTrace();
Utility.printExecption(LOG, e);
}
try {
mFriendRequestDuration = Long.parseLong(m_Server.getConfig().getProperty(FRIEND_Repeat_Request_Duration));
} catch (Exception e) {
e.printStackTrace();
Utility.printExecption(LOG, e);
}
try {
mFriendRejectDuration = Long.parseLong(m_Server.getConfig().getProperty(FRIEND_Reject_Request_Duration));
} catch (Exception e) {
e.printStackTrace();
Utility.printExecption(LOG, e);
}
try {
mFriendRequestExpiration = Long.parseLong(m_Server.getConfig().getProperty(FRIEND_Request_Expiration_Duration));
} catch (Exception e) {
e.printStackTrace();
Utility.printExecption(LOG, e);
}
try {
mChatroomRejoinWhenActive = Boolean.parseBoolean(m_Server.getConfig().getProperty(BrokerConstants.CHATROOM_Rejoin_When_Active));
} catch (Exception e) {
e.printStackTrace();
Utility.printExecption(LOG, e);
}
try {
mChatroomParticipantIdleTime = Long.parseLong(m_Server.getConfig().getProperty(CHATROOM_Participant_Idle_Time, "900000"));
} catch (Exception e) {
e.printStackTrace();
Utility.printExecption(LOG, e);
}
}
@Override
@@ -177,12 +231,24 @@ public class MemoryMessagesStore implements IMessagesStore {
@Override
public int getNotifyReceivers(String fromUser, WFCMessage.Message.Builder messageBuilder, Set<String> notifyReceivers) {
public int getNotifyReceivers(String fromUser, WFCMessage.Message.Builder messageBuilder, Set<String> notifyReceivers, boolean ignoreMsg) {
WFCMessage.Message message = messageBuilder.build();
HazelcastInstance hzInstance = m_Server.getHazelcastInstance();
int type = message.getConversation().getType();
int pullType = ProtoConstants.PullType.Pull_Normal;
if (ignoreMsg) {
if (type == ProtoConstants.ConversationType.ConversationType_ChatRoom) {
pullType = ProtoConstants.PullType.Pull_ChatRoom;
}
if (message.getContent().getPersistFlag() != Transparent) {
notifyReceivers.add(fromUser);
}
return pullType;
}
int pullType = 0;
if (type == ProtoConstants.ConversationType.ConversationType_Private) {
notifyReceivers.add(fromUser);
notifyReceivers.add(message.getConversation().getTarget());
@@ -192,8 +258,17 @@ public class MemoryMessagesStore implements IMessagesStore {
if (!StringUtil.isNullOrEmpty(message.getToUser())) {
notifyReceivers.add(message.getToUser());
} else if(!message.getToList().isEmpty()) {
notifyReceivers.addAll(message.getToList());
} else if(message.getToList()!=null && !message.getToList().isEmpty()) {
MultiMap<String, WFCMessage.GroupMember> groupMembers = hzInstance.getMultiMap(GROUP_MEMBERS);
Collection<WFCMessage.GroupMember> members = groupMembers.get(message.getConversation().getTarget());
if (members == null || members.size() == 0) {
members = loadGroupMemberFromDB(hzInstance, message.getConversation().getTarget());
}
for (WFCMessage.GroupMember member : members) {
if (member.getType() != GroupMemberType_Removed && message.getToList().contains(member.getMemberId())) {
notifyReceivers.add(member.getMemberId());
}
}
} else {
MultiMap<String, WFCMessage.GroupMember> groupMembers = hzInstance.getMultiMap(GROUP_MEMBERS);
Collection<WFCMessage.GroupMember> members = groupMembers.get(message.getConversation().getTarget());
@@ -230,14 +305,20 @@ public class MemoryMessagesStore implements IMessagesStore {
if (channelInfo != null) {
notifyReceivers.add(fromUser);
if (channelInfo.getOwner().equals(fromUser)) {
MultiMap<String, String> listeners = hzInstance.getMultiMap(CHANNEL_LISTENERS);
if (StringUtil.isNullOrEmpty(message.getToUser())) {
MultiMap<String, String> listeners = hzInstance.getMultiMap(CHANNEL_LISTENERS);
notifyReceivers.addAll(listeners.get(message.getConversation().getTarget()));
} else {
MultiMap<String, String> listeners = hzInstance.getMultiMap(CHANNEL_LISTENERS);
if (listeners.values().contains(message.getToUser())) {
notifyReceivers.add(message.getToUser());
}
} else if(message.getToList() != null && !message.getToList().isEmpty()) {
Collection<String> ls = listeners.get(message.getConversation().getTarget());
for (String to:message.getToList()) {
if (ls.contains(to)) {
notifyReceivers.add(to);
}
}
} else {
notifyReceivers.addAll(listeners.get(message.getConversation().getTarget()));
}
} else {
if (StringUtil.isNullOrEmpty(channelInfo.getCallback()) || channelInfo.getAutomatic() == 0) {
@@ -348,8 +429,17 @@ public class MemoryMessagesStore implements IMessagesStore {
continue;
}
if (bundle.getMessage().getContent().getExpireDuration() > 0) {
if (System.currentTimeMillis() < bundle.getMessage().getServerTimestamp() + bundle.getMessage().getContent().getExpireDuration()) {
continue;
}
}
size += bundle.getMessage().getSerializedSize();
if (size >= 512 * 1024) { //3M
if(builder.getMessageCount() == 0){
builder.addMessage(bundle.getMessage());
}
break;
}
builder.addMessage(bundle.getMessage());
@@ -528,13 +618,29 @@ public class MemoryMessagesStore implements IMessagesStore {
@Override
public boolean checkUserClientInChatroom(String user, String clientId, String chatroomId) {
MemorySessionStore.Session session = m_Server.getStore().sessionsStore().getSession(clientId);
if (session == null || System.currentTimeMillis() - session.getLastChatroomActiveTime() > 5 * 60 * 1000) {
handleQuitChatroom(user, clientId, chatroomId);
return false;
}
String existChatroomId = (String)m_Server.getHazelcastInstance().getMap(USER_CHATROOM).get(user);
if (StringUtil.isNullOrEmpty(existChatroomId) && (chatroomId == null || !existChatroomId.equals(chatroomId))) {
if (chatroomId == null) {
if (existChatroomId == null) {
return false;
} else {
chatroomId = existChatroomId;
}
}
if (StringUtil.isNullOrEmpty(existChatroomId) || !existChatroomId.equals(chatroomId)) {
if (mChatroomRejoinWhenActive) {
if (!StringUtil.isNullOrEmpty(existChatroomId)) {
handleQuitChatroom(user, clientId, existChatroomId);
}
handleJoinChatroom(user, clientId, chatroomId);
} else {
return false;
}
}
if (!mChatroomRejoinWhenActive && !checkChatroomParticipantIdelTime(session)) {
handleQuitChatroom(user, clientId, chatroomId);
return false;
}
@@ -684,7 +790,11 @@ public class MemoryMessagesStore implements IMessagesStore {
List<WFCMessage.GroupMember> tmp = new ArrayList<>();
ArrayList<String> newInviteUsers = new ArrayList<>();
for (WFCMessage.GroupMember member : memberList) {
member = member.toBuilder().setType(ProtoConstants.GroupMemberType.GroupMemberType_Normal).setUpdateDt(updateDt).setAlias("").build();
if (member.getMemberId().equals(groupInfo.getOwner())) {
member = member.toBuilder().setType(GroupMemberType_Owner).setUpdateDt(updateDt).setAlias("").build();
} else {
member = member.toBuilder().setType(GroupMemberType_Normal).setUpdateDt(updateDt).setAlias("").build();
}
tmp.add(member);
newInviteUsers.add(member.getMemberId());
}
@@ -772,7 +882,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 +915,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 +1117,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();
@@ -1144,8 +1295,9 @@ public class MemoryMessagesStore implements IMessagesStore {
if (members == null || members.size() == 0) {
members = loadGroupMemberFromDB(hzInstance, groupId);
}
for (WFCMessage.GroupMember member : members
) {
boolean isInGroup = false;
for (WFCMessage.GroupMember member : members) {
if (member.getMemberId().equals(memberId)) {
if (member.getType() == GroupMemberType_Silent) {
return ErrorCode.ERROR_CODE_NOT_RIGHT;
@@ -1158,11 +1310,16 @@ public class MemoryMessagesStore implements IMessagesStore {
if (member.getMemberId().equals(memberId) && member.getType() == GroupMemberType_Removed) {
return ErrorCode.ERROR_CODE_NOT_IN_GROUP;
}
isInGroup = true;
break;
}
}
if (!isInGroup) {
return ErrorCode.ERROR_CODE_NOT_IN_GROUP;
}
return ErrorCode.ERROR_CODE_SUCCESS;
}
@@ -1230,6 +1387,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();
@@ -1284,7 +1453,7 @@ public class MemoryMessagesStore implements IMessagesStore {
}
@Override
public ErrorCode modifyUserInfo(String userId, WFCMessage.ModifyMyInfoRequest request) {
public ErrorCode modifyUserInfo(String userId, WFCMessage.ModifyMyInfoRequest request) throws Exception {
HazelcastInstance hzInstance = m_Server.getHazelcastInstance();
IMap<String, WFCMessage.User> mUserMap = hzInstance.getMap(USERS);
@@ -1347,8 +1516,8 @@ public class MemoryMessagesStore implements IMessagesStore {
if (modified) {
builder.setUpdateDt(System.currentTimeMillis());
user = builder.build();
databaseStore.updateUser(user);
mUserMap.set(userId, user);
// databaseStore.updateUser(user);
return ErrorCode.ERROR_CODE_SUCCESS;
} else {
return ErrorCode.ERROR_CODE_NOT_MODIFIED;
@@ -1393,13 +1562,32 @@ public class MemoryMessagesStore implements IMessagesStore {
}
@Override
public void addUserInfo(WFCMessage.User user, String password) {
public void addUserInfo(WFCMessage.User user, String password) throws Exception {
HazelcastInstance hzInstance = m_Server.getHazelcastInstance();
IMap<String, WFCMessage.User> mUserMap = hzInstance.getMap(USERS);
databaseStore.updateUser(user);
mUserMap.put(user.getUid(), user);
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();
try {
databaseStore.updateUser(us);
} catch (Exception e) {
e.printStackTrace();
Utility.printExecption(LOG, e);
}
mUserMap.put(userId, us);
}
}
@Override
public WFCMessage.User getUserInfo(String userId) {
if (StringUtil.isNullOrEmpty(userId)) {
@@ -1424,8 +1612,8 @@ public class MemoryMessagesStore implements IMessagesStore {
return getUserInfo(userId);
}
@Override
public List<WFCMessage.User> searchUser(String keyword, boolean buzzy, int page) {
return databaseStore.searchUserFromDB(keyword, buzzy, page);
public List<WFCMessage.User> searchUser(String keyword, int searchType, int page) {
return databaseStore.searchUserFromDB(keyword, searchType, page);
}
@Override
@@ -1550,27 +1738,49 @@ public class MemoryMessagesStore implements IMessagesStore {
}
@Override
public boolean isBlacked(String fromUser, String userId) {
public ErrorCode isAllowUserMessage(String targetUser, String fromUser) {
HazelcastInstance hzInstance = m_Server.getHazelcastInstance();
MultiMap<String, FriendData> friendsMap = hzInstance.getMultiMap(USER_FRIENDS);
Collection<FriendData> friendDatas = friendsMap.get(fromUser);
Collection<FriendData> friendDatas = friendsMap.get(targetUser);
if (friendDatas == null || friendDatas.size() == 0) {
friendDatas = loadFriend(friendsMap, fromUser);
friendDatas = loadFriend(friendsMap, targetUser);
}
for (FriendData friendData : friendDatas) {
if (friendData.getFriendUid().equals(userId)) {
if (friendData.getState() == 2) {
return true;
if (friendData.getFriendUid().equals(fromUser)) {
if (friendData.getBlacked() == 1) {
return ErrorCode.ERROR_CODE_IN_BLACK_LIST;
} else {
return false;
if (friendData.getState() == 0) {
return ErrorCode.ERROR_CODE_SUCCESS;
} else {
break;
}
}
}
}
return false;
if (mDisableStrangerChat) {
//在禁止私聊时,允许机器人,物联网设备及管理员进行私聊。
IMap<String, WFCMessage.User> mUserMap = hzInstance.getMap(USERS);
WFCMessage.User target = mUserMap.get(targetUser);
if (target != null && target.getType() != ProtoConstants.UserType.UserType_Normal) {
return ErrorCode.ERROR_CODE_SUCCESS;
}
target = mUserMap.get(fromUser);
if (target != null && target.getType() != ProtoConstants.UserType.UserType_Normal) {
return ErrorCode.ERROR_CODE_SUCCESS;
}
//admin的usertype为3,所以不需要hardcode为admin添加例外
// if (targetUser.equals("admin") || fromUser.equals("admin")) {
// return ErrorCode.ERROR_CODE_SUCCESS;
// }
return ErrorCode.ERROR_CODE_NOT_RIGHT;
}
return ErrorCode.ERROR_CODE_SUCCESS;
}
@Override
@@ -1689,6 +1899,31 @@ public class MemoryMessagesStore implements IMessagesStore {
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<>();
@@ -1728,9 +1963,9 @@ public class MemoryMessagesStore implements IMessagesStore {
}
if (existRequest != null && existRequest.getStatus() != ProtoConstants.FriendRequestStatus.RequestStatus_Accepted) {
if (System.currentTimeMillis() - existRequest.getUpdateDt() > 7 * 24 * 60 * 60 * 1000) {
if (mFriendRequestDuration > 0 && System.currentTimeMillis() - existRequest.getUpdateDt() > mFriendRequestDuration) {
if (existRequest.getStatus() == ProtoConstants.FriendRequestStatus.RequestStatus_Rejected
&& System.currentTimeMillis() - existRequest.getUpdateDt() < 30 * 24 * 60 * 60 * 1000) {
&& System.currentTimeMillis() - existRequest.getUpdateDt() < mFriendRejectDuration) {
return ErrorCode.ERROR_CODE_FRIEND_REQUEST_BLOCKED;
}
} else {
@@ -1862,8 +2097,8 @@ public class MemoryMessagesStore implements IMessagesStore {
}
if (existRequest != null) {
if (System.currentTimeMillis() - existRequest.getUpdateDt() > 7 * 24 * 60 * 60 * 1000) {
return ErrorCode.ERROR_CODE_FRIEND_REQUEST_OVERTIME;
if (mFriendRequestExpiration > 0 && System.currentTimeMillis() - existRequest.getUpdateDt() > mFriendRequestExpiration) {
return ErrorCode.ERROR_CODE_FRIEND_REQUEST_EXPIRED;
} else {
existRequest = existRequest.toBuilder().setStatus(ProtoConstants.FriendRequestStatus.RequestStatus_Accepted).setUpdateDt(System.currentTimeMillis()).build();
databaseStore.persistOrUpdateFriendRequest(existRequest);
@@ -1896,7 +2131,10 @@ public class MemoryMessagesStore implements IMessagesStore {
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);
@@ -1992,6 +2230,18 @@ public class MemoryMessagesStore implements IMessagesStore {
return ErrorCode.ERROR_CODE_SUCCESS;
}
@Override
public boolean checkChatroomParticipantIdelTime(MemorySessionStore.Session session) {
if (session == null) {
return false;
}
if(mChatroomParticipantIdleTime > 0 && System.currentTimeMillis() - session.getLastChatroomActiveTime() > mChatroomParticipantIdleTime) {
return false;
}
return true;
}
@Override
public ErrorCode deleteFriend(String userId, String friendUid, long[] head) {
HazelcastInstance hzInstance = m_Server.getHazelcastInstance();
@@ -2037,6 +2287,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) {
@@ -2044,12 +2296,52 @@ 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();
Utility.printExecption(LOG, e);
}
}
}
if (!skip) {
builder.addEntry(entry);
}
}
}
}
@@ -2145,6 +2437,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);
@@ -2164,7 +2464,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 {
@@ -2202,6 +2502,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();
@@ -2236,8 +2542,9 @@ public class MemoryMessagesStore implements IMessagesStore {
else if(modifyType == Modify_Channel_OnlyCallback) {
try {
newInfoBuilder.setAutomatic(Integer.parseInt(value));
} catch (NumberFormatException e) {
} catch (Exception e) {
e.printStackTrace();
Utility.printExecption(LOG, e);
}
}
@@ -2487,7 +2794,7 @@ public class MemoryMessagesStore implements IMessagesStore {
}
return UUIDGenerator.getUUID();
}
@Override
public void storeRetained(Topic topic, StoredMessage storedMessage) {
LOG.debug("Store retained message for topic={}, CId={}", topic, storedMessage.getClientID());
@@ -16,17 +16,20 @@
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 win.liyufan.im.Utility;
import java.util.*;
import java.util.concurrent.*;
@@ -35,6 +38,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 +55,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 +145,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 +266,15 @@ 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();
Utility.printExecption(LOG, e);
}
}
@Override
@@ -261,6 +286,21 @@ public class MemorySessionStore implements ISessionsStore {
return session;
}
@Override
public void cleanDuplatedToken(String cid, int pushType, String token, boolean isVoip, String packageName) {
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)) && (!StringUtil.isNullOrEmpty(packageName) && packageName.equals(session.getAppName()))) {
session.deviceToken = null;
}
}
}
@Override
public void initStore() {
}
@@ -272,18 +312,91 @@ 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 updateOrCreateUserSession(String username, String clientID, int platform) {
LOG.debug("createUserSession for client <{}>, user <{}>", clientID, username);
Session session = sessions.get(clientID);
if (session != null && !session.username.equals(username)) {
if (userSessions.get(username) != null) {
userSessions.get(username).remove(clientID);
}
}
ClientSession clientSession = new ClientSession(clientID, this);
Session session = databaseStore.getSession(username, clientID, clientSession);
session = databaseStore.getSession(username, clientID, clientSession);
if (session == null) {
session = databaseStore.createSession(username, clientID, clientSession);
session = databaseStore.createSession(username, clientID, clientSession, platform);
}
sessions.put(clientID, session);
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 +404,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 +450,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 +496,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 +581,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 +659,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
@@ -19,7 +19,7 @@ import java.util.Map;
public class UserLoader implements MapStore<String, WFCMessage.User> {
@Override
public void store(String s, WFCMessage.User user) {
getDatabaseStore().updateUser(user);
// getDatabaseStore().updateUser(user);
}
@Override
@@ -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() {
}
@@ -103,7 +103,7 @@ public class NettyMQTTHandler extends ChannelInboundHandlerAdapter {
String clientID = NettyUtils.clientID(ctx.channel());
if (clientID != null && !clientID.isEmpty()) {
LOG.info("Notifying connection lost event. MqttClientId = {}.", clientID);
m_processor.processConnectionLost(clientID, ctx.channel());
m_processor.processConnectionLost(clientID, ctx.channel(), false);
}
ctx.close();
}
@@ -21,6 +21,7 @@ import cn.wildfirechat.proto.WFCMessage;
import com.xiaoleilu.loServer.model.FriendData;
import cn.wildfirechat.pojos.InputOutputUserBlockStatus;
import io.moquette.persistence.DatabaseStore;
import io.moquette.persistence.MemorySessionStore;
import io.moquette.persistence.UserClientEntry;
import io.moquette.spi.impl.subscriptions.Topic;
import io.netty.buffer.ByteBuf;
@@ -98,7 +99,7 @@ public interface IMessagesStore {
DatabaseStore getDatabaseStore();
WFCMessage.Message storeMessage(String fromUser, String fromClientId, WFCMessage.Message message);
void storeSensitiveMessage(WFCMessage.Message message);
int getNotifyReceivers(String fromUser, WFCMessage.Message.Builder message, Set<String> notifyReceivers);
int getNotifyReceivers(String fromUser, WFCMessage.Message.Builder message, Set<String> notifyReceivers, boolean ignoreMsg);
Set<String> getAllEnds();
WFCMessage.PullMessageResult fetchMessage(String user, String exceptClientId, long fromMessageId, int pullType);
WFCMessage.PullMessageResult loadRemoteMessages(String user, WFCMessage.Conversation conversation, long beforeUid, int count);
@@ -107,11 +108,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,20 +124,23 @@ 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);
ErrorCode modifyUserInfo(String userId, WFCMessage.ModifyMyInfoRequest request);
ErrorCode modifyUserInfo(String userId, WFCMessage.ModifyMyInfoRequest request) throws Exception;
ErrorCode modifyUserStatus(String userId, int status);
int getUserStatus(String userId);
List<InputOutputUserBlockStatus> getUserStatusList();
void addUserInfo(WFCMessage.User user, String password);
void addUserInfo(WFCMessage.User user, String password) throws Exception;
void destoryUser(String userId);
WFCMessage.User getUserInfo(String userId);
WFCMessage.User getUserInfoByName(String name);
WFCMessage.User getUserInfoByMobile(String mobile);
List<WFCMessage.User> searchUser(String keyword, boolean buzzy, int page);
List<WFCMessage.User> searchUser(String keyword, int searchType, int page);
boolean updateSystemSetting(int id, String value, String desc);
SystemSettingPojo getSystemSetting(int id);
@@ -154,6 +160,7 @@ public interface IMessagesStore {
ErrorCode login(String name, String password, List<String> userIdRet);
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);
@@ -161,22 +168,26 @@ public interface IMessagesStore {
ErrorCode deleteFriend(String userId, String friendUid, long[] head);
ErrorCode blackUserRequest(String fromUser, String targetUserId, int status, long[] head);
ErrorCode SyncFriendRequestUnread(String userId, long unreadDt, long[] head);
boolean isBlacked(String fromUser, String userId);
ErrorCode isAllowUserMessage(String fromUser, String userId);
ErrorCode setFriendAliasRequest(String fromUser, String targetUserId, String alias, long[] head);
ErrorCode handleJoinChatroom(String userId, String clientId, String chatroomId);
ErrorCode handleQuitChatroom(String userId, String clientId, String chatroomId);
boolean checkChatroomParticipantIdelTime(MemorySessionStore.Session session);
ErrorCode getUserSettings(String userId, long version, WFCMessage.GetUserSettingResult.Builder builder);
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);
@@ -38,15 +38,19 @@ public interface ISessionsStore {
*/
boolean contains(String clientID);
Session createUserSession(String username, String clientID);
Session updateOrCreateUserSession(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, String packageName);
void updateSessionToken(Session session, boolean voip);
void clearUserSession(String username);
/**
* @param clientID
* the client owning the session.
@@ -87,6 +87,7 @@ public class MessagesPublisher {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
Utility.printExecption(LOG, e);
}
chatroomScheduler.shutdown();
}
@@ -216,7 +217,7 @@ public class MessagesPublisher {
targetSession.refreshLastChatroomActiveTime();
}
if(System.currentTimeMillis() - targetSession.getLastChatroomActiveTime() > 5*60*1000) {
if (!m_messagesStore.checkChatroomParticipantIdelTime(targetSession)) {
m_messagesStore.handleQuitChatroom(user, targetSession.getClientID(), target);
continue;
}
@@ -16,7 +16,15 @@
package io.moquette.spi.impl;
import cn.wildfirechat.common.ErrorCode;
import cn.wildfirechat.pojos.SendMessageData;
import cn.wildfirechat.pojos.UserOnlineStatus;
import cn.wildfirechat.proto.ProtoConstants;
import cn.wildfirechat.proto.WFCMessage;
import com.google.gson.Gson;
import com.hazelcast.core.Member;
import com.hazelcast.util.StringUtil;
import io.moquette.BrokerConstants;
import io.moquette.persistence.RPCCenter;
import io.moquette.interception.InterceptHandler;
import io.moquette.interception.messages.InterceptAcknowledgedMessage;
@@ -38,9 +46,12 @@ import io.netty.handler.codec.mqtt.*;
import io.netty.handler.timeout.IdleStateHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import win.liyufan.im.HttpUtils;
import win.liyufan.im.Utility;
import java.util.*;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import static io.moquette.spi.impl.InternalRepublisher.createPublishForQos;
import static io.moquette.spi.impl.Utils.readBytesAndRewind;
@@ -55,7 +66,24 @@ import static io.moquette.server.ConnectionDescriptor.ConnectionState.*;
*
* Used by the front facing class ProtocolProcessorBootstrapper.
*/
public class ProtocolProcessor {
private static ExecutorService executorCallback = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());
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) {
@@ -135,14 +163,21 @@ public class ProtocolProcessor {
this.connectionDescriptors, this.messagesPublisher, sessionsStore, server.getImBusinessScheduler(), server);
mServer = server;
String onlineStatusCallback = server.getConfig().getProperty(BrokerConstants.USER_ONLINE_STATUS_CALLBACK);
if (!com.hazelcast.util.StringUtil.isNullOrEmpty(onlineStatusCallback)) {
mUserOnlineStatusCallback = onlineStatusCallback;
}
}
private String mUserOnlineStatusCallback;
public void processConnect(Channel channel, MqttConnectMessage msg) {
MqttConnectPayload payload = msg.payload();
String clientId = payload.clientIdentifier();
LOG.info("Processing CONNECT message. CId={}, username={}", clientId, payload.userName());
if (msg.variableHeader().version() <= MqttVersion.MQTT_3_1_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);
@@ -164,7 +199,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) {
@@ -210,11 +249,20 @@ public class ProtocolProcessor {
MemorySessionStore.Session session = m_sessionsStore.getSession(clientId);
if(session != null) {
session.refreshLastActiveTime();
forwardOnlineStatusEvent(payload.userName(), clientId, session.getPlatform(), UserOnlineStatus.ONLINE);
} else {
forwardOnlineStatusEvent(payload.userName(), clientId, ProtoConstants.Platform.Platform_UNSET, UserOnlineStatus.ONLINE);
}
LOG.info("The CONNECT message has been processed. CId={}, username={}", clientId, payload.userName());
}
public void forwardOnlineStatusEvent(String userId, String clientId, int platform, int status) {
if (!StringUtil.isNullOrEmpty(mUserOnlineStatusCallback)) {
executorCallback.execute(() -> HttpUtils.httpJsonPost(mUserOnlineStatusCallback, new Gson().toJson(new UserOnlineStatus(userId, clientId, platform, status))));
}
}
private MqttConnAckMessage connAck(MqttConnectReturnCode returnCode) {
return connAck(returnCode, false);
}
@@ -243,7 +291,7 @@ public class ProtocolProcessor {
// handle user authentication
if (msg.variableHeader().hasUserName()) {
int status = m_messagesStore.getUserStatus(msg.payload().userName());
if (status == 2) {
if (status == ProtoConstants.UserStatus.Forbidden) {
failedBlocked(channel);
return false;
}
@@ -253,9 +301,17 @@ 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) {
LOG.error("user {} session {} is deleted. login failure", msg.payload().userName(), clientId);
return false;
}
if (session != null && session.getUsername().equals(msg.payload().userName())) {
pwd = AES.AESDecrypt(pwd, session.getSecret(), true);
@@ -464,7 +520,7 @@ public class ProtocolProcessor {
}
if (!clearSession) {
processConnectionLost(clientID, channel);
processConnectionLost(clientID, channel, clearSession);
return;
}
@@ -501,16 +557,20 @@ 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);
String username = NettyUtils.userName(channel);
MemorySessionStore.Session session = m_sessionsStore.getSession(clientID);
if(session != null) {
forwardOnlineStatusEvent(username, clientID, session.getPlatform(), UserOnlineStatus.LOGOUT);
}
channel.closeFuture();
//disconnect the session
m_sessionsStore.sessionForClient(clientID).disconnect(clearSession);
}
@@ -541,17 +601,21 @@ public class ProtocolProcessor {
return true;
}
public void processConnectionLost(String clientID, Channel channel) {
public void processConnectionLost(String clientID, Channel channel, boolean clearSession) {
LOG.info("Processing connection lost event. CId={}", clientID);
String username = NettyUtils.userName(channel);
MemorySessionStore.Session session = m_sessionsStore.getSession(clientID);
if(session != null) {
session.refreshLastActiveTime();
forwardOnlineStatusEvent(username, clientID, session.getPlatform(), clearSession ? UserOnlineStatus.LOGOUT : UserOnlineStatus.OFFLINE);
}
ConnectionDescriptor oldConnDescr = new ConnectionDescriptor(clientID, channel);
if(connectionDescriptors.removeConnection(oldConnDescr)) {
MemorySessionStore.Session session = m_sessionsStore.getSession(clientID);
if(session != null) {
session.refreshLastActiveTime();
}
String username = NettyUtils.userName(channel);
m_interceptor.notifyClientConnectionLost(clientID, username);
}
}
/**
@@ -187,6 +187,7 @@ public class Qos1PublishHandler extends QosPublishHandler {
gzip.close();
} catch (Exception e) {
e.printStackTrace();
Utility.printExecption(LOG, e);
}
data = out.toByteArray();
code = (byte)ErrorCode.ERROR_CODE_SUCCESS_GZIPED.code;
@@ -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();
}
}
});
}
@@ -71,6 +71,7 @@ public interface IMTopic {
String ChannelPullTopic = "CHP";
String GetTokenTopic = "GETTOKEN";
String DestroyUserTopic = "DESTROYUSER";
String LoadRemoteMessagesTopic = "LRM";
}
+1 -1
View File
@@ -22,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 -->
+1 -1
View File
@@ -5,7 +5,7 @@
<parent>
<artifactId>wildfirechat-parent</artifactId>
<groupId>cn.wildfirechat</groupId>
<version>0.30</version>
<version>0.37</version>
</parent>
<modelVersion>4.0.0</modelVersion>
@@ -0,0 +1,22 @@
/*
* 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 cn.wildfirechat.pojos;
public class InputDestroyUser {
private String userId;
public String getUserId() {
return userId;
}
public void setUserId(String userId) {
this.userId = userId;
}
}
@@ -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";
@@ -22,6 +23,8 @@ public interface APIPath {
String Friend_Update_Status = "/admin/friend/status";
String Friend_Get_List = "/admin/friend/list";
String Friend_Get_Alias = "/admin/friend/get_alias";
String Friend_Set_Alias = "/admin/friend/set_alias";
String Msg_Send = "/admin/message/send";
String Msg_Recall = "/admin/message/recall";
@@ -36,15 +39,21 @@ 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_Member_Quit = "/admin/group/member/quit";
String Group_Set_Manager = "/admin/group/manager/set";
String Get_User_Groups = "/admin/group/of_user";
String Create_Channel = "/admin/channel/create";
String Get_System_Setting = "/admin/system/get_setting";
String Put_System_Setting = "/admin/system/put_setting";
String Channel_User_Info = "/channel/user_info";
String Channel_Update_Profile = "/channel/update_profile";
String Channel_Get_Profile = "/channel/get_profile";
String Channel_Message_Send = "/channel/message/send";
String Channel_Subscribe = "/channel/subscribe";
String Channel_Subscriber_List = "/channel/subscriber_list";
String Robot_User_Info = "/robot/user_info";
String Robot_Message_Send = "/robot/message/send";
@@ -36,7 +36,7 @@ public enum ErrorCode {
//user error
ERROR_CODE_FRIEND_ALREADY_REQUEST(16, "already send request"),
ERROR_CODE_FRIEND_REQUEST_BLOCKED(18, "friend request blocked"),
ERROR_CODE_FRIEND_REQUEST_OVERTIME(19, "friend request overtime"),
ERROR_CODE_FRIEND_REQUEST_EXPIRED(19, "friend request expired"),
ERROR_CODE_NOT_IN_CHATROOM(20, "not in chatroom"),
@@ -0,0 +1,36 @@
/*
* 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 cn.wildfirechat.pojos;
import cn.wildfirechat.proto.ProtoConstants;
import cn.wildfirechat.proto.WFCMessage;
import java.util.ArrayList;
import java.util.List;
public class InputChannelSubscribe {
private String target;
private int subscribe; //1,订阅;0,取消订阅
public String getTarget() {
return target;
}
public void setTarget(String target) {
this.target = target;
}
public int getSubscribe() {
return subscribe;
}
public void setSubscribe(int subscribe) {
this.subscribe = subscribe;
}
}
@@ -9,15 +9,10 @@
package cn.wildfirechat.pojos;
import cn.wildfirechat.proto.WFCMessage;
public class OutputCreateChannel {
public class InputGetAlias {
private String operator;
private String targetId;
public OutputCreateChannel(String targetId) {
this.targetId = targetId;
}
public String getTargetId() {
return targetId;
}
@@ -25,4 +20,12 @@ public class OutputCreateChannel {
public void setTargetId(String targetId) {
this.targetId = targetId;
}
public String getOperator() {
return operator;
}
public void setOperator(String operator) {
this.operator = operator;
}
}
@@ -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() {
@@ -0,0 +1,36 @@
/*
* 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 cn.wildfirechat.pojos;
import cn.wildfirechat.proto.WFCMessage;
import io.netty.util.internal.StringUtil;
import java.util.Arrays;
public class InputModifyChannelInfo {
private int type;
private String value;
public int getType() {
return type;
}
public void setType(int type) {
this.type = type;
}
public String getValue() {
return value;
}
public void setValue(String value) {
this.value = value;
}
}
@@ -24,6 +24,7 @@ public class InputOutputUserInfo {
private String company;
private String social;
private String extra;
private int type;
private long updateDt;
public static InputOutputUserInfo fromPbUser(WFCMessage.User pbUser) {
@@ -39,6 +40,7 @@ public class InputOutputUserInfo {
inputCreateUser.company = pbUser.getCompany();
inputCreateUser.social = pbUser.getSocial();
inputCreateUser.extra = pbUser.getExtra();
inputCreateUser.type = pbUser.getType();
inputCreateUser.updateDt = pbUser.getUpdateDt();
return inputCreateUser;
}
@@ -74,6 +76,7 @@ public class InputOutputUserInfo {
if (getExtra() != null)
newUserBuilder.setExtra(getExtra());
newUserBuilder.setGender(gender);
newUserBuilder.setType(type);
newUserBuilder.setUpdateDt(System.currentTimeMillis());
return newUserBuilder.build();
@@ -174,4 +177,12 @@ public class InputOutputUserInfo {
public void setGender(int gender) {
this.gender = gender;
}
public int getType() {
return type;
}
public void setType(int type) {
this.type = type;
}
}
@@ -0,0 +1,48 @@
/*
* 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 cn.wildfirechat.pojos;
import cn.wildfirechat.proto.WFCMessage;
import java.util.List;
public class InputQuitGroup extends InputGroupBase {
private String group_id;
public String getGroup_id() {
return group_id;
}
public void setGroup_id(String group_id) {
this.group_id = group_id;
}
public boolean isValide() {
return true;
}
public WFCMessage.QuitGroupRequest toProtoGroupRequest() {
WFCMessage.QuitGroupRequest.Builder removedGroupBuilder = WFCMessage.QuitGroupRequest.newBuilder();
removedGroupBuilder.setGroupId(group_id);
if (to_lines != null) {
for (Integer line : to_lines
) {
removedGroupBuilder.addToLine(line);
}
}
if (notify_message != null) {
removedGroupBuilder.setNotifyContent(notify_message.toProtoMessageContent());
}
return removedGroupBuilder.build();
}
}

Some files were not shown because too many files have changed in this diff Show More