Add Disruptor support for Frontend

Author: apanevin

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

@@ -21,6 +21,7 @@
  */
 
 import com.devicehive.api.RequestResponseMatcher;
+import com.devicehive.model.ServerEvent;
 import com.devicehive.proxy.api.ProxyClient;
 import com.devicehive.proxy.api.ProxyMessageBuilder;
 import com.devicehive.proxy.api.payload.NotificationCreatePayload;
@@ -31,6 +32,7 @@
 import com.devicehive.shim.api.Response;
 import com.devicehive.shim.api.client.RpcClient;
 import com.google.gson.Gson;
+import com.lmax.disruptor.RingBuffer;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -49,21 +51,23 @@ public class FrontendProxyClient implements RpcClient {
     private final ProxyClient client;
     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) {
+    public FrontendProxyClient(String requestTopic, String replyToTopic, ProxyClient client, RequestResponseMatcher requestResponseMatcher, Gson gson, RingBuffer<ServerEvent> ringBuffer) {
         this.requestTopic = requestTopic;
         this.replyToTopic = replyToTopic;
         this.client = client;
         this.requestResponseMatcher = requestResponseMatcher;
         this.gson = gson;
+        this.ringBuffer = ringBuffer;
     }
 
     @Override
     public void call(Request request, Consumer<Response> callback) {
         requestResponseMatcher.addRequestCallback(request.getCorrelationId(), callback);
         logger.debug("Request callback added for request: {}, correlationId: {}", request.getBody(), request.getCorrelationId());
 
-        push(request);
+        ringBuffer.publishEvent((serverEvent, sequence, response) -> serverEvent.set(response), request);
     }
 
     @Override

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

@@ -21,25 +21,41 @@
  */
 
 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.client.WebSocketKafkaProxyClient;
+import com.devicehive.proxy.config.WebSocketKafkaProxyConfig;
+import com.devicehive.shim.api.Request;
 import com.devicehive.shim.api.Response;
 import com.google.gson.Gson;
+import com.lmax.disruptor.WorkHandler;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
 
-public class ProxyResponseHandler implements NotificationHandler {
+public class ProxyResponseHandler implements NotificationHandler, WorkHandler<ServerEvent> {
 
     private static final Logger logger = LoggerFactory.getLogger(ProxyResponseHandler.class);
 
     private final Gson gson;
+    private final String requestTopic;
+    private final String replyToTopic;
+    private final ProxyClient proxyClient;
     private final RequestResponseMatcher requestResponseMatcher;
 
     @Autowired
-    public ProxyResponseHandler(Gson gson, RequestResponseMatcher requestResponseMatcher) {
+    public ProxyResponseHandler(Gson gson, String requestTopic, String replyToTopic, WebSocketKafkaProxyConfig proxyConfig, RequestResponseMatcher requestResponseMatcher) {
         this.gson = gson;
+        this.requestTopic = requestTopic;
+        this.replyToTopic = replyToTopic;
         this.requestResponseMatcher = requestResponseMatcher;
+        WebSocketKafkaProxyClient webSocketKafkaProxyClient = new WebSocketKafkaProxyClient((message, client) -> {});
+        webSocketKafkaProxyClient.setWebSocketKafkaProxyConfig(proxyConfig);
+        this.proxyClient = webSocketKafkaProxyClient;
+        this.proxyClient.start();
     }
 
     @Override
@@ -49,4 +65,16 @@ public void handle(String message, ProxyClient client) {
 
         requestResponseMatcher.offerResponse(response);
     }
+
+    @Override
+    public void onEvent(ServerEvent serverEvent) throws Exception {
+        final Request request = serverEvent.get();
+        if (request.getBody() == null) {
+            throw new NullPointerException("Request body must not be null.");
+        }
+        request.setReplyTo(replyToTopic);
+
+        proxyClient.push(ProxyMessageBuilder.notification(
+                new NotificationCreatePayload(requestTopic, gson.toJson(request), request.getPartitionKey())));
+    }
 }

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

@@ -21,13 +21,14 @@
  */
 
 import com.devicehive.api.RequestResponseMatcher;
+import com.devicehive.model.ServerEvent;
 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.proxy.config.WebSocketKafkaProxyConfig;
 import com.devicehive.shim.api.client.RpcClient;
 import com.google.gson.Gson;
+import com.lmax.disruptor.*;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.ComponentScan;
@@ -42,6 +43,9 @@
 import java.util.Base64;
 import java.util.Optional;
 import java.util.UUID;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.stream.IntStream;
 
 import static com.devicehive.configuration.Constants.REQUEST_TOPIC;
 
@@ -80,16 +84,54 @@ public RequestResponseMatcher requestResponseMatcher() {
     }
 
     @Bean
-    public NotificationHandler notificationHandler(Gson gson, RequestResponseMatcher requestResponseMatcher) {
-        return new ProxyResponseHandler(gson, requestResponseMatcher);
+    public NotificationHandler notificationHandler(Gson gson, RequestResponseMatcher requestResponseMatcher, WebSocketKafkaProxyConfig proxyConfig) {
+        return new ProxyResponseHandler(gson, REQUEST_TOPIC, RESPONSE_TOPIC, proxyConfig, requestResponseMatcher);
     }
 
     @Bean
-    public RpcClient rpcClient(NotificationHandler notificationHandler, WebSocketKafkaProxyConfig proxyConfig, RequestResponseMatcher requestResponseMatcher, Gson gson) {
+    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)
+        );
+        final RingBuffer<ServerEvent> ringBuffer = RingBuffer.createMultiProducer(ServerEvent::new, proxyConfig.getBufferSize(), getWaitStrategy(proxyConfig.getWaitStrategy()));
+        final SequenceBarrier barrier = ringBuffer.newBarrier();
+        WorkerPool<ServerEvent> workerPool = new WorkerPool<>(ringBuffer, barrier, new FatalExceptionHandler(), workHandlers);
+        ringBuffer.addGatingSequences(workerPool.getWorkerSequences());
+        return workerPool;
+    }
+
+    @Bean
+    public RpcClient rpcClient(NotificationHandler notificationHandler, WebSocketKafkaProxyConfig proxyConfig, RequestResponseMatcher requestResponseMatcher, Gson gson, WorkerPool<ServerEvent> workerPool) {
         WebSocketKafkaProxyClient proxyClient = new WebSocketKafkaProxyClient(notificationHandler);
         proxyClient.setWebSocketKafkaProxyConfig(proxyConfig);
-        RpcClient client = new FrontendProxyClient(REQUEST_TOPIC, RESPONSE_TOPIC, proxyClient, requestResponseMatcher, gson);
+        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);
         client.start();
         return client;
     }
+
+    private WaitStrategy getWaitStrategy(String strategy) {
+        WaitStrategy waitStrategy;
+
+        switch (strategy) {
+            case "blocking":
+                waitStrategy = new BlockingWaitStrategy();
+                break;
+            case "sleeping":
+                waitStrategy = new SleepingWaitStrategy();
+                break;
+            case "yielding":
+                waitStrategy = new YieldingWaitStrategy();
+                break;
+            case "busyspin":
+                waitStrategy = new BusySpinWaitStrategy();
+                break;
+            default:
+                waitStrategy = new BlockingWaitStrategy();
+                break;
+        }
+        return waitStrategy;
+    }
 }