transaction development

@@ -125,7 +125,7 @@ class DefaultTbContext implements TbContext {
 
 
     @Override
     @Override
     public TbMsg transformMsg(TbMsg origMsg, String type, EntityId originator, TbMsgMetaData metaData, String data) {
     public TbMsg transformMsg(TbMsg origMsg, String type, EntityId originator, TbMsgMetaData metaData, String data) {
-        return new TbMsg(origMsg.getId(), type, originator, metaData.copy(), data, origMsg.getRuleChainId(), origMsg.getRuleNodeId(), mainCtx.getQueuePartitionId());
+        return new TbMsg(origMsg.getId(), type, originator, metaData.copy(), origMsg.getDataType(), data, origMsg.getTransactionData(), origMsg.getRuleChainId(), origMsg.getRuleNodeId(), mainCtx.getQueuePartitionId());
     }
     }
 
 
     @Override
     @Override

@@ -279,6 +279,9 @@ public class TelemetryController extends BaseController {
             deleteFromTs = 0L;
             deleteFromTs = 0L;
             deleteToTs = System.currentTimeMillis();
             deleteToTs = System.currentTimeMillis();
         } else {
         } else {
+            if (startTs == null || endTs == null) {
+                return getImmediateDeferredResult("StartTs and endTs could not be empty when deleteAllDataForKeys equals [false]", HttpStatus.BAD_REQUEST);
+            }
             deleteFromTs = startTs;
             deleteFromTs = startTs;
             deleteToTs = endTs;
             deleteToTs = endTs;
         }
         }

@@ -15,40 +15,92 @@
  */
  */
 package org.thingsboard.server.service.transaction;
 package org.thingsboard.server.service.transaction;
 
 
-import com.google.common.util.concurrent.FutureCallback;
 import lombok.extern.slf4j.Slf4j;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.stereotype.Service;
 import org.springframework.stereotype.Service;
 import org.thingsboard.rule.engine.api.RuleChainTransactionService;
 import org.thingsboard.rule.engine.api.RuleChainTransactionService;
+import org.thingsboard.rule.engine.api.TbContext;
 import org.thingsboard.server.common.data.id.EntityId;
 import org.thingsboard.server.common.data.id.EntityId;
 import org.thingsboard.server.common.msg.TbMsg;
 import org.thingsboard.server.common.msg.TbMsg;
 
 
-import java.util.Queue;
+import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+import java.util.function.Consumer;
 
 
 @Service
 @Service
 @Slf4j
 @Slf4j
 public class BaseRuleChainTransactionService implements RuleChainTransactionService {
 public class BaseRuleChainTransactionService implements RuleChainTransactionService {
 
 
     @Value("${actors.rule.transaction.queue_size}")
     @Value("${actors.rule.transaction.queue_size}")
-    private int queueSize;
+    private int finalQueueSize;
 
 
-    private final ConcurrentMap<EntityId, Queue<TbMsg>> transactionMap = new ConcurrentHashMap<>();
+    private final Lock transactionLock = new ReentrantLock();
+    private final ConcurrentMap<EntityId, BlockingQueue<TbTransactionTask>> transactionMap = new ConcurrentHashMap<>();
 
 
-    @Override
-    public void beginTransaction(EntityId entityId, FutureCallback<Void> callback) {
-
-
-        transactionMap.computeIfAbsent(entityId, id -> new LinkedBlockingQueue<>(queueSize));
+    //TODO: add delete on timeout from queue -> onFailure accept
 
 
-        log.info("[{}]", queueSize);
+    @Override
+    public void beginTransaction(TbContext ctx, TbMsg msg, Consumer<TbMsg> onStart, Consumer<TbMsg> onEnd, Consumer<Throwable> onFailure) {
+        BlockingQueue<TbTransactionTask> queue = transactionMap.computeIfAbsent(msg.getTransactionData().getOriginatorId(), id ->
+                new LinkedBlockingQueue<>(finalQueueSize));
+        transactionLock.lock();
+        try {
+            TbTransactionTask task = new TbTransactionTask(msg, onStart, onEnd, onFailure);
+            int queueSize = queue.size();
+            if (queueSize >= finalQueueSize) {
+                onFailure.accept(new RuntimeException("Queue has no space!"));
+            } else {
+                addMsgToQueue(queue, task, onFailure);
+                if (queueSize == 0) {
+                    onStart.accept(msg);
+                } else {
+                    log.info("Msg [{}] [{}] is waiting to start transaction!", msg.getId(), msg.getType());
+                }
+            }
+        } finally {
+            transactionLock.unlock();
+        }
+    }
 
 
+    private void addMsgToQueue(BlockingQueue<TbTransactionTask> queue, TbTransactionTask task, Consumer<Throwable> onFailure) {
+        try {
+            queue.add(task);
+            log.info("Added msg to queue, size: [{}]", queue.size());
+        } catch (Exception e) {
+            log.error("Error when trying to add msg [{}] to the queue", task.getMsg(), e);
+            onFailure.accept(e);
+        }
     }
     }
 
 
     @Override
     @Override
-    public void endTransaction() {
+    public boolean endTransaction(TbContext ctx, TbMsg msg, Consumer<Throwable> onFailure) {
+        transactionLock.lock();
+        try {
+            BlockingQueue<TbTransactionTask> queue = transactionMap.get(msg.getTransactionData().getOriginatorId());
+            try {
+                TbTransactionTask currentTask = queue.element();
+                if (currentTask.getMsg().getTransactionData().getTransactionId().equals(msg.getTransactionData().getTransactionId())) {
+                    queue.remove();
+                    log.info("Removed msg from queue, size [{}]", queue.size());
+                    currentTask.getOnEnd().accept(currentTask.getMsg());
 
 
+                    TbTransactionTask nextTask = queue.peek();
+                    if (nextTask != null) {
+                        nextTask.getOnStart().accept(nextTask.getMsg());
+                    }
+                }
+            } catch (Exception e) {
+                log.error("Queue is empty!", queue);
+                onFailure.accept(e);
+                return true;
+            }
+        } finally {
+            transactionLock.unlock();
+        }
+        return false;
     }
     }
 }
 }

