Refactoring of Queue Interfaces

common/queue/src/main/java/org/thingsboard/server/TbQueueCallback.java

@@ -0,0 +1,8 @@
+package org.thingsboard.server;
+
+public interface TbQueueCallback {
+
+    void onSuccess();
+
+    void onFailure(Throwable t);
+}

common/queue/src/main/java/org/thingsboard/server/TbQueueConsumer.java

@@ -0,0 +1,11 @@
+package org.thingsboard.server;
+
+import java.util.List;
+
+public interface TbQueueConsumer<T extends TbQueueMsg> {
+
+    List<TbQueueMsg> poll();
+
+    void commit();
+
+}

common/queue/src/main/java/org/thingsboard/server/TbQueueMsg.java

@@ -0,0 +1,13 @@
+package org.thingsboard.server;
+
+import java.util.Map;
+import java.util.UUID;
+
+public interface TbQueueMsg {
+
+    UUID getKey();
+
+    Map<String, byte[]> getHeaders();
+
+    byte[] getData();
+}

common/queue/src/main/java/org/thingsboard/server/TbQueueMsgMetadata.java

@@ -0,0 +1,4 @@
+package org.thingsboard.server;
+
+public interface TbQueueMsgMetadata {
+}

common/queue/src/main/java/org/thingsboard/server/TbQueueProducer.java

@@ -0,0 +1,9 @@
+package org.thingsboard.server;
+
+import com.google.common.util.concurrent.ListenableFuture;
+
+public interface TbQueueProducer<T extends TbQueueMsg> {
+
+    ListenableFuture<TbQueueMsgMetadata> send(T msg, TbQueueCallback callback);
+
+}

common/queue/src/main/java/org/thingsboard/server/TbQueueRequestTemplate.java

@@ -0,0 +1,9 @@
+package org.thingsboard.server;
+
+import com.google.common.util.concurrent.ListenableFuture;
+
+public interface TbQueueRequestTemplate<Request extends TbQueueMsg, Response extends TbQueueMsg> {
+
+    ListenableFuture<Response> post(String key, Request request);
+
+}

common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/queue/TransportApiCall.java

@@ -0,0 +1,31 @@
+package org.thingsboard.server.common.transport.queue;
+
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+import java.util.UUID;
+
+public abstract class TransportApiCall {
+
+    protected byte[] uuidToBytes(UUID uuid) {
+        ByteBuffer buf = ByteBuffer.allocate(16);
+        buf.putLong(uuid.getMostSignificantBits());
+        buf.putLong(uuid.getLeastSignificantBits());
+        return buf.array();
+    }
+
+    protected static UUID bytesToUuid(byte[] bytes) {
+        ByteBuffer bb = ByteBuffer.wrap(bytes);
+        long firstLong = bb.getLong();
+        long secondLong = bb.getLong();
+        return new UUID(firstLong, secondLong);
+    }
+
+    protected byte[] stringToBytes(String string) {
+        return string.getBytes(StandardCharsets.UTF_8);
+    }
+
+    protected String bytesToString(byte[] data) {
+        return new String(data, StandardCharsets.UTF_8);
+    }
+
+}

common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/queue/TransportApiCallRequest.java

@@ -0,0 +1,40 @@
+package org.thingsboard.server.common.transport.queue;
+
+import org.thingsboard.server.TbQueueMsg;
+import org.thingsboard.server.gen.transport.TransportProtos;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.UUID;
+
+public class TransportApiCallRequest extends TransportApiCall implements TbQueueMsg {
+    public static final String REQUEST_ID_HEADER = "requestId";
+    public static final String RESPONSE_TOPIC_HEADER = "responseTopic";
+
+    private final UUID requestId;
+    private final Map<String, byte[]> headers;
+    private final TransportProtos.TransportApiRequestMsg msg;
+
+    public TransportApiCallRequest(UUID requestId, String responseTopic, TransportProtos.TransportApiRequestMsg msg) {
+        this.requestId = requestId;
+        this.headers = new HashMap<>();
+        this.headers.put(REQUEST_ID_HEADER, uuidToBytes(requestId));
+        this.headers.put(RESPONSE_TOPIC_HEADER, stringToBytes(responseTopic));
+        this.msg = msg;
+    }
+
+    @Override
+    public UUID getKey() {
+        return requestId;
+    }
+
+    @Override
+    public Map<String, byte[]> getHeaders() {
+        return null;
+    }
+
+    @Override
+    public byte[] getData() {
+        return msg.toByteArray();
+    }
+}

common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/queue/TransportQueueProvider.java

@@ -0,0 +1,19 @@
+package org.thingsboard.server.common.transport.queue;
+
+import org.thingsboard.server.TbQueueConsumer;
+import org.thingsboard.server.TbQueueMsg;
+import org.thingsboard.server.TbQueueProducer;
+
+public interface TransportQueueProvider {
+
+    TbQueueProducer<TransportApiCallRequest> getTransportApiCallRequestsProducer();
+
+    TbQueueConsumer<TbQueueMsg> getTransportApiCallResponsesConsumer();
+
+    TbQueueProducer<TbQueueMsg> getRuleEngineMsgProducer();
+
+    TbQueueProducer<TbQueueMsg> getTbCoreMsgProducer();
+
+    TbQueueConsumer<TbQueueMsg> getTransportNotificationsConsumer();
+
+}

