Make it possible to set idle WebSocket session timeout period (apache…

…#2118)

conf/broker.conf

@@ -462,6 +462,8 @@ webSocketNumIoThreads=8
 # Number of connections per Broker in Pulsar Client used in WebSocket proxy
 webSocketConnectionsPerBroker=8
 
+# Time in milliseconds that idle WebSocket session times out
+webSocketSessionIdleTimeoutMillis=300000
 
 ### --- Metrics --- ###
 

conf/standalone.conf

@@ -393,6 +393,8 @@ webSocketNumIoThreads=8
 # Number of connections per Broker in Pulsar Client used in WebSocket proxy
 webSocketConnectionsPerBroker=8
 
+# Time in milliseconds that idle WebSocket session times out
+webSocketSessionIdleTimeoutMillis=300000
 
 ### --- Metrics --- ###
 

conf/websocket.conf

@@ -52,6 +52,9 @@ numIoThreads=8
 # Number of connections per Broker in Pulsar Client used in WebSocket proxy
 connectionsPerBroker=8
 
+# Time in milliseconds that idle WebSocket session times out
+webSocketSessionIdleTimeoutMillis=300000
+
 ### --- Authentication --- ###
 
 # Enable authentication

pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java

@@ -464,6 +464,8 @@ public class ServiceConfiguration implements PulsarConfiguration {
     private int webSocketNumIoThreads = Runtime.getRuntime().availableProcessors();
     // Number of connections per Broker in Pulsar Client used in WebSocket proxy
     private int webSocketConnectionsPerBroker = Runtime.getRuntime().availableProcessors();
+    // Time in milliseconds that idle WebSocket session times out
+    private int webSocketSessionIdleTimeoutMillis = 300000;
 
     /**** --- Metrics --- ****/
     // If true, export topic level metrics otherwise namespace level
@@ -1582,6 +1584,14 @@ public void setPreferLaterVersions(boolean preferLaterVersions) {
 
     public void setWebSocketConnectionsPerBroker(int webSocketConnectionsPerBroker) { this.webSocketConnectionsPerBroker = webSocketConnectionsPerBroker; }
 
+    public int getWebSocketSessionIdleTimeoutMillis() {
+        return webSocketSessionIdleTimeoutMillis;
+    }
+
+    public void setWebSocketSessionIdleTimeoutMillis(int webSocketSessionIdleTimeoutMillis) {
+        this.webSocketSessionIdleTimeoutMillis = webSocketSessionIdleTimeoutMillis;
+    }
+
     public boolean exposeTopicLevelMetricsInPrometheus() {
         return exposeTopicLevelMetricsInPrometheus;
     }

pulsar-websocket/src/main/java/org/apache/pulsar/websocket/WebSocketConsumerServlet.java

@@ -37,8 +37,10 @@ public WebSocketConsumerServlet(WebSocketService service) {
     @Override
     public void configure(WebSocketServletFactory factory) {
         factory.getPolicy().setMaxTextMessageSize(WebSocketService.MaxTextFrameSize);
-
+        if (service.getConfig().getWebSocketSessionIdleTimeoutMillis() > 0) {
+            factory.getPolicy().setIdleTimeout(service.getConfig().getWebSocketSessionIdleTimeoutMillis());
+        }
         factory.setCreator(
                 (request, response) -> new ConsumerHandler(service, request.getHttpServletRequest(), response));
     }
-}
+}

pulsar-websocket/src/main/java/org/apache/pulsar/websocket/WebSocketProducerServlet.java

@@ -36,6 +36,9 @@ public WebSocketProducerServlet(WebSocketService service) {
     @Override
     public void configure(WebSocketServletFactory factory) {
         factory.getPolicy().setMaxTextMessageSize(WebSocketService.MaxTextFrameSize);
+        if (service.getConfig().getWebSocketSessionIdleTimeoutMillis() > 0) {
+            factory.getPolicy().setIdleTimeout(service.getConfig().getWebSocketSessionIdleTimeoutMillis());
+        }
         factory.setCreator((request, response) -> new ProducerHandler(service, request.getHttpServletRequest(), response));
     }
-}
+}

pulsar-websocket/src/main/java/org/apache/pulsar/websocket/WebSocketReaderServlet.java

@@ -37,8 +37,10 @@ public WebSocketReaderServlet(WebSocketService service) {
     @Override
     public void configure(WebSocketServletFactory factory) {
         factory.getPolicy().setMaxTextMessageSize(WebSocketService.MaxTextFrameSize);
-
+        if (service.getConfig().getWebSocketSessionIdleTimeoutMillis() > 0) {
+            factory.getPolicy().setIdleTimeout(service.getConfig().getWebSocketSessionIdleTimeoutMillis());
+        }
         factory.setCreator(
                 (request, response) -> new ReaderHandler(service, request.getHttpServletRequest(), response));
     }
-}
+}

pulsar-websocket/src/main/java/org/apache/pulsar/websocket/service/WebSocketProxyConfiguration.java

@@ -92,6 +92,8 @@ public class WebSocketProxyConfiguration implements PulsarConfiguration {
     private int numIoThreads = Runtime.getRuntime().availableProcessors();
     // Number of connections per Broker in Pulsar Client used in WebSocket proxy
     private int connectionsPerBroker = Runtime.getRuntime().availableProcessors();
+    // Time in milliseconds that idle WebSocket session times out
+    private int webSocketSessionIdleTimeoutMillis = 300000;
 
     // When this parameter is not empty, unauthenticated users perform as anonymousUserRole
     private String anonymousUserRole = null;
@@ -299,6 +301,14 @@ public void setConnectionsPerBroker(int connectionsPerBroker) {
         this.connectionsPerBroker = connectionsPerBroker;
     }
 
+    public int getWebSocketSessionIdleTimeoutMillis() {
+        return webSocketSessionIdleTimeoutMillis;
+    }
+
+    public void setWebSocketSessionIdleTimeoutMillis(int webSocketSessionIdleTimeoutMillis) {
+        this.webSocketSessionIdleTimeoutMillis = webSocketSessionIdleTimeoutMillis;
+    }
+
     public String getAnonymousUserRole() {
         return anonymousUserRole;
     }