Refactoring of Transport Interfaces

common/queue/pom.xml

@@ -94,6 +94,10 @@
             <artifactId>mockito-all</artifactId>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>com.google.protobuf</groupId>
+            <artifactId>protobuf-java</artifactId>
+        </dependency>
     </dependencies>
 
 </project>

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

@@ -4,6 +4,6 @@ import com.google.common.util.concurrent.ListenableFuture;
 
 public interface TbQueueRequestTemplate<Request extends TbQueueMsg, Response extends TbQueueMsg> {
 
-    ListenableFuture<Response> send(String key, Request request);
+    ListenableFuture<Response> send(Request request);
 
 }

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

@@ -0,0 +1,21 @@
+package org.thingsboard.server.common;
+
+import org.thingsboard.server.TbQueueMsgHeaders;
+
+import java.util.HashMap;
+import java.util.Map;
+
+public class DefaultTbQueueMsgHeaders implements TbQueueMsgHeaders {
+
+    private final Map<String, byte[]> data = new HashMap<>();
+
+    @Override
+    public byte[] put(String key, byte[] value) {
+        return data.put(key, value);
+    }
+
+    @Override
+    public byte[] get(String key) {
+        return data.get(key);
+    }
+}

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

@@ -133,7 +133,7 @@ public class DefaultTbQueueRequestTemplate<Request extends TbQueueMsg, Response
     }
 
     @Override
