Merge pull request #519 from devicehive/development

Reduce amount of proxy ws client

Author: tmatvienko

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

devicehive-proxy-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();