fixes #2127 - do not use a separate counter for batch ids

CHANGELOG.md

@@ -7,7 +7,7 @@ See [Conventional Commits](https://conventionalcommits.org) for commit guideline
 
 ### Bug Fixes
 
-* 
+* fixed subscription id conflict [#2127](https://github.com/hyperledger/web3j/pull/2127)
 
 ### Features
 

core/src/main/java/org/web3j/protocol/core/DefaultIdProvider.java

@@ -0,0 +1,26 @@
+/*
+ * Copyright 2020 Web3 Labs Ltd.
+ *
+ * Licensed 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.web3j.protocol.core;
+
+import java.util.concurrent.atomic.AtomicLong;
+
+public class DefaultIdProvider {
+    protected static final AtomicLong nextId = new AtomicLong(0);
+
+    protected DefaultIdProvider() {
+    }
+
+    public static long getNextId() {
+        return nextId.getAndIncrement();
+    }
+}

core/src/main/java/org/web3j/protocol/core/Request.java

@@ -15,16 +15,13 @@
 import java.io.IOException;
 import java.util.List;
 import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.atomic.AtomicLong;
 
 import com.fasterxml.jackson.annotation.JsonIgnore;
 import io.reactivex.Flowable;
 
 import org.web3j.protocol.Web3jService;
 
 public class Request<S, T extends Response> {
-    private static AtomicLong nextId = new AtomicLong(0);
-
     private String jsonrpc = "2.0";
     private String method;
     private List<S> params;
@@ -41,7 +38,7 @@ public Request() {}
     public Request(String method, List<S> params, Web3jService web3jService, Class<T> type) {
         this.method = method;
         this.params = params;
-        this.id = nextId.getAndIncrement();
+        this.id = DefaultIdProvider.getNextId();
         this.web3jService = web3jService;
         this.responseType = type;
     }

core/src/main/java/org/web3j/protocol/websocket/WebSocketService.java

@@ -26,7 +26,6 @@
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicLong;
 import java.util.function.Consumer;
 
 import com.fasterxml.jackson.core.JsonProcessingException;
@@ -44,6 +43,7 @@
 import org.web3j.protocol.Web3jService;
 import org.web3j.protocol.core.BatchRequest;
 import org.web3j.protocol.core.BatchResponse;
+import org.web3j.protocol.core.DefaultIdProvider;
 import org.web3j.protocol.core.Request;
 import org.web3j.protocol.core.Response;
 import org.web3j.protocol.core.methods.response.EthSubscribe;
@@ -65,8 +65,6 @@ public class WebSocketService implements Web3jService {
 
     // Timeout for JSON-RPC requests
     static final long REQUEST_TIMEOUT = 60;
-    // replaced batch's next id
-    static final AtomicLong nextBatchId = new AtomicLong(0);
 
     // WebSocket client
     private final WebSocketClient webSocketClient;
@@ -223,7 +221,7 @@ public CompletableFuture<BatchResponse> sendBatchAsync(BatchRequest requests) {
         CompletableFuture<BatchResponse> result = new CompletableFuture<>();
 
         // replace first batch elements's id to handle response
-        long requestId = nextBatchId.getAndIncrement();
+        long requestId = DefaultIdProvider.getNextId();
         Request<?, ? extends Response<?>> firstRequest = requests.getRequests().get(0);
         long originId = firstRequest.getId();
         requests.getRequests().get(0).setId(requestId);

core/src/test/java/org/web3j/protocol/websocket/WebSocketServiceTest.java

@@ -33,6 +33,7 @@
 
 import org.web3j.protocol.core.BatchRequest;
 import org.web3j.protocol.core.BatchResponse;
+import org.web3j.protocol.core.DefaultIdProvider;
 import org.web3j.protocol.core.Request;
 import org.web3j.protocol.core.Response;
 import org.web3j.protocol.core.methods.response.EthSubscribe;
@@ -213,6 +214,7 @@ void testBatchRequestReply() throws Exception {
                                 NetVersion.class));
         request.getRequests().get(0).setId(1L);
         request.getRequests().get(1).setId(1L);
+        DefaultIdProviderReset.resetNextId();
 
         CompletableFuture<BatchResponse> reply = service.sendBatchAsync(request);
 
@@ -625,4 +627,10 @@ private void sendWebSocketEvent() throws IOException {
                         + "  }"
                         + "}");
     }
+
+    private class DefaultIdProviderReset extends DefaultIdProvider {
+        static void resetNextId() {
+            nextId.set(0);
+        }
+    }
 }