common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/AbstractTransportService.java

@@ -16,6 +16,7 @@
 package org.thingsboard.server.common.transport.service;
 
 import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Value;
 import org.thingsboard.common.util.ThingsBoardThreadFactory;
 import org.thingsboard.server.common.data.EntityType;
@@ -26,6 +27,7 @@ import org.thingsboard.server.common.msg.tools.TbRateLimitsException;
 import org.thingsboard.server.common.transport.SessionMsgListener;
 import org.thingsboard.server.common.transport.TransportService;
 import org.thingsboard.server.common.transport.TransportServiceCallback;
+import org.thingsboard.server.common.transport.queue.TransportQueueProvider;
 import org.thingsboard.server.gen.transport.TransportProtos;
 
 import java.util.Random;
@@ -49,6 +51,9 @@ public abstract class AbstractTransportService implements TransportService {
     @Value("${transport.sessions.report_timeout}")
     private long sessionReportTimeout;
 
+    @Autowired
+    private TransportQueueProvider queueProvider;
+
     protected ScheduledExecutorService schedulerExecutor;
     protected ExecutorService transportCallbackExecutor;
 

transport/mqtt/src/main/resources/tb-mqtt-transport.yml

@@ -61,25 +61,24 @@ transport:
     type_cast_enabled: "${JSON_TYPE_CAST_ENABLED:true}"
     # Maximum allowed string value length when processing Telemetry/Attributes JSON (0 value disables string value length check)
     max_string_value_length: "${JSON_MAX_STRING_VALUE_LENGTH:0}"
-
-kafka:
-  enabled: true
-  bootstrap.servers: "${TB_KAFKA_SERVERS:localhost:9092}"
-  acks: "${TB_KAFKA_ACKS:all}"
-  retries: "${TB_KAFKA_RETRIES:1}"
-  batch.size: "${TB_KAFKA_BATCH_SIZE:16384}"
-  linger.ms: "${TB_KAFKA_LINGER_MS:1}"
-  buffer.memory: "${TB_BUFFER_MEMORY:33554432}"
-  transport_api:
-    requests_topic: "${TB_TRANSPORT_API_REQUEST_TOPIC:tb.transport.api.requests}"
-    responses_topic: "${TB_TRANSPORT_API_RESPONSE_TOPIC:tb.transport.api.responses}"
-    max_pending_requests: "${TB_TRANSPORT_MAX_PENDING_REQUESTS:10000}"
-    max_requests_timeout: "${TB_TRANSPORT_MAX_REQUEST_TIMEOUT:10000}"
-    response_poll_interval: "${TB_TRANSPORT_RESPONSE_POLL_INTERVAL_MS:25}"
-    response_auto_commit_interval: "${TB_TRANSPORT_RESPONSE_AUTO_COMMIT_INTERVAL_MS:100}"
-  rule_engine:
-    topic: "${TB_RULE_ENGINE_TOPIC:tb.rule-engine}"
-  notifications:
-    topic: "${TB_TRANSPORT_NOTIFICATIONS_TOPIC:tb.transport.notifications}"
-    poll_interval: "${TB_TRANSPORT_NOTIFICATIONS_POLL_INTERVAL_MS:25}"
-    auto_commit_interval: "${TB_TRANSPORT_NOTIFICATIONS_AUTO_COMMIT_INTERVAL_MS:100}"
+  queue:
+    type: kafka
+    bootstrap.servers: "${TB_KAFKA_SERVERS:localhost:9092}"
+    acks: "${TB_KAFKA_ACKS:all}"
+    retries: "${TB_KAFKA_RETRIES:1}"
+    batch.size: "${TB_KAFKA_BATCH_SIZE:16384}"
+    linger.ms: "${TB_KAFKA_LINGER_MS:1}"
+    buffer.memory: "${TB_BUFFER_MEMORY:33554432}"
+    transport_api:
+      requests_topic: "${TB_TRANSPORT_API_REQUEST_TOPIC:tb.transport.api.requests}"
+      responses_topic: "${TB_TRANSPORT_API_RESPONSE_TOPIC:tb.transport.api.responses}"
+      max_pending_requests: "${TB_TRANSPORT_MAX_PENDING_REQUESTS:10000}"
+      max_requests_timeout: "${TB_TRANSPORT_MAX_REQUEST_TIMEOUT:10000}"
+      response_poll_interval: "${TB_TRANSPORT_RESPONSE_POLL_INTERVAL_MS:25}"
+      response_auto_commit_interval: "${TB_TRANSPORT_RESPONSE_AUTO_COMMIT_INTERVAL_MS:100}"
+    rule_engine:
+      topic: "${TB_RULE_ENGINE_TOPIC:tb.rule-engine}"
+    notifications:
+      topic: "${TB_TRANSPORT_NOTIFICATIONS_TOPIC:tb.transport.notifications}"
+      poll_interval: "${TB_TRANSPORT_NOTIFICATIONS_POLL_INTERVAL_MS:25}"
+      auto_commit_interval: "${TB_TRANSPORT_NOTIFICATIONS_AUTO_COMMIT_INTERVAL_MS:100}"