Heap on master filled with JSON-RPC message Strings (eclipse-che#6676)

Author: Dmytro Kulieshov

core/che-core-api-core/pom.xml

@@ -86,6 +86,10 @@
             <groupId>org.eclipse.che.core</groupId>
             <artifactId>che-core-commons-lang</artifactId>
         </dependency>
+        <dependency>
+            <groupId>org.eclipse.che.core</groupId>
+            <artifactId>che-core-commons-schedule</artifactId>
+        </dependency>
         <dependency>
             <groupId>org.everrest</groupId>
             <artifactId>everrest-core</artifactId>

core/che-core-api-core/src/main/java/org/eclipse/che/api/core/websocket/impl/MessagesReSender.java

@@ -10,15 +10,15 @@
  */
 package org.eclipse.che.api.core.websocket.impl;
 
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.LinkedList;
-import java.util.List;
+import com.google.common.collect.EvictingQueue;
 import java.util.Map;
 import java.util.Optional;
+import java.util.Queue;
+import java.util.concurrent.ConcurrentHashMap;
 import javax.inject.Inject;
 import javax.inject.Singleton;
 import javax.websocket.Session;
+import org.eclipse.che.commons.schedule.ScheduleDelay;
 
 /**
  * Instance is responsible for re-sending messages that were not sent during the period when WEB
@@ -29,57 +29,77 @@
  */
 @Singleton
 public class MessagesReSender {
+
   private static final int MAX_MESSAGES = 100;
 
   private final WebSocketSessionRegistry registry;
 
-  private final Map<String, List<String>> messagesMap = new HashMap<>();
+  private final Map<String, Queue<DelayedMessage>> delayedMessageRegistry =
+      new ConcurrentHashMap<>();
 
   @Inject
   public MessagesReSender(WebSocketSessionRegistry registry) {
     this.registry = registry;
   }
 
-  public void add(String endpointId, String message) {
-    List<String> messages = messagesMap.get(endpointId);
+  @ScheduleDelay(initialDelay = 60, delay = 60)
+  void cleanStaleMessages() {
+    long currentTimeMillis = System.currentTimeMillis();
 
-    if (messages == null) {
-      messages = new LinkedList<>();
-      messagesMap.put(endpointId, messages);
-    }
+    delayedMessageRegistry
+        .values()
+        .forEach(it -> it.removeIf(m -> currentTimeMillis - m.timeMillis > 60_000));
 
-    if (messages.size() <= MAX_MESSAGES) {
-      messages.add(message);
-    }
+    delayedMessageRegistry.values().removeIf(Queue::isEmpty);
+  }
+
+  public void add(String endpointId, String message) {
+
+    delayedMessageRegistry
+        .computeIfAbsent(endpointId, k -> EvictingQueue.create(MAX_MESSAGES))
+        .offer(new DelayedMessage(message));
   }
 
   public void resend(String endpointId) {
-    final List<String> messages = messagesMap.remove(endpointId);
+    Queue<DelayedMessage> delayedMessages = delayedMessageRegistry.remove(endpointId);
 
-    if (messages == null || messages.isEmpty()) {
+    if (delayedMessages == null || delayedMessages.isEmpty()) {
       return;
     }
 
-    final Optional<Session> sessionOptional = registry.get(endpointId);
+    Optional<Session> sessionOptional = registry.get(endpointId);
 
     if (!sessionOptional.isPresent()) {
       return;
     }
 
-    final Session session = sessionOptional.get();
-
-    final List<String> backing = new ArrayList<>(messages);
-    messages.clear();
-
-    for (String message : backing) {
+    Queue<DelayedMessage> backingQueue = EvictingQueue.create(delayedMessages.size());
+    while (!delayedMessages.isEmpty()) {
+      backingQueue.offer(delayedMessages.poll());
+    }
 
+    Session session = sessionOptional.get();
+    for (DelayedMessage delayedMessage : backingQueue) {
       if (session.isOpen()) {
-        session.getAsyncRemote().sendText(message);
+        session.getAsyncRemote().sendText(delayedMessage.message);
       } else {
-        messages.add(message);
+        delayedMessages.add(delayedMessage);
       }
     }
 
-    messagesMap.put(endpointId, messages);
+    if (!delayedMessages.isEmpty()) {
+      delayedMessageRegistry.put(endpointId, delayedMessages);
+    }
+  }
+
+  private static class DelayedMessage {
+
+    private final long timeMillis;
+    private final String message;
+
+    private DelayedMessage(String message) {
+      this.message = message;
+      this.timeMillis = System.currentTimeMillis();
+    }
   }
 }