Merge pull request #170 from Zdanquxunhuan/feat/offlineMsg

Author: crossoverJie

README.md

@@ -125,9 +125,7 @@ docker run --rm --name redis -d -p 6379:6379 redis:7.4.0
 ```shell
 git clone https://github.com/crossoverJie/cim.git
 cd cim
-mvn clean package -DskipTests=true
-cd cim-common
-mvn install -DskipTests=true
+mvn clean install -DskipTests=true
 cd cim-server && cim-client && cim-forward-route
 mvn clean package spring-boot:repackage -DskipTests=true
 ```

cim-client-sdk/src/main/java/com/crossoverjie/cim/client/sdk/FetchOfflineMsgJob.java

@@ -0,0 +1,22 @@
+package com.crossoverjie.cim.client.sdk;
+
+import com.crossoverjie.cim.client.sdk.impl.ClientConfigurationData;
+import com.crossoverjie.cim.common.data.construct.RingBufferWheel;
+
+public class FetchOfflineMsgJob extends RingBufferWheel.Task{
+    private static final int INITIAL_DELAY_SECONDS = 5;
+
+    private RouteManager routeManager;
+    private ClientConfigurationData conf;
+
+    public FetchOfflineMsgJob(RouteManager routeManager, ClientConfigurationData conf) {
+        this.routeManager = routeManager;
+        this.conf = conf;
+        setKey(INITIAL_DELAY_SECONDS); //It will be sent with a 5-second delay
+    }
+
+    @Override
+    public void run() {
+        routeManager.fetchOfflineMsgs(conf.getAuth().getUserId());
+    }
+}

cim-client-sdk/src/main/java/com/crossoverjie/cim/client/sdk/RouteManager.java

@@ -10,6 +10,7 @@
 import com.crossoverjie.cim.route.api.RouteApi;
 import com.crossoverjie.cim.route.api.vo.req.ChatReqVO;
 import com.crossoverjie.cim.route.api.vo.req.LoginReqVO;
+import com.crossoverjie.cim.route.api.vo.req.OfflineMsgReqVO;
 import com.crossoverjie.cim.route.api.vo.req.P2PReqVO;
 import com.crossoverjie.cim.route.api.vo.res.CIMServerResVO;
 import java.util.Optional;
@@ -78,4 +79,9 @@ public Set<CIMUserInfo> onlineUser() throws Exception {
         BaseResponse<Set<CIMUserInfo>> onlineUsersResVO = routeApi.onlineUser();
         return onlineUsersResVO.getDataBody();
     }
+
+    public void fetchOfflineMsgs(Long userId){
+        OfflineMsgReqVO offlineMsgReqVO = OfflineMsgReqVO.builder().receiveUserId(userId).build();
+        routeApi.fetchOfflineMsgs(offlineMsgReqVO);
+    }
 }

cim-client-sdk/src/main/java/com/crossoverjie/cim/client/sdk/impl/ClientImpl.java

@@ -1,11 +1,14 @@
 package com.crossoverjie.cim.client.sdk.impl;
 
 import static com.crossoverjie.cim.common.enums.StatusEnum.RECONNECT_FAIL;
+
 import com.crossoverjie.cim.client.sdk.Client;
 import com.crossoverjie.cim.client.sdk.ClientState;
+import com.crossoverjie.cim.client.sdk.FetchOfflineMsgJob;
 import com.crossoverjie.cim.client.sdk.ReConnectManager;
 import com.crossoverjie.cim.client.sdk.RouteManager;
 import com.crossoverjie.cim.client.sdk.io.CIMClientHandleInitializer;
+import com.crossoverjie.cim.common.data.construct.RingBufferWheel;
 import com.crossoverjie.cim.common.exception.CIMException;
 import com.crossoverjie.cim.common.kit.HeartBeatHandler;
 import com.crossoverjie.cim.common.pojo.CIMUserInfo;
@@ -28,13 +31,7 @@
 import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
-import java.util.concurrent.BlockingQueue;
-import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.LinkedBlockingQueue;
-import java.util.concurrent.ThreadFactory;
-import java.util.concurrent.ThreadPoolExecutor;
-import java.util.concurrent.TimeUnit;
+import java.util.concurrent.*;
 import java.util.function.Consumer;
 import lombok.Getter;
 import lombok.extern.slf4j.Slf4j;
