handle data streams

Author: theo-s68

.github/workflows/java.yml

@@ -1,8 +1,10 @@
 name: Java Main Workflow
 
 on:
-#  pull_request:
-#    branches: [ master, rc-** ]
+  push:
+    branches: [ master ]
+  pull_request:
+    branches: [ master, rc-* ]
 
 jobs:
   detect-targets:
@@ -28,21 +30,21 @@ jobs:
           fi
           
           # Check each client dynamically
-          find clients/* -type d -maxdepth 0 | while read client; do
+          while read client; do
             CLIENT_NAME=$(basename "$client")
             if ! git diff --quiet "$BASE_COMMIT" HEAD -- "$client"; then
               echo "Changes detected in $CLIENT_NAME"
               MODIFIED_TARGETS+=("$CLIENT_NAME")
             fi
-          done
+          done < <(find clients/* -type d -maxdepth 0)
           
           # Convert to JSON array format
           MODIFIED_TARGETS_JSON=$(printf '%s\n' "${MODIFIED_TARGETS[@]}" | jq -R -s -c 'split("\n") | map(select(. != ""))')
           
           echo "Detected modified targets: $MODIFIED_TARGETS_JSON"
           echo "modified_targets=$MODIFIED_TARGETS_JSON" >> $GITHUB_ENV
           echo "::set-output name=modified_targets::$MODIFIED_TARGETS_JSON"
-  checkstyle:
+  verify:
     runs-on: ubuntu-latest
     needs: detect-targets
     if: ${{ needs.detect-targets.outputs.modified_targets != '[]' }}
@@ -59,29 +61,5 @@ jobs:
             'examples/pom.xml'
       - name: Validate modules
         run: |
-          mvn -f clients/pom.xml -N install
-          mvn -f clients/pom.xml -pl common install
-          mvn -f clients/pom.xml -pl `echo '${{ needs.detect-targets.outputs.modified_targets }}' | jq -r 'join(",")'` validate
-  build:
-    runs-on: ubuntu-latest
-    needs: detect-targets
-    if: ${{ needs.detect-targets.outputs.modified_targets != '[]' }}
-    strategy:
-      matrix:
-        target: ${{ fromJson(needs.detect-targets.outputs.modified_targets) }}
-        java-version: [ 11, 17 ]
-    steps:
-      - uses: actions/checkout@v3
-      - name: Set up JDK
-        uses: actions/setup-java@v3
-        with:
-          java-version: ${{ matrix.java-version }}
-          distribution: 'adopt'
-          cache: 'maven'
-          cache-dependency-path: |
-            'clients/pom.xml'
-            'examples/pom.xml'
-      - name: Build ${{ matrix.target }} module
-        run: |
-          mvn -f clients/pom.xml -N install
-          mvn -f clients/pom.xml -pl common,${{ matrix.target }} install -Dcheckstyle.skip=true
+          mvn -f clients/pom.xml -pl common install -Dgpg.skip
+          mvn -f clients/pom.xml -pl `echo '${{ needs.detect-targets.outputs.modified_targets }}' | jq -r 'join(",")'` verify -Dgpg.skip

clients/common/pom.xml

@@ -10,6 +10,6 @@
 
     <artifactId>binance-common</artifactId>
     <name>common</name>
-    <version>1.3.0</version>
+    <version>1.4.0</version>
     <packaging>jar</packaging>
 </project>

clients/common/src/main/java/com/binance/connector/client/common/ApiClient.java

@@ -134,7 +134,8 @@ public class ApiClient {
 
     private Gson json;
 
-    private Set<String> forbiddenHeaders = new HashSet<>(Arrays.asList("host", "authorization", "cookie", ":method", ":path"));
+    private Set<String> forbiddenHeaders =
+            new HashSet<>(Arrays.asList("host", "authorization", "cookie", ":method", ":path"));
 
     public ApiClient(ClientConfiguration configuration) {
         this(configuration, new BinanceAuthenticationFactory(), null);
@@ -188,8 +189,10 @@ public ApiClient(
             }
         }
 
-        if (configuration.getCustomHeaders() != null && !configuration.getCustomHeaders().isEmpty()) {
-            Interceptor customHeadersInterceptor = getCustomHeadersInterceptor(configuration.getCustomHeaders());
+        if (configuration.getCustomHeaders() != null
+                && !configuration.getCustomHeaders().isEmpty()) {
+            Interceptor customHeadersInterceptor =
+                    getCustomHeadersInterceptor(configuration.getCustomHeaders());
             builder.addInterceptor(customHeadersInterceptor);
         }
 
@@ -217,13 +220,15 @@ public ApiClient(
             if (authentication != null) {
                 authentications.put(BINANCE_SIGNATURE, authentication);
             }
+        }
 
-            Authentication binanceApiKeyOnly =
-                    (queryParams, headerParams, cookieParams, payload, method, uri) -> {
+        Authentication binanceApiKeyOnly =
+                (queryParams, headerParams, cookieParams, payload, method, uri) -> {
+                    if (signatureConfiguration != null && signatureConfiguration.getApiKey() != null) {
                         headerParams.put(HEADER_API_KEY, signatureConfiguration.getApiKey());
-                    };
-            authentications.put(BINANCE_API_KEY_ONLY, binanceApiKeyOnly);
-        }
+                    }
+                };
+        authentications.put(BINANCE_API_KEY_ONLY, binanceApiKeyOnly);
     }
 
     private void init() {
@@ -250,13 +255,15 @@ public void setJson(Gson json) {
 
     public Interceptor getCustomHeadersInterceptor(Map<String, String> customHeaders) {
         return chain -> {
-
             Request request = chain.request();
             Request.Builder newBuilder = request.newBuilder();
             for (String headerName : customHeaders.keySet()) {
                 String headerValue = customHeaders.get(headerName);
                 if (!validateHeader(headerName, headerValue)) {
-                    throw new ApiException("Invalid header " + headerName + ", it is forbidden or invalid (contains CR/LF)");
+                    throw new ApiException(
+                            "Invalid header "
+                                    + headerName
+                                    + ", it is forbidden or invalid (contains CR/LF)");
                 }
 
                 newBuilder.addHeader(headerName, headerValue);
@@ -1413,9 +1420,22 @@ public Request buildRequest(
 
         List<Pair> updatedQueryParams = new ArrayList<>(queryParams);
 
+        boolean hasAuth =
+                Arrays.stream(authNames)
+                        .anyMatch(
+                                s -> s.equals(BINANCE_SIGNATURE) || s.equals(BINANCE_API_KEY_ONLY));
+
+        // add api key to every request
+        String[] finalAuthNames;
+        if (!hasAuth) {
+            finalAuthNames = append(authNames, BINANCE_API_KEY_ONLY);
+        } else {
+            finalAuthNames = authNames;
+        }
+
         // update parameters with authentication settings
         updateParamsForAuth(
-                authNames,
+                finalAuthNames,
                 updatedQueryParams,
                 headerParams,
                 cookieParams,
@@ -1862,4 +1882,13 @@ private Boolean validateHeader(String name, String value) {
 
         return !value.contains("\n") && !value.contains("\t");
     }
+
+    private String[] append(String[] array, String value) {
+        if (array == null) {
+            return new String[] {value};
+        }
+        String[] newArray = Arrays.copyOf(array, array.length + 1);
+        newArray[newArray.length - 1] = value;
+        return newArray;
+    }
 }

clients/common/src/main/java/com/binance/connector/client/common/configuration/ClientConfiguration.java

@@ -3,7 +3,6 @@
 import com.binance.connector.client.common.dtos.TimeUnit;
 import java.net.Proxy;
 import java.util.Map;
-
 import okhttp3.Authenticator;
 import okhttp3.CertificatePinner;
 

clients/common/src/main/java/com/binance/connector/client/common/websocket/adapter/ConnectionInterface.java

@@ -2,12 +2,15 @@
 
 import com.binance.connector.client.common.websocket.dtos.ApiRequestWrapperDTO;
 import com.binance.connector.client.common.websocket.dtos.RequestWrapperDTO;
+import java.util.concurrent.BlockingQueue;
 
 public interface ConnectionInterface {
     void connect();
 
     void send(ApiRequestWrapperDTO request) throws InterruptedException;
 
+    BlockingQueue<String> sendForStream(ApiRequestWrapperDTO request) throws InterruptedException;
+
     void send(RequestWrapperDTO request) throws InterruptedException;
 
     void setUserAgent(String userAgent);

clients/common/src/main/java/com/binance/connector/client/common/websocket/adapter/ConnectionWrapper.java

@@ -28,6 +28,7 @@
 import java.nio.channels.ClosedChannelException;
 import java.text.DecimalFormat;
 import java.time.Duration;
+import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.HashMap;
@@ -36,8 +37,10 @@
 import java.util.Timer;
 import java.util.TimerTask;
 import java.util.UUID;
+import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.LinkedBlockingDeque;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Consumer;
 import java.util.logging.Level;
@@ -87,6 +90,8 @@ public class ConnectionWrapper implements WebSocketListener, ConnectionInterface
 
     private boolean pendingReconnect = false;
 
+    private List<BlockingQueue<String>> streamQueues = new ArrayList<>();
+
     public ConnectionWrapper(WebSocketClientConfiguration configuration, Gson gson) {
         this(configuration, null, gson);
     }
@@ -349,6 +354,15 @@ public void send(ApiRequestWrapperDTO request) {
         innerSend(request);
     }
 
+    @Override
+    public BlockingQueue<String> sendForStream(ApiRequestWrapperDTO request)
+            throws InterruptedException {
+        LinkedBlockingDeque<String> streamQueue = new LinkedBlockingDeque<>();
+        streamQueues.add(streamQueue);
+        send(request);
+        return streamQueue;
+    }
+
     public void innerSend(RequestWrapperDTO requestWrapperDTO) {
         send(requestWrapperDTO);
     }
@@ -431,12 +445,18 @@ public void onWebSocketText(String message) {
             JsonObject obj = root.getAsJsonObject();
             JsonElement idElem = obj.get("id");
             String id = idElem == null ? null : idElem.getAsString();
+            RequestWrapperDTO requestWrapperDTO = null;
+            if (id != null) {
+                requestWrapperDTO = pendingRequest.get(id);
+            }
 
-            if (id == null) {
+            if (requestWrapperDTO == null) {
+                for (BlockingQueue<String> streamQueue : streamQueues) {
+                    JsonElement eventElem = obj.get("event");
+                    streamQueue.offer(eventElem != null ? eventElem.toString() : message);
+                }
                 return;
             }
-
-            RequestWrapperDTO requestWrapperDTO = pendingRequest.get(id);
             Type responseType = requestWrapperDTO.getResponseType();
 
             Object responseResult = gson.fromJson(root, responseType);

clients/common/src/main/java/com/binance/connector/client/common/websocket/adapter/PoolConnectionWrapper.java

@@ -9,6 +9,7 @@
 import java.util.ListIterator;
 import java.util.Timer;
 import java.util.TimerTask;
+import java.util.concurrent.BlockingQueue;
 
 public class PoolConnectionWrapper implements ConnectionInterface {
     private final LinkedList<ConnectionWrapper> connectionList = new LinkedList<>();
@@ -86,6 +87,12 @@ public void send(RequestWrapperDTO request) throws InterruptedException {
         getConnection().send(request);
     }
 
+    @Override
+    public BlockingQueue<String> sendForStream(ApiRequestWrapperDTO request)
+            throws InterruptedException {
+        return getConnection().sendForStream(request);
+    }
+
     /**
      * @return the next connection from the pool, using round-robin
      */

