websocket improvements

Author: zb-sr

src/pom.xml

@@ -5,7 +5,7 @@
     <groupId>com.sportradar.mbs.sdk</groupId>
     <artifactId>mbs-sdk-parent</artifactId>
     <packaging>pom</packaging>
-    <version>0.9.7-SNAPSHOT</version>
+    <version>0.9.6-SNAPSHOT</version>
     <name>MBS SDK</name>
     <description>MBS SDK is a client library that enables easier integration with the Sportradar MBS</description>
     <inceptionYear>2024</inceptionYear>
@@ -34,7 +34,7 @@
     <scm>
         <url>https://github.com/sportradar/MbsSdkJava</url>
         <connection>scm:git:[email protected]:sportradar/MbsSdkJava.git</connection>
-    <tag>v0.9.7</tag>
+    <tag>v0.9.6</tag>
   </scm>
 
     <modules>

src/sdk/pom.xml

@@ -3,7 +3,7 @@
     <parent>
         <groupId>com.sportradar.mbs.sdk</groupId>
         <artifactId>mbs-sdk-parent</artifactId>
-        <version>0.9.7-SNAPSHOT</version>
+        <version>0.9.6-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 

src/sdk/src/main/java/com/sportradar/mbs/sdk/MbsSdkConfig.java

