From 8b552fa63235240d83aa005222c15e915f7a201f Mon Sep 17 00:00:00 2001 From: XSWL1018 <824576966@qq.com> Date: Mon, 14 Oct 2024 15:36:01 +0800 Subject: [PATCH] init --- ruoyi-plugins/pom.xml | 6 + ruoyi-plugins/ruoyi-websocket/pom.xml | 5 + .../ruoyi/websocket/NettyServerRunner.java | 22 +++ .../annotations/NettyWebSocketEndpoint.java | 12 ++ .../NettyWebSocketEndpointHandler.java | 74 ++++++++ .../nettyServer/NettyWebSocketServer.java | 64 +++++++ .../nettyServer/handler/WebSocketHandler.java | 168 ++++++++++++++++++ .../com/ruoyi/websocket/utils/CommonUtil.java | 78 ++++++++ 8 files changed, 429 insertions(+) create mode 100644 ruoyi-plugins/ruoyi-websocket/src/main/java/com/ruoyi/websocket/NettyServerRunner.java create mode 100644 ruoyi-plugins/ruoyi-websocket/src/main/java/com/ruoyi/websocket/annotations/NettyWebSocketEndpoint.java create mode 100644 ruoyi-plugins/ruoyi-websocket/src/main/java/com/ruoyi/websocket/nettyServer/NettyWebSocketEndpointHandler.java create mode 100644 ruoyi-plugins/ruoyi-websocket/src/main/java/com/ruoyi/websocket/nettyServer/NettyWebSocketServer.java create mode 100644 ruoyi-plugins/ruoyi-websocket/src/main/java/com/ruoyi/websocket/nettyServer/handler/WebSocketHandler.java create mode 100644 ruoyi-plugins/ruoyi-websocket/src/main/java/com/ruoyi/websocket/utils/CommonUtil.java diff --git a/ruoyi-plugins/pom.xml b/ruoyi-plugins/pom.xml index 358faf8..ba12644 100644 --- a/ruoyi-plugins/pom.xml +++ b/ruoyi-plugins/pom.xml @@ -13,6 +13,7 @@ 3.10.8 3.5.8 + 4.1.112.Final @@ -72,6 +73,11 @@ ruoyi-mybatis-interceptor ${ruoyi.version} + + io.netty + netty-all + ${netty.version} + diff --git a/ruoyi-plugins/ruoyi-websocket/pom.xml b/ruoyi-plugins/ruoyi-websocket/pom.xml index 66fccdb..ee7ac53 100644 --- a/ruoyi-plugins/ruoyi-websocket/pom.xml +++ b/ruoyi-plugins/ruoyi-websocket/pom.xml @@ -26,6 +26,11 @@ spring-boot-starter-websocket + + io.netty + netty-all + + diff --git a/ruoyi-plugins/ruoyi-websocket/src/main/java/com/ruoyi/websocket/NettyServerRunner.java b/ruoyi-plugins/ruoyi-websocket/src/main/java/com/ruoyi/websocket/NettyServerRunner.java new file mode 100644 index 0000000..9e5632c --- /dev/null +++ b/ruoyi-plugins/ruoyi-websocket/src/main/java/com/ruoyi/websocket/NettyServerRunner.java @@ -0,0 +1,22 @@ +package com.ruoyi.websocket; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.ApplicationArguments; +import org.springframework.boot.ApplicationRunner; +import org.springframework.boot.web.embedded.netty.NettyWebServer; +import org.springframework.stereotype.Component; + +import com.ruoyi.websocket.nettyServer.NettyWebSocketServer; + +@Component +public class NettyServerRunner implements ApplicationRunner { + + @Autowired + private NettyWebSocketServer server; + + @Override + public void run(ApplicationArguments args) throws Exception { + server.start(); + } + +} diff --git a/ruoyi-plugins/ruoyi-websocket/src/main/java/com/ruoyi/websocket/annotations/NettyWebSocketEndpoint.java b/ruoyi-plugins/ruoyi-websocket/src/main/java/com/ruoyi/websocket/annotations/NettyWebSocketEndpoint.java new file mode 100644 index 0000000..dbe0541 --- /dev/null +++ b/ruoyi-plugins/ruoyi-websocket/src/main/java/com/ruoyi/websocket/annotations/NettyWebSocketEndpoint.java @@ -0,0 +1,12 @@ +package com.ruoyi.websocket.annotations; + +import java.lang.annotation.ElementType; +import java.lang.annotation.Retention; +import java.lang.annotation.RetentionPolicy; +import java.lang.annotation.Target; + +@Retention(RetentionPolicy.RUNTIME) +@Target(ElementType.TYPE) +public @interface NettyWebSocketEndpoint { + String path(); +} diff --git a/ruoyi-plugins/ruoyi-websocket/src/main/java/com/ruoyi/websocket/nettyServer/NettyWebSocketEndpointHandler.java b/ruoyi-plugins/ruoyi-websocket/src/main/java/com/ruoyi/websocket/nettyServer/NettyWebSocketEndpointHandler.java new file mode 100644 index 0000000..b2e7eaf --- /dev/null +++ b/ruoyi-plugins/ruoyi-websocket/src/main/java/com/ruoyi/websocket/nettyServer/NettyWebSocketEndpointHandler.java @@ -0,0 +1,74 @@ +package com.ruoyi.websocket.nettyServer; + +import java.nio.channels.Channel; +import java.util.Map; + +import io.netty.channel.ChannelFutureListener; +import io.netty.channel.ChannelHandlerContext; +import io.netty.channel.group.ChannelGroup; +import io.netty.channel.group.DefaultChannelGroup; +import io.netty.handler.codec.http.FullHttpMessage; +import io.netty.handler.codec.http.websocketx.TextWebSocketFrame; +import io.netty.util.concurrent.GlobalEventExecutor; + +public abstract class NettyWebSocketEndpointHandler { + + private final ChannelGroup group = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE); + + private Map pathParam; + + private Map urlParam; + + public void sendAll(String msg) { + group.writeAndFlush(msg); + } + + public static void sendMsg(ChannelHandlerContext context, String msg) { + TextWebSocketFrame textWebSocketFrame = new TextWebSocketFrame(msg); + context.channel().writeAndFlush(textWebSocketFrame); + } + + public abstract void onMessage(ChannelHandlerContext channelHandlerContext, TextWebSocketFrame textWebSocketFrame); + + public abstract void onOpen(ChannelHandlerContext channelHandlerContext, FullHttpMessage fullHttpMessage); + + public abstract void onClose(ChannelHandlerContext channelHandlerContext); + + public abstract void onError(ChannelHandlerContext channelHandlerContext, Throwable throwable); + + public ChannelGroup getGroup() { + return group; + } + + public Map getPathParam() { + return pathParam; + } + + public void setPathParam(Map pathParam) { + this.pathParam = pathParam; + } + + public Map getUrlParam() { + return urlParam; + } + + public void setUrlParam(Map urlParam) { + this.urlParam = urlParam; + } + + public Long getLongPathParam(String key) { + return Long.valueOf(pathParam.get(key)); + } + + public String getPathParam(String key) { + return pathParam.get(key); + } + + public Double getDoublePathParam(String key) { + return Double.parseDouble(pathParam.get(key)); + } + + public void closeChannel(ChannelHandlerContext channelHandlerContext) { + channelHandlerContext.close().addListener(ChannelFutureListener.CLOSE); + } +} \ No newline at end of file diff --git a/ruoyi-plugins/ruoyi-websocket/src/main/java/com/ruoyi/websocket/nettyServer/NettyWebSocketServer.java b/ruoyi-plugins/ruoyi-websocket/src/main/java/com/ruoyi/websocket/nettyServer/NettyWebSocketServer.java new file mode 100644 index 0000000..94dd558 --- /dev/null +++ b/ruoyi-plugins/ruoyi-websocket/src/main/java/com/ruoyi/websocket/nettyServer/NettyWebSocketServer.java @@ -0,0 +1,64 @@ +package com.ruoyi.websocket.nettyServer; + +import org.springframework.beans.factory.annotation.Value; +import org.springframework.stereotype.Component; + +import com.ruoyi.websocket.nettyServer.handler.WebSocketHandler; +import io.netty.bootstrap.ServerBootstrap; +import io.netty.channel.ChannelInitializer; +import io.netty.channel.ChannelOption; +import io.netty.channel.ChannelPipeline; +import io.netty.channel.nio.NioEventLoopGroup; +import io.netty.channel.socket.nio.NioServerSocketChannel; +import io.netty.channel.socket.nio.NioSocketChannel; +import io.netty.handler.codec.http.HttpObjectAggregator; +import io.netty.handler.codec.http.HttpServerCodec; +import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler; + +@Component +public class NettyWebSocketServer { + + private static ServerBootstrap serverBootstrap; + + @Value("${netty.websocket.maxMessageSize}") + private Long messageSize; + + @Value("${netty.websocket.bossThreads}") + private Long bossThreads; + + @Value("${netty.websocket.workerThreads}") + private Long workerThreads; + + @Value("${netty.websocket.port}") + private Long port; + + @Value("${netty.websocket.enable}") + private Boolean enable; + + public ServerBootstrap start() throws InterruptedException { + if (!enable) { + return null; + } + ServerBootstrap serverBootstrap = new ServerBootstrap(); + NioEventLoopGroup boss = new NioEventLoopGroup(4); + NioEventLoopGroup worker = new NioEventLoopGroup(workerThreads.intValue()); + serverBootstrap.group(boss, worker); + serverBootstrap.channel(NioServerSocketChannel.class); + serverBootstrap.childOption(ChannelOption.SO_KEEPALIVE, true); + serverBootstrap.childHandler(new ChannelInitializer() { + @Override + protected void initChannel(NioSocketChannel channel) throws Exception { + ChannelPipeline pipeline = channel.pipeline(); + pipeline.addLast(new HttpServerCodec()); + pipeline.addLast(new HttpObjectAggregator(messageSize.intValue())); + pipeline.addLast(new WebSocketHandler()); + pipeline.addLast(new WebSocketServerProtocolHandler("/", true)); + } + }); + serverBootstrap.bind(port.intValue()).sync(); + System.out.println( + "----------------------------------------------------------------------------------- \n Arknights!"); + NettyWebSocketServer.serverBootstrap = serverBootstrap; + return serverBootstrap; + } +} diff --git a/ruoyi-plugins/ruoyi-websocket/src/main/java/com/ruoyi/websocket/nettyServer/handler/WebSocketHandler.java b/ruoyi-plugins/ruoyi-websocket/src/main/java/com/ruoyi/websocket/nettyServer/handler/WebSocketHandler.java new file mode 100644 index 0000000..79407c0 --- /dev/null +++ b/ruoyi-plugins/ruoyi-websocket/src/main/java/com/ruoyi/websocket/nettyServer/handler/WebSocketHandler.java @@ -0,0 +1,168 @@ +package com.ruoyi.websocket.nettyServer.handler; + +import java.util.concurrent.ConcurrentHashMap; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.stereotype.Component; +import com.ruoyi.common.utils.StringUtils; +import com.ruoyi.websocket.annotations.NettyWebSocketEndpoint; +import com.ruoyi.websocket.nettyServer.NettyWebSocketEndpointHandler; +import com.ruoyi.websocket.utils.CommonUtil; +import io.netty.channel.ChannelFutureListener; +import io.netty.channel.ChannelHandlerContext; +import io.netty.channel.ChannelId; +import io.netty.channel.SimpleChannelInboundHandler; +import io.netty.channel.group.ChannelGroup; +import io.netty.channel.group.DefaultChannelGroup; +import io.netty.handler.codec.http.DefaultFullHttpResponse; +import io.netty.handler.codec.http.FullHttpRequest; +import io.netty.handler.codec.http.HttpResponseStatus; +import io.netty.handler.codec.http.HttpVersion; +import io.netty.handler.codec.http.websocketx.TextWebSocketFrame; +import io.netty.util.concurrent.GlobalEventExecutor; +import jakarta.annotation.PostConstruct; + +import java.lang.reflect.Constructor; +import java.net.URI; +import java.net.URISyntaxException; +import java.util.*; + +@Component +public class WebSocketHandler extends SimpleChannelInboundHandler { + + @Autowired + private List handlers; + + private static final Map uriHandlerMapper = new ConcurrentHashMap<>(); + + public static final ChannelGroup channelGroup = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE); + + public static final Map channelHandlerMap = new ConcurrentHashMap<>(); + + @PostConstruct + private void init() throws URISyntaxException, NoSuchMethodException, SecurityException { + for (NettyWebSocketEndpointHandler handler : handlers) { + Class handlerClass = handler.getClass(); + NettyWebSocketEndpoint annotation = handlerClass.getAnnotation(NettyWebSocketEndpoint.class); + if (annotation == null || StringUtils.isEmpty(annotation.path())) { + throw new RuntimeException("未配置路径的 netty websocket endpoint "); + } + // uriHandlerMap.put(uri.getPath(), handler); + PathMatchModel pathMachModel = parseHandler(annotation.path(), handlerClass); + uriHandlerMapper.put(pathMachModel.path, pathMachModel); + } + } + + @Override + protected void channelRead0(ChannelHandlerContext context, TextWebSocketFrame webSocketFrame) throws Exception { + NettyWebSocketEndpointHandler handler = channelHandlerMap.get(context.channel().id()); + handler.onMessage(context, webSocketFrame); + } + + @Override + public void channelActive(ChannelHandlerContext ctx) throws Exception { + channelGroup.add(ctx.channel()); + } + + @Override + public void channelInactive(ChannelHandlerContext ctx) throws Exception { + NettyWebSocketEndpointHandler handler = channelHandlerMap.get(ctx.channel().id()); + if (handler != null) { + handler.onClose(ctx); + handler.getGroup().remove(ctx.channel()); + } + + channelHandlerMap.remove(ctx.channel().id()); + channelGroup.remove(ctx.channel()); + } + + @Override + public void channelRegistered(ChannelHandlerContext ctx) throws Exception { + + } + + @Override + public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { + channelHandlerMap.get(ctx.channel().id()).onError(ctx, cause); + } + + @Override + public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { + if (msg instanceof FullHttpRequest) { + FullHttpRequest fullHttpRequest = (FullHttpRequest) msg; + if (channelHandlerMap.get(ctx.channel().id()) != null) { + super.channelRead(ctx, fullHttpRequest); + return; + } + URI uri = new URI(fullHttpRequest.uri()); + PathMatchModel mathPathMachModel = mathPathMachModel(uri.getPath()); + if (mathPathMachModel == null) { + ctx.channel() + .writeAndFlush(new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.NOT_FOUND)); + ctx.close().addListener(ChannelFutureListener.CLOSE); + return; + } + NettyWebSocketEndpointHandler newInstance = (NettyWebSocketEndpointHandler) mathPathMachModel.handlerConstructor + .newInstance(); + if (!(mathPathMachModel.pathParams == null || mathPathMachModel.pathParams.isEmpty())) { + newInstance.setPathParam( + CommonUtil.parsePathParam(uri.getPath(), mathPathMachModel.pathParams, mathPathMachModel.path)); + super.channelRead(ctx, msg); + } + newInstance.setUrlParam(CommonUtil.parseQueryParameters(uri.getQuery())); + + channelHandlerMap.put(ctx.channel().id(), newInstance); + newInstance.onOpen(ctx, fullHttpRequest); + } else if (msg instanceof TextWebSocketFrame) { + super.channelRead(ctx, msg); + } + + } + + private static PathMatchModel parseHandler(String path, Class handlerClass) + throws NoSuchMethodException, SecurityException { + List paramName = new ArrayList<>(); + String[] split = path.split("/"); + for (int index = 1; index < split.length; index++) { + String item = split[index]; + if (item.startsWith("{") && item.endsWith("}")) { + paramName.add(item.substring(1, item.length() - 1).trim()); + split[index] = "?"; + } + } + StringBuilder finalPath = new StringBuilder(""); + for (int index = 1; index < split.length; index++) { + finalPath.append("/").append(split[index]); + } + return new PathMatchModel(paramName, finalPath.toString(), handlerClass.getDeclaredConstructor()); + } + + private static PathMatchModel mathPathMachModel(String uri) { + Map map = new HashMap<>(); + for (String key : uriHandlerMapper.keySet()) { + int mathUri = CommonUtil.mathUri(uri, key); + if (mathUri > 0) { + map.put(mathUri, uriHandlerMapper.get(key)); + } + } + if (map.keySet() == null || map.keySet().isEmpty()) { + return null; + } + Integer max = CommonUtil.getMax(map.keySet()); + return map.get(max); + } + + private static final class PathMatchModel { + private final List pathParams; + + private final String path; + + private final Constructor handlerConstructor; + + public PathMatchModel(List pathParams, String path, Constructor handlerConstructor) { + this.pathParams = pathParams; + this.path = path; + this.handlerConstructor = handlerConstructor; + } + + } +} diff --git a/ruoyi-plugins/ruoyi-websocket/src/main/java/com/ruoyi/websocket/utils/CommonUtil.java b/ruoyi-plugins/ruoyi-websocket/src/main/java/com/ruoyi/websocket/utils/CommonUtil.java new file mode 100644 index 0000000..d2b0228 --- /dev/null +++ b/ruoyi-plugins/ruoyi-websocket/src/main/java/com/ruoyi/websocket/utils/CommonUtil.java @@ -0,0 +1,78 @@ +package com.ruoyi.websocket.utils; + +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.Set; + +public class CommonUtil { + + /** + * @param uri + * @param uriTemplates + * @return + */ + public static int mathUri(String uri, String uriTemplate) { + String[] uriSplit = uri.split("/"); + String[] tempalteSplit = uriTemplate.split("/"); + if (uriSplit.length != tempalteSplit.length) { + return -1; + } + int mathLevel = 0; + for (int index = 1; index < tempalteSplit.length; index++) { + if (tempalteSplit[index].equals("?")) { + mathLevel = mathLevel + index; + continue; + } + if (!tempalteSplit[index].equals(uriSplit[index])) { + return -1; + } else { + mathLevel = mathLevel + tempalteSplit.length + 1; + } + } + return mathLevel; + } + + public static Map parseQueryParameters(String query) { + if (query == null || query.isEmpty()) { + return Map.of(); + } + + Map params = new HashMap<>(); + String[] pairs = query.split("&"); + for (String pair : pairs) { + String[] keyValue = pair.split("="); + if (keyValue.length > 1) { + params.put(keyValue[0], keyValue[1]); + } else { + params.put(keyValue[0], ""); + } + } + return params; + } + + public static Map parsePathParam(String uri, List pathParams, String uriTemplate) { + int index = 0; + String[] split = uriTemplate.split("/"); + String[] split2 = uri.split("/"); + Map map = new HashMap<>(); + for (int i = 1; i < split.length; i++) { + if (split[i].equals("?")) { + map.put(pathParams.get(index), split2[i]); + index++; + } + } + return map; + } + + public static Integer getMax(Set set) { + Optional maxNumber = set.stream().max(Integer::compare); + if (maxNumber.isPresent()) { + System.out.println("Max number: " + maxNumber.get()); + } else { + System.out.println("The list is empty"); + } + return maxNumber.get(); + } +}