clients/common/src/main/java/com/binance/connector/client/common/websocket/dtos/StreamResponse.java

@@ -0,0 +1,22 @@
+package com.binance.connector.client.common.websocket.dtos;
+
+import com.binance.connector.client.common.websocket.service.StreamBlockingQueueWrapper;
+import java.util.concurrent.CompletableFuture;
+
+public class StreamResponse<T, U> {
+    private final CompletableFuture<T> response;
+    private final StreamBlockingQueueWrapper<U> stream;
+
+    public StreamResponse(CompletableFuture<T> response, StreamBlockingQueueWrapper<U> stream) {
+        this.response = response;
+        this.stream = stream;
+    }
+
+    public CompletableFuture<T> getResponse() {
+        return response;
+    }
+
+    public StreamBlockingQueueWrapper<U> getStream() {
+        return stream;
+    }
+}

clients/common/src/main/java/com/binance/connector/client/common/websocket/service/StreamBlockingQueue.java

@@ -11,6 +11,10 @@ public class StreamBlockingQueue<T> implements BlockingQueue<T> {
     private final BlockingQueue<T> innerQueue;
     private final String operationId;
 
+    public StreamBlockingQueue(BlockingQueue<T> innerQueue) {
+        this(innerQueue, "");
+    }
+
     public StreamBlockingQueue(BlockingQueue<T> innerQueue, String operationId) {
         this.innerQueue = innerQueue;
         this.operationId = operationId;

clients/common/src/main/java/com/binance/connector/client/common/websocket/service/StreamBlockingQueueWrapper.java

@@ -1,15 +1,23 @@
 package com.binance.connector.client.common.websocket.service;
 
 import com.binance.connector.client.common.JSON;
+import com.google.gson.Gson;
 import com.google.gson.reflect.TypeToken;
 
 public class StreamBlockingQueueWrapper<T> {
     private final StreamBlockingQueue<String> innerQueue;
     private final TypeToken<T> convertType;
+    private final Gson gson;
 
     public StreamBlockingQueueWrapper(StreamBlockingQueue<String> innerQueue, TypeToken<T> type) {
+        this(innerQueue, type, JSON.getGson());
+    }
+
+    public StreamBlockingQueueWrapper(
+            StreamBlockingQueue<String> innerQueue, TypeToken<T> type, Gson gson) {
         this.innerQueue = innerQueue;
         this.convertType = type;
+        this.gson = gson;
     }
 
     public StreamBlockingQueue<String> getInnerQueue() {
@@ -18,6 +26,6 @@ public StreamBlockingQueue<String> getInnerQueue() {
 
     public T take() throws InterruptedException {
         String take = innerQueue.take();
-        return JSON.getGson().fromJson(take, convertType);
+        return gson.fromJson(take, convertType);
     }
 }