+package org.thingsboard.server.service.transaction;
+
+import lombok.Data;
+import org.thingsboard.server.common.msg.TbMsg;
+
+import java.util.function.Consumer;
+
+@Data
+public final class TbTransactionTask {
+
+    private final TbMsg msg;
+    private final Consumer<TbMsg> onStart;
+    private final Consumer<TbMsg> onEnd;
+    private final Consumer<Throwable> onFailure;
+
+}

@@ -41,6 +41,7 @@ public final class TbMsg implements Serializable {
     private final TbMsgMetaData metaData;
     private final TbMsgMetaData metaData;
     private final TbMsgDataType dataType;
     private final TbMsgDataType dataType;
     private final String data;
     private final String data;
+    private final TbMsgTransactionData transactionData;
 
 
     //The following fields are not persisted to DB, because they can always be recovered from the context;
     //The following fields are not persisted to DB, because they can always be recovered from the context;
     private final RuleChainId ruleChainId;
     private final RuleChainId ruleChainId;
@@ -55,11 +56,17 @@ public final class TbMsg implements Serializable {
         this.metaData = metaData;
         this.metaData = metaData;
         this.data = data;
         this.data = data;
         this.dataType = TbMsgDataType.JSON;
         this.dataType = TbMsgDataType.JSON;
+        this.transactionData = null;
         this.ruleChainId = ruleChainId;
         this.ruleChainId = ruleChainId;
         this.ruleNodeId = ruleNodeId;
         this.ruleNodeId = ruleNodeId;
         this.clusterPartition = clusterPartition;
         this.clusterPartition = clusterPartition;
     }
     }
 
 
