Merge pull request #38 from hengboy/master

Server only distributes messages to valid clients

Author: hengboy

message-pipe-server/src/main/java/org/minbox/framework/message/pipe/server/ClientExpiredExecutor.java

@@ -1,9 +1,8 @@
 package org.minbox.framework.message.pipe.server;
 
-import com.alibaba.fastjson.JSON;
 import lombok.extern.slf4j.Slf4j;
-import org.minbox.framework.message.pipe.core.information.ClientInformation;
 import org.minbox.framework.message.pipe.core.ClientStatus;
+import org.minbox.framework.message.pipe.core.information.ClientInformation;
 import org.minbox.framework.message.pipe.server.config.ServerConfiguration;
 import org.springframework.beans.factory.DisposableBean;
 import org.springframework.beans.factory.InitializingBean;
@@ -52,16 +51,22 @@ private void startEliminateExpiredClient() {
             List<ClientInformation> clients = ClientManager.getAllClient();
             if (clients != null && clients.size() > 0) {
                 clients.stream().forEach(client -> {
-                    long intervalSeconds = (client.getLastReportTime() - currentTime) / 1000;
+                    String clientId = ClientManager.getClientId(client.getAddress(), client.getPort());
+                    long intervalSeconds = (currentTime - client.getLastReportTime()) / 1000;
                     if (intervalSeconds > configuration.getExpiredExcludeThresholdSeconds()
                             && ClientStatus.ON_LINE.equals(client.getStatus())) {
                         client.setStatus(ClientStatus.OFF_LINE);
                         ClientManager.updateClientInformation(client);
-                        log.debug("MessagePipe Client：{}，status updated to offline.", JSON.toJSONString(client));
+                        log.info("MessagePipe Client：{}，status updated to offline.", clientId);
+                    } else if (intervalSeconds <= configuration.getExpiredExcludeThresholdSeconds()
+                            && ClientStatus.OFF_LINE.equals(client.getStatus())) {
+                        client.setStatus(ClientStatus.ON_LINE);
+                        ClientManager.updateClientInformation(client);
+                        log.info("MessagePipe Client：{}，status updated to online.", clientId);
                     }
                 });
             }
-        }, 5, 20, TimeUnit.SECONDS);
+        }, 5, configuration.getCheckClientExpiredIntervalSeconds(), TimeUnit.SECONDS);
     }
 
     @Override

message-pipe-server/src/main/java/org/minbox/framework/message/pipe/server/ClientManager.java

@@ -2,6 +2,7 @@
 
 import io.grpc.ManagedChannel;
 import io.grpc.ManagedChannelBuilder;
+import org.minbox.framework.message.pipe.core.ClientStatus;
 import org.minbox.framework.message.pipe.core.information.ClientInformation;
 import org.minbox.framework.message.pipe.core.exception.MessagePipeException;
 import org.springframework.util.ObjectUtils;
@@ -120,11 +121,16 @@ public static void bindClientToPipe(String pipeName, String clientId) {
      * @param pipeName message pipe name
      * @return The message pipe bind clients
      */
-    public static List<ClientInformation> getPipeBindClients(String pipeName) {
+    public static List<ClientInformation> getPipeBindOnLineClients(String pipeName) {
         List<ClientInformation> clientInformationList = new ArrayList<>();
         Set<String> clientIds = PIPE_CLIENTS.get(pipeName);
         if (!ObjectUtils.isEmpty(clientIds)) {
-            clientIds.stream().forEach(clientId -> clientInformationList.add(CLIENTS.get(clientId)));
+            clientIds.stream().forEach(clientId -> {
+                ClientInformation client = CLIENTS.get(clientId);
+                if (ClientStatus.ON_LINE == client.getStatus()) {
+                    clientInformationList.add(client);
+                }
+            });
         }
         return clientInformationList;
     }
@@ -151,7 +157,6 @@ public static synchronized ManagedChannel establishChannel(String clientId) {
     }
 
     /**
-     *
      * @param clientId
      */
     public static void removeChannel(String clientId) {

message-pipe-server/src/main/java/org/minbox/framework/message/pipe/server/config/ServerConfiguration.java

@@ -29,4 +29,8 @@ public class ServerConfiguration {
      * @see org.minbox.framework.message.pipe.server.ClientExpiredExecutor
      */
     private long expiredExcludeThresholdSeconds = 30;
+    /**
+     * Check the client timeout interval in seconds
+     */
+    private long checkClientExpiredIntervalSeconds = 10;
 }

message-pipe-server/src/main/java/org/minbox/framework/message/pipe/server/distribution/MessageDistributionExecutor.java

@@ -83,7 +83,7 @@ private void takeAndSend() {
         takeLock.lock(configuration.getLockTime().getLeaseTime(), configuration.getLockTime().getTimeUnit());
         if (!Thread.currentThread().isInterrupted()) {
             try {
-                List<ClientInformation> clients = ClientManager.getPipeBindClients(this.pipeName);
+                List<ClientInformation> clients = ClientManager.getPipeBindOnLineClients(this.pipeName);
                 String queueLockName = LockNames.MESSAGE_QUEUE.format(this.pipeName);
                 RBlockingQueue<Message> queue = redissonClient.getBlockingQueue(queueLockName);
                 message = queue.peek();
@@ -145,7 +145,9 @@ private boolean sendMessageToClient(Message message, ClientInformation clientInf
         } catch (Exception e) {
             throw e;
         }
-        log.debug("To the client: {}, sending the message is complete.", clientId);
+        if (isSendSuccessfully) {
+            log.debug("To the client: {}, sending the message is complete.", clientId);
+        }
         return isSendSuccessfully;
     }