@@ -31,6 +31,7 @@ public class MbsSdkConfig {
     private Duration wsReceiveMessageTimeout;
     private Duration wsConsumerGraceTimeout;
     private Duration wsRefreshConnectionTimeout;
+    private Duration wsPingInterval;
     private Integer wsNumberOfConnections;
     private BiConsumer<MbsSdk, Exception> unhandledExceptionHandler;
 
@@ -377,12 +378,30 @@ public Duration getWsRefreshConnectionTimeout() {
     /**
      * Sets the timeout for refreshing the WebSocket connection.
      *
-     * @param wsRefreshConnectionTimeout The timeout for refreshing the WebSocket connection to set.
+     * @param wsRefreshConnectionTimeout The timeout for refreshing the WebSocket connection.
      */
     public void setWsRefreshConnectionTimeout(Duration wsRefreshConnectionTimeout) {
         this.wsRefreshConnectionTimeout = wsRefreshConnectionTimeout;
     }
 
+    /**
+     * Gets the interval of WebSocket connection ping.
+     *
+     * @return  The interval of WebSocket connection ping.
+     */
+    public Duration getWsPingInterval() {
+        return wsPingInterval;
+    }
+
+    /**
+     * Sets the interval of WebSocket connection ping.
+     *
+     * @param wsPingInterval The interval of WebSocket connection ping.
+     */
+    public void setWsPingInterval(Duration wsPingInterval) {
+        this.wsPingInterval = wsPingInterval;
+    }
+
     /**
      * Gets the number of connections for the WebSocket.
      *

src/sdk/src/main/java/com/sportradar/mbs/sdk/internal/config/ImmutableConfig.java

@@ -37,6 +37,7 @@ public class ImmutableConfig implements
     private final Duration wsReceiveMessageTimeout;
     private final Duration wsConsumerGraceTimeout;
     private final Duration wsRefreshConnectionTimeout;
+    private final Duration wsPingInterval;
     private final int wsNumberOfConnections;
 
     public ImmutableConfig(final MbsSdkConfig config) {
@@ -47,20 +48,21 @@ public ImmutableConfig(final MbsSdkConfig config) {
         this.wsServer = notNull(config.getWsServer(), "wsServer");
         this.operatorId = config.getOperatorId();
         this.authRequestTimeout = withDefault(config.getAuthRetryDelay(), Duration.ofSeconds(5), MIN_DURATION);
-        this.authRetryDelay = withDefault(config.getAuthRetryDelay(), Duration.ofMillis(500), MIN_DURATION);
+        this.authRetryDelay = withDefault(config.getAuthRetryDelay(), Duration.ofSeconds(1), MIN_DURATION);
         this.wsNumberOfConnections = withDefault(config.getWsNumberOfConnections(), 1, 1);
         this.wsReconnectTimeout = withDefault(config.getWsReconnectTimeout(), Duration.ofSeconds(10), MIN_DURATION);
         this.wsFetchMessageTimeout = withDefault(config.getWsFetchMessageTimeout(), Duration.ofSeconds(1), MIN_DURATION);
         this.wsSendMessageTimeout = withDefault(config.getWsSendMessageTimeout(), Duration.ofSeconds(1), MIN_DURATION);
         this.wsReceiveMessageTimeout = withDefault(config.getWsReceiveMessageTimeout(), Duration.ofSeconds(30), MIN_DURATION);
-        this.wsConsumerGraceTimeout = withDefault(config.getWsConsumerGraceTimeout(), Duration.ofMinutes(1), MIN_DURATION);
+        this.wsConsumerGraceTimeout = withDefault(config.getWsConsumerGraceTimeout(), Duration.ofMinutes(10), MIN_DURATION);
         this.wsRefreshConnectionTimeout = withDefault(config.getWsRefreshConnectionTimeout(), Duration.ofMinutes(100), MIN_DURATION);
         this.protocolRetryCount = withDefault(config.getProtocolRetryCount(), 0, 0);
         this.protocolMaxSendBufferSize = withDefault(config.getProtocolMaxSendBufferSize(), 1_000, 1);
         this.protocolConnectTimeout = withDefault(config.getProtocolConnectTimeout(), Duration.ofSeconds(10), MIN_DURATION);
         this.protocolReceiveResponseTimeout = withDefault(config.getProtocolReceiveResponseTimeout(), Duration.ofSeconds(20), MIN_DURATION);
         this.protocolEnqueueTimeout = withDefault(config.getProtocolEnqueueTimeout(), Duration.ofMillis(100), MIN_DURATION);
         this.protocolDequeueTimeout = withDefault(config.getProtocolDequeueTimeout(), Duration.ofSeconds(1), MIN_DURATION);
+        this.wsPingInterval = withDefault(config.getWsPingInterval(), Duration.ofMinutes(1), MIN_DURATION);
         this.protocolNumberOfDispatchers = withDefault(config.getProtocolNumberOfDispatchers(), 1, 1);
     }
 
@@ -173,4 +175,9 @@ public Duration getWsRefreshConnectionTimeout() {
     public int getWsNumberOfConnections() {
         return wsNumberOfConnections;
     }
+
+    @Override
+    public Duration getWsPingInterval() {
+        return wsPingInterval;
+    }
 }

src/sdk/src/main/java/com/sportradar/mbs/sdk/internal/config/WebSocketConnectionConfig.java

@@ -18,4 +18,6 @@ public interface WebSocketConnectionConfig {
     Duration getWsConsumerGraceTimeout();
 
     Duration getWsRefreshConnectionTimeout();
+
+    Duration getWsPingInterval();
 }

src/sdk/src/main/java/com/sportradar/mbs/sdk/internal/connection/WebSocketConnection.java

@@ -100,7 +100,7 @@ private void sendLoop() {
 
     private void sendMsg(final WebSocket ws, final List<ByteBuffer> msgs) {
         for (int i = 0; i < msgs.size(); i++) {
-            ws.sendFragmentedFrame(Opcode.TEXT, msgs.get(i), i == (msgs.size() - 1));
+            ws.sendFragmentedFrame(Opcode.TEXT, msgs.get(i).duplicate(), i == (msgs.size() - 1));
         }
     }
 
@@ -110,7 +110,7 @@ private void reconnectWebSocket(final WebSocket ws, final boolean throwExc) {
         }
         final WebSocket newWs;
         try {
-            newWs = new WebSocket(this, config.getWsServer(), tokenProvider.getToken());
+            newWs = new WebSocket(this);
             if (!newWs.connectBlocking(config.getWsReconnectTimeout().toMillis(), MILLISECONDS)) {
                 throw new WebSocketConnectionException("Socket connect failed.");
             }
@@ -126,9 +126,9 @@ private void reconnectWebSocket(final WebSocket ws, final boolean throwExc) {
         }
         if (this.webSocket.compareAndSet(ws, newWs)) {
             if (ws != null) {
-                delay(ws::close, config.getWsConsumerGraceTimeout().toMillis(), MILLISECONDS);
+                delay(ws::close, config.getWsConsumerGraceTimeout());
             }
-            delay(() -> reconnectWebSocket(newWs, false), config.getWsRefreshConnectionTimeout().toMillis(), MILLISECONDS);
+            delay(() -> reconnectWebSocket(newWs, false), config.getWsRefreshConnectionTimeout());
         } else {
             newWs.close();
         }
@@ -156,12 +156,23 @@ public static class WebSocket extends WebSocketClient {
 
         private final WebSocketConnection connection;
 
-        public WebSocket(final WebSocketConnection connection, final URI serverUri, final String token) {
-            super(serverUri, headers(token));
+        public WebSocket(final WebSocketConnection connection) {
+            super(serverUri(connection), headers(connection));
+            this.setConnectionLostTimeout(getPingInterval(connection));
             this.connection = connection;
         }
 
-        private static Map<String, String> headers(final String token) {
+        private static URI serverUri(final WebSocketConnection connection) {
+            return connection.config.getWsServer();
+        }
+
+        private static int getPingInterval(final WebSocketConnection connection) {
+            final long interval = connection.config.getWsPingInterval().toSeconds();
+            return (int) Math.max(10, Math.min(300, interval));
+        }
+
+        private static Map<String, String> headers(final WebSocketConnection connection) {
+            final String token = connection.tokenProvider.getToken();
             final Map<String, String> headers = new HashMap<>();
             headers.put("Authorization", "Bearer " + token);
             return headers;

src/sdk/src/main/java/com/sportradar/mbs/sdk/internal/protocol/ProtocolEngine.java

@@ -132,8 +132,7 @@ private <R extends ContentResponse> void enqueueSendMsg(final Awaiter<R> awaiter
         sendQueue.add(awaiter.getSendWsInputMessage());
 
         final int nextRetryCount = retryCount + 1;
-        delay(() -> enqueueSendMsg(awaiter, nextRetryCount),
-                config.getProtocolReceiveResponseTimeout().toMillis(), TimeUnit.MILLISECONDS);
+        delay(() -> enqueueSendMsg(awaiter, nextRetryCount), config.getProtocolReceiveResponseTimeout());
     }
 
     private List<ByteBuffer> createFrames(final Request request) throws JsonProcessingException {

src/sdk/src/main/java/com/sportradar/mbs/sdk/internal/utils/Delayer.java

@@ -1,5 +1,6 @@
 package com.sportradar.mbs.sdk.internal.utils;
 
+import java.time.Duration;
 import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.ScheduledThreadPoolExecutor;
 import java.util.concurrent.ThreadFactory;
@@ -15,8 +16,8 @@ public class Delayer {
         EXECUTOR = executor;
     }
 
-    public static ScheduledFuture<?> delay(final Runnable command, final long delay, final TimeUnit unit) {
-        return EXECUTOR.schedule(command, delay, unit);
+    public static ScheduledFuture<?> delay(final Runnable command, final Duration delay) {
+        return EXECUTOR.schedule(command, delay.toMillis(), TimeUnit.MILLISECONDS);
     }
 
     private static final class SdkDelayerThreadFactory implements ThreadFactory {