partitions);
+
+}
diff --git a/common/queue/src/main/java/org/thingsboard/server/kafka/TbKafkaProperty.java b/common/queue/src/main/java/org/thingsboard/server/kafka/TbKafkaProperty.java
new file mode 100644
index 0000000000..784a5c5645
--- /dev/null
+++ b/common/queue/src/main/java/org/thingsboard/server/kafka/TbKafkaProperty.java
@@ -0,0 +1,32 @@
+/**
+ * Copyright © 2016-2018 The Thingsboard Authors
+ *
+ * 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.
+ */
+package org.thingsboard.server.kafka;
+
+import lombok.Data;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.stereotype.Component;
+
+/**
+ * Created by ashvayka on 25.09.18.
+ */
+@Data
+public class TbKafkaProperty {
+
+ private String key;
+ private String value;
+}
diff --git a/common/queue/src/main/java/org/thingsboard/server/kafka/TbKafkaRequestTemplate.java b/common/queue/src/main/java/org/thingsboard/server/kafka/TbKafkaRequestTemplate.java
new file mode 100644
index 0000000000..be8c087060
--- /dev/null
+++ b/common/queue/src/main/java/org/thingsboard/server/kafka/TbKafkaRequestTemplate.java
@@ -0,0 +1,173 @@
+/**
+ * Copyright © 2016-2018 The Thingsboard Authors
+ *
+ * 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.
+ */
+package org.thingsboard.server.kafka;
+
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.ListenableFuture;
+import com.google.common.util.concurrent.SettableFuture;
+import lombok.Builder;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.kafka.clients.admin.CreateTopicsResult;
+import org.apache.kafka.clients.admin.NewTopic;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.common.header.Header;
+import org.apache.kafka.common.header.internals.RecordHeader;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeoutException;
+
+/**
+ * Created by ashvayka on 25.09.18.
+ */
+@Slf4j
+public class TbKafkaRequestTemplate {
+
+ private final TBKafkaProducerTemplate requestTemplate;
+ private final TBKafkaConsumerTemplate responseTemplate;
+ private final ConcurrentMap> pendingRequests;
+ private final ExecutorService executor;
+ private final long maxRequestTimeout;
+ private final long maxPendingRequests;
+ private final long pollInterval;
+ private volatile long tickTs = 0L;
+ private volatile long tickSize = 0L;
+ private volatile boolean stopped = false;
+
+ @Builder
+ public TbKafkaRequestTemplate(TBKafkaProducerTemplate requestTemplate, TBKafkaConsumerTemplate responseTemplate,
+ long maxRequestTimeout,
+ long maxPendingRequests,
+ long pollInterval,
+ ExecutorService executor) {
+ this.requestTemplate = requestTemplate;
+ this.responseTemplate = responseTemplate;
+ this.pendingRequests = new ConcurrentHashMap<>();
+ this.maxRequestTimeout = maxRequestTimeout;
+ this.maxPendingRequests = maxPendingRequests;
+ this.pollInterval = pollInterval;
+ if (executor != null) {
+ this.executor = executor;
+ } else {
+ this.executor = Executors.newSingleThreadExecutor();
+ }
+ }
+
+ public void init() {
+ try {
+ TBKafkaAdmin admin = new TBKafkaAdmin();
+ CreateTopicsResult result = admin.createTopic(new NewTopic(responseTemplate.getTopic(), 1, (short) 1));
+ result.all().get();
+ } catch (Exception e) {
+ log.trace("Failed to create topic: {}", e.getMessage(), e);
+ }
+ tickTs = System.currentTimeMillis();
+ responseTemplate.subscribe();
+ executor.submit(() -> {
+ long nextCleanupMs = 0L;
+ while (!stopped) {
+ ConsumerRecords responses = responseTemplate.poll(Duration.ofMillis(pollInterval));
+ responses.forEach(response -> {
+ Header requestIdHeader = response.headers().lastHeader(TbKafkaSettings.REQUEST_ID_HEADER);
+ if (requestIdHeader == null) {
+ log.error("[{}] Missing requestIdHeader", response);
+ }
+ UUID requestId = bytesToUuid(requestIdHeader.value());
+ ResponseMetaData expectedResponse = pendingRequests.remove(requestId);
+ if (expectedResponse == null) {
+ log.trace("[{}] Invalid or stale request", requestId);
+ } else {
+ try {
+ expectedResponse.future.set(responseTemplate.decode(response));
+ } catch (IOException e) {
+ expectedResponse.future.setException(e);
+ }
+ }
+ });
+ tickTs = System.currentTimeMillis();
+ tickSize = pendingRequests.size();
+ if (nextCleanupMs < tickTs) {
+ //cleanup;
+ pendingRequests.entrySet().forEach(kv -> {
+ if (kv.getValue().expTime < tickTs) {
+ ResponseMetaData staleRequest = pendingRequests.remove(kv.getKey());
+ if (staleRequest != null) {
+ staleRequest.future.setException(new TimeoutException());
+ }
+ }
+ });
+ nextCleanupMs = tickTs + maxRequestTimeout;
+ }
+ }
+ });
+ }
+
+ public void stop() {
+ stopped = true;
+ }
+
+ public ListenableFuture post(String key, Request request) {
+ if (tickSize > maxPendingRequests) {
+ return Futures.immediateFailedFuture(new RuntimeException("Pending request map is full!"));
+ }
+ UUID requestId = UUID.randomUUID();
+ List headers = new ArrayList<>(2);
+ headers.add(new RecordHeader(TbKafkaSettings.REQUEST_ID_HEADER, uuidToBytes(requestId)));
+ headers.add(new RecordHeader(TbKafkaSettings.RESPONSE_TOPIC_HEADER, stringToBytes(responseTemplate.getTopic())));
+ SettableFuture future = SettableFuture.create();
+ pendingRequests.putIfAbsent(requestId, new ResponseMetaData<>(tickTs + maxRequestTimeout, future));
+ requestTemplate.send(key, request, headers);
+ return future;
+ }
+
+ private byte[] uuidToBytes(UUID uuid) {
+ ByteBuffer buf = ByteBuffer.allocate(16);
+ buf.putLong(uuid.getMostSignificantBits());
+ buf.putLong(uuid.getLeastSignificantBits());
+ return buf.array();
+ }
+
+ private static UUID bytesToUuid(byte[] bytes) {
+ ByteBuffer bb = ByteBuffer.wrap(bytes);
+ long firstLong = bb.getLong();
+ long secondLong = bb.getLong();
+ return new UUID(firstLong, secondLong);
+ }
+
+ private byte[] stringToBytes(String string) {
+ return string.getBytes(StandardCharsets.UTF_8);
+ }
+
+ private static class ResponseMetaData {
+ private final long expTime;
+ private final SettableFuture future;
+
+ ResponseMetaData(long ts, SettableFuture future) {
+ this.expTime = ts;
+ this.future = future;
+ }
+ }
+
+}
diff --git a/common/queue/src/main/java/org/thingsboard/server/kafka/TbKafkaSettings.java b/common/queue/src/main/java/org/thingsboard/server/kafka/TbKafkaSettings.java
new file mode 100644
index 0000000000..757902f174
--- /dev/null
+++ b/common/queue/src/main/java/org/thingsboard/server/kafka/TbKafkaSettings.java
@@ -0,0 +1,73 @@
+/**
+ * Copyright © 2016-2018 The Thingsboard Authors
+ *
+ * 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.
+ */
+package org.thingsboard.server.kafka;
+
+import lombok.extern.slf4j.Slf4j;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.stereotype.Component;
+
+import java.util.List;
+import java.util.Properties;
+
+/**
+ * Created by ashvayka on 25.09.18.
+ */
+@Slf4j
+@ConditionalOnProperty(prefix = "kafka", value = "enabled", havingValue = "true", matchIfMissing = false)
+@Component
+public class TbKafkaSettings {
+
+ public static final String REQUEST_ID_HEADER = "requestId";
+ public static final String RESPONSE_TOPIC_HEADER = "responseTopic";
+
+
+ @Value("${kafka.bootstrap.server}")
+ private String servers;
+
+ @Value("${kafka.acks}")
+ private String acks;
+
+ @Value("${kafka.retries}")
+ private int retries;
+
+ @Value("${kafka.batch.size}")
+ private long batchSize;
+
+ @Value("${kafka.linger.ms}")
+ private long lingerMs;
+
+ @Value("${kafka.buffer.memory}")
+ private long bufferMemory;
+
+ @Value("${kafka.other:null}")
+ private List other;
+
+ public Properties toProps() {
+ Properties props = new Properties();
+ props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, servers);
+ props.put(ProducerConfig.ACKS_CONFIG, acks);
+ props.put(ProducerConfig.RETRIES_CONFIG, retries);
+ props.put(ProducerConfig.BATCH_SIZE_CONFIG, batchSize);
+ props.put(ProducerConfig.LINGER_MS_CONFIG, lingerMs);
+ props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, bufferMemory);
+ if(other != null){
+ other.forEach(kv -> props.put(kv.getKey(), kv.getValue()));
+ }
+ return props;
+ }
+}
diff --git a/common/queue/src/main/java/org/thingsboard/server/kafka/TbRuleEngineEmulator.java b/common/queue/src/main/java/org/thingsboard/server/kafka/TbRuleEngineEmulator.java
new file mode 100644
index 0000000000..ecac2e6796
--- /dev/null
+++ b/common/queue/src/main/java/org/thingsboard/server/kafka/TbRuleEngineEmulator.java
@@ -0,0 +1,114 @@
+/**
+ * Copyright © 2016-2018 The Thingsboard Authors
+ *
+ * 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.
+ */
+package org.thingsboard.server.kafka;
+
+import lombok.extern.slf4j.Slf4j;
+import org.apache.kafka.clients.admin.CreateTopicsResult;
+import org.apache.kafka.clients.admin.NewTopic;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.header.Header;
+import org.apache.kafka.common.header.internals.RecordHeader;
+
+import java.nio.charset.StandardCharsets;
+import java.util.Collections;
+import java.util.List;
+import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.atomic.LongAdder;
+
+/**
+ * Created by ashvayka on 24.09.18.
+ */
+@Slf4j
+public class TbRuleEngineEmulator {
+//
+// public static void main(String[] args) throws InterruptedException, ExecutionException {
+// ConcurrentMap pendingRequestsMap = new ConcurrentHashMap<>();
+//
+// ExecutorService executorService = Executors.newCachedThreadPool();
+//
+// String responseTopic = "server" + Math.abs((int) (5000.0 * Math.random()));
+// try {
+// TBKafkaAdmin admin = new TBKafkaAdmin();
+// CreateTopicsResult result = admin.createTopic(new NewTopic(responseTopic, 1, (short) 1));
+// result.all().get();
+// } catch (Exception e) {
+// log.warn("Failed to create topic: {}", e.getMessage(), e);
+// }
+//
+// List headers = Collections.singletonList(new RecordHeader("responseTopic", responseTopic.getBytes(StandardCharsets.UTF_8)));
+//
+// TBKafkaConsumerTemplate responseConsumer = new TBKafkaConsumerTemplate();
+// TBKafkaProducerTemplate requestProducer = new TBKafkaProducerTemplate();
+//
+// LongAdder requestCounter = new LongAdder();
+// LongAdder responseCounter = new LongAdder();
+//
+// responseConsumer.subscribe(responseTopic);
+// executorService.submit((Runnable) () -> {
+// while (true) {
+// ConsumerRecords responses = responseConsumer.poll(100);
+// responses.forEach(response -> {
+// String expectedResponse = pendingRequestsMap.remove(response.key());
+// if (expectedResponse == null) {
+// log.error("[{}] Invalid request", response.key());
+// } else if (!expectedResponse.equals(response.value())) {
+// log.error("[{}] Invalid response: {} instead of {}", response.key(), response.value(), expectedResponse);
+// }
+// responseCounter.add(1);
+// });
+// }
+// });
+//
+// executorService.submit((Runnable) () -> {
+// int i = 0;
+// while (true) {
+// String requestId = UUID.randomUUID().toString();
+// String expectedResponse = UUID.randomUUID().toString();
+// pendingRequestsMap.put(requestId, expectedResponse);
+// requestProducer.send(new ProducerRecord<>("requests", null, requestId, expectedResponse, headers));
+// requestCounter.add(1);
+// i++;
+// if (i % 10000 == 0) {
+// try {
+// Thread.sleep(500L);
+// } catch (InterruptedException e) {
+// e.printStackTrace();
+// }
+// }
+// }
+// });
+//
+// executorService.submit((Runnable) () -> {
+// while (true) {
+// log.warn("Requests: [{}], Responses: [{}]", requestCounter.longValue(), responseCounter.longValue());
+// try {
+// Thread.sleep(1000L);
+// } catch (InterruptedException e) {
+// e.printStackTrace();
+// }
+// }
+// });
+//
+// Thread.sleep(60000);
+// }
+
+}
diff --git a/common/queue/src/main/resources/logback.xml b/common/queue/src/main/resources/logback.xml
new file mode 100644
index 0000000000..dcfc9301b2
--- /dev/null
+++ b/common/queue/src/main/resources/logback.xml
@@ -0,0 +1,35 @@
+
+
+
+
+
+
+
+ %d{ISO8601} [%thread] %-5level %logger{36} - %msg%n
+
+
+
+
+
+
+
+
+
+
+
\ No newline at end of file
diff --git a/pom.xml b/pom.xml
index aa54e72000..159dad189f 100755
--- a/pom.xml
+++ b/pom.xml
@@ -67,7 +67,6 @@
4.1.22.Final
1.5.0
3.6.5
- 0.9.0.0
2.19.1
3.0.2
2.6.1
@@ -82,6 +81,7 @@
5.0.2
0.1.14
+ 2.0.0
@@ -370,6 +370,11 @@
dao
${project.version}