package com.devicehive.shim.config.server;

/*
 * #%L
 * DeviceHive Shim Kafka Implementation
 * %%
 * Copyright (C) 2016 DataArt
 * %%
 * 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.
 * #L%
 */

import com.devicehive.shim.api.Response;
import com.devicehive.shim.api.server.RequestHandler;
import com.devicehive.shim.api.server.RpcServer;
import com.devicehive.shim.kafka.serializer.RequestSerializer;
import com.devicehive.shim.kafka.serializer.ResponseSerializer;
import com.devicehive.shim.kafka.server.KafkaRpcServer;
import com.devicehive.shim.kafka.server.RequestConsumer;
import com.devicehive.shim.kafka.server.ServerEvent;
import com.devicehive.shim.kafka.server.ServerEventHandler;
import com.google.gson.Gson;
import com.lmax.disruptor.*;
import com.lmax.disruptor.dsl.Disruptor;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringSerializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.*;
import org.springframework.core.env.Environment;

import java.util.Properties;
import java.util.concurrent.Executors;
import java.util.stream.IntStream;

@Configuration
@Profile("rpc-server")
@ComponentScan("com.devicehive.shim.config")
@PropertySource("classpath:kafka.properties")
public class KafkaRpcServerConfig {
    private static final Logger logger = LoggerFactory.getLogger(KafkaRpcServerConfig.class);

    public static final String REQUEST_TOPIC = "request_topic";

    @Autowired
    private Environment env;

    @Autowired
    private KafkaRpcConfig kafkaRpcConfig;

    @Value("${rpc.server.request-consumer.threads:1}")
    private int consumerThreads;

    @Value("${lmax.buffer-size:1024}")
    private int bufferSize;

    @Value("${lmax.wait.strategy:busyspin}")
    private String waitStrategy;

    @Bean(name = "server-producer")
    public Producer<String, Response> kafkaResponseProducer(Gson gson) {
        return new KafkaProducer<>(kafkaRpcConfig.producerProps(), new StringSerializer(), new ResponseSerializer(gson));
    }

    @Bean
    public WorkerPool<ServerEvent> workerPool(RequestHandler requestHandler,
                                                 @Qualifier("server-producer") Producer<String, Response> responseProducer) {
        final ServerEventHandler[] workHandlers = new ServerEventHandler[workerThreads];
        IntStream.range(0, workerThreads).forEach(
                nbr -> workHandlers[nbr] = new ServerEventHandler(requestHandler, responseProducer)
        );
        final RingBuffer<ServerEvent> ringBuffer = RingBuffer.createMultiProducer(ServerEvent::new, 1024, getWaitStrategy());
        final SequenceBarrier barrier = ringBuffer.newBarrier();
        WorkerPool<ServerEvent> workerPool = new WorkerPool<>(ringBuffer, barrier, new FatalExceptionHandler(), workHandlers);
        ringBuffer.addGatingSequences(workerPool.getWorkerSequences());
        return workerPool;
    }

    private WaitStrategy getWaitStrategy() {
        logger.info("RPC server wait strategy: {}", waitStrategy);
        WaitStrategy strategy;

        switch (waitStrategy) {
            case "blocking":
                strategy = new BlockingWaitStrategy(); break;
            case "sleeping":
                strategy =  new SleepingWaitStrategy(); break;
            case "yielding":
                strategy = new YieldingWaitStrategy(); break;
            case "busyspin":
                strategy =  new BusySpinWaitStrategy(); break;
            default:
                strategy =  new BusySpinWaitStrategy(); break;
        }
        return strategy;
    }

    @Bean
    public ServerEventHandler serverEventHandler(RequestHandler requestHandler,
                                                 @Qualifier("server-producer") Producer<String, Response> responseProducer) {
        return new ServerEventHandler(requestHandler, responseProducer);
    }

    @Bean
    public RequestConsumer requestConsumer(Gson gson) {
        return new RequestConsumer(REQUEST_TOPIC, kafkaRpcConfig.serverConsumerProps(), consumerThreads, new RequestSerializer(gson));
    }

    @Bean
    public RpcServer rpcServer(WorkerPool<ServerEvent> workerPool, RequestConsumer requestConsumer, ServerEventHandler eventHandler) {
        RpcServer server = new KafkaRpcServer(workerPool, requestConsumer, eventHandler, workerThreads);
        server.start();
        return server;
    }
}

@@ -1018,32 +1018,82 @@
 KafkaRpcServer;%0A
+import com.devicehive.shim.config.KafkaRpcConfig;%0A
 import com.devic
@@ -1129,24 +1129,24 @@
 stConsumer;%0A
-
 import com.d
@@ -1308,106 +1308,8 @@
 .*;%0A
-import com.lmax.disruptor.dsl.Disruptor;%0Aimport org.apache.kafka.clients.consumer.ConsumerConfig;%0A
 impo
@@ -1415,65 +1415,8 @@
 er;%0A
-import org.apache.kafka.clients.producer.ProducerConfig;%0A
 impo
@@ -1819,76 +1819,8 @@
 t;%0A%0A
-import java.util.Properties;%0Aimport java.util.concurrent.Executors;%0A
 impo
@@ -2331,24 +2331,24 @@
 hreads:1%7D%22)%0A
-
     private 
@@ -2369,16 +2369,93 @@
 reads;%0A%0A
+    @Value(%22$%7Brpc.server.worker.threads:1%7D%22)%0A    private int workerThreads;%0A%0A
     @Val

Fix disruptor build