+    public TbMsg(UUID id, String type, EntityId originator, TbMsgMetaData metaData, TbMsgDataType dataType, String data,
+                 RuleChainId ruleChainId, RuleNodeId ruleNodeId, long clusterPartition) {
+        this(id, type, originator, metaData, dataType, data, null, ruleChainId, ruleNodeId, clusterPartition);
+    }
+
     public static ByteBuffer toBytes(TbMsg msg) {
     public static ByteBuffer toBytes(TbMsg msg) {
         MsgProtos.TbMsgProto.Builder builder = MsgProtos.TbMsgProto.newBuilder();
         MsgProtos.TbMsgProto.Builder builder = MsgProtos.TbMsgProto.newBuilder();
         builder.setId(msg.getId().toString());
         builder.setId(msg.getId().toString());
@@ -82,6 +89,16 @@ public final class TbMsg implements Serializable {
             builder.setMetaData(MsgProtos.TbMsgMetaDataProto.newBuilder().putAllData(msg.getMetaData().getData()).build());
             builder.setMetaData(MsgProtos.TbMsgMetaDataProto.newBuilder().putAllData(msg.getMetaData().getData()).build());
         }
         }
 
 
+        TbMsgTransactionData transactionData = msg.getTransactionData();
+        if (transactionData != null) {
+            MsgProtos.TbMsgTransactionDataProto.Builder transactionBuilder = MsgProtos.TbMsgTransactionDataProto.newBuilder();
+            transactionBuilder.setId(transactionData.getTransactionId().toString());
+            transactionBuilder.setEntityType(transactionData.getOriginatorId().getEntityType().name());
+            transactionBuilder.setEntityIdMSB(transactionData.getOriginatorId().getId().getMostSignificantBits());
+            transactionBuilder.setEntityIdLSB(transactionData.getOriginatorId().getId().getLeastSignificantBits());
+            builder.setTransactionData(transactionBuilder.build());
+        }
+
         builder.setDataType(msg.getDataType().ordinal());
         builder.setDataType(msg.getDataType().ordinal());
         builder.setData(msg.getData());
         builder.setData(msg.getData());
         byte[] bytes = builder.build().toByteArray();
         byte[] bytes = builder.build().toByteArray();
@@ -92,6 +109,9 @@ public final class TbMsg implements Serializable {
         try {
         try {
             MsgProtos.TbMsgProto proto = MsgProtos.TbMsgProto.parseFrom(buffer.array());
             MsgProtos.TbMsgProto proto = MsgProtos.TbMsgProto.parseFrom(buffer.array());
             TbMsgMetaData metaData = new TbMsgMetaData(proto.getMetaData().getDataMap());
             TbMsgMetaData metaData = new TbMsgMetaData(proto.getMetaData().getDataMap());
+            EntityId transactionEntityId = EntityIdFactory.getByTypeAndUuid(proto.getTransactionData().getEntityType(),
+                    new UUID(proto.getTransactionData().getEntityIdMSB(), proto.getTransactionData().getEntityIdLSB()));
+            TbMsgTransactionData transactionData = new TbMsgTransactionData(UUID.fromString(proto.getTransactionData().getId()), transactionEntityId);
             EntityId entityId = EntityIdFactory.getByTypeAndUuid(proto.getEntityType(), new UUID(proto.getEntityIdMSB(), proto.getEntityIdLSB()));
             EntityId entityId = EntityIdFactory.getByTypeAndUuid(proto.getEntityType(), new UUID(proto.getEntityIdMSB(), proto.getEntityIdLSB()));
             RuleChainId ruleChainId = new RuleChainId(new UUID(proto.getRuleChainIdMSB(), proto.getRuleChainIdLSB()));
             RuleChainId ruleChainId = new RuleChainId(new UUID(proto.getRuleChainIdMSB(), proto.getRuleChainIdLSB()));
             RuleNodeId ruleNodeId = null;
             RuleNodeId ruleNodeId = null;
@@ -99,7 +119,7 @@ public final class TbMsg implements Serializable {
                  ruleNodeId = new RuleNodeId(new UUID(proto.getRuleNodeIdMSB(), proto.getRuleNodeIdLSB()));
                  ruleNodeId = new RuleNodeId(new UUID(proto.getRuleNodeIdMSB(), proto.getRuleNodeIdLSB()));
             }
             }
             TbMsgDataType dataType = TbMsgDataType.values()[proto.getDataType()];
             TbMsgDataType dataType = TbMsgDataType.values()[proto.getDataType()];
-            return new TbMsg(UUID.fromString(proto.getId()), proto.getType(), entityId, metaData, dataType, proto.getData(), ruleChainId, ruleNodeId, proto.getClusterPartition());
+            return new TbMsg(UUID.fromString(proto.getId()), proto.getType(), entityId, metaData, dataType, proto.getData(), transactionData, ruleChainId, ruleNodeId, proto.getClusterPartition());
         } catch (InvalidProtocolBufferException e) {
         } catch (InvalidProtocolBufferException e) {
             throw new IllegalStateException("Could not parse protobuf for TbMsg", e);
             throw new IllegalStateException("Could not parse protobuf for TbMsg", e);
         }
         }

+/**
+ * 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.common.msg;
+
+import lombok.Data;
+import org.thingsboard.server.common.data.id.EntityId;
+
+import java.io.Serializable;
+import java.util.UUID;
+
+@Data
+public final class TbMsgTransactionData implements Serializable {
+
+    private final UUID transactionId;
+    private final EntityId originatorId;
+
+}

@@ -23,6 +23,13 @@ message TbMsgMetaDataProto {
     map<string, string> data = 1;
     map<string, string> data = 1;
 }
 }
 
 
+message TbMsgTransactionDataProto {
+    string id = 1;
+    string entityType = 2;
+    int64 entityIdMSB = 3;
+    int64 entityIdLSB = 4;
+}
+
 message TbMsgProto {
 message TbMsgProto {
     string id = 1;
     string id = 1;
     string type = 2;
     string type = 2;
@@ -39,7 +46,9 @@ message TbMsgProto {
 
 
     TbMsgMetaDataProto metaData = 11;
     TbMsgMetaDataProto metaData = 11;
 
 
-    int32 dataType = 12;
-    string data = 13;
+    TbMsgTransactionDataProto transactionData = 12;
+
+    int32 dataType = 13;
+    string data = 14;
 
 
 }
 }

@@ -15,13 +15,15 @@
  */
  */
 package org.thingsboard.rule.engine.api;
 package org.thingsboard.rule.engine.api;
 
 
-import com.google.common.util.concurrent.FutureCallback;
-import org.thingsboard.server.common.data.id.EntityId;
+import org.thingsboard.server.common.msg.TbMsg;
+
+import java.util.concurrent.Callable;
+import java.util.function.Consumer;
 
 
 public interface RuleChainTransactionService {
 public interface RuleChainTransactionService {
 
 
-    void beginTransaction(EntityId entityId, FutureCallback<Void> callback);
+    void beginTransaction(TbContext ctx, TbMsg msg, Consumer<TbMsg> onStart, Consumer<TbMsg> onEnd, Consumer<Throwable> onFailure);
 
 
-    void endTransaction();
+    boolean endTransaction(TbContext ctx, TbMsg msg, Consumer<Throwable> onFailure);
 
 
 }
 }

 /**
 /**
  * Copyright © 2016-2018 The Thingsboard Authors
  * Copyright © 2016-2018 The Thingsboard Authors
- *
+ * <p>
  * Licensed under the Apache License, Version 2.0 (the "License");
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
  * you may not use this file except in compliance with the License.
  * You may obtain a copy of the License at
  * You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
  * Unless required by applicable law or agreed to in writing, software
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
@@ -15,7 +15,6 @@
  */
  */
 package org.thingsboard.rule.engine.transaction;
 package org.thingsboard.rule.engine.transaction;
 
 
-import com.google.common.util.concurrent.FutureCallback;
 import lombok.extern.slf4j.Slf4j;
 import lombok.extern.slf4j.Slf4j;
 import org.thingsboard.rule.engine.api.EmptyNodeConfiguration;
 import org.thingsboard.rule.engine.api.EmptyNodeConfiguration;
 import org.thingsboard.rule.engine.api.RuleNode;
 import org.thingsboard.rule.engine.api.RuleNode;
@@ -26,9 +25,14 @@ import org.thingsboard.rule.engine.api.TbNodeException;
 import org.thingsboard.rule.engine.api.util.TbNodeUtils;
 import org.thingsboard.rule.engine.api.util.TbNodeUtils;
 import org.thingsboard.server.common.data.plugin.ComponentType;
 import org.thingsboard.server.common.data.plugin.ComponentType;
 import org.thingsboard.server.common.msg.TbMsg;
 import org.thingsboard.server.common.msg.TbMsg;
+import org.thingsboard.server.common.msg.TbMsgDataType;
+import org.thingsboard.server.common.msg.TbMsgTransactionData;
 
 
+import java.util.UUID;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutionException;
 
 
+import static org.thingsboard.rule.engine.api.TbRelationTypes.SUCCESS;
+
 @Slf4j
 @Slf4j
 @RuleNode(
 @RuleNode(
         type = ComponentType.ACTION,
         type = ComponentType.ACTION,
@@ -50,9 +54,21 @@ public class TbTransactionBeginNode implements TbNode {
 
 
     @Override
     @Override
     public void onMsg(TbContext ctx, TbMsg msg) throws ExecutionException, InterruptedException, TbNodeException {
     public void onMsg(TbContext ctx, TbMsg msg) throws ExecutionException, InterruptedException, TbNodeException {
-        FutureCallback<Void> callback = null;
+        log.info("Msg in - [{}] [{}]", msg.getId(), msg.getType());
+
+        TbMsgTransactionData transactionData = new TbMsgTransactionData(UUID.randomUUID(), msg.getOriginator());
+
+        TbMsg tbMsg = new TbMsg(msg.getId(), msg.getType(), msg.getOriginator(), msg.getMetaData(), TbMsgDataType.JSON,
+                msg.getData(), transactionData, msg.getRuleChainId(), msg.getRuleNodeId(), msg.getClusterPartition());
 
 
-        ctx.getRuleChainTransactionService().beginTransaction(msg.getOriginator(), callback);
+        ctx.getRuleChainTransactionService().beginTransaction(ctx, tbMsg, onStart -> {
+                    log.info("Transaction starting... [{}] [{}]", tbMsg.getId(), tbMsg.getType());
+                    ctx.tellNext(tbMsg, SUCCESS);
+                }, onEnd -> log.info("Transaction ended successfully... [{}] [{}]", tbMsg.getId(), tbMsg.getType()),
+                throwable -> {
+                    log.error("Transaction failed due to queue size restriction! [{}] [{}]", tbMsg.getId(), tbMsg.getType(), throwable);
+                    ctx.tellFailure(tbMsg, throwable);
+                });
     }
     }
 
 
     @Override
     @Override

 /**
 /**
  * Copyright © 2016-2018 The Thingsboard Authors
  * Copyright © 2016-2018 The Thingsboard Authors
- *
+ * <p>
  * Licensed under the Apache License, Version 2.0 (the "License");
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
  * you may not use this file except in compliance with the License.
  * You may obtain a copy of the License at
  * You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
  * Unless required by applicable law or agreed to in writing, software
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
@@ -17,19 +17,19 @@ package org.thingsboard.rule.engine.transaction;
 
 
 import lombok.extern.slf4j.Slf4j;
 import lombok.extern.slf4j.Slf4j;
 import org.thingsboard.rule.engine.api.EmptyNodeConfiguration;
 import org.thingsboard.rule.engine.api.EmptyNodeConfiguration;
-import org.thingsboard.rule.engine.api.RuleChainTransactionService;
 import org.thingsboard.rule.engine.api.RuleNode;
 import org.thingsboard.rule.engine.api.RuleNode;
 import org.thingsboard.rule.engine.api.TbContext;
 import org.thingsboard.rule.engine.api.TbContext;
 import org.thingsboard.rule.engine.api.TbNode;
 import org.thingsboard.rule.engine.api.TbNode;
 import org.thingsboard.rule.engine.api.TbNodeConfiguration;
 import org.thingsboard.rule.engine.api.TbNodeConfiguration;
 import org.thingsboard.rule.engine.api.TbNodeException;
 import org.thingsboard.rule.engine.api.TbNodeException;
 import org.thingsboard.rule.engine.api.util.TbNodeUtils;
 import org.thingsboard.rule.engine.api.util.TbNodeUtils;
-import org.thingsboard.rule.engine.telemetry.TbMsgTimeseriesNodeConfiguration;
 import org.thingsboard.server.common.data.plugin.ComponentType;
 import org.thingsboard.server.common.data.plugin.ComponentType;
 import org.thingsboard.server.common.msg.TbMsg;
 import org.thingsboard.server.common.msg.TbMsg;
 
 
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutionException;
 
 
+import static org.thingsboard.rule.engine.api.TbRelationTypes.SUCCESS;
+
 @Slf4j
 @Slf4j
 @RuleNode(
 @RuleNode(
         type = ComponentType.ACTION,
         type = ComponentType.ACTION,
@@ -51,7 +51,11 @@ public class TbTransactionEndNode implements TbNode {
 
 
     @Override
     @Override
     public void onMsg(TbContext ctx, TbMsg msg) throws ExecutionException, InterruptedException, TbNodeException {
     public void onMsg(TbContext ctx, TbMsg msg) throws ExecutionException, InterruptedException, TbNodeException {
-
+        boolean isFailed = ctx.getRuleChainTransactionService().endTransaction(ctx, msg, throwable -> ctx.tellFailure(msg, throwable));
+        if (!isFailed) {
+            ctx.tellNext(msg, SUCCESS);
+        }
+        log.info("Msg out - [{}] [{}]", msg.getId(), msg.getType());
     }
     }
 
 
     @Override
     @Override