Fix review comment

Signed-off-by: Viet Nguyen Duc <[email protected]>

Author: VietND96

java/src/org/openqa/selenium/events/zeromq/BoundZmqEventBus.java

@@ -58,13 +58,13 @@ class BoundZmqEventBus implements EventBus {
     xpub = context.createSocket(SocketType.XPUB);
     xpub.setIPv6(xpubAddr.isIPv6);
     xpub.setImmediate(true);
-    configureHeartbeat(xpub, heartbeatPeriod, "XPUB");
+    ZmqUtils.configureHeartbeat(xpub, heartbeatPeriod, "XPUB");
     xpub.bind(xpubAddr.bindTo);
 
     xsub = context.createSocket(SocketType.XSUB);
     xsub.setIPv6(xsubAddr.isIPv6);
     xsub.setImmediate(true);
-    configureHeartbeat(xsub, heartbeatPeriod, "XSUB");
+    ZmqUtils.configureHeartbeat(xsub, heartbeatPeriod, "XSUB");
     xsub.bind(xsubAddr.bindTo);
 
     executor =
@@ -75,27 +75,11 @@ class BoundZmqEventBus implements EventBus {
               return thread;
             });
     executor.submit(() -> ZMQ.proxy(xsub, xpub, null));
-
     delegate =
         new UnboundZmqEventBus(
             context, xpubAddr.advertise, xsubAddr.advertise, secret, heartbeatPeriod);
   }
 