-    public ListenableFuture<Response> send(String key, Request request) {
+    public ListenableFuture<Response> send(Request request) {
         if (tickSize > maxPendingRequests) {
             return Futures.immediateFailedFuture(new RuntimeException("Pending request map is full!"));
         }
@@ -143,7 +143,7 @@ public class DefaultTbQueueRequestTemplate<Request extends TbQueueMsg, Response
         SettableFuture<Response> future = SettableFuture.create();
         ResponseMetaData<Response> responseMetaData = new ResponseMetaData<>(tickTs + maxRequestTimeout, future);
         pendingRequests.putIfAbsent(requestId, responseMetaData);
-        log.trace("[{}] Sending request, key [{}], expTime [{}]", requestId, key, responseMetaData.expTime);
+        log.trace("[{}] Sending request, key [{}], expTime [{}]", requestId, request.getKey(), responseMetaData.expTime);
         requestTemplate.send(request, new TbQueueCallback() {
             @Override
             public void onSuccess(TbQueueMsgMetadata metadata) {

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

@@ -0,0 +1,40 @@
+package org.thingsboard.server.common;
+
+import lombok.Data;
+import org.thingsboard.server.TbQueueMsg;
+import org.thingsboard.server.TbQueueMsgHeaders;
+
+import java.util.UUID;
+
+@Data
+public class TbProtoQueueMsg<T extends com.google.protobuf.GeneratedMessageV3> implements TbQueueMsg {
+
+    private final UUID key;
+    private final T value;
+    private final DefaultTbQueueMsgHeaders headers;
+
+    public TbProtoQueueMsg(UUID key, T value) {
+        this(key, value, new DefaultTbQueueMsgHeaders());
+    }
+
+    public TbProtoQueueMsg(UUID key, T value, DefaultTbQueueMsgHeaders headers) {
+        this.key = key;
+        this.value = value;
+        this.headers = headers;
+    }
+
+    @Override
+    public UUID getKey() {
+        return key;
+    }
+
+    @Override
+    public TbQueueMsgHeaders getHeaders() {
+        return headers;
+    }
+
+    @Override
+    public byte[] getData() {
+        return value.toByteArray();
+    }
+}

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

@@ -0,0 +1,31 @@
+package org.thingsboard.server.common.transport.queue;
+
+import org.springframework.stereotype.Component;
+import org.thingsboard.server.TbQueueConsumer;
+import org.thingsboard.server.TbQueueProducer;
+import org.thingsboard.server.TbQueueRequestTemplate;
+import org.thingsboard.server.common.TbProtoQueueMsg;
+import org.thingsboard.server.gen.transport.TransportProtos;
+
+@Component
+public class KafkaTransportQueueProvider implements TransportQueueProvider {
+    @Override
+    public TbQueueRequestTemplate<TbProtoQueueMsg<TransportProtos.TransportApiRequestMsg>, TbProtoQueueMsg<TransportProtos.TransportApiResponseMsg>> getTransportApiRequestTemplate() {
+        return null;
+    }
+
+    @Override
+    public TbQueueProducer<TbProtoQueueMsg<TransportProtos.ToRuleEngineMsg>> getRuleEngineMsgProducer() {
+        return null;
+    }
+
+    @Override
+    public TbQueueProducer<TbProtoQueueMsg<TransportProtos.ToRuleEngineMsg>> getTbCoreMsgProducer() {
+        return null;
+    }
+
+    @Override
+    public TbQueueConsumer<TbProtoQueueMsg<TransportProtos.ToTransportMsg>> getTransportNotificationsConsumer() {
+        return null;
+    }
+}

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

@@ -1,31 +0,0 @@
-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

@@ -1,40 +0,0 @@
-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

@@ -1,19 +1,22 @@
 package org.thingsboard.server.common.transport.queue;
 
 import org.thingsboard.server.TbQueueConsumer;
-import org.thingsboard.server.TbQueueMsg;
 import org.thingsboard.server.TbQueueProducer;
+import org.thingsboard.server.TbQueueRequestTemplate;
+import org.thingsboard.server.common.TbProtoQueueMsg;
+import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg;
+import org.thingsboard.server.gen.transport.TransportProtos.ToTransportMsg;
+import org.thingsboard.server.gen.transport.TransportProtos.TransportApiRequestMsg;
+import org.thingsboard.server.gen.transport.TransportProtos.TransportApiResponseMsg;
 
 public interface TransportQueueProvider {
 
-    TbQueueProducer<TransportApiCallRequest> getTransportApiCallRequestsProducer();
+    TbQueueRequestTemplate<TbProtoQueueMsg<TransportApiRequestMsg>, TbProtoQueueMsg<TransportApiResponseMsg>> getTransportApiRequestTemplate();
 
-    TbQueueConsumer<TbQueueMsg> getTransportApiCallResponsesConsumer();
+    TbQueueProducer<TbProtoQueueMsg<ToRuleEngineMsg>> getRuleEngineMsgProducer();
 
-    TbQueueProducer<TbQueueMsg> getRuleEngineMsgProducer();
+    TbQueueProducer<TbProtoQueueMsg<ToRuleEngineMsg>> getTbCoreMsgProducer();
 
-    TbQueueProducer<TbQueueMsg> getTbCoreMsgProducer();
+    TbQueueConsumer<TbProtoQueueMsg<ToTransportMsg>> getTransportNotificationsConsumer();
-
-    TbQueueConsumer<TbQueueMsg> getTransportNotificationsConsumer();
 
 }

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

@@ -19,6 +19,10 @@ 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.TbQueueConsumer;
+import org.thingsboard.server.TbQueueProducer;
+import org.thingsboard.server.TbQueueRequestTemplate;
+import org.thingsboard.server.common.TbProtoQueueMsg;
 import org.thingsboard.server.common.data.EntityType;
 import org.thingsboard.server.common.data.id.DeviceId;
 import org.thingsboard.server.common.data.id.TenantId;
@@ -29,6 +33,10 @@ 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 org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg;
+import org.thingsboard.server.gen.transport.TransportProtos.ToTransportMsg;
+import org.thingsboard.server.gen.transport.TransportProtos.TransportApiRequestMsg;
+import org.thingsboard.server.gen.transport.TransportProtos.TransportApiResponseMsg;
 
 import java.util.Random;
 import java.util.UUID;
@@ -54,6 +62,15 @@ public abstract class AbstractTransportService implements TransportService {
     @Autowired
     private TransportQueueProvider queueProvider;
 
+
+    protected TbQueueRequestTemplate<TbProtoQueueMsg<TransportApiRequestMsg>, TbProtoQueueMsg<TransportApiResponseMsg>> transportApiRequestTemplate;
+
+    protected TbQueueProducer<TbProtoQueueMsg<ToRuleEngineMsg>> ruleEngineMsgProducer;
+
+    protected TbQueueProducer<TbProtoQueueMsg<ToRuleEngineMsg>> tbCoreMsgProducer;
+
+    protected TbQueueConsumer<TbProtoQueueMsg<ToTransportMsg>> transportNotificationsConsumer;
+
     protected ScheduledExecutorService schedulerExecutor;
     protected ExecutorService transportCallbackExecutor;
 
@@ -296,8 +313,8 @@ public abstract class AbstractTransportService implements TransportService {
         return new UUID(sessionInfo.getSessionIdMSB(), sessionInfo.getSessionIdLSB());
     }
 
-    protected String getRoutingKey(TransportProtos.SessionInfoProto sessionInfo) {
+    protected UUID getRoutingKey(TransportProtos.SessionInfoProto sessionInfo) {
-        return new UUID(sessionInfo.getDeviceIdMSB(), sessionInfo.getDeviceIdLSB()).toString();
+        return new UUID(sessionInfo.getDeviceIdMSB(), sessionInfo.getDeviceIdLSB());
     }
 
     public void init() {
@@ -309,6 +326,10 @@ public abstract class AbstractTransportService implements TransportService {
         this.schedulerExecutor = Executors.newSingleThreadScheduledExecutor(ThingsBoardThreadFactory.forName("transport-scheduler"));
         this.transportCallbackExecutor = Executors.newWorkStealingPool(20);
         this.schedulerExecutor.scheduleAtFixedRate(this::checkInactivityAndReportActivity, new Random().nextInt((int) sessionReportTimeout), sessionReportTimeout, TimeUnit.MILLISECONDS);
+        transportApiRequestTemplate = queueProvider.getTransportApiRequestTemplate();
+        ruleEngineMsgProducer = queueProvider.getRuleEngineMsgProducer();
+        tbCoreMsgProducer = queueProvider.getTbCoreMsgProducer();
+        transportNotificationsConsumer = queueProvider.getTransportNotificationsConsumer();
     }
 
     public void destroy() {

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

@@ -16,16 +16,14 @@
 package org.thingsboard.server.common.transport.service;
 
 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.Callback;
-import org.apache.kafka.clients.producer.RecordMetadata;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
 import org.springframework.stereotype.Service;
 import org.thingsboard.common.util.ThingsBoardThreadFactory;
+import org.thingsboard.server.TbQueueCallback;
+import org.thingsboard.server.TbQueueMsgMetadata;
+import org.thingsboard.server.common.TbProtoQueueMsg;
 import org.thingsboard.server.common.transport.TransportServiceCallback;
 import org.thingsboard.server.gen.transport.TransportProtos.ClaimDeviceMsg;
 import org.thingsboard.server.gen.transport.TransportProtos.GetAttributeRequestMsg;
@@ -43,22 +41,18 @@ import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg;
 import org.thingsboard.server.gen.transport.TransportProtos.ToServerRpcRequestMsg;
 import org.thingsboard.server.gen.transport.TransportProtos.ToTransportMsg;
 import org.thingsboard.server.gen.transport.TransportProtos.TransportApiRequestMsg;
-import org.thingsboard.server.gen.transport.TransportProtos.TransportApiResponseMsg;
 import org.thingsboard.server.gen.transport.TransportProtos.TransportToDeviceActorMsg;
 import org.thingsboard.server.gen.transport.TransportProtos.ValidateDeviceCredentialsResponseMsg;
 import org.thingsboard.server.gen.transport.TransportProtos.ValidateDeviceTokenRequestMsg;
 import org.thingsboard.server.gen.transport.TransportProtos.ValidateDeviceX509CertRequestMsg;
 import org.thingsboard.server.kafka.AsyncCallbackTemplate;
-import org.thingsboard.server.kafka.TBKafkaAdmin;
-import org.thingsboard.server.kafka.TBKafkaConsumerTemplate;
-import org.thingsboard.server.kafka.TBKafkaProducerTemplate;
-import org.thingsboard.server.kafka.TbKafkaRequestTemplate;
 import org.thingsboard.server.kafka.TbKafkaSettings;
 import org.thingsboard.server.kafka.TbNodeIdProvider;
 
 import javax.annotation.PostConstruct;
 import javax.annotation.PreDestroy;
-import java.time.Duration;
+import java.util.List;
+import java.util.UUID;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 
@@ -97,10 +91,6 @@ public class RemoteTransportService extends AbstractTransportService {
     @Autowired
     private TbNodeIdProvider nodeIdProvider;
 
-    private TbKafkaRequestTemplate<TransportApiRequestMsg, TransportApiResponseMsg> transportApiTemplate;
-    private TBKafkaProducerTemplate<ToRuleEngineMsg> ruleEngineProducer;
-    private TBKafkaConsumerTemplate<ToTransportMsg> mainConsumer;
-
     private ExecutorService mainConsumerExecutor = Executors.newSingleThreadExecutor(ThingsBoardThreadFactory.forName("remote-transport-consumer"));
 
     private volatile boolean stopped = false;
@@ -109,67 +99,67 @@ public class RemoteTransportService extends AbstractTransportService {
     public void init() {
         super.init();
 
-        TBKafkaProducerTemplate.TBKafkaProducerTemplateBuilder<TransportApiRequestMsg> requestBuilder = TBKafkaProducerTemplate.builder();
+//        TBKafkaProducerTemplate.TBKafkaProducerTemplateBuilder<TransportApiRequestMsg> requestBuilder = TBKafkaProducerTemplate.builder();
-        requestBuilder.settings(kafkaSettings);
+//        requestBuilder.settings(kafkaSettings);
-        requestBuilder.clientId("producer-transport-api-request-" + nodeIdProvider.getNodeId());
+//        requestBuilder.clientId("producer-transport-api-request-" + nodeIdProvider.getNodeId());
-        requestBuilder.defaultTopic(transportApiRequestsTopic);
+//        requestBuilder.defaultTopic(transportApiRequestsTopic);
-        requestBuilder.encoder(new TransportApiRequestEncoder());
+//        requestBuilder.encoder(new TransportApiRequestEncoder());
-
+//
-        TBKafkaConsumerTemplate.TBKafkaConsumerTemplateBuilder<TransportApiResponseMsg> responseBuilder = TBKafkaConsumerTemplate.builder();
+//        TBKafkaConsumerTemplate.TBKafkaConsumerTemplateBuilder<TransportApiResponseMsg> responseBuilder = TBKafkaConsumerTemplate.builder();
-        responseBuilder.settings(kafkaSettings);
+//        responseBuilder.settings(kafkaSettings);
-        responseBuilder.topic(transportApiResponsesTopic + "." + nodeIdProvider.getNodeId());
+//        responseBuilder.topic(transportApiResponsesTopic + "." + nodeIdProvider.getNodeId());
-        responseBuilder.clientId("transport-api-client-" + nodeIdProvider.getNodeId());
+//        responseBuilder.clientId("transport-api-client-" + nodeIdProvider.getNodeId());
-        responseBuilder.groupId("transport-api-client");
+//        responseBuilder.groupId("transport-api-client");
-        responseBuilder.autoCommit(true);
+//        responseBuilder.autoCommit(true);
-        responseBuilder.autoCommitIntervalMs(autoCommitInterval);
+//        responseBuilder.autoCommitIntervalMs(autoCommitInterval);
-        responseBuilder.decoder(new TransportApiResponseDecoder());
+//        responseBuilder.decoder(new TransportApiResponseDecoder());
-
+//
-        TbKafkaRequestTemplate.TbKafkaRequestTemplateBuilder
+//        TbKafkaRequestTemplate.TbKafkaRequestTemplateBuilder
-                <TransportApiRequestMsg, TransportApiResponseMsg> builder = TbKafkaRequestTemplate.builder();
+//                <TransportApiRequestMsg, TransportApiResponseMsg> builder = TbKafkaRequestTemplate.builder();
-        builder.requestTemplate(requestBuilder.build());
+//        builder.requestTemplate(requestBuilder.build());
-        builder.responseTemplate(responseBuilder.build());
+//        builder.responseTemplate(responseBuilder.build());
-        builder.maxPendingRequests(maxPendingRequests);
+//        builder.maxPendingRequests(maxPendingRequests);
-        builder.maxRequestTimeout(maxRequestsTimeout);
+//        builder.maxRequestTimeout(maxRequestsTimeout);
-        builder.pollInterval(responsePollDuration);
+//        builder.pollInterval(responsePollDuration);
-        transportApiTemplate = builder.build();
+//        transportApiTemplate = builder.build();
-        transportApiTemplate.init();
+//        transportApiTemplate.init();
-
+//
-        TBKafkaProducerTemplate.TBKafkaProducerTemplateBuilder<ToRuleEngineMsg> ruleEngineProducerBuilder = TBKafkaProducerTemplate.builder();
+//        TBKafkaProducerTemplate.TBKafkaProducerTemplateBuilder<ToRuleEngineMsg> ruleEngineProducerBuilder = TBKafkaProducerTemplate.builder();
-        ruleEngineProducerBuilder.settings(kafkaSettings);
+//        ruleEngineProducerBuilder.settings(kafkaSettings);
-        ruleEngineProducerBuilder.clientId("producer-rule-engine-request-" + nodeIdProvider.getNodeId());
+//        ruleEngineProducerBuilder.clientId("producer-rule-engine-request-" + nodeIdProvider.getNodeId());
-        ruleEngineProducerBuilder.defaultTopic(ruleEngineTopic);
+//        ruleEngineProducerBuilder.defaultTopic(ruleEngineTopic);
-        ruleEngineProducerBuilder.encoder(new ToRuleEngineMsgEncoder());
+//        ruleEngineProducerBuilder.encoder(new ToRuleEngineMsgEncoder());
-        ruleEngineProducer = ruleEngineProducerBuilder.build();
+//        ruleEngineProducer = ruleEngineProducerBuilder.build();
-        ruleEngineProducer.init();
+//        ruleEngineProducer.init();
-
+//
-        String notificationsTopicName = notificationsTopic + "." + nodeIdProvider.getNodeId();
+//        String notificationsTopicName = notificationsTopic + "." + nodeIdProvider.getNodeId();
-
+//
-        try {
+//        try {
-            TBKafkaAdmin admin = new TBKafkaAdmin(kafkaSettings);
+//            TBKafkaAdmin admin = new TBKafkaAdmin(kafkaSettings);
-            CreateTopicsResult result = admin.createTopic(new NewTopic(notificationsTopicName, 1, (short) 1));
+//            CreateTopicsResult result = admin.createTopic(new NewTopic(notificationsTopicName, 1, (short) 1));
-            result.all().get();
+//            result.all().get();
-        } catch (Exception e) {
+//        } catch (Exception e) {
-            log.trace("Failed to create topic: {}", e.getMessage(), e);
+//            log.trace("Failed to create topic: {}", e.getMessage(), e);
-        }
+//        }
-
+//
-        TBKafkaConsumerTemplate.TBKafkaConsumerTemplateBuilder<ToTransportMsg> mainConsumerBuilder = TBKafkaConsumerTemplate.builder();
+//        TBKafkaConsumerTemplate.TBKafkaConsumerTemplateBuilder<ToTransportMsg> mainConsumerBuilder = TBKafkaConsumerTemplate.builder();
-        mainConsumerBuilder.settings(kafkaSettings);
+//        mainConsumerBuilder.settings(kafkaSettings);
-        mainConsumerBuilder.topic(notificationsTopicName);
+//        mainConsumerBuilder.topic(notificationsTopicName);
-        mainConsumerBuilder.clientId("transport-" + nodeIdProvider.getNodeId());
+//        mainConsumerBuilder.clientId("transport-" + nodeIdProvider.getNodeId());
-        mainConsumerBuilder.groupId("transport");
+//        mainConsumerBuilder.groupId("transport");
-        mainConsumerBuilder.autoCommit(true);
+//        mainConsumerBuilder.autoCommit(true);
-        mainConsumerBuilder.autoCommitIntervalMs(notificationsAutoCommitInterval);
+//        mainConsumerBuilder.autoCommitIntervalMs(notificationsAutoCommitInterval);
-        mainConsumerBuilder.decoder(new ToTransportMsgResponseDecoder());
+//        mainConsumerBuilder.decoder(new ToTransportMsgResponseDecoder());
-        mainConsumer = mainConsumerBuilder.build();
+//        mainConsumer = mainConsumerBuilder.build();
-        mainConsumer.subscribe();
+//        mainConsumer.subscribe();
 
         mainConsumerExecutor.execute(() -> {
             while (!stopped) {
                 try {
-                    ConsumerRecords<String, byte[]> records = mainConsumer.poll(Duration.ofMillis(notificationsPollDuration));
+                    List<TbProtoQueueMsg<ToTransportMsg>> records = transportNotificationsConsumer.poll(notificationsPollDuration);
                     records.forEach(record -> {
                         try {
-                            ToTransportMsg toTransportMsg = mainConsumer.decode(record);
+                            ToTransportMsg toTransportMsg = record.getValue();
                             if (toTransportMsg.hasToDeviceSessionMsg()) {
                                 processToTransportMsg(toTransportMsg.getToDeviceSessionMsg());
                             }
@@ -193,12 +183,12 @@ public class RemoteTransportService extends AbstractTransportService {
     public void destroy() {
         super.destroy();
         stopped = true;
-        if (transportApiTemplate != null) {
+//        if (transportApiTemplate != null) {
-            transportApiTemplate.stop();
+//            transportApiTemplate.stop();
-        }
+//        }
-        if (mainConsumer != null) {
+//        if (mainConsumer != null) {
-            mainConsumer.unsubscribe();
+//            mainConsumer.unsubscribe();
-        }
+//        }
         if (mainConsumerExecutor != null) {
             mainConsumerExecutor.shutdownNow();
         }
@@ -207,25 +197,25 @@ public class RemoteTransportService extends AbstractTransportService {
     @Override
     public void process(ValidateDeviceTokenRequestMsg msg, TransportServiceCallback<ValidateDeviceCredentialsResponseMsg> callback) {
         log.trace("Processing msg: {}", msg);
-        AsyncCallbackTemplate.withCallback(transportApiTemplate.post(msg.getToken(),
+        TbProtoQueueMsg<TransportApiRequestMsg> protoMsg = new TbProtoQueueMsg<>(UUID.randomUUID(), TransportApiRequestMsg.newBuilder().setValidateTokenRequestMsg(msg).build());
-                TransportApiRequestMsg.newBuilder().setValidateTokenRequestMsg(msg).build()),
+        AsyncCallbackTemplate.withCallback(transportApiRequestTemplate.send(protoMsg),
-                response -> callback.onSuccess(response.getValidateTokenResponseMsg()), callback::onError, transportCallbackExecutor);
+                response -> callback.onSuccess(response.getValue().getValidateTokenResponseMsg()), callback::onError, transportCallbackExecutor);
     }
 
     @Override
     public void process(ValidateDeviceX509CertRequestMsg msg, TransportServiceCallback<ValidateDeviceCredentialsResponseMsg> callback) {
         log.trace("Processing msg: {}", msg);
-        AsyncCallbackTemplate.withCallback(transportApiTemplate.post(msg.getHash(),
+        TbProtoQueueMsg<TransportApiRequestMsg> protoMsg = new TbProtoQueueMsg<>(UUID.randomUUID(), TransportApiRequestMsg.newBuilder().setValidateX509CertRequestMsg(msg).build());
-                TransportApiRequestMsg.newBuilder().setValidateX509CertRequestMsg(msg).build()),
+        AsyncCallbackTemplate.withCallback(transportApiRequestTemplate.send(protoMsg),
-                response -> callback.onSuccess(response.getValidateTokenResponseMsg()), callback::onError, transportCallbackExecutor);
+                response -> callback.onSuccess(response.getValue().getValidateTokenResponseMsg()), callback::onError, transportCallbackExecutor);
     }
 
     @Override
     public void process(GetOrCreateDeviceFromGatewayRequestMsg msg, TransportServiceCallback<GetOrCreateDeviceFromGatewayResponseMsg> callback) {
         log.trace("Processing msg: {}", msg);
-        AsyncCallbackTemplate.withCallback(transportApiTemplate.post(msg.getDeviceName(),
+        TbProtoQueueMsg<TransportApiRequestMsg> protoMsg = new TbProtoQueueMsg<>(UUID.randomUUID(), TransportApiRequestMsg.newBuilder().setGetOrCreateDeviceRequestMsg(msg).build());
-                TransportApiRequestMsg.newBuilder().setGetOrCreateDeviceRequestMsg(msg).build()),
+        AsyncCallbackTemplate.withCallback(transportApiRequestTemplate.send(protoMsg),
-                response -> callback.onSuccess(response.getGetOrCreateDeviceResponseMsg()), callback::onError, transportCallbackExecutor);
+                response -> callback.onSuccess(response.getValue().getGetOrCreateDeviceResponseMsg()), callback::onError, transportCallbackExecutor);
     }
 
     @Override
@@ -322,18 +312,30 @@ public class RemoteTransportService extends AbstractTransportService {
     }
 
     private void send(SessionInfoProto sessionInfo, ToRuleEngineMsg toRuleEngineMsg, TransportServiceCallback<Void> callback) {
-        ruleEngineProducer.send(getRoutingKey(sessionInfo), toRuleEngineMsg, (metadata, exception) -> {
+        ruleEngineMsgProducer.send(new TbProtoQueueMsg<>(getRoutingKey(sessionInfo), toRuleEngineMsg), callback != null ?
-            if (callback != null) {
+                new TransportTbQueueCallback(callback) : null);
-                if (exception == null) {
+
-                    this.transportCallbackExecutor.submit(() -> {
+    }
+
+    private class TransportTbQueueCallback implements TbQueueCallback {
+        private final TransportServiceCallback<Void> callback;
+
+        private TransportTbQueueCallback(TransportServiceCallback<Void> callback) {
+            this.callback = callback;
+        }
+
+        @Override
+        public void onSuccess(TbQueueMsgMetadata metadata) {
+            RemoteTransportService.this.transportCallbackExecutor.submit(() -> {
                 callback.onSuccess(null);
             });
-                } else {
+        }
-                    this.transportCallbackExecutor.submit(() -> {
+
-                        callback.onError(exception);
+        @Override
+        public void onFailure(Throwable t) {
+            RemoteTransportService.this.transportCallbackExecutor.submit(() -> {
+                callback.onError(t);
             });
         }
     }
-        });
-    }
 }