mirror of
https://github.com/zongzibinbin/MallChat.git
synced 2026-03-14 06:03:42 +08:00
@@ -6,7 +6,6 @@ import com.auth0.jwt.algorithms.Algorithm;
|
|||||||
import com.auth0.jwt.interfaces.DecodedJWT;
|
import com.auth0.jwt.interfaces.DecodedJWT;
|
||||||
import com.auth0.jwt.interfaces.JWTVerifier;
|
import com.auth0.jwt.interfaces.JWTVerifier;
|
||||||
import lombok.extern.slf4j.Slf4j;
|
import lombok.extern.slf4j.Slf4j;
|
||||||
import org.apache.commons.lang3.StringUtils;
|
|
||||||
import org.junit.Test;
|
import org.junit.Test;
|
||||||
|
|
||||||
import java.util.Date;
|
import java.util.Date;
|
||||||
@@ -18,7 +17,7 @@ public class CreateTokenTest {
|
|||||||
@Test
|
@Test
|
||||||
public void create(){
|
public void create(){
|
||||||
String token = JWT.create()
|
String token = JWT.create()
|
||||||
.withClaim("uid", 123L) // 只存一个uid信息,其他的自己去redis查
|
.withClaim("uid", 10004L) // 只存一个uid信息,其他的自己去redis查
|
||||||
.withClaim("createTime", new Date())
|
.withClaim("createTime", new Date())
|
||||||
.sign(Algorithm.HMAC256("dsfsdfsdfsdfsd")); // signature
|
.sign(Algorithm.HMAC256("dsfsdfsdfsdfsd")); // signature
|
||||||
log.info("生成的token为 {}",token);
|
log.info("生成的token为 {}",token);
|
||||||
@@ -32,13 +31,4 @@ public class CreateTokenTest {
|
|||||||
log.info("decode error,token:{}", token, e);
|
log.info("decode error,token:{}", token, e);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@Test
|
|
||||||
public void verifyToken(){
|
|
||||||
String token = JWT.create()
|
|
||||||
.withClaim("uid", 1) // 只存一个uid信息,其他的自己去redis查
|
|
||||||
.withClaim("createTime", new Date())
|
|
||||||
.sign(Algorithm.HMAC256("dsfsdfsdfsdfsd")); // signature
|
|
||||||
log.info("生成的token为{}",token);
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -32,7 +32,11 @@ import org.springframework.context.ApplicationEventPublisher;
|
|||||||
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
|
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
|
||||||
import org.springframework.stereotype.Component;
|
import org.springframework.stereotype.Component;
|
||||||
|
|
||||||
import java.util.*;
|
import java.util.Date;
|
||||||
|
import java.util.Map;
|
||||||
|
import java.util.Objects;
|
||||||
|
import java.util.Optional;
|
||||||
|
import java.util.Set;
|
||||||
import java.util.concurrent.ConcurrentHashMap;
|
import java.util.concurrent.ConcurrentHashMap;
|
||||||
import java.util.concurrent.CopyOnWriteArrayList;
|
import java.util.concurrent.CopyOnWriteArrayList;
|
||||||
import java.util.concurrent.locks.Condition;
|
import java.util.concurrent.locks.Condition;
|
||||||
|
|||||||
@@ -1,5 +1,6 @@
|
|||||||
package com.abin.mallchat.custom.user.websocket;
|
package com.abin.mallchat.custom.user.websocket;
|
||||||
|
|
||||||
|
import cn.hutool.core.net.url.UrlBuilder;
|
||||||
import io.netty.channel.ChannelHandlerContext;
|
import io.netty.channel.ChannelHandlerContext;
|
||||||
import io.netty.channel.ChannelInboundHandlerAdapter;
|
import io.netty.channel.ChannelInboundHandlerAdapter;
|
||||||
import io.netty.handler.codec.http.FullHttpRequest;
|
import io.netty.handler.codec.http.FullHttpRequest;
|
||||||
@@ -13,7 +14,16 @@ public class HttpHeadersHandler extends ChannelInboundHandlerAdapter {
|
|||||||
@Override
|
@Override
|
||||||
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
|
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
|
||||||
if (msg instanceof FullHttpRequest) {
|
if (msg instanceof FullHttpRequest) {
|
||||||
HttpHeaders headers = ((FullHttpRequest) msg).headers();
|
FullHttpRequest request = (FullHttpRequest) msg;
|
||||||
|
UrlBuilder urlBuilder = UrlBuilder.ofHttp(request.uri());
|
||||||
|
|
||||||
|
// 获取token参数
|
||||||
|
String token = urlBuilder.getQuery().get("token").toString();
|
||||||
|
NettyUtil.setAttr(ctx.channel(), NettyUtil.TOKEN, token);
|
||||||
|
|
||||||
|
// 获取请求路径
|
||||||
|
request.setUri(urlBuilder.getPath().toString());
|
||||||
|
HttpHeaders headers = request.headers();
|
||||||
String ip = headers.get("X-Real-IP");
|
String ip = headers.get("X-Real-IP");
|
||||||
if (StringUtils.isEmpty(ip)) {//如果没经过nginx,就直接获取远端地址
|
if (StringUtils.isEmpty(ip)) {//如果没经过nginx,就直接获取远端地址
|
||||||
InetSocketAddress address = (InetSocketAddress) ctx.channel().remoteAddress();
|
InetSocketAddress address = (InetSocketAddress) ctx.channel().remoteAddress();
|
||||||
@@ -21,7 +31,10 @@ public class HttpHeadersHandler extends ChannelInboundHandlerAdapter {
|
|||||||
}
|
}
|
||||||
NettyUtil.setAttr(ctx.channel(), NettyUtil.IP, ip);
|
NettyUtil.setAttr(ctx.channel(), NettyUtil.IP, ip);
|
||||||
ctx.pipeline().remove(this);
|
ctx.pipeline().remove(this);
|
||||||
|
ctx.fireChannelRead(request);
|
||||||
|
}else
|
||||||
|
{
|
||||||
|
ctx.fireChannelRead(msg);
|
||||||
}
|
}
|
||||||
ctx.fireChannelRead(msg);
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -1,6 +1,7 @@
|
|||||||
package com.abin.mallchat.custom.user.websocket;
|
package com.abin.mallchat.custom.user.websocket;
|
||||||
|
|
||||||
import io.netty.channel.Channel;
|
import io.netty.channel.Channel;
|
||||||
|
import io.netty.handler.codec.http.websocketx.WebSocketServerHandshaker;
|
||||||
import io.netty.util.Attribute;
|
import io.netty.util.Attribute;
|
||||||
import io.netty.util.AttributeKey;
|
import io.netty.util.AttributeKey;
|
||||||
|
|
||||||
@@ -15,6 +16,7 @@ public class NettyUtil {
|
|||||||
public static AttributeKey<String> TOKEN = AttributeKey.valueOf("token");
|
public static AttributeKey<String> TOKEN = AttributeKey.valueOf("token");
|
||||||
public static AttributeKey<String> IP = AttributeKey.valueOf("ip");
|
public static AttributeKey<String> IP = AttributeKey.valueOf("ip");
|
||||||
public static AttributeKey<Long> UID = AttributeKey.valueOf("uid");
|
public static AttributeKey<Long> UID = AttributeKey.valueOf("uid");
|
||||||
|
public static AttributeKey<WebSocketServerHandshaker> HANDSHAKER_ATTR_KEY = AttributeKey.valueOf(WebSocketServerHandshaker.class, "HANDSHAKER");
|
||||||
|
|
||||||
public static <T> void setAttr(Channel channel, AttributeKey<T> attributeKey, T data) {
|
public static <T> void setAttr(Channel channel, AttributeKey<T> attributeKey, T data) {
|
||||||
Attribute<T> attr = channel.attr(attributeKey);
|
Attribute<T> attr = channel.attr(attributeKey);
|
||||||
|
|||||||
@@ -10,6 +10,7 @@ import io.netty.channel.socket.SocketChannel;
|
|||||||
import io.netty.channel.socket.nio.NioServerSocketChannel;
|
import io.netty.channel.socket.nio.NioServerSocketChannel;
|
||||||
import io.netty.handler.codec.http.HttpObjectAggregator;
|
import io.netty.handler.codec.http.HttpObjectAggregator;
|
||||||
import io.netty.handler.codec.http.HttpServerCodec;
|
import io.netty.handler.codec.http.HttpServerCodec;
|
||||||
|
import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;
|
||||||
import io.netty.handler.logging.LogLevel;
|
import io.netty.handler.logging.LogLevel;
|
||||||
import io.netty.handler.logging.LoggingHandler;
|
import io.netty.handler.logging.LoggingHandler;
|
||||||
import io.netty.handler.stream.ChunkedWriteHandler;
|
import io.netty.handler.stream.ChunkedWriteHandler;
|
||||||
@@ -88,7 +89,7 @@ public class NettyWebSocketServer {
|
|||||||
* 4. WebSocketServerProtocolHandler 核心功能是把 http协议升级为 ws 协议,保持长连接;
|
* 4. WebSocketServerProtocolHandler 核心功能是把 http协议升级为 ws 协议,保持长连接;
|
||||||
* 是通过一个状态码 101 来切换的
|
* 是通过一个状态码 101 来切换的
|
||||||
*/
|
*/
|
||||||
pipeline.addLast(new WebSocketHandshakeHandler());
|
pipeline.addLast(new WebSocketServerProtocolHandler("/"));
|
||||||
// 自定义handler ,处理业务逻辑
|
// 自定义handler ,处理业务逻辑
|
||||||
pipeline.addLast(new NettyWebSocketServerHandler());
|
pipeline.addLast(new NettyWebSocketServerHandler());
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -19,10 +19,12 @@ import lombok.extern.slf4j.Slf4j;
|
|||||||
@Slf4j
|
@Slf4j
|
||||||
public class NettyWebSocketServerHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> {
|
public class NettyWebSocketServerHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> {
|
||||||
|
|
||||||
|
private WebSocketService webSocketService;
|
||||||
|
|
||||||
// 当web客户端连接后,触发该方法
|
// 当web客户端连接后,触发该方法
|
||||||
@Override
|
@Override
|
||||||
public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
|
public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
|
||||||
// getService().connect(ctx.channel());
|
this.webSocketService = getService();
|
||||||
}
|
}
|
||||||
|
|
||||||
// 客户端离线
|
// 客户端离线
|
||||||
@@ -45,7 +47,7 @@ public class NettyWebSocketServerHandler extends SimpleChannelInboundHandler<Tex
|
|||||||
}
|
}
|
||||||
|
|
||||||
private void userOffLine(ChannelHandlerContext ctx) {
|
private void userOffLine(ChannelHandlerContext ctx) {
|
||||||
getService().removed(ctx.channel());
|
this.webSocketService.removed(ctx.channel());
|
||||||
ctx.channel().close();
|
ctx.channel().close();
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -66,10 +68,10 @@ public class NettyWebSocketServerHandler extends SimpleChannelInboundHandler<Tex
|
|||||||
userOffLine(ctx);
|
userOffLine(ctx);
|
||||||
}
|
}
|
||||||
} else if (evt == WebSocketServerProtocolHandler.ServerHandshakeStateEvent.HANDSHAKE_COMPLETE) {
|
} else if (evt == WebSocketServerProtocolHandler.ServerHandshakeStateEvent.HANDSHAKE_COMPLETE) {
|
||||||
getService().connect(ctx.channel());
|
this.webSocketService.connect(ctx.channel());
|
||||||
String token = NettyUtil.getAttr(ctx.channel(), NettyUtil.TOKEN);
|
String token = NettyUtil.getAttr(ctx.channel(), NettyUtil.TOKEN);
|
||||||
if (StrUtil.isNotBlank(token)) {
|
if (StrUtil.isNotBlank(token)) {
|
||||||
getService().authorize(ctx.channel(), new WSAuthorize(token));
|
this.webSocketService.authorize(ctx.channel(), new WSAuthorize(token));
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
super.userEventTriggered(ctx, evt);
|
super.userEventTriggered(ctx, evt);
|
||||||
@@ -93,13 +95,13 @@ public class NettyWebSocketServerHandler extends SimpleChannelInboundHandler<Tex
|
|||||||
WSReqTypeEnum wsReqTypeEnum = WSReqTypeEnum.of(wsBaseReq.getType());
|
WSReqTypeEnum wsReqTypeEnum = WSReqTypeEnum.of(wsBaseReq.getType());
|
||||||
switch (wsReqTypeEnum) {
|
switch (wsReqTypeEnum) {
|
||||||
case LOGIN:
|
case LOGIN:
|
||||||
getService().handleLoginReq(ctx.channel());
|
this.webSocketService.handleLoginReq(ctx.channel());
|
||||||
log.info("请求二维码 = " + msg.text());
|
log.info("请求二维码 = " + msg.text());
|
||||||
break;
|
break;
|
||||||
case HEARTBEAT:
|
case HEARTBEAT:
|
||||||
break;
|
break;
|
||||||
case AUTHORIZE:
|
case AUTHORIZE:
|
||||||
getService().authorize(ctx.channel(), JSONUtil.toBean(wsBaseReq.getData(), WSAuthorize.class));
|
this.webSocketService.authorize(ctx.channel(), JSONUtil.toBean(wsBaseReq.getData(), WSAuthorize.class));
|
||||||
log.info("主动认证 = " + msg.text());
|
log.info("主动认证 = " + msg.text());
|
||||||
break;
|
break;
|
||||||
default:
|
default:
|
||||||
|
|||||||
@@ -1,42 +0,0 @@
|
|||||||
package com.abin.mallchat.custom.user.websocket;
|
|
||||||
|
|
||||||
|
|
||||||
import io.netty.channel.ChannelFuture;
|
|
||||||
import io.netty.channel.ChannelFutureListener;
|
|
||||||
import io.netty.channel.ChannelHandlerContext;
|
|
||||||
import io.netty.channel.ChannelInboundHandlerAdapter;
|
|
||||||
import io.netty.handler.codec.http.FullHttpRequest;
|
|
||||||
import io.netty.handler.codec.http.websocketx.WebSocketServerHandshaker;
|
|
||||||
import io.netty.handler.codec.http.websocketx.WebSocketServerHandshakerFactory;
|
|
||||||
import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;
|
|
||||||
|
|
||||||
public class WebSocketHandshakeHandler extends ChannelInboundHandlerAdapter {
|
|
||||||
@Override
|
|
||||||
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
|
|
||||||
if (msg instanceof FullHttpRequest) {
|
|
||||||
FullHttpRequest request = (FullHttpRequest) msg;
|
|
||||||
String token = request.headers().get("Sec-Websocket-Protocol");
|
|
||||||
NettyUtil.setAttr(ctx.channel(), NettyUtil.TOKEN, token);
|
|
||||||
// 构建WebSocket握手处理器
|
|
||||||
WebSocketServerHandshakerFactory handshakeFactory = new WebSocketServerHandshakerFactory(
|
|
||||||
request.uri(), token, false);
|
|
||||||
WebSocketServerHandshaker handshake = handshakeFactory.newHandshaker(request);
|
|
||||||
final ChannelFuture handshakeFuture = handshake.handshake(ctx.channel(), request);
|
|
||||||
ctx.pipeline().remove(this);
|
|
||||||
handshakeFuture.addListener(new ChannelFutureListener() {
|
|
||||||
@Override
|
|
||||||
public void operationComplete(ChannelFuture future) {
|
|
||||||
if (!future.isSuccess()) {
|
|
||||||
ctx.fireExceptionCaught(future.cause());
|
|
||||||
} else {
|
|
||||||
// 手动触发WebSocket握手状态事件
|
|
||||||
ctx.fireUserEventTriggered(
|
|
||||||
WebSocketServerProtocolHandler.ServerHandshakeStateEvent.HANDSHAKE_COMPLETE);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
});
|
|
||||||
} else {
|
|
||||||
super.channelRead(ctx, msg);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
Reference in New Issue
Block a user