-  private void configureHeartbeat(ZMQ.Socket socket, Duration heartbeatPeriod, String socketType) {
-    if (heartbeatPeriod != null && !heartbeatPeriod.isZero() && !heartbeatPeriod.isNegative()) {
-      int heartbeatIvl = (int) heartbeatPeriod.toMillis();
-      socket.setHeartbeatIvl(heartbeatIvl);
-      // Set heartbeat timeout to 3x the interval
-      socket.setHeartbeatTimeout(heartbeatIvl * 3);
-      // Set heartbeat TTL to 6x the interval
-      socket.setHeartbeatTtl(heartbeatIvl * 6);
-      LOG.info(
-          String.format(
-              "Event bus %s socket heartbeat configured: interval=%dms, timeout=%dms, ttl=%dms",
-              socketType, heartbeatIvl, heartbeatIvl * 3, heartbeatIvl * 6));
-    }
-  }
-
   @Override
   public boolean isReady() {
     return !executor.isShutdown();

java/src/org/openqa/selenium/events/zeromq/UnboundZmqEventBus.java

@@ -141,13 +141,13 @@ class UnboundZmqEventBus implements EventBus {
             () -> {
               sub = context.createSocket(SocketType.SUB);
               sub.setIPv6(isSubAddressIPv6(publishConnection));
-              configureHeartbeat(sub, heartbeatPeriod, "SUB");
+              ZmqUtils.configureHeartbeat(sub, heartbeatPeriod, "SUB");
               sub.connect(publishConnection);
               sub.subscribe(new byte[0]);
 
               pub = context.createSocket(SocketType.PUB);
               pub.setIPv6(isSubAddressIPv6(subscribeConnection));
-              configureHeartbeat(pub, heartbeatPeriod, "PUB");
+              ZmqUtils.configureHeartbeat(pub, heartbeatPeriod, "PUB");
               pub.connect(subscribeConnection);
             });
     // Connections are already established
@@ -176,22 +176,7 @@ class UnboundZmqEventBus implements EventBus {
 
   @Override
   public boolean isReady() {
-    return !socketPollingExecutor.isShutdown();
-  }
-
-  private void configureHeartbeat(ZMQ.Socket socket, Duration heartbeatPeriod, String socketType) {
-    if (heartbeatPeriod != null && !heartbeatPeriod.isZero() && !heartbeatPeriod.isNegative()) {
-      int heartbeatIvl = (int) heartbeatPeriod.toMillis();
-      socket.setHeartbeatIvl(heartbeatIvl);
-      // Set heartbeat timeout to 3x the interval
-      socket.setHeartbeatTimeout(heartbeatIvl * 3);
-      // Set heartbeat TTL to 6x the interval
-      socket.setHeartbeatTtl(heartbeatIvl * 6);
-      LOG.info(
-          String.format(
-              "Event bus %s socket heartbeat configured: interval=%dms, timeout=%dms, ttl=%dms",
-              socketType, heartbeatIvl, heartbeatIvl * 3, heartbeatIvl * 6));
-    }
+    return !socketPollingExecutor.isShutdown() && pollingStarted.get();
   }
 
   private boolean isSubAddressIPv6(String connection) {

java/src/org/openqa/selenium/events/zeromq/ZeroMqEventBus.java

@@ -39,14 +39,21 @@
 public class ZeroMqEventBus {
 
   private static final String EVENTS_SECTION = "events";
+  private static final int DEFAULT_HEARTBEAT_PERIOD_SECONDS = 60;
 
   private ZeroMqEventBus() {
     // Use the create method.
   }
 
   public static EventBus create(
       ZContext context, String publish, String subscribe, boolean bind, Secret secret) {
-    return create(context, publish, subscribe, bind, secret, Duration.ofSeconds(60));
+    return create(
+        context,
+        publish,
+        subscribe,
+        bind,
+        secret,
+        Duration.ofSeconds(DEFAULT_HEARTBEAT_PERIOD_SECONDS));
   }
 
   public static EventBus create(
@@ -96,9 +103,7 @@ public static EventBus create(Config config) {
                 });
 
     boolean bind = config.getBool(EVENTS_SECTION, "bind").orElse(false);
-    int heartbeatPeriodSeconds =
-        config.getInt(EVENTS_SECTION, "eventbus-heartbeat-period").orElse(60);
-    Duration heartbeatPeriod = Duration.ofSeconds(heartbeatPeriodSeconds);
+    Duration heartbeatPeriod = getHeartbeatPeriod(config);
 
     SecretOptions secretOptions = new SecretOptions(config);
 
@@ -111,6 +116,14 @@ public static EventBus create(Config config) {
         heartbeatPeriod);
   }
 
+  private static Duration getHeartbeatPeriod(Config config) {
+    int periodSeconds =
+        config
+            .getInt(EVENTS_SECTION, "eventbus-heartbeat-period")
+            .orElse(DEFAULT_HEARTBEAT_PERIOD_SECONDS);
+    return Duration.ofSeconds(periodSeconds);
+  }
+
   private static String mungeUri(URI base, String scheme, int port) {
     try {
       return new URI(scheme, null, base.getHost(), port, null, null, null).toString();

java/src/org/openqa/selenium/events/zeromq/ZmqUtils.java

@@ -0,0 +1,52 @@
+// Licensed to the Software Freedom Conservancy (SFC) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The SFC licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.openqa.selenium.events.zeromq;
+
+import java.time.Duration;
+import java.util.logging.Logger;
+import org.zeromq.ZMQ;
+
+/** Utility methods for ZeroMQ socket configuration. */
+class ZmqUtils {
+
+  private static final Logger LOG = Logger.getLogger(ZmqUtils.class.getName());
+
+  private ZmqUtils() {
+    // Utility class
+  }
+
+  /**
+   * Configures ZeroMQ heartbeat settings on a socket to prevent stale connections.
+   *
+   * @param socket The ZMQ socket to configure
+   * @param heartbeatPeriod The heartbeat interval duration
+   * @param socketType The socket type name for logging (e.g., "SUB", "PUB", "XPUB", "XSUB")
+   */
+  static void configureHeartbeat(ZMQ.Socket socket, Duration heartbeatPeriod, String socketType) {
+    if (heartbeatPeriod != null && !heartbeatPeriod.isZero() && !heartbeatPeriod.isNegative()) {
+      int heartbeatIvl = (int) heartbeatPeriod.toMillis();
+      socket.setHeartbeatIvl(heartbeatIvl);
+      socket.setHeartbeatTimeout(heartbeatIvl * 3);
+      socket.setHeartbeatTtl(heartbeatIvl * 6);
+      LOG.info(
+          String.format(
+              "ZMQ %s socket heartbeat configured: interval=%dms, timeout=%dms, ttl=%dms",
+              socketType, heartbeatIvl, heartbeatIvl * 3, heartbeatIvl * 6));
+    }
+  }
+}