[fix][websocket] Fix webSocketPingDurationSeconds config (#19256)

Signed-off-by: Zixuan Liu <nodeces@gmail.com>
apache · Jan 20, 2023 · 516437e · 516437e
1 parent e3b76d4
commit 516437e
Show file tree

Hide file tree

Showing 4 changed files with 21 additions and 17 deletions.
diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
@@ -2613,6 +2613,12 @@ The delayed message index bucket time step(in seconds) in per bucket snapshot se
     )
     private int webSocketSessionIdleTimeoutMillis = 300000;
 
+    @FieldContext(
+            category = CATEGORY_WEBSOCKET,
+            doc = "Interval of time to sending the ping to keep alive in WebSocket proxy. "
+                    + "This value greater than 0 means enabled")
+    private int webSocketPingDurationSeconds = -1;
+
     @FieldContext(
         category = CATEGORY_WEBSOCKET,
         doc = "The maximum size of a text message during parsing in WebSocket proxy."

diff --git a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyConfiguration.java b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyConfiguration.java
@@ -811,6 +811,12 @@ public class ProxyConfiguration implements PulsarConfiguration {
     )
     private boolean webSocketServiceEnabled = false;
 
+    @FieldContext(
+            category = CATEGORY_WEBSOCKET,
+            doc = "Interval of time to sending the ping to keep alive in WebSocket proxy. "
+                    + "This value greater than 0 means enabled")
+    private int webSocketPingDurationSeconds = -1;
+
     @FieldContext(
             category = CATEGORY_WEBSOCKET,
             doc = "Name of the cluster to which this broker belongs to"

diff --git a/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/AbstractWebSocketHandler.java b/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/AbstractWebSocketHandler.java
@@ -56,7 +56,6 @@
 import org.apache.pulsar.common.util.Codec;
 import org.apache.pulsar.common.util.ObjectMapperFactory;
 import org.apache.pulsar.websocket.data.ConsumerCommand;
-import org.apache.pulsar.websocket.service.WebSocketProxyConfiguration;
 import org.eclipse.jetty.websocket.api.Session;
 import org.eclipse.jetty.websocket.api.WebSocketAdapter;
 import org.eclipse.jetty.websocket.servlet.ServletUpgradeResponse;
@@ -192,18 +191,15 @@ private void closePingFuture() {
     @Override
     public void onWebSocketConnect(Session session) {
         super.onWebSocketConnect(session);
-        WebSocketProxyConfiguration webSocketProxyConfig = service.getWebSocketProxyConfig();
-        if (webSocketProxyConfig != null) {
-            int webSocketPingDurationSeconds = webSocketProxyConfig.getWebSocketPingDurationSeconds();
-            if (webSocketPingDurationSeconds > 0) {
-                pingFuture = service.getExecutor().scheduleAtFixedRate(() -> {
-                    try {
-                        session.getRemote().sendPing(ByteBuffer.wrap("PING".getBytes(StandardCharsets.UTF_8)));
-                    } catch (IOException e) {
-                        log.warn("[{}] WebSocket send ping", getSession().getRemoteAddress(), e);
-                    }
-                }, webSocketPingDurationSeconds, webSocketPingDurationSeconds, TimeUnit.SECONDS);
-            }
+        int webSocketPingDurationSeconds = service.getConfig().getWebSocketPingDurationSeconds();
+        if (webSocketPingDurationSeconds > 0) {
+            pingFuture = service.getExecutor().scheduleAtFixedRate(() -> {
+                try {
+                    session.getRemote().sendPing(ByteBuffer.wrap("PING".getBytes(StandardCharsets.UTF_8)));
+                } catch (IOException e) {
+                    log.warn("[{}] WebSocket send ping", getSession().getRemoteAddress(), e);
+                }
+            }, webSocketPingDurationSeconds, webSocketPingDurationSeconds, TimeUnit.SECONDS);
         }
         log.info("[{}] New WebSocket session on topic {}", session.getRemoteAddress(), topic);
     }

diff --git a/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/WebSocketService.java b/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/WebSocketService.java
@@ -69,9 +69,6 @@ public class WebSocketService implements Closeable {
     private MetadataStoreExtended configMetadataStore;
     private ServiceConfiguration config;
 
-    @Getter
-    private WebSocketProxyConfiguration webSocketProxyConfig;
-
     @Getter
     private Optional<CryptoKeyReader> cryptoKeyReader = Optional.empty();
 
@@ -83,7 +80,6 @@ public class WebSocketService implements Closeable {
 
     public WebSocketService(WebSocketProxyConfiguration config) {
         this(createClusterData(config), PulsarConfigurationLoader.convertFrom(config));
-        this.webSocketProxyConfig = config;
     }
 
     public WebSocketService(ClusterData localCluster, ServiceConfiguration config) {