fix(ws): implement periodic ping frame (#460)

* fix(ws): implement periodic ping frame * remove unused workers/schedulers
cryostatio · May 14, 2024 · 94134bd · 94134bd
1 parent d67c297
commit 94134bd
Show file tree

Hide file tree

Showing 3 changed files with 17 additions and 25 deletions.
diff --git a/src/main/java/io/cryostat/Producers.java b/src/main/java/io/cryostat/Producers.java
@@ -16,10 +16,8 @@
 package io.cryostat;
 
 import java.net.URI;
-import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ForkJoinPool;
-import java.util.concurrent.ScheduledExecutorService;
 
 import io.cryostat.core.reports.InterruptibleReportGenerator;
 import io.cryostat.core.sys.Clock;
@@ -32,7 +30,6 @@
 import jakarta.enterprise.context.RequestScoped;
 import jakarta.enterprise.inject.Produces;
 import jakarta.inject.Named;
-import org.apache.commons.codec.binary.Base32;
 import org.apache.commons.codec.binary.Base64;
 import org.eclipse.microprofile.config.inject.ConfigProperty;
 import org.projectnessie.cel.tools.ScriptHost;
@@ -60,12 +57,6 @@ public static FileSystem produceFileSystem() {
         return new FileSystem();
     }
 
-    @Produces
-    @ApplicationScoped
-    public static Base32 produceBase32() {
-        return new Base32();
-    }
-
     @Produces
     @ApplicationScoped
     @DefaultBean
@@ -74,20 +65,6 @@ public static Base64 produceBase64Url() {
         return new Base64(0, null, true);
     }
 
-    @Produces
-    @ApplicationScoped
-    @DefaultBean
-    public static ExecutorService produceExecutorService() {
-        return ForkJoinPool.commonPool();
-    }
-
-    @Produces
-    @ApplicationScoped
-    @DefaultBean
-    public static ScheduledExecutorService produceScheduledExecutorService() {
-        return Executors.newSingleThreadScheduledExecutor();
-    }
-
     @Produces
     // RequestScoped so that each individual report generation request has its own interruptible
     // generator with an independent task queueing thread which dispatches to the shared common pool

diff --git a/src/main/java/io/cryostat/reports/StorageCachingReportsService.java b/src/main/java/io/cryostat/reports/StorageCachingReportsService.java
@@ -20,7 +20,6 @@
 import java.time.Instant;
 import java.util.Map;
 import java.util.concurrent.CompletionException;
-import java.util.concurrent.ExecutorService;
 import java.util.function.Predicate;
 
 import org.openjdk.jmc.flightrecorder.rules.IRule;
@@ -66,7 +65,6 @@ class StorageCachingReportsService implements ReportsService {
     Duration expiry;
 
     @Inject S3Client storage;
-    @Inject ExecutorService worker;
     @Inject RecordingHelper recordingHelper;
     @Inject ObjectMapper mapper;
 

diff --git a/src/main/java/io/cryostat/ws/MessagingServer.java b/src/main/java/io/cryostat/ws/MessagingServer.java
@@ -16,12 +16,15 @@
 package io.cryostat.ws;
 
 import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.CopyOnWriteArraySet;
 
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
+import io.quarkus.scheduler.Scheduled;
 import io.quarkus.vertx.ConsumeEvent;
 import jakarta.enterprise.context.ApplicationScoped;
 import jakarta.inject.Inject;
@@ -38,6 +41,8 @@
 public class MessagingServer {
 
     private static final String CLIENT_ACTIVITY_CATEGORY = "WsClientActivity";
+    private static final ByteBuffer PING_MSG =
+            ByteBuffer.wrap("ping".getBytes(StandardCharsets.UTF_8));
 
     @Inject ObjectMapper mapper;
     @Inject Logger logger;
@@ -106,4 +111,16 @@ void broadcast(Notification notification) {
                                             }
                                         }));
     }
+
+    @Scheduled(every = "${cryostat.websocket.ping-period:20s}")
+    void pingClients() {
+        sessions.forEach(
+                session -> {
+                    try {
+                        session.getBasicRemote().sendPing(PING_MSG);
+                    } catch (IOException e) {
+                        logger.debug(e);
+                    }
+                });
+    }
 }