Reduce amount of proxy ws client

devicehive · May 25, 2018 · f25a5ac · f25a5ac
1 parent d74c180
commit f25a5ac
Show file tree

Hide file tree

Showing 2 changed files with 9 additions and 2 deletions.
diff --git a/devicehive-proxy-ws-kafka-impl/src/main/java/com/devicehive/proxy/ProxyResponseHandler.java b/devicehive-proxy-ws-kafka-impl/src/main/java/com/devicehive/proxy/ProxyResponseHandler.java
@@ -55,7 +55,10 @@ public ProxyResponseHandler(Gson gson, String requestTopic, String replyToTopic,
         WebSocketKafkaProxyClient webSocketKafkaProxyClient = new WebSocketKafkaProxyClient((message, client) -> {});
         webSocketKafkaProxyClient.setWebSocketKafkaProxyConfig(proxyConfig);
         this.proxyClient = webSocketKafkaProxyClient;
-        this.proxyClient.start();
+    }
+
+    public void start() {
+        proxyClient.start();
     }
 
     @Override

diff --git a/...xy-ws-kafka-impl/src/main/java/com/devicehive/proxy/config/FrontendProxyClientConfig.java b/...xy-ws-kafka-impl/src/main/java/com/devicehive/proxy/config/FrontendProxyClientConfig.java
@@ -91,7 +91,11 @@ public NotificationHandler notificationHandler(Gson gson, RequestResponseMatcher
     public WorkerPool<ServerEvent> workerPool(Gson gson, RequestResponseMatcher requestResponseMatcher, WebSocketKafkaProxyConfig proxyConfig) {
         final ProxyResponseHandler[] workHandlers = new ProxyResponseHandler[proxyConfig.getWorkerThreads()];
         IntStream.range(0, proxyConfig.getWorkerThreads()).forEach(
-                nbr -> workHandlers[nbr] = new ProxyResponseHandler(gson, REQUEST_TOPIC, RESPONSE_TOPIC, proxyConfig, requestResponseMatcher)
+                nbr -> {
+                    ProxyResponseHandler handler = new ProxyResponseHandler(gson, REQUEST_TOPIC, RESPONSE_TOPIC, proxyConfig, requestResponseMatcher);
+                    handler.start();
+                    workHandlers[nbr] = handler;
+                }
         );
         final RingBuffer<ServerEvent> ringBuffer = RingBuffer.createMultiProducer(ServerEvent::new, proxyConfig.getBufferSize(), getWaitStrategy(proxyConfig.getWaitStrategy()));
         final SequenceBarrier barrier = ringBuffer.newBarrier();