Add pool for proxy subscription

Author: apanevin

devicehive-proxy-ws-kafka-impl/src/main/java/com/devicehive/proxy/FrontendProxyClient.java

@@ -22,11 +22,14 @@
 
 import com.devicehive.api.RequestResponseMatcher;
 import com.devicehive.model.ServerEvent;
+import com.devicehive.proxy.api.NotificationHandler;
 import com.devicehive.proxy.api.ProxyClient;
 import com.devicehive.proxy.api.ProxyMessageBuilder;
 import com.devicehive.proxy.api.payload.NotificationCreatePayload;
 import com.devicehive.proxy.api.payload.SubscribePayload;
 import com.devicehive.proxy.api.payload.TopicsPayload;
+import com.devicehive.proxy.client.WebSocketKafkaProxyClient;
+import com.devicehive.proxy.config.WebSocketKafkaProxyConfig;
 import com.devicehive.shim.api.Request;
 import com.devicehive.shim.api.RequestType;
 import com.devicehive.shim.api.Response;
@@ -37,29 +40,32 @@
 import org.slf4j.LoggerFactory;
 
 import java.util.Arrays;
-import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
+import java.util.UUID;
+import java.util.concurrent.*;
 import java.util.function.Consumer;
 
 public class FrontendProxyClient implements RpcClient {
     private static final Logger logger = LoggerFactory.getLogger(FrontendProxyClient.class);
 
     private final String requestTopic;
     private final String replyToTopic;
-    private final ProxyClient client;
+    private final WebSocketKafkaProxyClient client;
+    private final WebSocketKafkaProxyConfig proxyConfig;
+    private final NotificationHandler notificationHandler;
     private final RequestResponseMatcher requestResponseMatcher;
     private final Gson gson;
     private final RingBuffer<ServerEvent> ringBuffer;
 
-    public FrontendProxyClient(String requestTopic, String replyToTopic, ProxyClient client, RequestResponseMatcher requestResponseMatcher, Gson gson, RingBuffer<ServerEvent> ringBuffer) {
+    public FrontendProxyClient(String requestTopic, String replyToTopic, WebSocketKafkaProxyConfig proxyConfig, NotificationHandler notificationHandler, RequestResponseMatcher requestResponseMatcher, Gson gson, RingBuffer<ServerEvent> ringBuffer) {
         this.requestTopic = requestTopic;
         this.replyToTopic = replyToTopic;
-        this.client = client;
+        this.proxyConfig = proxyConfig;
+        this.notificationHandler = notificationHandler;
         this.requestResponseMatcher = requestResponseMatcher;
         this.gson = gson;
         this.ringBuffer = ringBuffer;
+        this.client = new WebSocketKafkaProxyClient((message, client) -> {});
+        client.setWebSocketKafkaProxyConfig(proxyConfig);
     }
 
     @Override
@@ -85,7 +91,17 @@ public void push(Request request) {
     public void start() {
         client.start();
         client.push(ProxyMessageBuilder.create(new TopicsPayload(Arrays.asList(requestTopic, replyToTopic)))).join();
-        client.push(ProxyMessageBuilder.subscribe(new SubscribePayload(replyToTopic))).join();
+
+        UUID uuid = UUID.randomUUID();
+        Executor executionPool = Executors.newFixedThreadPool(proxyConfig.getWorkerThreads());
+        for (int i = 0; i < proxyConfig.getWorkerThreads(); i++) {
+            executionPool.execute(() -> {
+                WebSocketKafkaProxyClient client = new WebSocketKafkaProxyClient(notificationHandler);
+                client.setWebSocketKafkaProxyConfig(proxyConfig);
+                client.start();
+                client.push(ProxyMessageBuilder.subscribe(new SubscribePayload(replyToTopic, uuid.toString()))).join();
+            });
+        }
 
         pingServer();
     }

devicehive-proxy-ws-kafka-impl/src/main/java/com/devicehive/proxy/config/BackendProxyClientConfig.java

@@ -42,6 +42,7 @@
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.Profile;
 
+import java.util.concurrent.Executor;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.stream.IntStream;
@@ -77,13 +78,18 @@ public NotificationHandler notificationHandler(Gson gson, WorkerPool<ServerEvent
     }
 
     @Bean
-    public ProxyClient proxyClient(NotificationHandler notificationHandler, WebSocketKafkaProxyConfig proxyConfig) {
-        WebSocketKafkaProxyClient client = new WebSocketKafkaProxyClient(notificationHandler);
-        client.setWebSocketKafkaProxyConfig(proxyConfig);
-        client.start();
-        client.push(ProxyMessageBuilder.create(new TopicsPayload(REQUEST_TOPIC))).join();
-        client.push(ProxyMessageBuilder.subscribe(new SubscribePayload(REQUEST_TOPIC, proxyConfig.getConsumerGroup()))).join();
-        return client;
+    public Executor executionPool(NotificationHandler notificationHandler, WebSocketKafkaProxyConfig proxyConfig) {
+        Executor executionPool = Executors.newFixedThreadPool(proxyConfig.getWorkerThreads());
+        for (int i = 0; i < proxyConfig.getWorkerThreads(); i++) {
+            executionPool.execute(() -> {
+                WebSocketKafkaProxyClient client = new WebSocketKafkaProxyClient(notificationHandler);
+                client.setWebSocketKafkaProxyConfig(proxyConfig);
+                client.start();
+                client.push(ProxyMessageBuilder.create(new TopicsPayload(REQUEST_TOPIC))).join();
+                client.push(ProxyMessageBuilder.subscribe(new SubscribePayload(REQUEST_TOPIC, proxyConfig.getConsumerGroup()))).join();
+            });
+        }
+        return executionPool;
     }
 
     @Bean

devicehive-proxy-ws-kafka-impl/src/main/java/com/devicehive/proxy/config/FrontendProxyClientConfig.java

@@ -25,7 +25,6 @@
 import com.devicehive.proxy.FrontendProxyClient;
 import com.devicehive.proxy.ProxyResponseHandler;
 import com.devicehive.proxy.api.NotificationHandler;
-import com.devicehive.proxy.client.WebSocketKafkaProxyClient;
 import com.devicehive.shim.api.client.RpcClient;
 import com.google.gson.Gson;
 import com.lmax.disruptor.*;
@@ -103,11 +102,9 @@ public WorkerPool<ServerEvent> workerPool(Gson gson, RequestResponseMatcher requ
 
     @Bean
     public RpcClient rpcClient(NotificationHandler notificationHandler, WebSocketKafkaProxyConfig proxyConfig, RequestResponseMatcher requestResponseMatcher, Gson gson, WorkerPool<ServerEvent> workerPool) {
-        WebSocketKafkaProxyClient proxyClient = new WebSocketKafkaProxyClient(notificationHandler);
-        proxyClient.setWebSocketKafkaProxyConfig(proxyConfig);
         final ExecutorService execService = Executors.newFixedThreadPool(proxyConfig.getWorkerThreads());
         RingBuffer<ServerEvent> ringBuffer = workerPool.start(execService);
-        RpcClient client = new FrontendProxyClient(REQUEST_TOPIC, RESPONSE_TOPIC, proxyClient, requestResponseMatcher, gson, ringBuffer);
+        RpcClient client = new FrontendProxyClient(REQUEST_TOPIC, RESPONSE_TOPIC, proxyConfig, notificationHandler, requestResponseMatcher, gson, ringBuffer);
         client.start();
         return client;
     }