diff --git a/kernel-d-auth/auth-api/pom.xml b/kernel-d-auth/auth-api/pom.xml index a70084690..d1aed3af1 100644 --- a/kernel-d-auth/auth-api/pom.xml +++ b/kernel-d-auth/auth-api/pom.xml @@ -38,13 +38,6 @@ scanner-api ${roses.version} - - - - cn.stylefeng.roses - socket-api - ${roses.version} - diff --git a/kernel-d-auth/auth-api/src/main/java/cn/stylefeng/roses/kernel/auth/api/pojo/login/LoginUser.java b/kernel-d-auth/auth-api/src/main/java/cn/stylefeng/roses/kernel/auth/api/pojo/login/LoginUser.java index a018aae36..23563947c 100644 --- a/kernel-d-auth/auth-api/src/main/java/cn/stylefeng/roses/kernel/auth/api/pojo/login/LoginUser.java +++ b/kernel-d-auth/auth-api/src/main/java/cn/stylefeng/roses/kernel/auth/api/pojo/login/LoginUser.java @@ -32,7 +32,6 @@ import cn.stylefeng.roses.kernel.auth.api.pojo.login.basic.SimpleRoleInfo; import cn.stylefeng.roses.kernel.auth.api.pojo.login.basic.SimpleUserInfo; import cn.stylefeng.roses.kernel.rule.constants.RuleConstants; import cn.stylefeng.roses.kernel.scanner.api.annotation.field.ChineseDescription; -import cn.stylefeng.roses.kernel.socket.api.expander.SocketConfigExpander; import lombok.Data; import java.io.Serializable; @@ -174,7 +173,6 @@ public class LoginUser implements Serializable { Map params = new HashMap<>(1); params.put("userId", this.userId); - params.put("port", Long.valueOf(SocketConfigExpander.getSocketPort())); return StrUtil.format(this.wsUrl, params); } diff --git a/kernel-d-socket/pom.xml b/kernel-d-socket/pom.xml index e1214e45b..2a3380671 100644 --- a/kernel-d-socket/pom.xml +++ b/kernel-d-socket/pom.xml @@ -17,7 +17,6 @@ socket-api - socket-sdk-websocket socket-business-websocket socket-spring-boot-starter diff --git a/kernel-d-socket/socket-business-websocket/pom.xml b/kernel-d-socket/socket-business-websocket/pom.xml index 48ed6002b..deef4e3fb 100644 --- a/kernel-d-socket/socket-business-websocket/pom.xml +++ b/kernel-d-socket/socket-business-websocket/pom.xml @@ -14,11 +14,11 @@ socket-business-websocket - + + - cn.stylefeng.roses - socket-sdk-websocket - ${roses.version} + org.springframework.boot + spring-boot-starter-websocket @@ -26,6 +26,20 @@ org.springframework.boot spring-boot-starter-web + + + + cn.stylefeng.roses + socket-api + ${roses.version} + + + + + cn.stylefeng.roses + jwt-sdk + ${roses.version} + \ No newline at end of file diff --git a/kernel-d-socket/socket-business-websocket/src/main/java/cn/stylefeng/roses/kernel/socket/business/websocket/config/WebSocketConfig.java b/kernel-d-socket/socket-business-websocket/src/main/java/cn/stylefeng/roses/kernel/socket/business/websocket/config/WebSocketConfig.java new file mode 100644 index 000000000..44fcd185a --- /dev/null +++ b/kernel-d-socket/socket-business-websocket/src/main/java/cn/stylefeng/roses/kernel/socket/business/websocket/config/WebSocketConfig.java @@ -0,0 +1,20 @@ +package cn.stylefeng.roses.kernel.socket.business.websocket.config; + +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.web.socket.server.standard.ServerEndpointExporter; + +/** + * 开启WebSocket支持 + * + * @author majianguo + * @date 2021/6/21 下午5:01 + */ +@Configuration +public class WebSocketConfig { + + @Bean + public ServerEndpointExporter serverEndpointExporter() { + return new ServerEndpointExporter(); + } +} diff --git a/kernel-d-socket/socket-sdk-websocket/src/main/java/cn/stylefeng/roses/kernel/socket/websocket/message/SocketMessageCenter.java b/kernel-d-socket/socket-business-websocket/src/main/java/cn/stylefeng/roses/kernel/socket/business/websocket/message/SocketMessageCenter.java similarity index 94% rename from kernel-d-socket/socket-sdk-websocket/src/main/java/cn/stylefeng/roses/kernel/socket/websocket/message/SocketMessageCenter.java rename to kernel-d-socket/socket-business-websocket/src/main/java/cn/stylefeng/roses/kernel/socket/business/websocket/message/SocketMessageCenter.java index cfa55d114..20bc0f322 100644 --- a/kernel-d-socket/socket-sdk-websocket/src/main/java/cn/stylefeng/roses/kernel/socket/websocket/message/SocketMessageCenter.java +++ b/kernel-d-socket/socket-business-websocket/src/main/java/cn/stylefeng/roses/kernel/socket/business/websocket/message/SocketMessageCenter.java @@ -1,4 +1,4 @@ -package cn.stylefeng.roses.kernel.socket.websocket.message; +package cn.stylefeng.roses.kernel.socket.business.websocket.message; import cn.stylefeng.roses.kernel.socket.api.message.SocketMsgCallbackInterface; import java.util.HashMap; diff --git a/kernel-d-socket/socket-sdk-websocket/src/main/java/cn/stylefeng/roses/kernel/socket/websocket/operator/WebSocketOperator.java b/kernel-d-socket/socket-business-websocket/src/main/java/cn/stylefeng/roses/kernel/socket/business/websocket/operator/WebSocketOperator.java similarity index 87% rename from kernel-d-socket/socket-sdk-websocket/src/main/java/cn/stylefeng/roses/kernel/socket/websocket/operator/WebSocketOperator.java rename to kernel-d-socket/socket-business-websocket/src/main/java/cn/stylefeng/roses/kernel/socket/business/websocket/operator/WebSocketOperator.java index cc1cb1249..282f46194 100644 --- a/kernel-d-socket/socket-sdk-websocket/src/main/java/cn/stylefeng/roses/kernel/socket/websocket/operator/WebSocketOperator.java +++ b/kernel-d-socket/socket-business-websocket/src/main/java/cn/stylefeng/roses/kernel/socket/business/websocket/operator/WebSocketOperator.java @@ -1,4 +1,4 @@ -package cn.stylefeng.roses.kernel.socket.websocket.operator; +package cn.stylefeng.roses.kernel.socket.business.websocket.operator; import cn.hutool.core.util.ObjectUtil; import cn.stylefeng.roses.kernel.socket.api.SocketOperatorApi; @@ -6,10 +6,10 @@ import cn.stylefeng.roses.kernel.socket.api.exception.SocketException; import cn.stylefeng.roses.kernel.socket.api.exception.enums.SocketExceptionEnum; import cn.stylefeng.roses.kernel.socket.api.message.SocketMsgCallbackInterface; import cn.stylefeng.roses.kernel.socket.api.session.pojo.SocketSession; -import cn.stylefeng.roses.kernel.socket.websocket.message.SocketMessageCenter; -import cn.stylefeng.roses.kernel.socket.websocket.operator.channel.GettySocketOperator; -import cn.stylefeng.roses.kernel.socket.websocket.pojo.WebSocketMessageDTO; -import cn.stylefeng.roses.kernel.socket.websocket.session.SessionCenter; +import cn.stylefeng.roses.kernel.socket.business.websocket.message.SocketMessageCenter; +import cn.stylefeng.roses.kernel.socket.business.websocket.pojo.WebSocketMessageDTO; +import cn.stylefeng.roses.kernel.socket.business.websocket.session.SessionCenter; +import cn.stylefeng.roses.kernel.socket.business.websocket.operator.channel.GettySocketOperator; import java.util.Collection; import java.util.List; diff --git a/kernel-d-socket/socket-sdk-websocket/src/main/java/cn/stylefeng/roses/kernel/socket/websocket/operator/channel/GettyChannelExpandInterFace.java b/kernel-d-socket/socket-business-websocket/src/main/java/cn/stylefeng/roses/kernel/socket/business/websocket/operator/channel/GettyChannelExpandInterFace.java similarity index 81% rename from kernel-d-socket/socket-sdk-websocket/src/main/java/cn/stylefeng/roses/kernel/socket/websocket/operator/channel/GettyChannelExpandInterFace.java rename to kernel-d-socket/socket-business-websocket/src/main/java/cn/stylefeng/roses/kernel/socket/business/websocket/operator/channel/GettyChannelExpandInterFace.java index f4a043d66..98e170b31 100644 --- a/kernel-d-socket/socket-sdk-websocket/src/main/java/cn/stylefeng/roses/kernel/socket/websocket/operator/channel/GettyChannelExpandInterFace.java +++ b/kernel-d-socket/socket-business-websocket/src/main/java/cn/stylefeng/roses/kernel/socket/business/websocket/operator/channel/GettyChannelExpandInterFace.java @@ -1,4 +1,4 @@ -package cn.stylefeng.roses.kernel.socket.websocket.operator.channel; +package cn.stylefeng.roses.kernel.socket.business.websocket.operator.channel; import cn.stylefeng.roses.kernel.socket.api.session.SocketSessionOperatorApi; diff --git a/kernel-d-socket/socket-business-websocket/src/main/java/cn/stylefeng/roses/kernel/socket/business/websocket/operator/channel/GettySocketOperator.java b/kernel-d-socket/socket-business-websocket/src/main/java/cn/stylefeng/roses/kernel/socket/business/websocket/operator/channel/GettySocketOperator.java new file mode 100644 index 000000000..e66decd53 --- /dev/null +++ b/kernel-d-socket/socket-business-websocket/src/main/java/cn/stylefeng/roses/kernel/socket/business/websocket/operator/channel/GettySocketOperator.java @@ -0,0 +1,53 @@ +package cn.stylefeng.roses.kernel.socket.business.websocket.operator.channel; + +import com.alibaba.fastjson.JSON; +import javax.websocket.Session; +import java.io.IOException; + +/** + * Socket操作类实现 + *

+ * 简单封装Spring Boot的默认WebSocket + * + * @author majianguo + * @date 2021/6/1 下午3:41 + */ +public class GettySocketOperator implements GettyChannelExpandInterFace { + + /** + * 实际操作的通道 + */ + private Session socketChannel; + + public GettySocketOperator(Session socketChannel) { + this.socketChannel = socketChannel; + } + + @Override + public void writeAndFlush(Object obj) { + try { + socketChannel.getBasicRemote().sendText(JSON.toJSONString(obj)); + } catch (IOException e) { + e.printStackTrace(); + } + } + + @Override + public void writeToChannel(Object obj) { + socketChannel.getAsyncRemote().sendText(JSON.toJSONString(obj)); + } + + @Override + public void close() { + try { + socketChannel.close(); + } catch (IOException e) { + e.printStackTrace(); + } + } + + @Override + public boolean isInvalid() { + return socketChannel.isOpen(); + } +} diff --git a/kernel-d-socket/socket-sdk-websocket/src/main/java/cn/stylefeng/roses/kernel/socket/websocket/pojo/WebSocketMessageDTO.java b/kernel-d-socket/socket-business-websocket/src/main/java/cn/stylefeng/roses/kernel/socket/business/websocket/pojo/WebSocketMessageDTO.java similarity index 95% rename from kernel-d-socket/socket-sdk-websocket/src/main/java/cn/stylefeng/roses/kernel/socket/websocket/pojo/WebSocketMessageDTO.java rename to kernel-d-socket/socket-business-websocket/src/main/java/cn/stylefeng/roses/kernel/socket/business/websocket/pojo/WebSocketMessageDTO.java index d8c6ebbf7..7b024860c 100644 --- a/kernel-d-socket/socket-sdk-websocket/src/main/java/cn/stylefeng/roses/kernel/socket/websocket/pojo/WebSocketMessageDTO.java +++ b/kernel-d-socket/socket-business-websocket/src/main/java/cn/stylefeng/roses/kernel/socket/business/websocket/pojo/WebSocketMessageDTO.java @@ -1,4 +1,4 @@ -package cn.stylefeng.roses.kernel.socket.websocket.pojo; +package cn.stylefeng.roses.kernel.socket.business.websocket.pojo; import cn.stylefeng.roses.kernel.socket.api.SocketOperatorApi; import lombok.Data; diff --git a/kernel-d-socket/socket-business-websocket/src/main/java/cn/stylefeng/roses/kernel/socket/business/websocket/server/WebSocketServer.java b/kernel-d-socket/socket-business-websocket/src/main/java/cn/stylefeng/roses/kernel/socket/business/websocket/server/WebSocketServer.java new file mode 100644 index 000000000..e3c6c0c2b --- /dev/null +++ b/kernel-d-socket/socket-business-websocket/src/main/java/cn/stylefeng/roses/kernel/socket/business/websocket/server/WebSocketServer.java @@ -0,0 +1,149 @@ +package cn.stylefeng.roses.kernel.socket.business.websocket.server; + +import cn.hutool.core.util.ObjectUtil; +import cn.stylefeng.roses.kernel.jwt.api.context.JwtContext; +import cn.stylefeng.roses.kernel.socket.api.enums.ClientMessageTypeEnum; +import cn.stylefeng.roses.kernel.socket.api.enums.ServerMessageTypeEnum; +import cn.stylefeng.roses.kernel.socket.api.message.SocketMsgCallbackInterface; +import cn.stylefeng.roses.kernel.socket.api.session.pojo.SocketSession; +import cn.stylefeng.roses.kernel.socket.business.websocket.message.SocketMessageCenter; +import cn.stylefeng.roses.kernel.socket.business.websocket.operator.channel.GettySocketOperator; +import cn.stylefeng.roses.kernel.socket.business.websocket.pojo.WebSocketMessageDTO; +import cn.stylefeng.roses.kernel.socket.business.websocket.session.SessionCenter; +import com.alibaba.fastjson.JSON; +import lombok.extern.slf4j.Slf4j; +import org.springframework.stereotype.Component; + +import javax.websocket.*; +import javax.websocket.server.PathParam; +import javax.websocket.server.ServerEndpoint; + +/** + * 消息监听处理器 + * + * @author majianguo + * @date 2021/6/1 下午2:35 + */ +@Slf4j +@ServerEndpoint(value = "/webSocket/{userId}") +@Component +public class WebSocketServer { + + /** + * 连接建立调用的方法 + *

+ * 暂时无用，需要在建立连接的时候做一些事情的话可以修改这里 + * + * @param session 会话信息 + * @author majianguo + * @date 2021/6/21 下午5:14 + **/ + @OnOpen + public void onOpen(Session session, @PathParam("userId") String userId) { + } + + /** + * 连接关闭调用的方法 + * + * @param session 会话信息 + * @author majianguo + * @date 2021/6/21 下午5:14 + **/ + @OnClose + public void onClose(Session session) { + SessionCenter.closed(session.getId()); + } + + /** + * 收到消息调用的方法 + * + * @param message 　接收到的消息 + * @param socketChannel 会话信息 + * @author majianguo + * @date 2021/6/21 下午5:14 + **/ + @OnMessage + public void onMessage(String message, Session socketChannel) { + + // 转换为Java对象 + WebSocketMessageDTO WebSocketMessageDTO = JSON.parseObject(message, WebSocketMessageDTO.class); + + // 心跳包 + if (ClientMessageTypeEnum.USER_HEART.getCode().equals(WebSocketMessageDTO.getClientMsgType())) { + // 更新会话最后活跃时间 + SocketSession session = SessionCenter.getSessionBySessionId(socketChannel.getId()); + if (ObjectUtil.isNotEmpty(session)) { + session.setLastActiveTime(System.currentTimeMillis()); + } + } + + // 用户ID为空不处理直接跳过 + if (ObjectUtil.isEmpty(WebSocketMessageDTO.getFormUserId())) { + return; + } + + // 维护通道是否已初始化 + SocketSession socketSession = SessionCenter.getSessionBySessionId(socketChannel.getId()); + if (ObjectUtil.isEmpty(socketSession) && ClientMessageTypeEnum.USER_CONNECTION_AUTHENTICATION.getCode().equals(WebSocketMessageDTO.getClientMsgType())) { + // 操作api包装 + GettySocketOperator gettySocketOperator = new GettySocketOperator(socketChannel); + + // 回复消息 + WebSocketMessageDTO replyMsg = new WebSocketMessageDTO(); + replyMsg.setServerMsgType(ServerMessageTypeEnum.SYS_REPLY_MSG_TYPE.getCode()); + replyMsg.setToUserId(WebSocketMessageDTO.getFormUserId()); + + try { + // 校验token是否合法 + JwtContext.me().validateTokenWithException(WebSocketMessageDTO.getData().toString()); + + // 设置回复内容 + replyMsg.setData(socketChannel.getId()); + + // 创建会话对象 + socketSession = new SocketSession<>(); + socketSession.setSessionId(socketChannel.getId()); + socketSession.setUserId(WebSocketMessageDTO.getFormUserId()); + socketSession.setSocketOperatorApi(gettySocketOperator); + socketSession.setConnectionTime(System.currentTimeMillis()); + + // 维护会话 + SessionCenter.addSocketSession(socketSession); + } finally { + // 回复消息 + gettySocketOperator.writeAndFlush(replyMsg); + } + return; + } + + // 会话建立成功执行业务逻辑 + if (ObjectUtil.isNotEmpty(socketSession)) { + + // 更新最后会话时间 + socketSession.setLastActiveTime(System.currentTimeMillis()); + + // 找到该消息的处理器 + SocketMsgCallbackInterface socketMsgCallbackInterface = SocketMessageCenter.getSocketMsgCallbackInterface(WebSocketMessageDTO.getClientMsgType()); + if (ObjectUtil.isNotEmpty(socketMsgCallbackInterface)) { + // 触发回调 + socketMsgCallbackInterface.callback(WebSocketMessageDTO.getClientMsgType(), WebSocketMessageDTO, socketSession); + } else { + socketChannel.getAsyncRemote().sendText("{\"serverMsgType\":\"404\"}"); + } + } + } + + /** + * 会话发送异常调用的方法 + * + * @param session 会话信息 + * @param error 　错误信息 + * @author majianguo + * @date 2021/6/21 下午5:14 + **/ + @OnError + public void onError(Session session, Throwable error) { + log.error("发生错误"); + error.printStackTrace(); + } +} diff --git a/kernel-d-socket/socket-sdk-websocket/src/main/java/cn/stylefeng/roses/kernel/socket/websocket/session/SessionCenter.java b/kernel-d-socket/socket-business-websocket/src/main/java/cn/stylefeng/roses/kernel/socket/business/websocket/session/SessionCenter.java similarity index 95% rename from kernel-d-socket/socket-sdk-websocket/src/main/java/cn/stylefeng/roses/kernel/socket/websocket/session/SessionCenter.java rename to kernel-d-socket/socket-business-websocket/src/main/java/cn/stylefeng/roses/kernel/socket/business/websocket/session/SessionCenter.java index e5e2d266b..97d93d04a 100644 --- a/kernel-d-socket/socket-sdk-websocket/src/main/java/cn/stylefeng/roses/kernel/socket/websocket/session/SessionCenter.java +++ b/kernel-d-socket/socket-business-websocket/src/main/java/cn/stylefeng/roses/kernel/socket/business/websocket/session/SessionCenter.java @@ -1,8 +1,8 @@ -package cn.stylefeng.roses.kernel.socket.websocket.session; +package cn.stylefeng.roses.kernel.socket.business.websocket.session; import cn.hutool.core.util.ObjectUtil; import cn.stylefeng.roses.kernel.socket.api.session.pojo.SocketSession; -import cn.stylefeng.roses.kernel.socket.websocket.operator.channel.GettySocketOperator; +import cn.stylefeng.roses.kernel.socket.business.websocket.operator.channel.GettySocketOperator; import java.util.ArrayList; import java.util.List; diff --git a/kernel-d-socket/socket-business-websocket/src/main/java/cn/stylefeng/roses/kernel/socket/business/websocket/spring/WebSocketApplicationRunnerImpl.java b/kernel-d-socket/socket-business-websocket/src/main/java/cn/stylefeng/roses/kernel/socket/business/websocket/spring/WebSocketApplicationRunnerImpl.java deleted file mode 100644 index cf542c83c..000000000 --- a/kernel-d-socket/socket-business-websocket/src/main/java/cn/stylefeng/roses/kernel/socket/business/websocket/spring/WebSocketApplicationRunnerImpl.java +++ /dev/null @@ -1,45 +0,0 @@ -package cn.stylefeng.roses.kernel.socket.business.websocket.spring; - -import cn.stylefeng.roses.kernel.socket.api.expander.SocketConfigExpander; -import cn.stylefeng.roses.kernel.socket.websocket.server.WebSocketServer; -import com.gettyio.core.channel.config.ServerConfig; -import lombok.extern.slf4j.Slf4j; -import org.springframework.boot.ApplicationArguments; -import org.springframework.boot.ApplicationRunner; -import org.springframework.stereotype.Component; -import java.net.StandardSocketOptions; - -/** - * Spring Boot启动完成拉起WebSocket - * - * @author majianguo - * @date 2021/6/2 上午11:06 - */ -@Component -@Slf4j -public class WebSocketApplicationRunnerImpl implements ApplicationRunner { - - @Override - public void run(ApplicationArguments args) { - // 初始化配置对象 - ServerConfig aioServerConfig = new ServerConfig(); - - // 设置host,默认0.0.0.0 - aioServerConfig.setHost(SocketConfigExpander.getSocketHost()); - - // 设置端口号,默认11130 - aioServerConfig.setPort(SocketConfigExpander.getSocketPort()); - - // 设置服务器端内存池最大可分配空间大小，默认512mb，内存池空间可以根据吞吐量设置。 - // 尽量可以设置大一点，因为这不会真正的占用系统内存，只有真正使用时才会分配 - aioServerConfig.setServerChunkSize(SocketConfigExpander.getSocketServerChunkSize()); - - // 设置SocketOptions - aioServerConfig.setOption(StandardSocketOptions.SO_RCVBUF, 8192); - - // 启动 - WebSocketServer.run(aioServerConfig); - - log.info("WebSocket Server Start Success!"); - } -} diff --git a/kernel-d-socket/socket-sdk-websocket/README.md b/kernel-d-socket/socket-sdk-websocket/README.md deleted file mode 100644 index 5f1b090ab..000000000 --- a/kernel-d-socket/socket-sdk-websocket/README.md +++ /dev/null @@ -1 +0,0 @@ -socket模块的websocket实现 \ No newline at end of file diff --git a/kernel-d-socket/socket-sdk-websocket/pom.xml b/kernel-d-socket/socket-sdk-websocket/pom.xml deleted file mode 100644 index 1ffc0143b..000000000 --- a/kernel-d-socket/socket-sdk-websocket/pom.xml +++ /dev/null @@ -1,31 +0,0 @@ - - - - cn.stylefeng.roses - kernel-d-socket - 7.0.4 - ../pom.xml - - - 4.0.0 - - socket-sdk-websocket - - - - - cn.stylefeng.roses - socket-api - ${roses.version} - - - - - cn.stylefeng.roses - jwt-sdk - ${roses.version} - - - \ No newline at end of file diff --git a/kernel-d-socket/socket-sdk-websocket/src/main/java/cn/stylefeng/roses/kernel/socket/websocket/operator/channel/GettySocketOperator.java b/kernel-d-socket/socket-sdk-websocket/src/main/java/cn/stylefeng/roses/kernel/socket/websocket/operator/channel/GettySocketOperator.java deleted file mode 100644 index 0e2cbe4a3..000000000 --- a/kernel-d-socket/socket-sdk-websocket/src/main/java/cn/stylefeng/roses/kernel/socket/websocket/operator/channel/GettySocketOperator.java +++ /dev/null @@ -1,46 +0,0 @@ -package cn.stylefeng.roses.kernel.socket.websocket.operator.channel; - -import com.alibaba.fastjson.JSON; -import com.gettyio.core.channel.SocketChannel; -import com.gettyio.expansion.handler.codec.websocket.frame.TextWebSocketFrame; - -/** - * Socket操作类实现 - *

- * 这里使用的是Getty,所以对Getty的SocketChannel对象做简单封装 - * - * @author majianguo - * @date 2021/6/1 下午3:41 - */ -public class GettySocketOperator implements GettyChannelExpandInterFace { - - /** - * 实际操作的通道 - */ - private SocketChannel socketChannel; - - public GettySocketOperator(SocketChannel socketChannel) { - this.socketChannel = socketChannel; - } - - @Override - public void writeAndFlush(Object obj) { - TextWebSocketFrame textWebSocketFrame = new TextWebSocketFrame(JSON.toJSONString(obj)); - socketChannel.writeAndFlush(textWebSocketFrame); - } - - @Override - public void writeToChannel(Object obj) { - socketChannel.writeToChannel(obj); - } - - @Override - public void close() { - socketChannel.close(); - } - - @Override - public boolean isInvalid() { - return socketChannel.isInvalid(); - } -} diff --git a/kernel-d-socket/socket-sdk-websocket/src/main/java/cn/stylefeng/roses/kernel/socket/websocket/server/WebSocketInitializer.java b/kernel-d-socket/socket-sdk-websocket/src/main/java/cn/stylefeng/roses/kernel/socket/websocket/server/WebSocketInitializer.java deleted file mode 100644 index 55d6ba81a..000000000 --- a/kernel-d-socket/socket-sdk-websocket/src/main/java/cn/stylefeng/roses/kernel/socket/websocket/server/WebSocketInitializer.java +++ /dev/null @@ -1,31 +0,0 @@ -package cn.stylefeng.roses.kernel.socket.websocket.server; - -import cn.stylefeng.roses.kernel.socket.websocket.server.handler.WebSocketMessageHandler; -import com.gettyio.core.channel.SocketChannel; -import com.gettyio.core.pipeline.ChannelInitializer; -import com.gettyio.core.pipeline.DefaultChannelPipeline; -import com.gettyio.expansion.handler.codec.websocket.WebSocketDecoder; -import com.gettyio.expansion.handler.codec.websocket.WebSocketEncoder; - -/** - * WebSocket通道责任链对象 - * - * @author majianguo - * @date 2021/6/1 下午2:36 - */ -public class WebSocketInitializer extends ChannelInitializer { - - @Override - public void initChannel(SocketChannel channel) { - // 获取责任链对象 - DefaultChannelPipeline pipeline = channel.getDefaultChannelPipeline(); - - // 先把ws的编解码器添加到责任链前面。注意，只有先通过ws的编解码器，才能解析ws的消息帧， - // 后续的解码器才能继续解析期望得到的结果 - pipeline.addLast(new WebSocketEncoder()); - pipeline.addLast(new WebSocketDecoder()); - - // 添加自定义的消息处理器 - pipeline.addLast(new WebSocketMessageHandler()); - } -} diff --git a/kernel-d-socket/socket-sdk-websocket/src/main/java/cn/stylefeng/roses/kernel/socket/websocket/server/WebSocketServer.java b/kernel-d-socket/socket-sdk-websocket/src/main/java/cn/stylefeng/roses/kernel/socket/websocket/server/WebSocketServer.java deleted file mode 100644 index 3eaed6e89..000000000 --- a/kernel-d-socket/socket-sdk-websocket/src/main/java/cn/stylefeng/roses/kernel/socket/websocket/server/WebSocketServer.java +++ /dev/null @@ -1,70 +0,0 @@ -package cn.stylefeng.roses.kernel.socket.websocket.server; - -import com.gettyio.core.channel.config.ServerConfig; -import com.gettyio.core.channel.starter.AioServerStarter; - -import java.net.StandardSocketOptions; - -/** - * WebSocket服务端 - * - * @author majianguo - * @date 2021/6/1 下午2:40 - */ -public class WebSocketServer { - - /** - * 无参数启动(开发测试使用) - * - * @author majianguo - * @date 2021/6/2 上午11:10 - **/ - public static void start() { - // 初始化配置对象 - ServerConfig aioServerConfig = new ServerConfig(); - - // 设置host,不设置默认localhost - aioServerConfig.setHost("0.0.0.0"); - - // 设置端口号 - aioServerConfig.setPort(11130); - - // 设置服务器端内存池最大可分配空间大小，默认256mb，内存池空间可以根据吞吐量设置。 - // 尽量可以设置大一点，因为这不会真正的占用系统内存，只有真正使用时才会分配 - aioServerConfig.setServerChunkSize(512 * 1024 * 1024); - - // 设置数据输出器队列大小，一般不用设置这个参数，默认是10*1024*1024 - aioServerConfig.setBufferWriterQueueSize(10 * 1024 * 1024); - - // 设置读取缓存块大小，一般不用设置这个参数，默认128字节 - aioServerConfig.setReadBufferSize(2048); - - // 设置内存池等待分配内存的最大阻塞时间，默认是1秒 - aioServerConfig.setChunkPoolBlockTime(2000); - - // 设置SocketOptions - aioServerConfig.setOption(StandardSocketOptions.SO_RCVBUF, 8192); - - // 启动 - run(aioServerConfig); - } - - /** - * 启动Socket服务 - * - * @param aioServerConfig 服务器配置 - * @author majianguo - * @date 2021/6/1 下午2:40 - **/ - public static void run(ServerConfig aioServerConfig) { - final AioServerStarter starter = new AioServerStarter(aioServerConfig); - starter.channelInitializer(new WebSocketInitializer()); - try { - // 启动服务 - starter.start(); - } catch (Exception e) { - e.printStackTrace(); - } - } - -} diff --git a/kernel-d-socket/socket-sdk-websocket/src/main/java/cn/stylefeng/roses/kernel/socket/websocket/server/handler/WebSocketMessageHandler.java b/kernel-d-socket/socket-sdk-websocket/src/main/java/cn/stylefeng/roses/kernel/socket/websocket/server/handler/WebSocketMessageHandler.java deleted file mode 100644 index ebf9bee60..000000000 --- a/kernel-d-socket/socket-sdk-websocket/src/main/java/cn/stylefeng/roses/kernel/socket/websocket/server/handler/WebSocketMessageHandler.java +++ /dev/null @@ -1,116 +0,0 @@ -package cn.stylefeng.roses.kernel.socket.websocket.server.handler; - -import cn.hutool.core.util.ObjectUtil; -import cn.stylefeng.roses.kernel.jwt.api.context.JwtContext; -import cn.stylefeng.roses.kernel.socket.api.enums.ClientMessageTypeEnum; -import cn.stylefeng.roses.kernel.socket.api.enums.ServerMessageTypeEnum; -import cn.stylefeng.roses.kernel.socket.api.message.SocketMsgCallbackInterface; -import cn.stylefeng.roses.kernel.socket.websocket.message.SocketMessageCenter; -import cn.stylefeng.roses.kernel.socket.api.session.pojo.SocketSession; -import cn.stylefeng.roses.kernel.socket.websocket.pojo.WebSocketMessageDTO; -import cn.stylefeng.roses.kernel.socket.websocket.session.SessionCenter; -import cn.stylefeng.roses.kernel.socket.websocket.operator.channel.GettySocketOperator; -import com.alibaba.fastjson.JSON; -import com.gettyio.core.channel.SocketChannel; -import com.gettyio.core.pipeline.in.SimpleChannelInboundHandler; -import com.gettyio.expansion.handler.codec.websocket.frame.TextWebSocketFrame; -import com.gettyio.expansion.handler.codec.websocket.frame.WebSocketFrame; -import lombok.extern.slf4j.Slf4j; - -import java.nio.charset.StandardCharsets; - -/** - * 消息监听处理器 - * - * @author majianguo - * @date 2021/6/1 下午2:35 - */ -@Slf4j -public class WebSocketMessageHandler extends SimpleChannelInboundHandler { - - @Override - public void channelAdded(SocketChannel aioChannel) { - log.info(aioChannel.getChannelId() + " connection successful."); - } - - @Override - public void channelClosed(SocketChannel aioChannel) { - log.info(aioChannel.getChannelId() + " disconnected"); - SessionCenter.closed(aioChannel.getChannelId()); - } - - @Override - public void channelRead0(SocketChannel socketChannel, WebSocketFrame webSocketFrame) { - - if (webSocketFrame instanceof TextWebSocketFrame) { - String data = new String(webSocketFrame.getPayloadData(), StandardCharsets.UTF_8); - - // 转换为Java对象 - WebSocketMessageDTO WebSocketMessageDTO = JSON.toJavaObject(JSON.parseObject(data), WebSocketMessageDTO.class); - - // 心跳包 - if (ClientMessageTypeEnum.USER_HEART.getCode().equals(WebSocketMessageDTO.getClientMsgType())) { - // 更新会话最后活跃时间 - SocketSession session = SessionCenter.getSessionBySessionId(socketChannel.getChannelId()); - if (ObjectUtil.isNotEmpty(session)) { - session.setLastActiveTime(System.currentTimeMillis()); - } - } - - // 用户ID为空不处理直接跳过 - if (ObjectUtil.isEmpty(WebSocketMessageDTO.getFormUserId())) { - return; - } - - // 维护通道是否已初始化 - SocketSession socketSession = SessionCenter.getSessionBySessionId(socketChannel.getChannelId()); - if (ObjectUtil.isEmpty(socketSession) && ClientMessageTypeEnum.USER_CONNECTION_AUTHENTICATION.getCode().equals(WebSocketMessageDTO.getClientMsgType())) { - // 操作api包装 - GettySocketOperator gettySocketOperator = new GettySocketOperator(socketChannel); - - // 回复消息 - WebSocketMessageDTO replyMsg = new WebSocketMessageDTO(); - replyMsg.setServerMsgType(ServerMessageTypeEnum.SYS_REPLY_MSG_TYPE.getCode()); - replyMsg.setToUserId(WebSocketMessageDTO.getFormUserId()); - - try { - // 校验token是否合法 - JwtContext.me().validateTokenWithException(WebSocketMessageDTO.getData().toString()); - - // 设置回复内容 - replyMsg.setData(socketChannel.getChannelId()); - - // 创建会话对象 - socketSession = new SocketSession<>(); - socketSession.setSessionId(socketChannel.getChannelId()); - socketSession.setUserId(WebSocketMessageDTO.getFormUserId()); - socketSession.setSocketOperatorApi(gettySocketOperator); - socketSession.setConnectionTime(System.currentTimeMillis()); - - // 维护会话 - SessionCenter.addSocketSession(socketSession); - } finally { - // 回复消息 - gettySocketOperator.writeAndFlush(replyMsg); - } - return; - } - - // 会话建立成功执行业务逻辑 - if (ObjectUtil.isNotEmpty(socketSession)) { - - // 更新最后会话时间 - socketSession.setLastActiveTime(System.currentTimeMillis()); - - // 找到该消息的处理器 - SocketMsgCallbackInterface socketMsgCallbackInterface = SocketMessageCenter.getSocketMsgCallbackInterface(WebSocketMessageDTO.getClientMsgType()); - if (ObjectUtil.isNotEmpty(socketMsgCallbackInterface)) { - // 触发回调 - socketMsgCallbackInterface.callback(WebSocketMessageDTO.getClientMsgType(), WebSocketMessageDTO, socketSession); - } else { - socketChannel.writeAndFlush(new TextWebSocketFrame("{\"serverMsgType\":\"404\"}")); - } - } - } - } -} diff --git a/kernel-d-socket/socket-spring-boot-starter/src/main/java/cn/stylefeng/roses/kernel/socket/starter/GunsSocketAutoConfiguration.java b/kernel-d-socket/socket-spring-boot-starter/src/main/java/cn/stylefeng/roses/kernel/socket/starter/GunsSocketAutoConfiguration.java index b5b6eead3..3213ef664 100644 --- a/kernel-d-socket/socket-spring-boot-starter/src/main/java/cn/stylefeng/roses/kernel/socket/starter/GunsSocketAutoConfiguration.java +++ b/kernel-d-socket/socket-spring-boot-starter/src/main/java/cn/stylefeng/roses/kernel/socket/starter/GunsSocketAutoConfiguration.java @@ -25,7 +25,7 @@ package cn.stylefeng.roses.kernel.socket.starter; import cn.stylefeng.roses.kernel.socket.api.SocketOperatorApi; -import cn.stylefeng.roses.kernel.socket.websocket.operator.WebSocketOperator; +import cn.stylefeng.roses.kernel.socket.business.websocket.operator.WebSocketOperator; import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration;