diff --git a/code/link56/link56-backend/src/main/java/cn/link56/common/config/WebMvcConfig.java b/code/link56/link56-backend/src/main/java/cn/link56/common/config/WebMvcConfig.java index 9dab5cc4..e58c2754 100644 --- a/code/link56/link56-backend/src/main/java/cn/link56/common/config/WebMvcConfig.java +++ b/code/link56/link56-backend/src/main/java/cn/link56/common/config/WebMvcConfig.java @@ -24,6 +24,8 @@ public class WebMvcConfig implements WebMvcConfigurer { "/company/recommend", "/company/recommend/hot", + "/post/tag", + "/forum/post/page", "/forum/post/detail", "/forum/post/getRecommendList", diff --git a/code/link56/link56-backend/src/main/java/cn/link56/im/NettyServer.java b/code/link56/link56-backend/src/main/java/cn/link56/im/NettyServer.java index c7658b15..cb0be399 100644 --- a/code/link56/link56-backend/src/main/java/cn/link56/im/NettyServer.java +++ b/code/link56/link56-backend/src/main/java/cn/link56/im/NettyServer.java @@ -1,7 +1,11 @@ package cn.link56.im; import cn.link56.im.config.NettyConfig; +import cn.link56.im.handler.AuthHandler; +import cn.link56.im.handler.DispatchHandler; +import cn.link56.im.handler.MessageHandler; import cn.link56.im.handler.ServerHandler; +import io.netty.handler.timeout.IdleStateHandler; import org.springframework.stereotype.Component; import io.netty.bootstrap.ServerBootstrap; import io.netty.channel.*; @@ -18,6 +22,7 @@ import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler; import javax.annotation.PostConstruct; import javax.annotation.PreDestroy; import javax.annotation.Resource; +import java.util.concurrent.TimeUnit; @Component public class NettyServer { @@ -27,6 +32,13 @@ public class NettyServer { private EventLoopGroup bossGroup; private EventLoopGroup workerGroup; + @Resource + private AuthHandler authHandler; + @Resource + private ServerHandler serverHandler; + @Resource + private DispatchHandler dispatchHandler; + @PostConstruct public void start() { new Thread(this::runServer, "Netty-ImServer-Thread").start(); @@ -52,7 +64,17 @@ public class NettyServer { pipeline.addLast(new ChunkedWriteHandler()); pipeline.addLast(new CorsHandler(CorsConfig.withAnyOrigin().build())); pipeline.addLast(new WebSocketServerProtocolHandler(nettyConfig.getPath(), null, true, 65536)); - pipeline.addLast(new ServerHandler()); + // 4. 空闲检测（60 秒没收到客户端任何消息就断） + pipeline.addLast("idle-handler", + new IdleStateHandler( + 60, + 0, + 0, + TimeUnit.SECONDS + )); + pipeline.addLast(serverHandler); + pipeline.addLast(authHandler); + pipeline.addLast(dispatchHandler); } }); diff --git a/code/link56/link56-backend/src/main/java/cn/link56/im/enums/MessageTypeEnums.java b/code/link56/link56-backend/src/main/java/cn/link56/im/enums/MessageTypeEnums.java index 5fc54428..fe1a88d7 100644 --- a/code/link56/link56-backend/src/main/java/cn/link56/im/enums/MessageTypeEnums.java +++ b/code/link56/link56-backend/src/main/java/cn/link56/im/enums/MessageTypeEnums.java @@ -2,6 +2,29 @@ package cn.link56.im.enums; import lombok.Getter; +@Getter public enum MessageTypeEnums { - PING,PONG,ONLINE,OFFLINE,PRIVATE,GROUP + PING(0), + PONG(1), + LOGIN(2), + LOGOUT(3), + ONLINE(4), + OFFLINE(5), + PRIVATE(6), + GROUP(7); + + private final int code; + + MessageTypeEnums(int code) { + this.code = code; + } + + public static MessageTypeEnums valueOf(int code) { + for (MessageTypeEnums type : values()) { + if (type.code == code) { + return type; + } + } + throw new IllegalArgumentException("未知消息类型: " + code); + } } diff --git a/code/link56/link56-backend/src/main/java/cn/link56/im/handler/AuthHandler.java b/code/link56/link56-backend/src/main/java/cn/link56/im/handler/AuthHandler.java new file mode 100644 index 00000000..1b3aef5d --- /dev/null +++ b/code/link56/link56-backend/src/main/java/cn/link56/im/handler/AuthHandler.java @@ -0,0 +1,49 @@ +package cn.link56.im.handler; + +import cn.link56.auth.token.model.TokenPayload; +import cn.link56.auth.token.service.AuthTokenService; +import cn.link56.im.enums.MessageTypeEnums; +import cn.link56.im.message.MessageRequest; +import cn.link56.im.model.dto.OnlineDTO; +import cn.link56.im.storage.SessionManager; +import com.alibaba.druid.util.StringUtils; +import io.netty.channel.ChannelHandler; +import io.netty.channel.ChannelHandlerContext; +import io.netty.channel.ChannelInboundHandlerAdapter; +import org.springframework.stereotype.Component; + +import javax.annotation.Resource; + +@ChannelHandler.Sharable +@Component +public class AuthHandler extends ChannelInboundHandlerAdapter { + @Resource + private AuthTokenService authTokenService; + + @Override + public void channelRead(ChannelHandlerContext ctx, Object msg) + throws Exception { + + if (!(msg instanceof MessageRequest)) { + ctx.fireChannelRead(msg); + return; + } + + + MessageRequest req = (MessageRequest) msg; + Integer type = req.getType(); + + if (type == MessageTypeEnums.PING.getCode() || type == MessageTypeEnums.ONLINE.getCode()){ + ctx.fireChannelRead(req); + return; + } + + // 如果该通道未验证 + if (!SessionManager.isAuthenticated(ctx.channel())) { + ctx.close(); + return; + } + + ctx.fireChannelRead(req); + } +} diff --git a/code/link56/link56-backend/src/main/java/cn/link56/im/handler/DispatchHandler.java b/code/link56/link56-backend/src/main/java/cn/link56/im/handler/DispatchHandler.java new file mode 100644 index 00000000..756fe622 --- /dev/null +++ b/code/link56/link56-backend/src/main/java/cn/link56/im/handler/DispatchHandler.java @@ -0,0 +1,26 @@ +package cn.link56.im.handler; + +import cn.link56.im.message.MessageRequest; +import io.netty.channel.ChannelHandler; +import io.netty.channel.ChannelHandlerContext; +import io.netty.channel.SimpleChannelInboundHandler; +import org.springframework.stereotype.Component; + +import javax.annotation.Resource; + +/** + * 转发消息 + */ +@ChannelHandler.Sharable +@Component +public class DispatchHandler extends SimpleChannelInboundHandler { + @Resource + private MessageHandler messageHandler; + + @Override + protected void channelRead0(ChannelHandlerContext ctx, + MessageRequest request) { + + messageHandler.handleMessage(ctx.channel(), request); + } +} diff --git a/code/link56/link56-backend/src/main/java/cn/link56/im/handler/MessageHandler.java b/code/link56/link56-backend/src/main/java/cn/link56/im/handler/MessageHandler.java index cbf9f71c..34a12b5d 100644 --- a/code/link56/link56-backend/src/main/java/cn/link56/im/handler/MessageHandler.java +++ b/code/link56/link56-backend/src/main/java/cn/link56/im/handler/MessageHandler.java @@ -1,18 +1,28 @@ package cn.link56.im.handler; +import cn.link56.auth.token.model.TokenPayload; +import cn.link56.auth.token.service.AuthTokenService; import cn.link56.im.enums.MessageTypeEnums; import cn.link56.im.message.MessageRequest; import cn.link56.im.message.MessageResponse; +import cn.link56.im.model.dto.OnlineDTO; +import cn.link56.im.storage.SessionManager; +import cn.link56.im.utils.MessageDataConverter; import cn.link56.im.utils.MessageWriter; import io.netty.channel.Channel; import io.netty.channel.ChannelHandlerContext; import org.springframework.stereotype.Component; +import javax.annotation.Resource; + @Component public class MessageHandler { + @Resource + private AuthTokenService authTokenService; // 处理消息 public void handleMessage(Channel channel, MessageRequest request) { + System.out.println(request); switch (MessageTypeEnums.valueOf(request.getType())) { case PING: handlePing(channel, request); @@ -41,22 +51,32 @@ public class MessageHandler { // 上线 private void handleOnline(Channel channel, MessageRequest request) { - // 验证 - Long from = request.getFrom(); - Object data = request.getData(); // companyId userId + OnlineDTO dto = MessageDataConverter.convert(request.getData(), OnlineDTO.class); + TokenPayload tokenPayload = authTokenService.parsePayload(dto.getToken()); - // 账号 Token + if (tokenPayload == null) { + MessageWriter.writeToChannel(channel,MessageResponse.error("Token无效")); + return; + } + + Long companyId = tokenPayload.getCompanyId(); + Long userId = tokenPayload.getUserId(); + + // 添加 Session 并自动标记已认证 + SessionManager.addSession(companyId, userId, channel); + + MessageWriter.writeToChannel(channel, MessageResponse.success(MessageTypeEnums.ONLINE)); } + // 下线逻辑 private void handleOffline(Channel channel, MessageRequest request) { - // 下线逻辑 } + // 私聊逻辑 private void handlePrivate(Channel channel, MessageRequest request) { - // 私聊逻辑 } + // 群聊逻辑 private void handleGroup(Channel channel, MessageRequest request) { - // 群聊逻辑 } } diff --git a/code/link56/link56-backend/src/main/java/cn/link56/im/handler/ServerHandler.java b/code/link56/link56-backend/src/main/java/cn/link56/im/handler/ServerHandler.java index 233cfa04..c5c7c4fc 100644 --- a/code/link56/link56-backend/src/main/java/cn/link56/im/handler/ServerHandler.java +++ b/code/link56/link56-backend/src/main/java/cn/link56/im/handler/ServerHandler.java @@ -1,19 +1,20 @@ package cn.link56.im.handler; import cn.link56.im.message.MessageRequest; +import cn.link56.im.storage.SessionManager; import com.fasterxml.jackson.databind.ObjectMapper; import io.netty.channel.ChannelHandler; import io.netty.channel.ChannelHandlerContext; import io.netty.channel.SimpleChannelInboundHandler; import io.netty.handler.codec.http.websocketx.TextWebSocketFrame; import io.netty.handler.timeout.IdleStateEvent; +import org.springframework.stereotype.Component; +@Component @ChannelHandler.Sharable public class ServerHandler extends SimpleChannelInboundHandler { private static final ObjectMapper MAPPER = new ObjectMapper(); - private final MessageHandler messageHandler = new MessageHandler(); - @Override public void handlerAdded(ChannelHandlerContext ctx) { System.out.println("客户端连接: " + ctx.channel().id().asShortText()); @@ -22,13 +23,15 @@ public class ServerHandler extends SimpleChannelInboundHandler Session */ private static final ConcurrentHashMap sessions = new ConcurrentHashMap<>(); + /** Channel 是否已认证 */ + private static final AttributeKey AUTH_KEY = + AttributeKey.valueOf("AUTHENTICATED"); + + /** Channel 绑定的 userId */ + private static final AttributeKey USER_ID_KEY = + AttributeKey.valueOf("USER_ID"); + + /** Channel 绑定的 companyId */ + private static final AttributeKey COMPANY_ID_KEY = + AttributeKey.valueOf("COMPANY_ID"); + + /* ================= 认证相关 ================= */ + + public static void markAuthenticated(Channel channel, Long companyId, Long userId) { + if (channel == null || companyId == null || userId == null) return; + + channel.attr(AUTH_KEY).set(true); + channel.attr(USER_ID_KEY).set(userId); + channel.attr(COMPANY_ID_KEY).set(companyId); + } + + public static boolean isAuthenticated(Channel channel) { + if (channel == null) return false; + return Boolean.TRUE.equals(channel.attr(AUTH_KEY).get()); + } + + public static Long getUserId(Channel channel) { + return channel.attr(USER_ID_KEY).get(); + } + + public static Long getCompanyId(Channel channel) { + return channel.attr(COMPANY_ID_KEY).get(); + } + + /* ================= Session 管理 ================= */ - /** 添加会话，支持同一个用户多端登录 */ public static void addSession(Long companyId, Long userId, Channel channel) { if (companyId == null || userId == null || channel == null) return; - // 通过 compute 方法保证线程安全 + // 线程安全地添加 Channel 到 Session sessions.compute(userId, (key, session) -> { if (session == null) { - // 新建 Session session = new Session(companyId, userId); } - // 添加新的 Channel 到 Session session.addChannel(channel); + + markAuthenticated(channel, companyId, userId); + return session; }); } - /** 移除某个 Channel */ - public static void removeChannel(Long userId, Channel channel) { - if (userId == null || channel == null) return; + + public static void removeChannel(Channel channel) { + if (channel == null) return; + + Long userId = getUserId(channel); + if (userId == null) return; Session session = sessions.get(userId); if (session != null) { session.removeChannel(channel); - // 如果用户所有连接都断开，删除 Session if (!session.isOnline()) { sessions.remove(userId); } } + + // 清理 Channel 上的状态 + channel.attr(AUTH_KEY).set(null); + channel.attr(USER_ID_KEY).set(null); + channel.attr(COMPANY_ID_KEY).set(null); } /** 获取用户 Session */ @@ -43,7 +89,6 @@ public class SessionManager { return sessions.get(userId); } - /** 检查用户是否在线 */ public static boolean isOnline(Long userId) { Session session = sessions.get(userId); return session != null && session.isOnline(); diff --git a/code/link56/link56-backend/src/main/java/cn/link56/im/utils/MessageDataConverter.java b/code/link56/link56-backend/src/main/java/cn/link56/im/utils/MessageDataConverter.java new file mode 100644 index 00000000..c77f3723 --- /dev/null +++ b/code/link56/link56-backend/src/main/java/cn/link56/im/utils/MessageDataConverter.java @@ -0,0 +1,11 @@ +package cn.link56.im.utils; + +import com.fasterxml.jackson.databind.ObjectMapper; + +public class MessageDataConverter { + private static final ObjectMapper MAPPER = new ObjectMapper(); + + public static T convert(Object data, Class clazz) { + return MAPPER.convertValue(data, clazz); + } +} diff --git a/code/link56/link56-backend/src/main/java/cn/link56/im/utils/MessageWriter.java b/code/link56/link56-backend/src/main/java/cn/link56/im/utils/MessageWriter.java index ef29d95e..d50fd787 100644 --- a/code/link56/link56-backend/src/main/java/cn/link56/im/utils/MessageWriter.java +++ b/code/link56/link56-backend/src/main/java/cn/link56/im/utils/MessageWriter.java @@ -14,18 +14,20 @@ import java.util.Set; public class MessageWriter { private static final ObjectMapper MAPPER = new ObjectMapper(); - - public static void writeToUser(Long userId,Object message){ - if (ObjectUtils.isEmpty(userId) || ObjectUtils.isEmpty(message)){ + public static void writeToUser(Long userId, Object message) { + if (userId == null || message == null) { return; } Session session = SessionManager.getSession(userId); - Set channels = session.getChannels(); - if (channels.isEmpty()) return; + if (session == null || session.getChannels().isEmpty()) { + return; + } - for (Channel ch : channels) { - writeToChannel(ch, message); + for (Channel ch : session.getChannels()) { + if (ch != null && ch.isActive()) { + writeToChannel(ch, message); + } } } @@ -40,6 +42,8 @@ public class MessageWriter { String payload = MAPPER.writeValueAsString(message); // 发送给客户端 channel.writeAndFlush(new TextWebSocketFrame(payload)); + System.out.println("===发送消息==="); + System.out.println(payload); } catch (Exception e) { System.out.println("===发送消息失败==="); e.printStackTrace(); diff --git a/code/link56/link56-backend/src/main/resources/application.yml b/code/link56/link56-backend/src/main/resources/application.yml index 0800ae45..006f6ba6 100644 --- a/code/link56/link56-backend/src/main/resources/application.yml +++ b/code/link56/link56-backend/src/main/resources/application.yml @@ -35,7 +35,7 @@ mybatis-plus: cache-enabled: true use-generated-keys: true default-executor-type: simple - log-impl: org.apache.ibatis.logging.stdout.StdOutImpl +# log-impl: org.apache.ibatis.logging.stdout.StdOutImpl map-underscore-to-camel-case: true global-config: db-config: diff --git a/code/link56/link56-frontend-new/.env.test b/code/link56/link56-frontend-new/.env.test index 88c5adfe..9ab3d032 100644 --- a/code/link56/link56-frontend-new/.env.test +++ b/code/link56/link56-frontend-new/.env.test @@ -1,2 +1,2 @@ VITE_SERVER_BASEURL = 'http://127.0.0.1:8084' -VITE_WEBSOCKET_BASEURL = 'ws://127.0.0.1:8083/websocket' \ No newline at end of file +VITE_WEBSOCKET_BASEURL = 'ws://127.0.0.1:8090/ws' \ No newline at end of file diff --git a/code/link56/link56-frontend-new/src/App.vue b/code/link56/link56-frontend-new/src/App.vue index 35b218ae..36d3e56b 100644 --- a/code/link56/link56-frontend-new/src/App.vue +++ b/code/link56/link56-frontend-new/src/App.vue @@ -1,20 +1,32 @@ + + + +