@@ -65,6 +62,8 @@ public class ClientImpl extends ClientState implements Client {
     @Getter
     private final Request heartBeatPacket;
 
+    private RingBufferWheel ringBufferWheel;
+
     // Client connected server info
     private CIMServerResVO serverInfo;
 
@@ -93,6 +92,16 @@ public ClientImpl(ClientConfigurationData conf) {
         clientMap.put(conf.getAuth().getUserId(), this);
 
         connectServer(v -> this.conf.getEvent().info("Login success!"));
+
+        postConnectionSetup();
+    }
+
+    /**
+     * 1. Pull offline messages from the server
+     */
+    private void postConnectionSetup() {
+        ringBufferWheel = new RingBufferWheel(Executors.newFixedThreadPool(1));
+        ringBufferWheel.addTask(new FetchOfflineMsgJob(routeManager, conf));
     }
 
     private void connectServer(Consumer<Void> success) {
@@ -226,6 +235,7 @@ public void close() {
         super.setState(ClientState.State.Closed);
         this.routeManager.offLine(this.getAuth().getUserId());
         this.clientMap.remove(this.getAuth().getUserId());
+        ringBufferWheel.stop(true);
     }
 
     @Override

cim-client-sdk/src/main/java/com/crossoverjie/cim/client/sdk/io/CIMClientHandle.java

@@ -67,6 +67,10 @@ protected void channelRead0(ChannelHandlerContext ctx, Response msg) {
         if (msg.getCmd() != BaseCommand.PING) {
             String receiveUserId = msg.getPropertiesMap().get(Constants.MetaKey.RECEIVE_USER_ID);
             ClientImpl client = ClientImpl.getClientMap().get(Long.valueOf(receiveUserId));
+            if (client == null) {
+                log.error("client not found for userId: {}", receiveUserId);
+                return;
+            }
             // callback
             client.getConf().getCallbackThreadPool().execute(() -> {
                 log.info("client address: {} :{}", ctx.channel().remoteAddress(), client);

cim-client-sdk/src/test/java/com/crossoverjie/cim/client/sdk/ClientTest.java

@@ -15,6 +15,8 @@
 import java.util.Set;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicReference;
+
+import com.crossoverjie.cim.route.constant.Constant;
 import lombok.Cleanup;
 import lombok.extern.slf4j.Slf4j;
 import org.awaitility.Awaitility;
@@ -34,7 +36,7 @@ public void tearDown() {
     @Test
     public void groupChat() throws Exception {
         super.starSingleServer();
-        super.startRoute();
+        super.startRoute(Constant.OfflineStoreMode.REDIS);
         String routeUrl = "http://localhost:8083";
         String cj = "crossoverJie";
         String zs = "zs";
@@ -102,12 +104,14 @@ public void groupChat() throws Exception {
         Awaitility.await().untilAsserted(
                 () -> Assertions.assertEquals(msg, client2Receive.get()));
         super.stopSingle();
+        client1.close();
+        client2.close();
     }
 
     @Test
     public void testP2PChat() throws Exception {
         super.starSingleServer();
-        super.startRoute();
+        super.startRoute(Constant.OfflineStoreMode.REDIS);
         String routeUrl = "http://localhost:8083";
         String cj = "cj";
         String zs = "zs";
@@ -238,7 +242,7 @@ public void testP2PChat() throws Exception {
     @Test
     public void testReconnect() throws Exception {
         super.startTwoServer();
-        super.startRoute();
+        super.startRoute(Constant.OfflineStoreMode.REDIS);
 
         String routeUrl = "http://localhost:8083";
         String cj = "cj";
@@ -317,12 +321,14 @@ public void testReconnect() throws Exception {
         Awaitility.await()
                 .untilAsserted(() -> Assertions.assertEquals(msg, client2Receive.get()));
         super.stopTwoServer();
+        client1.close();
+        client2.close();
     }
 
     @Test
     public void offLineAndOnline() throws Exception {
         super.starSingleServer();
-        super.startRoute();
+        super.startRoute(Constant.OfflineStoreMode.REDIS);
         String routeUrl = "http://localhost:8083";
         String cj = "crossoverJie";
         String zs = "zs";
@@ -395,12 +401,14 @@ public void offLineAndOnline() throws Exception {
                 () -> Assertions.assertEquals(msg, client2Receive.get()));
 
         super.stopSingle();
+        client1.close();
+        client2.close();
     }
 
     @Test
     public void testClose() throws Exception {
         super.starSingleServer();
-        super.startRoute();
+        super.startRoute(Constant.OfflineStoreMode.REDIS);
         String routeUrl = "http://localhost:8083";
         String cj = "crossoverJie";
         Long id = super.registerAccount(cj);
@@ -430,7 +438,7 @@ public void testClose() throws Exception {
     @Test
     public void testIncorrectUser() throws Exception {
         super.starSingleServer();
-        super.startRoute();
+        super.startRoute(Constant.OfflineStoreMode.REDIS);
         String routeUrl = "http://localhost:8083";
         String cj = "xx";
         long id = 100L;