Skip to content

W
wecloud_im_server
Commit c307c58c by giaogiao
Options

im

parent 0ba3859c
Showing with 2347 additions and 185 deletions
api-app/.gitignore deleted 100755 → 0
/target/
/classes
!.mvn/wrapper/maven-wrapper.jar
### STS ###
.apt_generated
.classpath
.factorypath
.project
.settings
.springBeans
.sts4-cache
### IntelliJ IDEA ###
.idea
*.iws
*.iml
*.ipr
### NetBeans ###
/nbproject/private/
/build/
/nbbuild/
/dist/
/nbdist/
/.nb-gradle/
.DS_Store
*.log
logs
*.rdb
api-app/README.md deleted 100755 → 0
# system 项目系统模块
\ No newline at end of file
api-app/pom.xml deleted 100755 → 0
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns="http://maven.apache.org/POM/4.0.0"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>io.geekidea.springbootplus</groupId>
<artifactId>parent</artifactId>
<version>2.0</version>
</parent>
<artifactId>api-app</artifactId>
<name>api-app</name>
<description>app的api服务模块</description>
<dependencies>
<dependency>
<groupId>io.geekidea.springbootplus</groupId>
<artifactId>framework</artifactId>
</dependency>
<dependency>
<groupId>io.geekidea.springbootplus</groupId>
<artifactId>common</artifactId>
</dependency>
</dependencies>
</project>
\ No newline at end of file
api-app/src/main/resources/rebel.xml deleted 100644 → 0
<?xml version="1.0" encoding="UTF-8"?>
<!--
This is the JRebel configuration file. It maps the running application to your IDE workspace, enabling JRebel reloading for this project.
Refer to https://manuals.jrebel.com/jrebel/standalone/config.html for more information.
-->
<application xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" generated-by="intellij"
xmlns="http://www.zeroturnaround.com"
xsi:schemaLocation="http://www.zeroturnaround.com http://update.zeroturnaround.com/jrebel/rebel-2_1.xsd">
<classpath>
<dir name="/Users/giaogiao/Documents/hewei/code/gitPull/spring-boot-plus-master/service/target/classes">
</dir>
</classpath>
</application>
bootstrap/pom.xml
...@@ -15,31 +15,22 @@ ...@@ -15,31 +15,22 @@
<description>项目启动模块</description> <description>项目启动模块</description>
<dependencies> <dependencies>
<!-- <dependency>-->
<!-- <groupId>de.codecentric</groupId>-->
<!-- <artifactId>spring-boot-admin-starter-client</artifactId>-->
<!-- <version>${spring-boot-admin.version}</version>-->
<!-- </dependency>-->
<dependency> <dependency>
<groupId>io.geekidea.springbootplus</groupId> <groupId>io.geekidea.springbootplus</groupId>
<artifactId>framework</artifactId> <artifactId>framework</artifactId>
</dependency> </dependency>
<!-- <dependency>--> <!-- <dependency>-->
<!-- <groupId>io.geekidea.springbootplus</groupId>--> <!-- <groupId>io.geekidea.springbootplus</groupId>-->
<!-- <artifactId>api-merchant</artifactId>--> <!-- <artifactId>api-app</artifactId>-->
<!-- </dependency>--> <!-- </dependency>-->
<dependency> <dependency>
<groupId>io.geekidea.springbootplus</groupId> <groupId>io.geekidea.springbootplus</groupId>
<artifactId>api-app</artifactId> <artifactId>common</artifactId>
</dependency> </dependency>
<!-- <dependency>-->
<!-- <groupId>io.geekidea.springbootplus</groupId>-->
<!-- <artifactId>api-system</artifactId>-->
<!-- </dependency>-->
<!-- <dependency>-->
<!-- <groupId>io.geekidea.springbootplus</groupId>-->
<!-- <artifactId>scheduled</artifactId>-->
<!-- </dependency>-->
</dependencies> </dependencies>
<build> <build>
......
bootstrap/src/main/java/io/geekidea/springbootplus/StartNettyService.java 0 → 100644
package io.geekidea.springbootplus;
import com.sien.common.tillo.netty.core.NettyStart;
import org.springframework.boot.CommandLineRunner;
import org.springframework.stereotype.Component;
import javax.annotation.Resource;
@Component
public class StartNettyService implements CommandLineRunner {
// @Value("${netty.port}")
// private int port;
@Resource
private NettyStart nettyServer;
@Override
public void run(String... args) throws Exception {
// nettyServer.run(port);
nettyServer.run(8899);
}
}
common/pom.xml
...@@ -22,6 +22,46 @@ ...@@ -22,6 +22,46 @@
</dependency> </dependency>
<!-- netty-->
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-codec</artifactId>
<version>4.1.25.Final</version>
</dependency>
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-transport</artifactId>
<version>4.1.25.Final</version>
</dependency>
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-transport-native-epoll</artifactId>
<version>4.1.25.Final</version>
<classifier>linux-x86_64</classifier>
</dependency>
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-codec-http</artifactId>
<version>4.1.25.Final</version>
</dependency>
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-handler</artifactId>
<version>4.1.25.Final</version>
</dependency>
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-transport-udt</artifactId>
<version>4.1.25.Final</version>
</dependency>
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-transport-sctp</artifactId>
<version>4.1.25.Final</version>
</dependency>
<!-- netty-->
<!-- 公众号（包括订阅号和服务号）：weixin-java-mp --> <!-- 公众号（包括订阅号和服务号）：weixin-java-mp -->
<dependency> <dependency>
<groupId>com.github.binarywang</groupId> <groupId>com.github.binarywang</groupId>
......
common/src/main/java/com/sien/common/controller/AppHelloWorldController.java
//package com.sien.common.controller; package com.sien.common.controller;
//
//import io.geekidea.springbootplus.framework.common.api.ApiResult; import io.geekidea.springbootplus.framework.common.api.ApiResult;
//import io.geekidea.springbootplus.framework.log.annotation.OperationLog; import io.geekidea.springbootplus.framework.log.annotation.OperationLog;
//import io.swagger.annotations.Api; import io.swagger.annotations.Api;
//import io.swagger.annotations.ApiOperation; import io.swagger.annotations.ApiOperation;
//import lombok.extern.slf4j.Slf4j; import lombok.extern.slf4j.Slf4j;
//import org.apache.shiro.authz.annotation.RequiresRoles; import org.apache.shiro.authz.annotation.RequiresRoles;
//import org.springframework.web.bind.annotation.GetMapping; import org.springframework.web.bind.annotation.GetMapping;
//import org.springframework.web.bind.annotation.RequestMapping; import org.springframework.web.bind.annotation.RequestMapping;
//import org.springframework.web.bind.annotation.RestController; import org.springframework.web.bind.annotation.RestController;
//
//import java.io.IOException; import java.io.IOException;
//
///** /**
// * *
// * Hello World Controller * Hello World Controller
// **/ **/
//@Slf4j @Slf4j
//@Api(value = "权限测试", tags = {"APP Hello World2"}) @Api(value = "权限测试", tags = {"APP Hello World2"})
//@RestController @RestController
//@RequestMapping("/app") @RequestMapping("/app")
//public class AppHelloWorldController { public class AppHelloWorldController {
//
// /** /**
// * Hello World * Hello World
// * *
// * @return * @return
// * @throws IOException * @throws IOException
// */ */
// @GetMapping(value = "/world") @GetMapping(value = "/world")
// @OperationLog(name = "helloWorld") @OperationLog(name = "helloWorld")
// @ApiOperation(value = "helloWorld", response = String.class) @ApiOperation(value = "helloWorld", response = String.class)
// public ApiResult<String> helloWorld() throws IOException { public ApiResult<String> helloWorld() throws IOException {
// log.debug("Hello World...app"); log.debug("Hello World...app");
// return ApiResult.ok("Hello World app"); return ApiResult.ok("Hello World app");
// } }
//
//
// @GetMapping(value = "/needRole") @GetMapping(value = "/needRole")
// @OperationLog(name = "needRole") @OperationLog(name = "needRole")
// @ApiOperation(value = "needRole", response = String.class) @ApiOperation(value = "needRole", response = String.class)
// public ApiResult<String> needRole() throws IOException { public ApiResult<String> needRole() throws IOException {
// log.debug("Hello World...app"); log.debug("Hello World...app");
// return ApiResult.ok("Hello World app"); return ApiResult.ok("Hello World app");
// } }
//
// @GetMapping(value = "/needRoleAdmin") @GetMapping(value = "/needRoleAdmin")
// @OperationLog(name = "needRoleAdmin") @OperationLog(name = "needRoleAdmin")
// @ApiOperation(value = "needRoleAdmin", response = String.class) @ApiOperation(value = "needRoleAdmin", response = String.class)
// @RequiresRoles("app:admin") @RequiresRoles("app:admin")
// public ApiResult<String> needRoleAdmin() throws IOException { public ApiResult<String> needRoleAdmin() throws IOException {
// log.debug("Hello World...app"); log.debug("Hello World...app");
// return ApiResult.ok("Hello World needRoleAdmin"); return ApiResult.ok("Hello World needRoleAdmin");
// } }
//
// @GetMapping(value = "/needRoleAll") @GetMapping(value = "/needRoleAll")
// @OperationLog(name = "needRoleAll") @OperationLog(name = "needRoleAll")
// @ApiOperation(value = "needRoleAll", response = String.class) @ApiOperation(value = "needRoleAll", response = String.class)
// @RequiresRoles("app:all") @RequiresRoles("app:all")
// public ApiResult<String> needRoleAll() throws IOException { public ApiResult<String> needRoleAll() throws IOException {
// log.debug("Hello World...app"); log.debug("Hello World...app");
// return ApiResult.ok("Hello World needRoleAll"); return ApiResult.ok("Hello World needRoleAll");
// } }
//
// @GetMapping(value = "/noRole") @GetMapping(value = "/noRole")
// @OperationLog(name = "noRole") @OperationLog(name = "noRole")
// @ApiOperation(value = "noRole", response = String.class) @ApiOperation(value = "noRole", response = String.class)
// public ApiResult<String> noRole() throws IOException { public ApiResult<String> noRole() throws IOException {
// log.debug("Hello World...app"); log.info("Hello World...app");
// return ApiResult.ok("Hello World app noRole"); return ApiResult.ok("Hello World app noRole");
// } }
//
//} }
common/src/main/java/com/sien/common/tillo/app_ws/AppImHandler.java 0 → 100644
package com.sien.common.tillo.app_ws;
import cn.hutool.core.thread.ThreadFactoryBuilder;
import com.sien.common.tillo.app_ws.model.Constants;
import com.sien.common.tillo.app_ws.receive.ReadWsData;
import com.sien.common.tillo.app_ws.service.AppUserChannelsService;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import javax.annotation.Resource;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
/**
* @Description app端 长连接事件处理
* @Author hewei hwei1233@163.com
* @Date 2019-07-26
*/
@Component
@ChannelHandler.Sharable
@Slf4j
public class AppImHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> {
@Resource
private ReadWsData readWsData;
@Resource
private AppUserChannelsService appUserChannelsService;
private final static ThreadFactory NAMED_THREAD_FACTORY = new ThreadFactoryBuilder()
.setNamePrefix("business-").build();
/**
* 核心业务处理线程池
* 属于io密集型业务
* io密集型任务配置尽可能多的线程数量
*/
private final static ExecutorService THREAD_POOL_EXECUTOR =
new ThreadPoolExecutor(Constants.CPU_PROCESSORS * 120, Constants.CPU_PROCESSORS * 130 + 2,
1L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>(), NAMED_THREAD_FACTORY, new ThreadPoolExecutor.CallerRunsPolicy());
@Override
protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame msg) {
String data = msg.text();
ChannelFuture channelFuture = ctx.writeAndFlush(new TextWebSocketFrame(data));
log.debug("data:" + data);
try {
if (data.isEmpty()) {
return;
}
/*
* 在此进入耗时业务线程池， 将不再阻塞netty的I/O线程，提高网络吞吐
*/
THREAD_POOL_EXECUTOR.execute(() ->
execute(ctx, data)
);
} catch (Exception e) {
//返回错误
ctx.channel().writeAndFlush(new TextWebSocketFrame("error=" + e.toString() + ",data=" + data));
log.error(e.getMessage() + data, e);
}
}
private void execute(ChannelHandlerContext ctx, String data) {
// Long userIdByChannel = appUserChannelsService.getUserIdByChannel(ctx);
//
// log.debug("appWS收到" + userIdByChannel + ":" + data + ",channelId:" + ctx.channel().id().asLongText());
System.out.println("appWS收到" + data);
String language = ctx.channel().attr(AppUserChannelsService.LANGUAGE).get();
// readWsData.convertModel(data, userIdByChannel, language);
}
/**
* 检测到异常
*
* @param ctx
* @param cause
* @throws Exception
*/
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
//排除当客户端意外关闭的情况，不是发送指定指令通知服务器退出，就会产生此错误。
if (ctx.channel().isActive()) {
Long userIdByChannel = appUserChannelsService.getUserIdByChannel(ctx);
log.error("uid:" + userIdByChannel + ",ws异常,channelId:" + ctx.channel().id().asLongText(), cause);
}
}
@Override
public void handlerAdded(ChannelHandlerContext ctx) {
// Long userIdByChannel = appUserChannelsService.getUserIdByChannel(ctx);
//
// log.debug("uid:" + userIdByChannel + "，app端连接WS成功" + ",channelId:" + ctx.channel().id().asLongText());
System.out.println("连接WS成功");
}
/**
* 客户端不活跃
*
* @param ctx
* @throws Exception
*/
@Override
public void channelInactive(ChannelHandlerContext ctx) {
log.debug("连接WS成功");
System.out.println("连接WS成功");
// Long userIdByChannel = appUserChannelsService.getUserIdByChannel(ctx);
// log.debug("uid:" + userIdByChannel + "," + "不活跃" + ",channelId:" + ctx.channel().id().asLongText());
//
// appUserChannelsService.remove(ctx);
}
/**
* 移除时触发， 不活跃的情况下会移除，会再次触发该事件
*/
@Override
public void handlerRemoved(ChannelHandlerContext ctx) {
System.out.println("handlerRemoved");
// Long userIdByChannel = appUserChannelsService.getUserIdByChannel(ctx);
// log.debug("uid:" + userIdByChannel + "," + "handlerRemoved" + ",channelId:" + ctx.channel().id().asLongText());
}
}
common/src/main/java/com/sien/common/tillo/app_ws/annotation/ReceiveTypeAnnotation.java 0 → 100644
package com.sien.common.tillo.app_ws.annotation;
import com.sien.common.tillo.app_ws.enums.WsRequestCmdEnum;
import java.lang.annotation.Documented;
import java.lang.annotation.ElementType;
import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;
/**
* @Description netty 接收类型注解，用于策略模式
* @Author hewei hwei1233@163.com
* @Date 2020-01-02
*/
@Retention(RetentionPolicy.RUNTIME)//RUNTIME运行时保留
@Target(ElementType.TYPE) //type描述类、接口
@Documented
public @interface ReceiveTypeAnnotation {
WsRequestCmdEnum type();
}
common/src/main/java/com/sien/common/tillo/app_ws/cache/UserCache.java 0 → 100644
package com.sien.common.tillo.app_ws.cache;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.stereotype.Service;
import java.net.InetAddress;
import java.net.NetworkInterface;
import java.net.SocketException;
import java.util.Enumeration;
/**
* @author hewei123@163.com
* @Description 用户与redis绑定
* @createTime 2020年04月14日 16:21:00
*/
@Service
@Slf4j
public class UserCache {
// @Autowired
// private RedisUtils redisUtils;
/**
* 在线状态
*/
public static final Integer ONLINE = 1;
/**
* 离线状态
*/
public static final Integer OFFLINE = 0;
/**
* key name
*/
private static final String KEY_BASE = "wsu:";
/**
* 用户在线状态fieldKey
*/
private static final String ONLINE_STATUS_KEY = "st";
/**
* 用户所连机器ip的fieldKey
*/
private static final String PRIVATE_IP_KEY = "lip";
// 用户语言暂时没用到
// private static final String LANGUAGE = "la";
// 用户公网ip，在公网部署集群需要用到
// private static final String PUBLIC_IP = "iip";
/**
* 排除无效的mac地址
*/
private final static byte[][] INVALID_MACS = {
{0x00, 0x05, 0x69}, // VMWare
{0x00, 0x1C, 0x14}, // VMWare
{0x00, 0x0C, 0x29}, // VMWare
{0x00, 0x50, 0x56}, // VMWare
{0x08, 0x00, 0x27}, // Virtualbox
{0x0A, 0x00, 0x27}, // Virtualbox
{0x00, 0x03, (byte) 0xFF}, // Virtual-PC
{0x00, 0x15, 0x5D} // Hyper-V
};
/**
* 内网ip
*/
private static String lAN_IP = "";
// 获取机器内网ip
static {
lAN_IP = getLocalIpAddress();
log.debug("lAN_IP:" + lAN_IP);
}
/**
* 判断是否为虚拟mac地址
*
* @param mac
* @return
*/
public static boolean isVmMac(byte[] mac) {
if (null == mac) {
return false;
}
for (byte[] invalid : INVALID_MACS) {
if (invalid[0] == mac[0] && invalid[1] == mac[1] && invalid[2] == mac[2]) {
return true;
}
}
return false;
}
/**
* 获取本机地址
*/
public static String getLocalIpAddress() {
try {
Enumeration<NetworkInterface> networkInterfaces = NetworkInterface.getNetworkInterfaces();
while (networkInterfaces.hasMoreElements()) {
NetworkInterface ni = networkInterfaces.nextElement();
/*
排除docker虚拟网卡
*/
String docker0 = "docker0";
if (ni.getName().equals(docker0)) {
continue;
}
if (!ni.isUp() || ni.isLoopback() || ni.isVirtual()) {
continue;
}
if (isVmMac(ni.getHardwareAddress())) {
continue;
}
Enumeration<InetAddress> inetAddresses = ni.getInetAddresses();
while (inetAddresses.hasMoreElements()) {
InetAddress inetAddress = inetAddresses.nextElement();
if (inetAddress.isLinkLocalAddress()) {
continue;
}
return inetAddress.getHostAddress();
}
}
} catch (SocketException e) {
log.debug("获取本机IP地址失败。" + e);
}
return StringUtils.EMPTY;
}
/**
* 用户上线绑定机器ip
*
* @param id
*/
public void online(String id) {
log.debug("ws用户上线保存redis连接ip:" + lAN_IP, ",uid:" );
// redisUtils.hset(KEY_BASE + id, PRIVATE_IP_KEY, lAN_IP);
// redisUtils.hset(KEY_BASE + id, ONLINE_STATUS_KEY, String.valueOf(ONLINE));
}
/**
* 用户下线删除绑定机器ip
*
* @param id
*/
public void offline(String id) {
log.debug("ws用户离线删除redis key,uid:" + id);
// redisUtils.kdel(KEY_BASE + id);
}
// /**
// * 根据用户id获取存在redis中的数据 例如绑定的服务器ip地址
// *
// * @param id
// * @return
// */
// public AppHashValueModel getById(String id) {
//
// Map<String, String> hgetll = redisUtils.hgetll(KEY_BASE + id);
// if (hgetll.isEmpty()) {
// return null;
// }
// AppHashValueModel appHashValueModel = new AppHashValueModel();
// appHashValueModel.setLanIp(hgetll.get(PRIVATE_IP_KEY));
// appHashValueModel.setOnlineStatus(Integer.parseInt(hgetll.get(ONLINE_STATUS_KEY)));
// return appHashValueModel;
//
// }
//
}
common/src/main/java/com/sien/common/tillo/app_ws/enums/WsRequestCmdEnum.java 0 → 100644
package com.sien.common.tillo.app_ws.enums;
/**
* @Description ws请求类型
* @Author hewei hwei1233@163.com
* @Date 2019-12-05
*/
public enum WsRequestCmdEnum {
/**
* 数据
*/
DATA(1),
/**
* ping
*/
PING(2);
private final int cmdCode;
WsRequestCmdEnum(int uriCode) {
this.cmdCode = uriCode;
}
public int getCmdCode() {
return cmdCode;
}
/**
* 根据uriCode获取
*
* @param uriCode
* @return
*/
public static WsRequestCmdEnum getByCode(int uriCode) {
for (WsRequestCmdEnum wsRequestUriPathEnum : values()) {
if (wsRequestUriPathEnum.getCmdCode() == uriCode) {
return wsRequestUriPathEnum;
}
}
return null;
}
}
common/src/main/java/com/sien/common/tillo/app_ws/enums/WsResponseCmdEnum.java 0 → 100644
package com.sien.common.tillo.app_ws.enums;
/**
* @Description ws响应类型
* @Author hewei hwei1233@163.com
* @Date 2019-12-05
*/
public enum WsResponseCmdEnum {
/**
* 离线消息下发完成指令
*/
OFFLINE_MSG_SUC(100),
/**
* 主动下发消息指令
*/
WRITE_MSG(101);
private final int cmdCode;
WsResponseCmdEnum(int uriCode) {
this.cmdCode = uriCode;
}
/**
* 根据uriCode获取
*
* @param uriCode
* @return
*/
public static WsResponseCmdEnum getByCode(int uriCode) {
for (WsResponseCmdEnum wsResponsePathEnum : values()) {
if (wsResponsePathEnum.getCmdCode() == uriCode) {
return wsResponsePathEnum;
}
}
return null;
}
public int getCmdCode() {
return cmdCode;
}
}
common/src/main/java/com/sien/common/tillo/app_ws/model/Constants.java 0 → 100644
package com.sien.common.tillo.app_ws.model;
import lombok.extern.slf4j.Slf4j;
/**
* @author wjm
* @Description 常量
* @date 2018/12/20
*/
@Slf4j
public class Constants {
/*
* 当前服务器cpu核心数量（）
*/
public static final Integer CPU_PROCESSORS = Runtime.getRuntime().availableProcessors();
static {
log.info("CPU_PROCESSORS:" + CPU_PROCESSORS);
}
/**
* 加密消息
*/
public static final Integer IS_CRYPTO_MESSAGE = 1;
/**
* 加密消息类型
*/
public static final Integer CRYPTO_TYPE = 3;
/**
* 网页版长连接url
*/
public static final String WEB_WS_URL = "/webws";
/**
* WEBWS_V_2
*/
public static final String WEBWS_V_2 = "/webwsV2";
/**
* app长连接url
*/
public static final String APP_WS_URL = "/appws";
/**
* 存储当前登录用户id的字段名
*/
public static final String CURRENT_USER_ID = "CURRENT_USER_ID";
/**
* token有效期（小时）30天
*/
public static final int TOKEN_EXPIRES_HOUR = 720;
/**
* token
*/
public static final String TOKEN = "token";
/**
* 密码加密盐
*/
public static final String SALT = "tillo2018";
/**
* http字符传
*/
public static final String HTTP = "http";
/**
* 标点符号--逗号
*/
public static final String COMMA = ",";
/**
* 标点符号--百分号
*/
public static final String PERCENT = "%";
/**
* 标点符号--问号
*/
public static final String MARK = "?";
/**
* 群聊取模数
*/
public static final Integer GROUP_MODULUS = 32;
/**
* 字符串0
*/
public static final String ZERO = "0";
/**
* 横线
*/
public static final String LINE = "-";
}
common/src/main/java/com/sien/common/tillo/app_ws/model/FileHelperOfflineModel.java 0 → 100644
package com.sien.common.tillo.app_ws.model;
import io.swagger.annotations.ApiModelProperty;
import lombok.Data;
import java.io.Serializable;
import java.util.Date;
@Data
public class FileHelperOfflineModel implements Serializable {
private String messageId;
@ApiModelProperty(value = "消息内容")
private String content;
@ApiModelProperty(value = "消息类型 text, audio, image, file, video, location")
private String type;
@ApiModelProperty(value = "资源id")
private String sourceId;
@ApiModelProperty(value = "时长（s）")
private Integer duration;
@ApiModelProperty(value = "位置信息")
private String locationInfo;
@ApiModelProperty(value = "图片/视频尺寸相关信息")
private String measureInfo;
@ApiModelProperty(value = "消息回执id")
private String backId;
@ApiModelProperty(value = "文件名")
private String fileName;
@ApiModelProperty(value = "文件大小")
private String fileSize;
@ApiModelProperty(value = "消息渠道 singleChat, groupChat,systemNotice,syncMySend")
private String route;
@ApiModelProperty(value = "时间戳")
private Date timestamp;
}
common/src/main/java/com/sien/common/tillo/app_ws/model/RequestHeaderConstants.java 0 → 100644
package com.sien.common.tillo.app_ws.model;
/**
* @author wjm
* @Description 请求头常量管理
* @date 2020/4/14
*/
public class RequestHeaderConstants {
/**
* 存放Authorization的header字段
*/
public static final String AUTHORIZATION = "Authorization";
/**
* 存放userIdn的header字段
*/
public static final String USER_ID = "userId";
/**
* 存放platform的header字段
*/
public static final String PLATFORM = "platform";
/**
* 存放手机唯一设备ID的deviceId的header字段
*/
public static final String DEVICE_ID = "deviceId";
/**
* 存放language的header字段
*/
public static final String LANGUAGE = "language";
/**
* 存放版本号versionNo的header字段
*/
public static final String VERSION_NO = "versionNo";
/**
* 存放版本号IOS版本号的header字段 ps: ios13
*/
public static final String IOS_VERSION = "iOSVersion";
/**
* 存放手机机型信息的header字段
*/
public static final String MOBILE_MODEL = "model";
}
common/src/main/java/com/sien/common/tillo/app_ws/model/ResponseModel.java 0 → 100644
package com.sien.common.tillo.app_ws.model;
import lombok.Data;
import lombok.experimental.Accessors;
import java.io.Serializable;
/**
* @Description null
* @Author hewei hwei1233@163.com
* @Date 2019-12-05
*/
@Data
@Accessors(chain = true)
public class ResponseModel implements Serializable {
/**
* 枚举类UriPathEnum 请求uri的编码
* 由于websocket使用同一个通道发送数据，需要区分不同类型请求
*/
private long path;
/**
* 状态码
*/
private int code;
/**
* 状态描述
*/
private String msg;
/**
* json数据
*/
private Object data;
/**
* 请求id, 以判空是否发送成功, 服务端处理完成后返回
* 可以以当前时间戳为id
*/
private String reqId;
}
common/src/main/java/com/sien/common/tillo/app_ws/model/ResultStatus.java 0 → 100644
package com.sien.common.tillo.app_ws.model;
import lombok.AllArgsConstructor;
import lombok.Getter;
/**
* 自定义请求状态码
*
* @author ScienJus
* @date 2015/7/15.
*/
@Getter
@AllArgsConstructor
public enum ResultStatus {
/**
* 成功
*/
SUCCESS(200, "Success", "成功"),
PRIVILEGE_IS_ERROR(401, "Unauthorized", "未授权限，请联系管理员");
/**
* 返回码
*/
private int code;
/**
* 默认返回英文结果描述 en
*/
private String message;
/**
* 返回中文结果描述 zh_simple
*/
private String zhMessage;
}
common/src/main/java/com/sien/common/tillo/app_ws/model/request/ReceiveModel.java 0 → 100644
package com.sien.common.tillo.app_ws.model.request;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import java.io.Serializable;
/**
* @Description null
* @Author hewei hwei1233@163.com
* @Date 2019-12-05
*/
@Data
@NoArgsConstructor
@AllArgsConstructor
public class ReceiveModel implements Serializable {
/**
* 枚举类UriPathEnum 请求uri的编码
* 由于websocket使用同一个通道发送数据，需要区分不同类型请求
*/
private Integer path;
/**
* json数据
*/
private String data;
/**
* 请求id, 以判空是否请求成功, 服务端处理完成后 返回此id
* 由前端生成，可以用uuid，也可以用时间戳
*/
private String reqId;
}
common/src/main/java/com/sien/common/tillo/app_ws/model/request/RequestOfflineMsgModel.java 0 → 100644
package com.sien.common.tillo.app_ws.model.request;
import lombok.Data;
import java.io.Serializable;
import java.util.List;
/**
* @Description 拉取所有房间离线消息
* @Author hewei hwei1233@163.com
* @Date 2020-01-03
*/
@Data
public class RequestOfflineMsgModel implements Serializable {
/**
* 优先权,需要优先拉取的房间id
* 如为空，可不传
*/
private List<Long> priority;
/**
* 身份公钥
*/
private String identityKey;
}
common/src/main/java/com/sien/common/tillo/app_ws/receive/ReadWsData.java 0 → 100644
package com.sien.common.tillo.app_ws.receive;
import com.alibaba.fastjson.JSON;
import com.sien.common.tillo.app_ws.enums.WsRequestCmdEnum;
import com.sien.common.tillo.app_ws.model.request.ReceiveModel;
import com.sien.common.tillo.app_ws.service.WriteDataService;
import com.sien.common.tillo.app_ws.strategy.AbstractReceiveStrategy;
import com.sien.common.tillo.app_ws.strategy.ReceiveStrategyContext;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import javax.annotation.Resource;
/**
* @Description ws 数据接收 转换类
* @Author hewei hwei1233@163.com
* @Date 2019-12-03
*/
@Service
public class ReadWsData {
private static Logger log = LoggerFactory.getLogger(ReadWsData.class);
// idea此处报红 属于正常
// @SuppressWarnings("SpringJavaInjectionPointsAutowiringInspection")
@Autowired
private ReceiveStrategyContext receiveStrategyContext;
@Resource
private WriteDataService writeDataService;
private static final String PING = "p";
/**
* 在此开始进入业务流程子线程，将不占netty的io线程
*
* @param data
* @param userId
* @param language
* @throws Exception
*/
public void convertModel(String data, Long userId, String language) {
if (PING.equals(data)) {
log.debug("收到心跳：" + userId);
return;
}
ReceiveModel requestModel = JSON.parseObject(data, ReceiveModel.class);
if (null == requestModel || null == requestModel.getPath()) {
return;
}
try {
// User user = userService.findById(userId);
this.doProcess(language, requestModel);
} catch (Exception e) {
log.error("系统繁忙:" + data + ",userId:" + userId, e);
// writeDataService.nullDataSuccess(requestModel, ResultStatus.SYS_BUSY, userId, language);
}
}
private void doProcess(String language, ReceiveModel requestModel) {
WsRequestCmdEnum wsRequestUriPathEnum = WsRequestCmdEnum.getByCode(requestModel.getPath());
// 使用策略模式， 根据不同类型请求调用不同实现类
AbstractReceiveStrategy receiveStrategy = receiveStrategyContext.getStrategy(wsRequestUriPathEnum);
receiveStrategy.process(requestModel, language);
}
}
common/src/main/java/com/sien/common/tillo/app_ws/service/AppUserChannelsService.java 0 → 100644
package com.sien.common.tillo.app_ws.service;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.util.AttributeKey;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
/**
* @Description 维护netty用户channel对象
* @Author hewei hwei1233@163.com
* @Date 2019-08-01
*/
public interface AppUserChannelsService {
/**
* channel对象
* 用户id为key
* context为值
*/
Map<String, NioSocketChannel> CHANNEL_MAP = new ConcurrentHashMap<>();
/**
* USER_ID
*/
AttributeKey<Long> USER_ID = AttributeKey.valueOf("userId");
/**
* LANGUAGE
*/
AttributeKey<String> LANGUAGE = AttributeKey.valueOf("language");
/**
* APP_VERSION
*/
AttributeKey<String> APP_VERSION = AttributeKey.valueOf("appVersion");
AttributeKey<String> TOKEN = AttributeKey.valueOf("TOKEN");
AttributeKey<String> DEVICEID = AttributeKey.valueOf("DEVICEID");
AttributeKey<String> PLATFORM = AttributeKey.valueOf("PLATFORM");
/**
* 根据userID获取channel
* @param userId
* @return
*/
NioSocketChannel get(String userId);
/**
* userID绑定channel
* @param userId
* @param channel
*/
void put(String userId, NioSocketChannel channel);
/**
* 移除channel
* @param channelHandlerContext
*/
void remove(ChannelHandlerContext channelHandlerContext);
/**
* 根据channel返回userIds
*
* @param channelHandlerContext
* @return
*/
Long getUserIdByChannel(ChannelHandlerContext channelHandlerContext);
/**
* 下发数据
*
* @param msg
* @param userId
* @return
*/
boolean writeData(String msg, Long userId);
/**
* rpc异步下发数据
*
* @param msg
* @param userId
* @return
*/
boolean rpcWriteData(String msg, Long userId);
/**
* rpc-异步下发踢人数据及关闭旧通道
*
* @param msg
* @param userId
* @return
*/
boolean rpcKickWriteData(String msg, Long userId);
/**
* rpc-异步关闭旧通道
*
* @param userId
* @return
*/
boolean rpcCloseOldChannel(Long userId);
Boolean isOnLocal(Long userId);
}
common/src/main/java/com/sien/common/tillo/app_ws/service/WriteDataService.java 0 → 100644
package com.sien.common.tillo.app_ws.service;
import com.sien.common.tillo.app_ws.model.ResponseModel;
import com.sien.common.tillo.app_ws.model.ResultStatus;
import com.sien.common.tillo.app_ws.model.request.ReceiveModel;
/**
* @Description ws响应数据，各种状态码封装
* @Author hewei hwei1233@163.com
* @Date 2019-12-05
*/
public interface WriteDataService {
/**
* 可自定义状态码 带data
*
* @param requestModel
* @param resultStatus
* @param data
* @param userId
* @param language
*/
void dataAndStatus(ReceiveModel requestModel, ResultStatus resultStatus, Object data, Long userId, String language);
/**
* 固定"成功"状态码 带data
*
* @param requestModel
* @param data
* @param userId
* @param language
*/
void successAndData(ReceiveModel requestModel, Object data, Long userId, String language);
/**
* 固定"成功"状态码 无data
*
* @param requestModel
* @param resultStatus
* @param userId
* @param language
*/
void nullDataSuccess(ReceiveModel requestModel, ResultStatus resultStatus, Long userId, String language);
/**
* 固定"参数错误"状态码 无data
*
* @param requestModel
* @param userId
* @param language
*/
void paramErrorAndNullData(ReceiveModel requestModel, Long userId, String language);
/**
* 调用ws处理响应逻辑
*
* @param responseModel
* @param userId
*/
void write(ResponseModel responseModel, Long userId);
}
common/src/main/java/com/sien/common/tillo/app_ws/service/impl/AppUserChannelsServiceImpl.java 0 → 100644
package com.sien.common.tillo.app_ws.service.impl;
import cn.hutool.core.thread.ThreadFactoryBuilder;
import com.sien.common.tillo.app_ws.cache.UserCache;
import com.sien.common.tillo.app_ws.model.Constants;
import com.sien.common.tillo.app_ws.service.AppUserChannelsService;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
/**
* @Description 维护netty用户channel对象
* @Author hewei hwei1233@163.com
* @Date 2019-12-03
*/
@Component
@Slf4j
public class AppUserChannelsServiceImpl implements AppUserChannelsService {
private final static ThreadFactory NAMED_THREAD_FACTORY = new ThreadFactoryBuilder()
.setNamePrefix("rpcWrite-").build();
// .setPriority(3)
/**
* 远程调用ws下发数据线程池
* 属于IO密集型任务
* IO密集型任务线程并不是一直在执行任务，则应配置尽可能多的线程，如CPU核数*2
* 某大厂设置策略：IO密集型时，大部分线程都阻塞，故需要多配置线程数： 公式：CPU核数/1-阻塞系数 阻塞系数：0.8-0.9
* 比如8核CPU： 8/1-0.9 = 80 个线程数
* <p>
* 由于来自远程端调用下发数据 如果是群聊1000人群则调用1000次 为不要占用太多资源 需要排队下发
* 经过并发测试 200并发1000人群消息 需要调用200x1000=20w次 考虑单机cpu性能还要顾及本机api业务 设置阻塞队列
* 为避免过多占用本地io线程导致response慢，设置LinkedBlockingQueue数量多可以避免抢占，TODO （队列数量需要测试调试到最优数量 ）
* 最大线程数量不要设置太多 数量、优先级一定要比本地io线程低级
*
*
* <p>
* TODO 待完善：消息发送投递至MQ， 消费方从MQ队列获取下发任务 本地队列不宜缓存太多(机器死机则会全丢失) 堆积的请求处理队列可能会耗费非常大的内存甚至死机
*/
// private final static ExecutorService THREAD_POOL_RPC_WRITE_EXECUTOR =
// new ThreadPoolExecutor(Constants.CPU_PROCESSORS * 4, Constants.CPU_PROCESSORS * 4 + 1,
// 1L, TimeUnit.MILLISECONDS,
// new LinkedBlockingQueue<Runnable>(), NAMED_THREAD_FACTORY, new ThreadPoolExecutor.CallerRunsPolicy());
// private final static ExecutorService THREAD_POOL_RPC_WRITE_EXECUTOR =
// new ThreadPoolExecutor(4, 6,
// 0L, TimeUnit.MILLISECONDS,
// new LinkedBlockingQueue<Runnable>(), NAMED_THREAD_FACTORY, new ThreadPoolExecutor.AbortPolicy());
private final static ExecutorService THREAD_POOL_RPC_WRITE_EXECUTOR =
new ThreadPoolExecutor(Constants.CPU_PROCESSORS * 100, Constants.CPU_PROCESSORS * 100,
1L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>(), NAMED_THREAD_FACTORY, new ThreadPoolExecutor.CallerRunsPolicy());
@Autowired
private UserCache userCache;
@Override
public NioSocketChannel get(String userId) {
return AppUserChannelsService.CHANNEL_MAP.get(userId);
}
@Override
public void put(String userId, NioSocketChannel channel) {
// AppHashValueModel appHashValueModel = new AppHashValueModel();
// appHashValueModel.setOnlineStatus(UserCache.ONLINE);
userCache.online(userId);
AppUserChannelsService.CHANNEL_MAP.put(userId, channel);
}
@Override
public void remove(ChannelHandlerContext channelHandlerContext) {
Channel channel = channelHandlerContext.channel();
/*
channel != null : 通道不能为空
!channel.isActive() 通道不能是活跃状态的
!channel.isOpen() 通道不能是打开状态的
// !channel.isWritable() 通道是否可写数据
*/
if (channel != null && !channel.isActive() && !channel.isOpen()) {
String userId = String.valueOf(this.getUserIdByChannel(channelHandlerContext));
userCache.offline(userId);
log.debug("不活跃remove,uid:" + userId);
AppUserChannelsService.CHANNEL_MAP.remove(userId);
channelHandlerContext.channel().close();
}
}
@Override
public Long getUserIdByChannel(ChannelHandlerContext channelHandlerContext) {
return channelHandlerContext.channel().attr(AppUserChannelsService.USER_ID).get();
}
@Override
public Boolean isOnLocal(Long userId) {
NioSocketChannel nioSocketChannel = this.get(String.valueOf(userId));
return null != nioSocketChannel;
}
/**
* TODO 待完成: 根据ACK回执 以及线程等待超时机制来判断客户端是否离线和超时;
* TODO 待完成: 发送后阻塞当前子线程2秒后获取ack回执 如客户端发起ack回执则需要主动唤醒当前子线程 立马唤醒当前子线程， 判断如果已回执则返回发送成功， 如果未回执则判断客户端是否断线或发送错误
*
* @param msg
* @param userId
* @return
*/
@Override
public boolean rpcWriteData(String msg, Long userId) {
Future<Boolean> future = THREAD_POOL_RPC_WRITE_EXECUTOR.submit(() -> {
NioSocketChannel nioSocketChannel = get(String.valueOf(userId));
if (null == nioSocketChannel) {
userCache.offline(String.valueOf(userId));
log.debug("rpc-writeData连接为空:" + userId + "," + msg);
return false;
}
// 判断连接是否断开
if (nioSocketChannel.isShutdown()) {
log.debug("rpc-writeData连接断开:" + userId + "," + msg + ",\nchannelId:" + nioSocketChannel.id().asLongText());
return false;
}
if (log.isInfoEnabled()) {
log.info("rpc-writeData:" + userId + "," + msg + ",\nchannelId:" + nioSocketChannel.id().asLongText());
}
ChannelFuture channelFuture = nioSocketChannel.writeAndFlush(new TextWebSocketFrame(msg));
channelFuture.addListener(
//执行后回调的方法
(ChannelFutureListener) channelFuture1 -> {
if (log.isInfoEnabled()) {
log.info("rpc-netty线程异步执行结果:" + channelFuture1.isDone() + ",业务执行结果:" + channelFuture1.isSuccess()
+ "；\nwriteData:" + userId + "," + msg + ",channelId:" + nioSocketChannel.id().asLongText());
}
});
channelFuture.get();
return true;
});
boolean resultStatus = false;
try {
resultStatus = future.get();
} catch (InterruptedException | ExecutionException e) {
e.printStackTrace();
}
return resultStatus;
}
@Override
public boolean rpcKickWriteData(String msg, Long userId) {
Future<Boolean> future = THREAD_POOL_RPC_WRITE_EXECUTOR.submit(() -> {
NioSocketChannel nioSocketChannel = get(String.valueOf(userId));
if (null == nioSocketChannel) {
log.debug("rpc-kickWriteData连接为空:" + userId + "," + msg);
return false;
}
// 判断连接是否断开
if (nioSocketChannel.isShutdown()) {
log.debug("rpc-kickWriteData连接断开:" + userId + "," + msg + ",\nchannelId:" + nioSocketChannel.id().asLongText());
nioSocketChannel.close();
return false;
}
if (log.isDebugEnabled()) {
log.debug("rpc-kickWriteData:" + userId + "," + msg + ",\nchannelId:" + nioSocketChannel.id().asLongText());
}
ChannelFuture channelFuture = nioSocketChannel.writeAndFlush(new TextWebSocketFrame(msg));
channelFuture.addListener(
//执行后回调的方法
(ChannelFutureListener) channelFuture1 -> {
if (log.isDebugEnabled()) {
log.debug("rpc-netty踢人线程异步执行结果:" + channelFuture1.isDone() + ",业务执行结果:" + channelFuture1.isSuccess()
+ "；\nkickWriteData:" + userId + "," + msg + ",channelId:" + nioSocketChannel.id().asLongText());
}
});
channelFuture.get();
// 关闭
nioSocketChannel.close();
return true;
});
boolean resultStatus = false;
try {
resultStatus = future.get();
} catch (InterruptedException | ExecutionException e) {
e.printStackTrace();
}
return resultStatus;
}
@Override
public boolean rpcCloseOldChannel(Long userId) {
Future<Boolean> future = THREAD_POOL_RPC_WRITE_EXECUTOR.submit(() -> {
NioSocketChannel nioSocketChannel = get(String.valueOf(userId));
if (null == nioSocketChannel) {
log.debug("rpc-closeOldChannel连接为空:" + userId);
return false;
}
// 关闭
nioSocketChannel.close();
return true;
});
boolean resultStatus = false;
try {
resultStatus = future.get();
} catch (InterruptedException | ExecutionException e) {
e.printStackTrace();
}
return resultStatus;
}
@Override
public boolean writeData(String msg, Long userId) {
// Future<Boolean> future = THREAD_POOL_EXECUTOR.submit(() -> {
NioSocketChannel nioSocketChannel = get(String.valueOf(userId));
if (null == nioSocketChannel) {
userCache.offline(String.valueOf(userId));
log.debug("writeData连接为空:" + userId + "," + msg);
return false;
}
// 判断连接是否断开
if (nioSocketChannel.isShutdown()) {
log.debug("writeData连接断开:" + userId + "," + msg + ",\nchannelId:" + nioSocketChannel.id().asLongText());
return false;
}
if (log.isDebugEnabled()) {
log.debug("writeData:" + userId + "," + msg + ",\nchannelId:" + nioSocketChannel.id().asLongText());
}
ChannelFuture channelFuture = nioSocketChannel.writeAndFlush(new TextWebSocketFrame(msg));
channelFuture.addListener(
//执行后回调的方法
(ChannelFutureListener) channelFuture1 -> {
if (log.isDebugEnabled()) {
log.debug("netty线程异步执行结果:" + channelFuture1.isDone() + ",业务执行结果:" + channelFuture1.isSuccess()
+ "；\nwriteData:" + userId + "," + msg + ",channelId:" + nioSocketChannel.id().asLongText());
}
});
// channelFuture.get();
return true;
// });
// Boolean resultStatus = false;
// try {
// resultStatus = future.get();
// } catch (InterruptedException | ExecutionException e) {
// e.printStackTrace();
// }
// return resultStatus;
}
}
common/src/main/java/com/sien/common/tillo/app_ws/service/impl/WriteDataServiceImpl.java 0 → 100644
package com.sien.common.tillo.app_ws.service.impl;
import com.alibaba.fastjson.JSON;
import com.sien.common.tillo.app_ws.model.ResponseModel;
import com.sien.common.tillo.app_ws.model.ResultStatus;
import com.sien.common.tillo.app_ws.model.request.ReceiveModel;
import com.sien.common.tillo.app_ws.service.AppUserChannelsService;
import com.sien.common.tillo.app_ws.service.WriteDataService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import java.util.HashMap;
/**
* @Description 下发数据
* @Author hewei hwei1233@163.com
* @Date 2019-12-05
*/
@Component
public class WriteDataServiceImpl implements WriteDataService {
@Autowired
private AppUserChannelsService appUserChannelsService;
@Override
public void successAndData(ReceiveModel requestModel, Object data, Long userId, String language) {
this.dataAndStatus(requestModel, ResultStatus.SUCCESS, data, userId, language);
}
@Override
public void nullDataSuccess(ReceiveModel requestModel, ResultStatus resultStatus, Long userId, String language) {
this.dataAndStatus(requestModel, resultStatus, new HashMap<>(1), userId, language);
}
@Override
public void paramErrorAndNullData(ReceiveModel requestModel, Long userId, String language) {
// this.nullDataSuccess(requestModel, ResultStatus.PARAM_ERROR, userId, language);
}
@Override
public void dataAndStatus(ReceiveModel requestModel, ResultStatus resultStatus, Object data, Long userId, String language) {
// ResultModel<String> resultModel = new ResultModel<>(resultStatus, language);
//
// ResponseModel responseModel = new ResponseModel();
// responseModel.setMsg(resultModel.getMessage());
// responseModel.setPath(requestModel.getPath());
// responseModel.setReqId(requestModel.getReqId());
// responseModel.setData(data);
// responseModel.setCode(resultModel.getStatus());
// this.write(responseModel, userId);
}
@Override
public void write(ResponseModel responseModel, Long userId) {
String json = JSON.toJSONString(responseModel);
appUserChannelsService.writeData(json, userId);
}
}
common/src/main/java/com/sien/common/tillo/app_ws/strategy/AbstractReceiveStrategy.java 0 → 100644
package com.sien.common.tillo.app_ws.strategy;
import com.sien.common.tillo.app_ws.model.request.ReceiveModel;
/**
* @Description 接收netty不同类型请求
* 抽象类 策略设计模式
* @Author hewei hwei1233@163.com
* @Date 2020-01-02
*/
public abstract class AbstractReceiveStrategy {
/**
* 处理业务流程
*
* @param requestModel
* @param language
* @throws Exception
*/
abstract public void process(ReceiveModel requestModel, String language);
}
common/src/main/java/com/sien/common/tillo/app_ws/strategy/ReceiveStrategyContext.java 0 → 100644
package com.sien.common.tillo.app_ws.strategy;
import com.sien.common.tillo.app_ws.enums.WsRequestCmdEnum;
import com.sien.common.tillo.app_ws.utils.SpringBeanUtils;
import org.springframework.util.CollectionUtils;
import java.util.Map;
/**
* @Description 策略模式 上下文
* 维护指令码与策略实现的对应
* @Author hewei hwei1233@163.com
*/
public class ReceiveStrategyContext {
private Map<WsRequestCmdEnum, Class> strategyMap;
public ReceiveStrategyContext(Map<WsRequestCmdEnum, Class> strategyMap) {
this.strategyMap = strategyMap;
}
public AbstractReceiveStrategy getStrategy(WsRequestCmdEnum wsRequestPathEnum) {
if (wsRequestPathEnum == null) {
throw new IllegalArgumentException("not fond enum");
}
if (CollectionUtils.isEmpty(strategyMap)) {
throw new IllegalArgumentException("strategy map is empty,please check you strategy package path");
}
Class aClass = strategyMap.get(wsRequestPathEnum);
if (aClass == null) {
throw new IllegalArgumentException("not fond strategy for type:" + wsRequestPathEnum.getCmdCode());
}
return (AbstractReceiveStrategy) SpringBeanUtils.getBean(aClass);
}
}
common/src/main/java/com/sien/common/tillo/app_ws/strategy/ReceiveStrategyProcessor.java 0 → 100644
package com.sien.common.tillo.app_ws.strategy;
import com.google.common.collect.Maps;
import com.sien.common.tillo.app_ws.annotation.ReceiveTypeAnnotation;
import com.sien.common.tillo.app_ws.enums.WsRequestCmdEnum;
import com.sien.common.tillo.app_ws.utils.ClassScanner;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.config.BeanFactoryPostProcessor;
import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
import org.springframework.stereotype.Component;
import java.util.Map;
import java.util.Set;
/**
* @Description 策略 bean注解扫描注册类
* 扫描自定义注解，将指令码与实现类绑定，将对应关系添加到上下文对象中
* <p>
* BeanStrategyProcessor是spring在容器初始化后对外暴露的扩展点，
* spring ioc容器允许beanFactoryPostProcessor在容器加载注册BeanDefinition后读取BeanDefinition，并能修改它。
* @Author hewei hwei1233@163.com
* @Date 2020-01-02
*/
@Component
public class ReceiveStrategyProcessor implements BeanFactoryPostProcessor {
// 扫码注解的包路径
private static final String STRATEGY_PACK = "com.sql.tillo.app_ws.strategy,com.sql.tillo.app_ws.process";
@Override
public void postProcessBeanFactory(ConfigurableListableBeanFactory configurableListableBeanFactory) throws BeansException {
Map<WsRequestCmdEnum, Class> handlerMap = Maps.newHashMapWithExpectedSize(5);
// 扫码ReceiveTypeAnnotation注解的类
Set<Class<?>> classSet = ClassScanner.scan(STRATEGY_PACK, ReceiveTypeAnnotation.class);
classSet.forEach(clazz -> {
// 获取注解中的类型值，与枚举类一一对应
WsRequestCmdEnum type = clazz.getAnnotation(ReceiveTypeAnnotation.class).type();
handlerMap.put(type, clazz);
});
// 初始化Contenxt， 将其注册到spring容器当中
ReceiveStrategyContext context = new ReceiveStrategyContext(handlerMap);
try {
configurableListableBeanFactory.registerResolvableDependency(Class.forName(ReceiveStrategyContext.class.getName()), context);
} catch (ClassNotFoundException e) {
e.printStackTrace();
}
}
}
common/src/main/java/com/sien/common/tillo/app_ws/strategy/concrete/SingleConcreteReceiveStrategy.java 0 → 100644
package com.sien.common.tillo.app_ws.strategy.concrete;
import com.sien.common.tillo.app_ws.annotation.ReceiveTypeAnnotation;
import com.sien.common.tillo.app_ws.enums.WsRequestCmdEnum;
import com.sien.common.tillo.app_ws.model.request.ReceiveModel;
import com.sien.common.tillo.app_ws.strategy.AbstractReceiveStrategy;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
/**
* @Description 处理app单聊消息
*/
@ReceiveTypeAnnotation(type = WsRequestCmdEnum.DATA)
@Service
@Slf4j
public class SingleConcreteReceiveStrategy extends AbstractReceiveStrategy {
@Override
public void process(ReceiveModel requestModel, String language) {
}
}
common/src/main/java/com/sien/common/tillo/app_ws/utils/ClassScanner.java 0 → 100644
package com.sien.common.tillo.app_ws.utils;
import org.apache.commons.lang3.ArrayUtils;
import org.springframework.beans.factory.BeanDefinitionStoreException;
import org.springframework.context.ResourceLoaderAware;
import org.springframework.core.io.Resource;
import org.springframework.core.io.ResourceLoader;
import org.springframework.core.io.support.PathMatchingResourcePatternResolver;
import org.springframework.core.io.support.ResourcePatternResolver;
import org.springframework.core.io.support.ResourcePatternUtils;
import org.springframework.core.type.classreading.CachingMetadataReaderFactory;
import org.springframework.core.type.classreading.MetadataReader;
import org.springframework.core.type.classreading.MetadataReaderFactory;
import org.springframework.core.type.filter.AnnotationTypeFilter;
import org.springframework.core.type.filter.TypeFilter;
import org.springframework.util.StringUtils;
import org.springframework.util.SystemPropertyUtils;
import java.io.IOException;
import java.lang.annotation.Annotation;
import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Set;
/**
* @Description 扫描注解 工具类
* @Author hewei hwei1233@163.com
* @Date 2020-01-03
*/
public class ClassScanner implements ResourceLoaderAware {
private final List<TypeFilter> includeFilters = new LinkedList<TypeFilter>();
private final List<TypeFilter> excludeFilters = new LinkedList<TypeFilter>();
private ResourcePatternResolver resourcePatternResolver = new PathMatchingResourcePatternResolver();
private MetadataReaderFactory metadataReaderFactory = new CachingMetadataReaderFactory(this.resourcePatternResolver);
@SafeVarargs
public static Set<Class<?>> scan(String[] basePackages, Class<? extends Annotation>... annotations) {
ClassScanner cs = new ClassScanner();
if (ArrayUtils.isNotEmpty(annotations)) {
for (Class anno : annotations) {
cs.addIncludeFilter(new AnnotationTypeFilter(anno));
}
}
Set<Class<?>> classes = new HashSet<>();
for (String s : basePackages) {
classes.addAll(cs.doScan(s));
}
return classes;
}
@SafeVarargs
public static Set<Class<?>> scan(String basePackages, Class<? extends Annotation>... annotations) {
return ClassScanner.scan(StringUtils.tokenizeToStringArray(basePackages, ",; \t\n"), annotations);
}
public final ResourceLoader getResourceLoader() {
return this.resourcePatternResolver;
}
@Override
public void setResourceLoader(ResourceLoader resourceLoader) {
this.resourcePatternResolver = ResourcePatternUtils
.getResourcePatternResolver(resourceLoader);
this.metadataReaderFactory = new CachingMetadataReaderFactory(
resourceLoader);
}
public void addIncludeFilter(TypeFilter includeFilter) {
this.includeFilters.add(includeFilter);
}
public void addExcludeFilter(TypeFilter excludeFilter) {
this.excludeFilters.add(0, excludeFilter);
}
public void resetFilters(boolean useDefaultFilters) {
this.includeFilters.clear();
this.excludeFilters.clear();
}
public Set<Class<?>> doScan(String basePackage) {
Set<Class<?>> classes = new HashSet<>();
try {
String packageSearchPath = ResourcePatternResolver.CLASSPATH_ALL_URL_PREFIX
+ org.springframework.util.ClassUtils
.convertClassNameToResourcePath(SystemPropertyUtils
.resolvePlaceholders(basePackage))
+ "/**/*.class";
Resource[] resources = this.resourcePatternResolver
.getResources(packageSearchPath);
for (int i = 0; i < resources.length; i++) {
Resource resource = resources[i];
if (resource.isReadable()) {
MetadataReader metadataReader = this.metadataReaderFactory.getMetadataReader(resource);
if ((includeFilters.size() == 0 && excludeFilters.size() == 0) || matches(metadataReader)) {
try {
classes.add(Class.forName(metadataReader
.getClassMetadata().getClassName()));
} catch (ClassNotFoundException e) {
e.printStackTrace();
}
}
}
}
} catch (IOException ex) {
throw new BeanDefinitionStoreException(
"I/O failure during classpath scanning", ex);
}
return classes;
}
protected boolean matches(MetadataReader metadataReader) throws IOException {
for (TypeFilter tf : this.excludeFilters) {
if (tf.match(metadataReader, this.metadataReaderFactory)) {
return false;
}
}
for (TypeFilter tf : this.includeFilters) {
if (tf.match(metadataReader, this.metadataReaderFactory)) {
return true;
}
}
return false;
}
}
common/src/main/java/com/sien/common/tillo/app_ws/utils/FullHttpRequestUtils.java 0 → 100644
package com.sien.common.tillo.app_ws.utils;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.http.DefaultFullHttpResponse;
import io.netty.handler.codec.http.FullHttpRequest;
import io.netty.handler.codec.http.FullHttpResponse;
import io.netty.handler.codec.http.HttpHeaderNames;
import io.netty.handler.codec.http.HttpMethod;
import io.netty.handler.codec.http.HttpResponseStatus;
import io.netty.handler.codec.http.HttpVersion;
import io.netty.handler.codec.http.QueryStringDecoder;
import io.netty.util.CharsetUtil;
import org.apache.http.MethodNotSupportedException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import springfox.documentation.RequestHandler;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import static io.netty.buffer.Unpooled.copiedBuffer;
/**
* @Description netty请求工具类
* @Author hewei hwei1233@163.com
* @Date 2019-07-19
*/
public class FullHttpRequestUtils {
private static Logger logger = LoggerFactory.getLogger(RequestHandler.class);
/**
* 参数解析
*
* @param fullReq
* @return
* @throws MethodNotSupportedException
*/
public static Map<String, String> parameterParse(FullHttpRequest fullReq) throws MethodNotSupportedException {
HttpMethod method = fullReq.method();
Map<String, String> parmMap = new HashMap<>(10);
// GET请求
if (HttpMethod.GET.equals(method) || HttpMethod.DELETE.equals(method)) {
QueryStringDecoder decoder = new QueryStringDecoder(fullReq.uri());
Map<String, List<String>> parameters = decoder.parameters();
parameters.forEach((key, value) -> {
// entry.getValue()是一个List, 只取第一个元素
parmMap.put(key, value.get(0));
});
} else {
// 不支持其它方法
throw new MethodNotSupportedException("");
}
return parmMap;
}
/**
* 获取body参数
*
* @param request
* @return
*/
public static String getBody(FullHttpRequest request) {
ByteBuf buf = request.content();
return buf.toString(CharsetUtil.UTF_8);
}
/**
* 发送的返回值
*
* @param ctx 返回
* @param context 消息
* @param status 状态
*/
public static void send(ChannelHandlerContext ctx, String context, HttpResponseStatus status) {
FullHttpResponse response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, status, copiedBuffer(context, CharsetUtil.UTF_8));
response.headers().set(HttpHeaderNames.CONTENT_TYPE, "application/json;charset=utf-8");
logger.debug("response:\n" + response.toString() + "\ncontext:" + context);
ctx.writeAndFlush(response).addListener(ChannelFutureListener.CLOSE);
}
}
common/src/main/java/com/sien/common/tillo/app_ws/utils/RedisUtils.java 0 → 100644
package com.sien.common.tillo.app_ws.utils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
/**
* @author hewei123@163.com
* @Description redis 通用工具类
* @createTime 2020年04月14日 16:07:00
*/
@Component
public class RedisUtils {
@Autowired
private StringRedisTemplate redisTemplate;
/**
* 获取hash中field对应的值
*
* @param key
* @param field
* @return
*/
public String hget(String key, String field) {
Object val = redisTemplate.opsForHash().get(key, field);
return val == null ? null : val.toString();
}
/**
* 添加or更新hash的值
*
* @param key
* @param field
* @param value
*/
public void hset(String key, String field, String value) {
redisTemplate.opsForHash().put(key, field, value);
}
/**
* 删除hash中field这一对kv
*
* @param key
* @param field
*/
public void hdel(String key, String field) {
redisTemplate.opsForHash().delete(key, field);
}
/**
* 删除key
*
* @param key 如果传入hash类型的key，则把整个hash中所有field删除
*/
public void kdel(String key) {
redisTemplate.delete(key);
}
/**
* 取hash类型 该key所有参数
*
* @param key
* @return
*/
public Map<String, String> hgetll(String key) {
return redisTemplate.execute((RedisCallback<Map<String, String>>) con -> {
Map<byte[], byte[]> result = con.hGetAll(key.getBytes());
if (CollectionUtils.isEmpty(result)) {
return new HashMap<>(0);
}
Map<String, String> ans = new HashMap<>(result.size());
for (Map.Entry<byte[], byte[]> entry : result.entrySet()) {
ans.put(new String(entry.getKey()), new String(entry.getValue()));
}
return ans;
});
}
/**
* 取该hash的key中指定多个参数
*
* @param key
* @param fields
* @return
*/
public Map<String, String> hmget(String key, List<String> fields) {
List<String> result = redisTemplate.<String, String>opsForHash().multiGet(key, fields);
Map<String, String> ans = new HashMap<>(fields.size());
int index = 0;
for (String field : fields) {
if (result.get(index) == null) {
continue;
}
ans.put(field, result.get(index));
}
return ans;
}
/**
* 向指定key中存放set<String>集合
*
* @param key
* @param value
*/
public void addForSet(String key, String value) {
redisTemplate.opsForSet().add(key, value);
}
/**
* 移除指定key中et<String>集合的某值
*
* @param key
*/
public void removeForSet(String key, String value) {
redisTemplate.opsForSet().remove(key, value);
}
/**
* 获取指定key中存放set<String>的集合
*
* @param key
*/
public Set<String> getForSetMembers(String key) {
return redisTemplate.opsForSet().members(key);
}
/**
* 删除指定key缓存
*
* @param key
*/
public void deleteByKey(String key) {
redisTemplate.delete(key);
}
}
common/src/main/java/com/sien/common/tillo/app_ws/utils/SpringBeanUtils.java 0 → 100644
package com.sien.common.tillo.app_ws.utils;
import org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.stereotype.Component;
/**
* @Description 在非spring管理的类中获取spring注册的bean
* @Author hewei hwei1233@163.com
* @Date 2020-01-03
*/
@Component
public class SpringBeanUtils implements ApplicationContextAware {
private static ApplicationContext applicationContext;
@Override
public void setApplicationContext(ApplicationContext context) throws BeansException {
if (context != null) {
applicationContext = context;
}
}
public static Object getBean(String name) {
return applicationContext.getBean(name);
}
public static <T> T getBean(Class<T> clazz) {
return applicationContext.getBean(clazz);
}
}
common/src/main/java/com/sien/common/tillo/netty/core/ChannelInboundHandler.java 0 → 100644
package com.sien.common.tillo.netty.core;
import com.sien.common.tillo.netty.handler.NettyApiRequest;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.handler.codec.http.FullHttpRequest;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;
import java.net.InetAddress;
@Component
@ChannelHandler.Sharable
public class ChannelInboundHandler extends ChannelInboundHandlerAdapter {
private final Logger logger = LoggerFactory.getLogger(ChannelInboundHandler.class);
private final NettyApiRequest singleRequest;
public ChannelInboundHandler(NettyApiRequest singleRequest) {
this.singleRequest = singleRequest;
}
/**
* 收到消息时，返回信息
*/
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) {
FullHttpRequest httpRequest = (FullHttpRequest) msg;
try {
singleRequest.handle(ctx, msg, httpRequest);
} catch (Exception e) {
logger.error("SingleNettyServer处理请求失败!", e);
// this.sendBad(ctx);
} finally {
//释放请求
httpRequest.release();
}
}
/**
* 发送错误信息
*
* @param ctx
*/
// private void sendBad(ChannelHandlerContext ctx) {
// String result = null;
//
// try {
// result = JsonUtil.obj2Json(ResultModel.error(ResultStatus.REQUEST_ERROR));
// } catch (IOException ex) {
// ex.printStackTrace();
// }
// FullHttpRequestUtils.send(ctx, result, HttpResponseStatus.OK);
// }
/**
* 建立连接时，返回消息
*/
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
if (logger.isDebugEnabled()) {
logger.debug("连接的客户端地址:{}", ctx.channel().remoteAddress());
}
ctx.writeAndFlush("客户端" + InetAddress.getLocalHost().getHostName() + "成功与服务端建立连接！ ");
super.channelActive(ctx);
}
}
common/src/main/java/com/sien/common/tillo/netty/core/NettyChannelInitializer.java 0 → 100644
package com.sien.common.tillo.netty.core;
import com.sien.common.tillo.app_ws.model.Constants;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.codec.http.HttpObjectAggregator;
import io.netty.handler.codec.http.HttpServerCodec;
import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;
import io.netty.handler.stream.ChunkedWriteHandler;
import org.springframework.stereotype.Component;
import javax.annotation.Resource;
@Component
public class NettyChannelInitializer extends ChannelInitializer<SocketChannel> {
@Resource
private ChannelInboundHandler channelInboundHandler;
@Override
protected void initChannel(SocketChannel ch) {
ChannelPipeline pipeline = ch.pipeline();
// http
pipeline.addLast(new HttpServerCodec());
pipeline.addLast(new ChunkedWriteHandler());
pipeline.addLast(new HttpObjectAggregator(2048576000));
// 服务端api接口
pipeline.addLast("SingleHttpRequestHandler", channelInboundHandler);
// "/appws"路径 升级长连接
pipeline.addLast("appWebSocketServerProtocolHandler", new WebSocketServerProtocolHandler(Constants.APP_WS_URL));
}
}
common/src/main/java/com/sien/common/tillo/netty/core/NettyStart.java 0 → 100644
package com.sien.common.tillo.netty.core;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
@Component
@Slf4j
public class NettyStart {
private final NettyChannelInitializer nettyChannelInitializer;
private static EventLoopGroup boss = new NioEventLoopGroup(1);
private static EventLoopGroup work = new NioEventLoopGroup();
private static ServerBootstrap serverBootstrap = new ServerBootstrap();
static {
serverBootstrap.group(boss, work);
//Netty4使用对象池，重用缓冲区
serverBootstrap.childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
serverBootstrap.option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
//设置 心跳保活 socket 的参数选项 keepAlive
serverBootstrap.childOption(ChannelOption.SO_KEEPALIVE, true);
// 设置不延迟发送TCP_NODELAY=true
serverBootstrap.childOption(ChannelOption.TCP_NODELAY, true);
// 初始化服务端可连接队列
serverBootstrap.option(ChannelOption.SO_BACKLOG, 1000);
// 配置io模型为nio非阻塞
serverBootstrap.channel(NioServerSocketChannel.class);
}
public NettyStart(NettyChannelInitializer nettyChannelInitializer) {
this.nettyChannelInitializer = nettyChannelInitializer;
}
/**
* Netty创建全部都是实现自AbstractBootstrap。
* 客户端的是Bootstrap，服务端的则是 ServerBootstrap。
**/
public void run(int port) {
log.info( "启动netty");
try {
//设置过滤器
serverBootstrap.childHandler(nettyChannelInitializer);
// 服务器绑定端口监听
ChannelFuture f = serverBootstrap.bind(port).sync();
f.channel().closeFuture().sync();
} catch (Exception e) {
e.printStackTrace();
} finally {
//关闭EventLoopGroup，释放掉所有资源包括创建的线程
boss.shutdownGracefully();
work.shutdownGracefully();
}
}
}
common/src/main/java/com/sien/common/tillo/netty/handler/NettyApiRequest.java 0 → 100644
package com.sien.common.tillo.netty.handler;
import com.sien.common.tillo.app_ws.AppImHandler;
import com.sien.common.tillo.app_ws.model.Constants;
import com.sien.common.tillo.app_ws.model.RequestHeaderConstants;
import com.sien.common.tillo.app_ws.utils.FullHttpRequestUtils;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.http.FullHttpRequest;
import io.netty.handler.codec.http.HttpResponseStatus;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;
import javax.annotation.Resource;
import java.util.Map;
/**
* @Description 聊天模块 http请求处理
* @Author hewei hwei1233@163.com
* @Date 2019-07-19
*/
@Component
public class NettyApiRequest {
private final Logger logger = LoggerFactory.getLogger(NettyApiRequest.class);
// @Resource
// private AppUserChannelsService appUserChannelsService;
@Resource
private AppImHandler appImHandler;
/**
* http请求接收
*
* @param ctx
* @param msg
* @param httpRequest
* @throws Exception
*/
public void handle(ChannelHandlerContext ctx, Object msg, FullHttpRequest httpRequest) throws Exception {
if (!(msg instanceof FullHttpRequest)) {
// String context = JsonUtil.obj2Json(ResultModel.error(ResultStatus.REQUEST_ERROR));
String context = "JsonUtil.obj2Json(ResultModel.error(ResultStatus.REQUEST_ERROR))";
FullHttpRequestUtils.send(ctx, context, HttpResponseStatus.OK);
return;
}
String path = httpRequest.uri();
String body = FullHttpRequestUtils.getBody(httpRequest);
if (logger.isDebugEnabled()) {
logger.debug("httpRequest:\n" + httpRequest.toString() + "\n" + body);
}
if (path.contains(Constants.APP_WS_URL)) {
/*
app聊天http升级webSocket
*/
this.initAppWs(ctx, httpRequest);
}
}
/**
* app 初始化websocket
*
* @param ctx
* @param httpRequest
* @throws org.apache.http.MethodNotSupportedException
*/
private void initAppWs(ChannelHandlerContext ctx, FullHttpRequest httpRequest) throws Exception {
Map<String, String> paramMap = FullHttpRequestUtils.parameterParse(httpRequest);
String token = paramMap.get(Constants.TOKEN);
String deviceId = paramMap.get(RequestHeaderConstants.DEVICE_ID);
String language = paramMap.get(RequestHeaderConstants.LANGUAGE);
String platform = paramMap.get(RequestHeaderConstants.PLATFORM);
// 当前版本号 VERSIONNO
String versionNo = paramMap.get(RequestHeaderConstants.VERSION_NO);
// 请求头userId
String headerUserId = paramMap.get(RequestHeaderConstants.USER_ID);
// 设置uri前缀
httpRequest.setUri(Constants.APP_WS_URL);
// 保持当前连接
ctx.fireChannelRead(httpRequest.retain());
// 设置属性值 userid - channel
// ctx.channel().attr(AppUserChannelsService.USER_ID).set(userId);
// ctx.channel().attr(AppUserChannelsService.LANGUAGE).set(language);
// 添加长连接handler
// ctx.pipeline().addAfter("appWebSocketServerProtocolHandler","appImHandler", appImHandler);
ctx.pipeline().addLast("appImHandler", appImHandler);
// 保存用户上下文对象
// appUserChannelsService.put(String.valueOf(headerUserId), (NioSocketChannel) ctx.channel());
//移除当前api处理handler， 不再参与长连接处理
ctx.pipeline().remove("SingleHttpRequestHandler");
}
}
config/src/main/resources/config/application-dev.yml
...@@ -12,13 +12,13 @@ spring-boot-plus: ...@@ -12,13 +12,13 @@ spring-boot-plus:
request-log-format: false request-log-format: false
response-log-format: false response-log-format: false
domain: https://c7a30ed052d8.ngrok.io domain: https://767827238379.ngrok.io
spring: spring:
datasource: datasource:
url: jdbc:mysql://47.99.47.225:3306/sien?serverTimezone=UTC&useUnicode=true&characterEncoding=utf8&useSSL=false&allowPublicKeyRetrieval=true url: jdbc:mysql://localhost:3306/wecloud_im?serverTimezone=UTC&useUnicode=true&characterEncoding=utf8&useSSL=false&allowPublicKeyRetrieval=true
username: root username: root
password: temple123456 password: 123
# Redis配置 # Redis配置
redis: redis:
......
config/src/main/resources/config/application.yml
...@@ -2,7 +2,7 @@ ...@@ -2,7 +2,7 @@
############################# 访问路径、端口tomcat start ############################# ############################# 访问路径、端口tomcat start #############################
server: server:
port: 80 port: 8082
servlet: servlet:
context-path: /api context-path: /api
tomcat: tomcat:
...@@ -186,6 +186,8 @@ spring-boot-plus: ...@@ -186,6 +186,8 @@ spring-boot-plus:
- /user/registerOrLogin,/user/login - /user/registerOrLogin,/user/login
- /sms/registerOrLoginCode - /sms/registerOrLoginCode
- /wechatUser/check - /wechatUser/check
- /app/noRole
- /wxapi/checkToken - /wxapi/checkToken
# 排除静态资源 # 排除静态资源
- /static/**,/templates/** - /static/**,/templates/**
......
pom.xml
...@@ -65,7 +65,7 @@ ...@@ -65,7 +65,7 @@
<module>framework</module> <module>framework</module>
<module>generator</module> <module>generator</module>
<module>common</module> <module>common</module>
<module>api-app</module> <!-- <module>api-app</module>-->
<!-- <module>distribution</module>--> <!-- <module>distribution</module>-->
<!-- <module>admin</module>--> <!-- <module>admin</module>-->
<!-- <module>api-system</module>--> <!-- <module>api-system</module>-->
...@@ -294,11 +294,11 @@ ...@@ -294,11 +294,11 @@
<version>${project.version}</version> <version>${project.version}</version>
</dependency> </dependency>
<dependency> <!-- <dependency>-->
<groupId>io.geekidea.springbootplus</groupId> <!-- <groupId>io.geekidea.springbootplus</groupId>-->
<artifactId>api-app</artifactId> <!-- <artifactId>api-app</artifactId>-->
<version>${project.version}</version> <!-- <version>${project.version}</version>-->
</dependency> <!-- </dependency>-->
<dependency> <dependency>
<groupId>io.geekidea.springbootplus</groupId> <groupId>io.geekidea.springbootplus</groupId>
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment