diff --git a/im-client/src/main/java/net/sopod/soim/client/cmd/CmdEnum.java b/im-client/src/main/java/net/sopod/soim/client/cmd/CmdEnum.java index e165e8d..9adaf6c 100644 --- a/im-client/src/main/java/net/sopod/soim/client/cmd/CmdEnum.java +++ b/im-client/src/main/java/net/sopod/soim/client/cmd/CmdEnum.java @@ -31,6 +31,8 @@ public enum CmdEnum { friends(FriendsHandler.class), + me(MeHandler.class), + help, /** 退出 */ diff --git a/im-client/src/main/java/net/sopod/soim/client/handler/cmd/LoginHandler.java b/im-client/src/main/java/net/sopod/soim/client/handler/cmd/LoginHandler.java index 46e4b9b..13e092f 100644 --- a/im-client/src/main/java/net/sopod/soim/client/handler/cmd/LoginHandler.java +++ b/im-client/src/main/java/net/sopod/soim/client/handler/cmd/LoginHandler.java @@ -50,7 +50,7 @@ public class LoginHandler implements CmdHandler { .setUid(loginRes.getUid()) .setToken(loginRes.getAuthToken()) .build(); - soImSession.connect(clientConfig.getHost(), clientConfig.getPort(), reqTokenAuth); + soImSession.connect(clientConfig.getHost(), clientConfig.getPort(), reqTokenAuth, args.getAccount()); } } diff --git a/im-client/src/main/java/net/sopod/soim/client/handler/cmd/MeHandler.java b/im-client/src/main/java/net/sopod/soim/client/handler/cmd/MeHandler.java new file mode 100644 index 0000000..c413803 --- /dev/null +++ b/im-client/src/main/java/net/sopod/soim/client/handler/cmd/MeHandler.java @@ -0,0 +1,28 @@ +package net.sopod.soim.client.handler.cmd; + +import com.google.inject.Inject; +import com.google.inject.Singleton; +import net.sopod.soim.client.cmd.handler.NonArgsHandler; +import net.sopod.soim.client.logger.Logger; +import net.sopod.soim.client.session.SoImSession; + +/** + * MeHandler + * + * @author tmy + * @date 2022-06-02 15:49 + */ +@Singleton +public class MeHandler extends NonArgsHandler { + + @Inject + private SoImSession soImSession; + + @Override + public void handle() { + Long uid = soImSession.getUid(); + String account = soImSession.getAccount(); + Logger.info("me: " + uid + "|" + account); + } + +} diff --git a/im-client/src/main/java/net/sopod/soim/client/protocol/ImMessageInboundHandler.java b/im-client/src/main/java/net/sopod/soim/client/protocol/ImMessageInboundHandler.java new file mode 100644 index 0000000..39e9e45 --- /dev/null +++ b/im-client/src/main/java/net/sopod/soim/client/protocol/ImMessageInboundHandler.java @@ -0,0 +1,22 @@ +package net.sopod.soim.client.protocol; + +import io.netty.channel.ChannelHandlerContext; +import io.netty.channel.SimpleChannelInboundHandler; +import net.sopod.soim.data.serialize.ImMessage; + +/** + * ImMessageInboundHandler + * + * @author tmy + * @date 2022-06-02 17:51 + */ +public class ImMessageInboundHandler extends SimpleChannelInboundHandler { + + @Override + protected void channelRead0(ChannelHandlerContext channelHandlerContext, ImMessage imMessage) throws Exception { + // 请求序列号，complete 对应 CompletableFuture + int serialNo = imMessage.getSerialNo(); + + } + +} diff --git a/im-client/src/main/java/net/sopod/soim/client/protocol/ImMessageOutboundHandler.java b/im-client/src/main/java/net/sopod/soim/client/protocol/ImMessageOutboundHandler.java new file mode 100644 index 0000000..534aa11 --- /dev/null +++ b/im-client/src/main/java/net/sopod/soim/client/protocol/ImMessageOutboundHandler.java @@ -0,0 +1,28 @@ +package net.sopod.soim.client.protocol; + +import io.netty.channel.ChannelHandlerContext; +import io.netty.channel.ChannelOutboundHandlerAdapter; +import io.netty.channel.ChannelPromise; +import net.sopod.soim.data.serialize.ImMessage; + +/** + * ImMessageOutboundHandler + * + * @author tmy + * @date 2022-06-02 17:53 + */ +public class ImMessageOutboundHandler extends ChannelOutboundHandlerAdapter { + + @Override + public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception { + if (!(msg instanceof ImMessage)) { + return; + } + ImMessage imMessage = (ImMessage) msg; + imMessage.setSerialNo(10086); + // queue.add something..., 调用的地方，通过阻塞方式 FastThreadLocal 获取一个 CompletableFuture + + super.write(ctx, msg, promise); + } + +} diff --git a/im-client/src/main/java/net/sopod/soim/client/protocol/ImMessageRegistry.java b/im-client/src/main/java/net/sopod/soim/client/protocol/ImMessageRegistry.java new file mode 100644 index 0000000..a88e6e8 --- /dev/null +++ b/im-client/src/main/java/net/sopod/soim/client/protocol/ImMessageRegistry.java @@ -0,0 +1,10 @@ +package net.sopod.soim.client.protocol; + +/** + * ImMessageRegistry + * + * @author tmy + * @date 2022-06-02 17:32 + */ +public class ImMessageRegistry { +} diff --git a/im-client/src/main/java/net/sopod/soim/client/protocol/MessageQueueHolder.java b/im-client/src/main/java/net/sopod/soim/client/protocol/MessageQueueHolder.java new file mode 100644 index 0000000..0ee47cd --- /dev/null +++ b/im-client/src/main/java/net/sopod/soim/client/protocol/MessageQueueHolder.java @@ -0,0 +1,18 @@ +package net.sopod.soim.client.protocol; + +/** + * MessageQueueHolder + * 发送一条消息产生一个序列号，在队列中等待响应消息 + * + * @author tmy + * @date 2022-06-02 17:29 + */ +public class MessageQueueHolder { + + + + public void a() { + + } + +} diff --git a/im-client/src/main/java/net/sopod/soim/client/session/SoImSession.java b/im-client/src/main/java/net/sopod/soim/client/session/SoImSession.java index a44383c..d7fca2b 100644 --- a/im-client/src/main/java/net/sopod/soim/client/session/SoImSession.java +++ b/im-client/src/main/java/net/sopod/soim/client/session/SoImSession.java @@ -10,12 +10,14 @@ import io.netty.channel.nio.NioEventLoopGroup; import io.netty.channel.socket.SocketChannel; import io.netty.channel.socket.nio.NioSocketChannel; import net.sopod.soim.client.logger.Logger; +import net.sopod.soim.common.res.R; import net.sopod.soim.common.util.netty.Varint32FrameCodec; import net.sopod.soim.data.serialize.ImMessageCodec; import net.sopod.soim.data.msg.auth.Auth; import net.sopod.soim.data.msg.chat.Chat; import org.slf4j.LoggerFactory; +import java.util.concurrent.CompletableFuture; import java.util.concurrent.atomic.AtomicBoolean; /** @@ -39,12 +41,14 @@ public class SoImSession { private Long uid; + private String account; + @Inject public SoImSession(MessageDispatcher messageDispatcher) { this.messageDispatcher = messageDispatcher; } - public void connect(String host, Integer port, Auth.ReqTokenAuth tokenAuth) { + public void connect(String host, Integer port, Auth.ReqTokenAuth tokenAuth, String account) { // 关闭旧连接 this.close(); @@ -57,7 +61,9 @@ public class SoImSession { protected void initChannel(SocketChannel ch) throws Exception { ch.pipeline() .addLast(new Varint32FrameCodec()) - .addLast(new ImMessageCodec()) + .addLast(new ImMessageCodec.ProtoMsg2ImMessageEncoder()) + .addLast(new ImMessageCodec.ImMessage2ByteEncoder()) + .addLast(new ImMessageCodec.ImMessageDecoder()) .addLast(messageDispatcher); } }); @@ -66,6 +72,7 @@ public class SoImSession { clientChannel = b.connect(host, port).await().channel(); // 连接后立即发送认证消息，10s未认证连接关闭 clientChannel.writeAndFlush(tokenAuth); + this.account = account; } catch (Exception e) { Logger.error("连接服务器失败: {}", e.getMessage()); } @@ -88,17 +95,18 @@ public class SoImSession { /** * 发送消息 */ - public void send(MessageLite message) { + public CompletableFuture send(MessageLite message) { if (!auth.get()) { Logger.error("请先登录"); - return; + return null; } if (clientChannel == null || !clientChannel.isActive()) { Logger.error("连接已关闭"); - return; + return null; } clientChannel.writeAndFlush(message); + return new CompletableFuture<>(); } public void textChat(String receiverName, String message) { @@ -114,6 +122,14 @@ public class SoImSession { this.send(textChat); } + public Long getUid() { + return uid; + } + + public String getAccount() { + return account; + } + /** * 关闭 tcp 连接 */ diff --git a/im-common/pom.xml b/im-common/pom.xml index b4deffc..e8a9208 100644 --- a/im-common/pom.xml +++ b/im-common/pom.xml @@ -42,6 +42,12 @@ ${netty.version} provided + + org.apache.dubbo + dubbo-rpc-api + ${dubbo.version} + provided + \ No newline at end of file diff --git a/im-common/src/main/java/net/sopod/soim/common/dubbo/SoimExceptionFilter.java b/im-common/src/main/java/net/sopod/soim/common/dubbo/SoimExceptionFilter.java new file mode 100644 index 0000000..48e0710 --- /dev/null +++ b/im-common/src/main/java/net/sopod/soim/common/dubbo/SoimExceptionFilter.java @@ -0,0 +1,32 @@ +package net.sopod.soim.common.dubbo; + +import net.sopod.soim.common.dubbo.exception.SoimException; +import org.apache.dubbo.rpc.Invocation; +import org.apache.dubbo.rpc.Invoker; +import org.apache.dubbo.rpc.Result; +import org.apache.dubbo.rpc.filter.ExceptionFilter; +import org.apache.dubbo.rpc.service.GenericService; + +/** + * SoimExceptionFilter + * + * @author tmy + * @date 2022-06-02 11:08 + */ +public class SoimExceptionFilter extends ExceptionFilter { + + private static final String exceptPack = SoimException.class.getPackageName(); + + @Override + public void onResponse(Result appResponse, Invoker invoker, Invocation invocation) { + if (appResponse.hasException() && GenericService.class != invoker.getInterface()) { + // 是自定义异常直接返回 + if (appResponse.getException().getClass().getName() + .startsWith(exceptPack)) { + return; + } + } + super.onResponse(appResponse, invoker, invocation); + } + +} diff --git a/im-common/src/main/java/net/sopod/soim/common/dubbo/exception/DasException.java b/im-common/src/main/java/net/sopod/soim/common/dubbo/exception/DasException.java new file mode 100644 index 0000000..4707f64 --- /dev/null +++ b/im-common/src/main/java/net/sopod/soim/common/dubbo/exception/DasException.java @@ -0,0 +1,33 @@ +package net.sopod.soim.common.dubbo.exception; + +/** + * DasException + * + * @author tmy + * @date 2022-06-02 12:02 + */ +public class DasException extends SoimException { + + private static final long serialVersionUID = -533031075736547194L; + + public DasException() { + super(); + } + + public DasException(String message) { + super(message); + } + + public DasException(String message, Throwable cause) { + super(message, cause); + } + + public DasException(Throwable cause) { + super(cause); + } + + public DasException(String message, Throwable cause, boolean writableStackTrace) { + super(message, cause, writableStackTrace); + } + +} diff --git a/im-common/src/main/java/net/sopod/soim/common/dubbo/exception/LogicException.java b/im-common/src/main/java/net/sopod/soim/common/dubbo/exception/LogicException.java new file mode 100644 index 0000000..d54421d --- /dev/null +++ b/im-common/src/main/java/net/sopod/soim/common/dubbo/exception/LogicException.java @@ -0,0 +1,33 @@ +package net.sopod.soim.common.dubbo.exception; + +/** + * LogicException + * + * @author tmy + * @date 2022-06-02 12:03 + */ +public class LogicException extends SoimException { + + private static final long serialVersionUID = 8123008638375992724L; + + public LogicException() { + super(); + } + + public LogicException(String message) { + super(message); + } + + public LogicException(String message, Throwable cause) { + super(message, cause); + } + + public LogicException(Throwable cause) { + super(cause); + } + + public LogicException(String message, Throwable cause, boolean writableStackTrace) { + super(message, cause, writableStackTrace); + } + +} diff --git a/im-common/src/main/java/net/sopod/soim/common/dubbo/exception/ServiceException.java b/im-common/src/main/java/net/sopod/soim/common/dubbo/exception/ServiceException.java new file mode 100644 index 0000000..efe2739 --- /dev/null +++ b/im-common/src/main/java/net/sopod/soim/common/dubbo/exception/ServiceException.java @@ -0,0 +1,34 @@ +package net.sopod.soim.common.dubbo.exception; + +/** + * 通用服务异常 + * 默认不生成堆栈信息，设置 writableStackTrace 为 true 则生成堆栈信息 + * + * @author 晚星 + * @date 2019/12/30 16:30 + */ +public class ServiceException extends SoimException { + + private static final long serialVersionUID = 6555896535065406344L; + + public ServiceException() { + super(); + } + + public ServiceException(String message) { + super(message); + } + + public ServiceException(String message, Throwable cause) { + super(message, cause); + } + + public ServiceException(Throwable cause) { + super(cause); + } + + public ServiceException(String message, Throwable cause, boolean writableStackTrace) { + super(message, cause, writableStackTrace); + } + +} diff --git a/im-common/src/main/java/net/sopod/soim/common/dubbo/exception/SoimException.java b/im-common/src/main/java/net/sopod/soim/common/dubbo/exception/SoimException.java new file mode 100644 index 0000000..7e62369 --- /dev/null +++ b/im-common/src/main/java/net/sopod/soim/common/dubbo/exception/SoimException.java @@ -0,0 +1,39 @@ +package net.sopod.soim.common.dubbo.exception; + +import java.io.Serializable; + +/** + * SoimException + * + * @author tmy + * @date 2022-06-02 11:29 + */ +public class SoimException extends RuntimeException implements Serializable { + + private static final long serialVersionUID = 7148131411144540939L; + + public SoimException() {} + + public SoimException(String message) { + super(message, null, false, false); + } + + public SoimException(String message, Throwable cause) { + super(message, cause, false, false); + } + + public SoimException(Throwable cause) { + super(cause == null ? null : cause.getMessage(), cause, false, false); + } + + /** + * + * @param message 错误信息 + * @param cause 错误对象 + * @param writableStackTrace 是否启用堆栈追踪 + */ + public SoimException(String message, Throwable cause , boolean writableStackTrace) { + super(message, cause, false, writableStackTrace); + } + +} diff --git a/im-common/src/main/java/net/sopod/soim/common/res/R.java b/im-common/src/main/java/net/sopod/soim/common/res/R.java new file mode 100644 index 0000000..7dd6b23 --- /dev/null +++ b/im-common/src/main/java/net/sopod/soim/common/res/R.java @@ -0,0 +1,11 @@ +package net.sopod.soim.common.res; + +/** + * R + * + * @author tmy + * @date 2022-06-02 15:31 + */ +public class R { + +} diff --git a/im-common/src/main/resources/META-INF/dubbo/org.apache.dubbo.rpc.Filter b/im-common/src/main/resources/META-INF/dubbo/org.apache.dubbo.rpc.Filter new file mode 100644 index 0000000..959e0af --- /dev/null +++ b/im-common/src/main/resources/META-INF/dubbo/org.apache.dubbo.rpc.Filter @@ -0,0 +1 @@ +soimExceptionFilter=net.sopod.soim.common.dubbo.SoimExceptionFilter \ No newline at end of file diff --git a/im-das-api/im-das-user-api/src/main/java/net/sopod/soim/das/user/api/config/ChatMQAutoConfiguration.java b/im-das-api/im-das-user-api/src/main/java/net/sopod/soim/das/user/api/config/ChatMQAutoConfiguration.java new file mode 100644 index 0000000..0d6d557 --- /dev/null +++ b/im-das-api/im-das-user-api/src/main/java/net/sopod/soim/das/user/api/config/ChatMQAutoConfiguration.java @@ -0,0 +1,16 @@ +package net.sopod.soim.das.user.api.config; + +import org.springframework.boot.autoconfigure.condition.ConditionalOnClass; +import org.springframework.context.annotation.Import; + +/** + * ChatRabbitMQAutoConfiguration + * + * @author tmy + * @date 2022-06-02 10:01 + */ +@ConditionalOnClass(name = "org.springframework.amqp.core.Queue") +@Import(ChatPersistentRabbitMQConfiguration.class) +public class ChatMQAutoConfiguration { + +} diff --git a/im-das-api/im-das-user-api/src/main/java/net/sopod/soim/das/user/api/config/ChatRabbitMQAutoConfiguration.java b/im-das-api/im-das-user-api/src/main/java/net/sopod/soim/das/user/api/config/ChatPersistentRabbitMQConfiguration.java similarity index 91% rename from im-das-api/im-das-user-api/src/main/java/net/sopod/soim/das/user/api/config/ChatRabbitMQAutoConfiguration.java rename to im-das-api/im-das-user-api/src/main/java/net/sopod/soim/das/user/api/config/ChatPersistentRabbitMQConfiguration.java index 3644a58..5e057bf 100644 --- a/im-das-api/im-das-user-api/src/main/java/net/sopod/soim/das/user/api/config/ChatRabbitMQAutoConfiguration.java +++ b/im-das-api/im-das-user-api/src/main/java/net/sopod/soim/das/user/api/config/ChatPersistentRabbitMQConfiguration.java @@ -8,6 +8,7 @@ import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.amqp.support.converter.MessageConverter; import org.springframework.beans.factory.support.BeanDefinitionRegistry; import org.springframework.beans.factory.support.RootBeanDefinition; +import org.springframework.boot.autoconfigure.condition.ConditionalOnClass; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.context.annotation.Import; @@ -15,15 +16,14 @@ import org.springframework.context.annotation.ImportBeanDefinitionRegistrar; import org.springframework.core.type.AnnotationMetadata; /** - * ChatRabbitMQAutoConfiguration + * ChatPersistentRabbitMQConfiguration * RabbitMQ的配置类，用来配置队列、交换器、路由等高级信息 * * @author tmy * @date 2022-05-30 23:14 */ -@Import(ChatRabbitMQAutoConfiguration.class) @Configuration -public class ChatRabbitMQAutoConfiguration implements ImportBeanDefinitionRegistrar { +public class ChatPersistentRabbitMQConfiguration implements ImportBeanDefinitionRegistrar { /** * direct直连模式, 按照routingkey分发到指定队列 diff --git a/im-das-api/im-das-user-api/src/main/resources/META-INF/spring.factories b/im-das-api/im-das-user-api/src/main/resources/META-INF/spring.factories index df33bab..d1555ad 100644 --- a/im-das-api/im-das-user-api/src/main/resources/META-INF/spring.factories +++ b/im-das-api/im-das-user-api/src/main/resources/META-INF/spring.factories @@ -1 +1 @@ -org.springframework.boot.autoconfigure.EnableAutoConfiguration=net.sopod.soim.das.user.api.config.ChatRabbitMQAutoConfiguration \ No newline at end of file +org.springframework.boot.autoconfigure.EnableAutoConfiguration=net.sopod.soim.das.user.api.config.ChatMQAutoConfiguration diff --git a/im-das/im-das-user/src/main/java/net/sopod/soim/das/user/amqp/Sender.java b/im-das/im-das-user/src/main/java/net/sopod/soim/das/user/amqp/Sender.java index 737909c..7dbff13 100644 --- a/im-das/im-das-user/src/main/java/net/sopod/soim/das/user/amqp/Sender.java +++ b/im-das/im-das-user/src/main/java/net/sopod/soim/das/user/amqp/Sender.java @@ -23,7 +23,7 @@ import java.util.concurrent.TimeUnit; * @author tmy * @date 2022-05-28 16:23 */ -//@Component +@Component @AllArgsConstructor public class Sender implements ApplicationListener { diff --git a/im-das/im-das-user/src/main/java/net/sopod/soim/das/user/service/FriendDasImpl.java b/im-das/im-das-user/src/main/java/net/sopod/soim/das/user/service/FriendDasImpl.java index f552cec..33d1efa 100644 --- a/im-das/im-das-user/src/main/java/net/sopod/soim/das/user/service/FriendDasImpl.java +++ b/im-das/im-das-user/src/main/java/net/sopod/soim/das/user/service/FriendDasImpl.java @@ -53,6 +53,9 @@ public class FriendDasImpl implements FriendDas { return friendMapper.insert(imFriend); } + /** + * TODO cache + */ @Override public Long getFriendId(Long uid, Long fid) { LambdaQueryWrapper friendQuery = new QueryWrapper().lambda() diff --git a/im-service-api/im-entry-protocol/src/main/java/net/sopod/soim/data/serialize/ImMessage.java b/im-service-api/im-entry-protocol/src/main/java/net/sopod/soim/data/serialize/ImMessage.java index be2a1c2..f2519cf 100644 --- a/im-service-api/im-entry-protocol/src/main/java/net/sopod/soim/data/serialize/ImMessage.java +++ b/im-service-api/im-entry-protocol/src/main/java/net/sopod/soim/data/serialize/ImMessage.java @@ -1,5 +1,6 @@ package net.sopod.soim.data.serialize; +import com.google.protobuf.MessageLite; import io.netty.buffer.ByteBuf; import java.util.Random; @@ -55,6 +56,8 @@ public class ImMessage { private byte[] body; + private transient MessageLite decodeBody; + /** * 从 bytebuf 流读取 */ @@ -159,4 +162,13 @@ public class ImMessage { return MESSAGE_HEAD_LEN + body.length; } + public MessageLite getDecodeBody() { + return decodeBody; + } + + public ImMessage setDecodeBody(MessageLite decodeBody) { + this.decodeBody = decodeBody; + return this; + } + } diff --git a/im-service-api/im-entry-protocol/src/main/java/net/sopod/soim/data/serialize/ImMessageCodec.java b/im-service-api/im-entry-protocol/src/main/java/net/sopod/soim/data/serialize/ImMessageCodec.java index 4ec580a..10110f4 100644 --- a/im-service-api/im-entry-protocol/src/main/java/net/sopod/soim/data/serialize/ImMessageCodec.java +++ b/im-service-api/im-entry-protocol/src/main/java/net/sopod/soim/data/serialize/ImMessageCodec.java @@ -6,6 +6,7 @@ import io.netty.channel.ChannelHandlerContext; import io.netty.channel.CombinedChannelDuplexHandler; import io.netty.handler.codec.ByteToMessageDecoder; import io.netty.handler.codec.MessageToByteEncoder; +import io.netty.handler.codec.MessageToMessageEncoder; import net.sopod.soim.data.proto.ProtoMessageManager; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -22,21 +23,36 @@ import java.util.List; */ public class ImMessageCodec //extends MessageToMessageCodec { extends CombinedChannelDuplexHandler { - public static void main(String[] args) throws ClassNotFoundException { - Type superType = ImMessageCodec.class.getGenericSuperclass(); - String typeName = superType.getTypeName(); - int idx = typeName.indexOf('<'); - String genericName = typeName.substring(idx + 1, typeName.length() - 1); - System.out.println(genericName.trim()); - System.out.println(Arrays.toString(genericName.split(", "))); - } private static final Logger logger = LoggerFactory.getLogger(ImMessageCodec.class); public ImMessageCodec() { super(new ProtoMsgDecoder(), new ProtoMsgEncoder()); } + public static class ImMessageDecoder extends ByteToMessageDecoder { + @Override + protected void decode(ChannelHandlerContext ctx, ByteBuf byteBuf, List out) throws Exception { + ImMessage message = ImMessage.read(byteBuf); + boolean isMagicError; + if ((isMagicError = (message == ImMessage.MAGIC_ERROR)) + || message == ImMessage.PROTOCOL_ERROR) { + logger.warn("decode im message error: {}, remote={}, closing channel.", + isMagicError ? "MagicError" : "ProtocolError", + ctx.channel().remoteAddress()); + ctx.channel().close(); + return; + } + // 解码 protobuf 消息体 + int serviceNo = message.getServiceNo(); + byte[] protoByte = message.getBody(); + MessageLite protoClass = ProtoMessageManager.getProtoInstance(serviceNo); + MessageLite protoMsg = protoClass.getParserForType().parseFrom(protoByte); + message.setDecodeBody(protoMsg); + out.add(message); + } + } + public static class ProtoMsgDecoder extends ByteToMessageDecoder { @Override protected void decode(ChannelHandlerContext ctx, ByteBuf byteBuf, List list) throws Exception { @@ -52,6 +68,7 @@ public class ImMessageCodec //extends MessageToMessageCodec { + @Override + protected void encode(ChannelHandlerContext ctx, MessageLite message, List out) throws Exception { + Integer serialNo = ProtoMessageManager.getSerialNo(message.getClass()); + // TODO unknow class serialNo + ImMessage imMessage = new ImMessage() + .setServiceNo(serialNo) + .setBody(message.toByteArray()); + out.add(imMessage); + } + } + + public static class ImMessage2ByteEncoder extends MessageToByteEncoder { + @Override + protected void encode(ChannelHandlerContext ctx, ImMessage imMessage, ByteBuf out) throws Exception { + imMessage.write(out); + } + } + public static class ProtoMsgEncoder extends MessageToByteEncoder { @Override protected void encode(ChannelHandlerContext ctx, MessageLite message, ByteBuf byteBuf) throws Exception { diff --git a/im-service-api/im-router-api/src/main/java/net/sopod/soim/router/api/service/UserRouteService.java b/im-service-api/im-router-api/src/main/java/net/sopod/soim/router/api/service/UserRouteService.java index 68cfbc4..da7a93d 100644 --- a/im-service-api/im-router-api/src/main/java/net/sopod/soim/router/api/service/UserRouteService.java +++ b/im-service-api/im-router-api/src/main/java/net/sopod/soim/router/api/service/UserRouteService.java @@ -25,7 +25,11 @@ public interface UserRouteService { */ List onlineUserList(String keyword); - Boolean routeTextChat(TextChat textChat); + /** + * @param friendId 好友关系id + * @param textChat 聊天内容 + */ + Boolean routeTextChat(Long friendId, TextChat textChat); List isOnlineUsers(List userIds); diff --git a/im-service/im-logic-user/src/main/java/net/sopod/soim/logic/user/service/ChatServiceImpl.java b/im-service/im-logic-user/src/main/java/net/sopod/soim/logic/user/service/ChatServiceImpl.java index dde6836..c492b1a 100644 --- a/im-service/im-logic-user/src/main/java/net/sopod/soim/logic/user/service/ChatServiceImpl.java +++ b/im-service/im-logic-user/src/main/java/net/sopod/soim/logic/user/service/ChatServiceImpl.java @@ -1,5 +1,8 @@ package net.sopod.soim.logic.user.service; +import net.sopod.soim.common.dubbo.exception.LogicException; +import net.sopod.soim.common.dubbo.exception.ServiceException; +import net.sopod.soim.common.dubbo.exception.SoimException; import net.sopod.soim.das.user.api.model.entity.ImUser; import net.sopod.soim.das.user.api.service.FriendDas; import net.sopod.soim.das.user.api.service.UserDas; @@ -9,6 +12,8 @@ import net.sopod.soim.logic.common.util.RpcContextUtil; import net.sopod.soim.router.api.service.UserRouteService; import org.apache.dubbo.config.annotation.DubboReference; import org.apache.dubbo.config.annotation.DubboService; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import java.util.Objects; @@ -21,6 +26,8 @@ import java.util.Objects; @DubboService public class ChatServiceImpl implements ChatService { + private static final Logger logger = LoggerFactory.getLogger(ChatServiceImpl.class); + @DubboReference private UserRouteService userRouteService; @@ -36,7 +43,8 @@ public class ChatServiceImpl implements ChatService { || Objects.equals(textChat.getReceiverUid(), 0L)) { ImUser receiverUser = userDas.getNormalUserByAccount(textChat.getReceiverName()); if (receiverUser == null) { - return false; + // TODO receiverUid 由客户端传递 + throw new LogicException("聊天对象不存在"); } textChat.setReceiverUid(receiverUser.getId()); } @@ -44,11 +52,11 @@ public class ChatServiceImpl implements ChatService { Long friendId = friendDas.getFriendId(textChat.getUid(), textChat.getReceiverUid()); // 不是好友 if (friendId == null) { - return false; + throw new LogicException("请添加好友后发送消息"); } // 设置调用 router 为消息接受者地址 RpcContextUtil.setContextUid(textChat.getReceiverUid()); - return userRouteService.routeTextChat(textChat); + return userRouteService.routeTextChat(friendId, textChat); } } diff --git a/im-service/im-logic-user/src/main/resources/application.yml b/im-service/im-logic-user/src/main/resources/application.yml index 2c0d44f..da07499 100644 --- a/im-service/im-logic-user/src/main/resources/application.yml +++ b/im-service/im-logic-user/src/main/resources/application.yml @@ -14,4 +14,5 @@ dubbo: check: false provider: retries: 0 - timeout: 2000 \ No newline at end of file + timeout: 2000 + filter: soimExceptionFilter,-exception \ No newline at end of file diff --git a/im-service/im-router/pom.xml b/im-service/im-router/pom.xml index 3d254d5..3d1907c 100644 --- a/im-service/im-router/pom.xml +++ b/im-service/im-router/pom.xml @@ -22,6 +22,11 @@ im-entry-api ${soim.version} + + net.sopod + im-segment-id-api + ${soim.version} + net.sopod im-das-user-api diff --git a/im-service/im-router/src/main/java/net/sopod/soim/router/service/UserRouteServiceImpl.java b/im-service/im-router/src/main/java/net/sopod/soim/router/service/UserRouteServiceImpl.java index af75312..55751ee 100644 --- a/im-service/im-router/src/main/java/net/sopod/soim/router/service/UserRouteServiceImpl.java +++ b/im-service/im-router/src/main/java/net/sopod/soim/router/service/UserRouteServiceImpl.java @@ -1,12 +1,19 @@ package net.sopod.soim.router.service; import net.sopod.soim.common.constant.DubboConstant; +import net.sopod.soim.common.dubbo.exception.ServiceException; import net.sopod.soim.common.util.ImClock; import net.sopod.soim.common.util.StringUtil; +import net.sopod.soim.das.user.api.config.LogicTables; +import net.sopod.soim.das.user.api.model.entity.ImMessage; import net.sopod.soim.das.user.api.model.entity.ImUser; +import net.sopod.soim.das.user.api.mq.ChatQueue; +import net.sopod.soim.das.user.api.mq.ChatQueueType; +import net.sopod.soim.das.user.api.service.FriendDas; import net.sopod.soim.das.user.api.service.UserDas; import net.sopod.soim.entry.api.service.OnlineUserService; import net.sopod.soim.entry.api.service.TextChatService; +import net.sopod.soim.logic.api.segmentid.core.SegmentIdGenerator; import net.sopod.soim.logic.common.model.TextChat; import net.sopod.soim.logic.common.model.UserInfo; import net.sopod.soim.logic.common.util.RpcContextUtil; @@ -20,7 +27,9 @@ import org.apache.dubbo.config.annotation.DubboService; import org.apache.dubbo.rpc.RpcContext; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import org.springframework.amqp.rabbit.core.RabbitTemplate; +import javax.annotation.Resource; import java.util.ArrayList; import java.util.List; import java.util.stream.Collectors; @@ -32,7 +41,7 @@ import java.util.stream.Stream; * @author tmy * @date 2022-04-28 9:47 */ -@DubboService() +@DubboService public class UserRouteServiceImpl implements UserRouteService { private static final Logger logger = LoggerFactory.getLogger(UserRouteServiceImpl.class); @@ -40,12 +49,20 @@ public class UserRouteServiceImpl implements UserRouteService { @DubboReference private UserDas userDas; + @DubboReference + private FriendDas friendDas; + @DubboReference private TextChatService textChatService; @DubboReference private OnlineUserService onlineUserService; + @Resource + private SegmentIdGenerator segmentIdGenerator; + + @Resource + private RabbitTemplate rabbitTemplate; @Override public RegistryRes registryUserEntry(Long uid, String imEntryAddr) { @@ -81,8 +98,16 @@ public class UserRouteServiceImpl implements UserRouteService { * 调用该方法时，将到 im-router 服务的路由 uid 设置为消息接受者的 uid */ @Override - public Boolean routeTextChat(TextChat textChat) { - // TODO 消息队列存储 + public Boolean routeTextChat(Long friendId, TextChat textChat) { + // 通过消息队列持久化存储到db + ImMessage imMessage = new ImMessage() + .setFriendId(friendId) + .setId(segmentIdGenerator.nextId(LogicTables.IM_MESSAGE)) + .setContent(textChat.getMessage()) + .setSender(textChat.getUid()) + .setReceiver(textChat.getReceiverUid()) + .setCreateTime(ImClock.date()); + rabbitTemplate.convertAndSend(ChatQueueType.IM_MESSAGE.getQueueName(), imMessage); RpcContextUtil.setContextUid(textChat.getReceiverUid()); Boolean send = textChatService.sendTextChat(textChat); diff --git a/im-service/im-router/src/main/resources/application.yml b/im-service/im-router/src/main/resources/application.yml index 7812230..9614a91 100644 --- a/im-service/im-router/src/main/resources/application.yml +++ b/im-service/im-router/src/main/resources/application.yml @@ -11,6 +11,11 @@ spring: max-idle: 100 max-wait: 1000 min-idle: 2 + rabbitmq: + host: 124.222.131.236 + port: 3672 + username: soim + password: sopod@rabbit# dubbo: application: @@ -31,3 +36,5 @@ dubbo: retries: 0 # 这里服务重试时会路由到非uid所在对应im-router timeout: 2000 register: false # 不自动注册，数据初始化后注册 + filter: soimExceptionFilter,-exception +