Merge branch 'develop/2.0' of https://github.com/thingsboard/thingsboard into develop/2.0

application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java

@@ -235,6 +235,10 @@ public class ActorSystemContext {
     @Getter
     private boolean tenantComponentsInitEnabled;
 
+    @Value("${actors.rule.allow_system_mail_service}")
+    @Getter
+    private boolean allowSystemMailService;
+
     @Getter
     @Setter
     private ActorSystem actorSystem;

application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java

@@ -209,7 +209,11 @@ class DefaultTbContext implements TbContext {
 
     @Override
     public MailService getMailService() {
-        return mainCtx.getMailService();
+        if (mainCtx.isAllowSystemMailService()) {
+            return mainCtx.getMailService();
+        } else {
+            throw new RuntimeException("Access to System Mail Service is forbidden!");
+        }
     }
 
     @Override

application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleChainActorMessageProcessor.java

@@ -183,23 +183,35 @@ public class RuleChainActorMessageProcessor extends ComponentMsgProcessor<RuleCh
 
     void onServiceToRuleEngineMsg(ServiceToRuleEngineMsg envelope) {
         checkActive();
-        putToQueue(enrichWithRuleChainId(envelope.getTbMsg()), msg -> pushMsgToNode(firstNode, msg, ""));
+        if (firstNode != null) {
+            putToQueue(enrichWithRuleChainId(envelope.getTbMsg()), msg -> pushMsgToNode(firstNode, msg, ""));
+        }
     }
 
     void onDeviceActorToRuleEngineMsg(DeviceActorToRuleEngineMsg envelope) {
         checkActive();
-        putToQueue(enrichWithRuleChainId(envelope.getTbMsg()), msg -> {
+        if (firstNode != null) {
-            pushMsgToNode(firstNode, msg, "");
+            putToQueue(enrichWithRuleChainId(envelope.getTbMsg()), msg -> {
-            envelope.getCallbackRef().tell(new RuleEngineQueuePutAckMsg(msg.getId()), self);
+                pushMsgToNode(firstNode, msg, "");
-        });
+                envelope.getCallbackRef().tell(new RuleEngineQueuePutAckMsg(msg.getId()), self);
+            });
+        }
     }
 
     void onRuleChainToRuleChainMsg(RuleChainToRuleChainMsg envelope) {
         checkActive();
         if (envelope.isEnqueue()) {
-            putToQueue(enrichWithRuleChainId(envelope.getMsg()), msg -> pushMsgToNode(firstNode, msg, envelope.getFromRelationType()));
+            if (firstNode != null) {
+                putToQueue(enrichWithRuleChainId(envelope.getMsg()), msg -> pushMsgToNode(firstNode, msg, envelope.getFromRelationType()));
+            }
         } else {
-            pushMsgToNode(firstNode, envelope.getMsg(), envelope.getFromRelationType());
+            if (firstNode != null) {
+                pushMsgToNode(firstNode, envelope.getMsg(), envelope.getFromRelationType());
+            } else {
+                TbMsg msg = envelope.getMsg();
+                EntityId ackId = msg.getRuleNodeId() != null ? msg.getRuleNodeId() : msg.getRuleChainId();
+                queue.ack(tenantId, envelope.getMsg(), ackId.getId(), msg.getClusterPartition());
+            }
         }
     }
 

application/src/main/java/org/thingsboard/server/install/ThingsboardInstallService.java

@@ -82,7 +82,7 @@ public class ThingsboardInstallService {
                         databaseUpgradeService.upgradeDatabase("1.3.1");
 
                     case "1.4.0":
-                        log.info("Upgrading ThingsBoard from version 1.4.0 to 1.5.0 ...");
+                        log.info("Upgrading ThingsBoard from version 1.4.0 to 2.0.0 ...");
 
                         databaseUpgradeService.upgradeDatabase("1.4.0");
 

application/src/main/java/org/thingsboard/server/service/install/CassandraDatabaseUpgradeService.java

@@ -198,7 +198,7 @@ public class CassandraDatabaseUpgradeService implements DatabaseUpgradeService {
             case "1.4.0":
 
                 log.info("Updating schema ...");
-                schemaUpdateFile = Paths.get(installScripts.getDataDir(), "upgrade", "1.5.0", SCHEMA_UPDATE_CQL);
+                schemaUpdateFile = Paths.get(installScripts.getDataDir(), "upgrade", "2.0.0", SCHEMA_UPDATE_CQL);
                 loadCql(schemaUpdateFile);
                 log.info("Schema updated.");
 

application/src/main/java/org/thingsboard/server/service/install/DefaultDataUpdateService.java

@@ -47,7 +47,7 @@ public class DefaultDataUpdateService implements DataUpdateService {
     public void updateData(String fromVersion) throws Exception {
         switch (fromVersion) {
             case "1.4.0":
-                log.info("Updating data from version 1.4.0 to 1.5.0 ...");
+                log.info("Updating data from version 1.4.0 to 2.0.0 ...");
                 tenantsDefaultRuleChainUpdater.updateEntities(null);
                 break;
             default:

application/src/main/java/org/thingsboard/server/service/install/SqlDatabaseUpgradeService.java

@@ -104,7 +104,7 @@ public class SqlDatabaseUpgradeService implements DatabaseUpgradeService {
             case "1.4.0":
                 try (Connection conn = DriverManager.getConnection(dbUrl, dbUserName, dbPassword)) {
                     log.info("Updating schema ...");
-                    schemaUpdateFile = Paths.get(installScripts.getDataDir(), "upgrade", "1.5.0", SCHEMA_UPDATE_SQL);
+                    schemaUpdateFile = Paths.get(installScripts.getDataDir(), "upgrade", "2.0.0", SCHEMA_UPDATE_SQL);
                     String sql = new String(Files.readAllBytes(schemaUpdateFile), Charset.forName("UTF-8"));
                     conn.createStatement().execute(sql); //NOSONAR, ignoring because method used to execute thingsboard database upgrade script
                     log.info("Schema updated.");

application/src/main/java/org/thingsboard/server/service/queue/DefaultMsgQueueService.java

@@ -40,10 +40,10 @@ import java.util.concurrent.atomic.AtomicLong;
 @Slf4j
 public class DefaultMsgQueueService implements MsgQueueService {
 
-    @Value("${rule.queue.max_size}")
+    @Value("${actors.rule.queue.max_size}")
     private long queueMaxSize;
 
-    @Value("${rule.queue.cleanup_period}")
+    @Value("${actors.rule.queue.cleanup_period}")
     private long queueCleanUpPeriod;
 
     @Autowired

application/src/main/resources/thingsboard.yml

@@ -203,6 +203,7 @@ cassandra:
     default_fetch_size: "${CASSANDRA_DEFAULT_FETCH_SIZE:2000}"
     # Specify partitioning size for timestamp key-value storage. Example MINUTES, HOURS, DAYS, MONTHS
     ts_key_value_partitioning: "${TS_KV_PARTITIONING:MONTHS}"
+    ts_key_value_ttl: "${TS_KV_TTL:0}"
     buffer_size: "${CASSANDRA_QUERY_BUFFER_SIZE:200000}"
     concurrent_limit: "${CASSANDRA_QUERY_CONCURRENT_LIMIT:1000}"
     permit_max_wait_time: "${PERMIT_MAX_WAIT_TIME:120000}"
@@ -236,6 +237,8 @@ actors:
     js_thread_pool_size: "${ACTORS_RULE_JS_THREAD_POOL_SIZE:10}"
     # Specify thread pool size for mail sender executor service
     mail_thread_pool_size: "${ACTORS_RULE_MAIL_THREAD_POOL_SIZE:10}"
+    # Whether to allow usage of system mail service for rules
+    allow_system_mail_service: "${ACTORS_RULE_ALLOW_SYSTEM_MAIL_SERVICE:true}"
     # Specify thread pool size for external call service
     external_call_thread_pool_size: "${ACTORS_RULE_EXTERNAL_CALL_THREAD_POOL_SIZE:10}"
     js_sandbox:
@@ -253,6 +256,13 @@ actors:
     node:
       # Errors for particular actor are persisted once per specified amount of milliseconds
       error_persist_frequency: "${ACTORS_RULE_NODE_ERROR_FREQUENCY:3000}"
+    queue:
+      # Message queue type
+      type: "${ACTORS_RULE_QUEUE_TYPE:memory}"
+      # Message queue maximum size (per tenant)
+      max_size: "${ACTORS_RULE_QUEUE_MAX_SIZE:100}"
+      # Message queue cleanup period in seconds
+      cleanup_period: "${ACTORS_RULE_QUEUE_CLEANUP_PERIOD:3600}"
   statistics:
     # Enable/disable actor statistics
     enabled: "${ACTORS_STATISTICS_ENABLED:true}"
@@ -333,16 +343,6 @@ spring:
     username: "${SPRING_DATASOURCE_USERNAME:sa}"
     password: "${SPRING_DATASOURCE_PASSWORD:}"
 
-rule:
-  queue:
-    #Message queue type (memory or db)
-    type: "${RULE_QUEUE_TYPE:memory}"
-    #Message queue maximum size (per tenant)
-    max_size: "${RULE_QUEUE_MAX_SIZE:100}"
-    #Message queue cleanup period in seconds
-    cleanup_period: "${RULE_QUEUE_CLEANUP_PERIOD:3600}"
-
-
 # PostgreSQL DAO Configuration
 #spring:
 #  data:

application/src/test/java/org/thingsboard/server/controller/AbstractRuleEngineControllerTest.java

@@ -32,6 +32,7 @@ import org.thingsboard.server.dao.rule.RuleChainService;
 import org.thingsboard.server.service.queue.MsgQueueService;
 
 import java.io.IOException;
+import java.util.function.Predicate;
 
 /**
  * Created by ashvayka on 20.03.18.
@@ -75,4 +76,9 @@ public class AbstractRuleEngineControllerTest extends AbstractControllerTest {
             throw new RuntimeException(e);
         }
     }
+
+    protected Predicate<Event> filterByCustomEvent() {
+        return event -> event.getBody().get("msgType").textValue().equals("CUSTOM");
+    }
+
 }

application/src/test/java/org/thingsboard/server/rules/flow/AbstractRuleEngineFlowIntegrationTest.java

@@ -47,6 +47,8 @@ import java.io.IOException;
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
+import java.util.function.Predicate;
+import java.util.stream.Collectors;
 
 import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status;
 
@@ -157,15 +159,15 @@ public abstract class AbstractRuleEngineFlowIntegrationTest extends AbstractRule
 
         Thread.sleep(3000);
 
-        TimePageData<Event> events = getDebugEvents(savedTenant.getId(), ruleChain.getFirstRuleNodeId(), 1000);
+        TimePageData<Event> eventsPage = getDebugEvents(savedTenant.getId(), ruleChain.getFirstRuleNodeId(), 1000);
+        List<Event> events = eventsPage.getData().stream().filter(filterByCustomEvent()).collect(Collectors.toList());
+        Assert.assertEquals(2, events.size());
 
-        Assert.assertEquals(2, events.getData().size());
+        Event inEvent = events.stream().filter(e -> e.getBody().get("type").asText().equals(DataConstants.IN)).findFirst().get();
-
-        Event inEvent = events.getData().stream().filter(e -> e.getBody().get("type").asText().equals(DataConstants.IN)).findFirst().get();
         Assert.assertEquals(ruleChain.getFirstRuleNodeId(), inEvent.getEntityId());
         Assert.assertEquals(device.getId().getId().toString(), inEvent.getBody().get("entityId").asText());
 
-        Event outEvent = events.getData().stream().filter(e -> e.getBody().get("type").asText().equals(DataConstants.OUT)).findFirst().get();
+        Event outEvent = events.stream().filter(e -> e.getBody().get("type").asText().equals(DataConstants.OUT)).findFirst().get();
         Assert.assertEquals(ruleChain.getFirstRuleNodeId(), outEvent.getEntityId());
         Assert.assertEquals(device.getId().getId().toString(), outEvent.getBody().get("entityId").asText());
 
@@ -174,15 +176,16 @@ public abstract class AbstractRuleEngineFlowIntegrationTest extends AbstractRule
         RuleChain finalRuleChain = ruleChain;
         RuleNode lastRuleNode = metaData.getNodes().stream().filter(node -> !node.getId().equals(finalRuleChain.getFirstRuleNodeId())).findFirst().get();
 
-        events = getDebugEvents(savedTenant.getId(), lastRuleNode.getId(), 1000);
+        eventsPage = getDebugEvents(savedTenant.getId(), lastRuleNode.getId(), 1000);
+        events = eventsPage.getData().stream().filter(filterByCustomEvent()).collect(Collectors.toList());
 
-        Assert.assertEquals(2, events.getData().size());
+        Assert.assertEquals(2, events.size());
 
-        inEvent = events.getData().stream().filter(e -> e.getBody().get("type").asText().equals(DataConstants.IN)).findFirst().get();
+        inEvent = events.stream().filter(e -> e.getBody().get("type").asText().equals(DataConstants.IN)).findFirst().get();
         Assert.assertEquals(lastRuleNode.getId(), inEvent.getEntityId());
         Assert.assertEquals(device.getId().getId().toString(), inEvent.getBody().get("entityId").asText());
 
-        outEvent = events.getData().stream().filter(e -> e.getBody().get("type").asText().equals(DataConstants.OUT)).findFirst().get();
+        outEvent = events.stream().filter(e -> e.getBody().get("type").asText().equals(DataConstants.OUT)).findFirst().get();
         Assert.assertEquals(lastRuleNode.getId(), outEvent.getEntityId());
         Assert.assertEquals(device.getId().getId().toString(), outEvent.getBody().get("entityId").asText());
 
@@ -274,15 +277,16 @@ public abstract class AbstractRuleEngineFlowIntegrationTest extends AbstractRule
 
         Thread.sleep(3000);
 
-        TimePageData<Event> events = getDebugEvents(savedTenant.getId(), rootRuleChain.getFirstRuleNodeId(), 1000);
+        TimePageData<Event> eventsPage = getDebugEvents(savedTenant.getId(), rootRuleChain.getFirstRuleNodeId(), 1000);
+        List<Event> events = eventsPage.getData().stream().filter(filterByCustomEvent()).collect(Collectors.toList());
 
-        Assert.assertEquals(2, events.getData().size());
+        Assert.assertEquals(2, events.size());
 
-        Event inEvent = events.getData().stream().filter(e -> e.getBody().get("type").asText().equals(DataConstants.IN)).findFirst().get();
+        Event inEvent = events.stream().filter(e -> e.getBody().get("type").asText().equals(DataConstants.IN)).findFirst().get();
         Assert.assertEquals(rootRuleChain.getFirstRuleNodeId(), inEvent.getEntityId());
         Assert.assertEquals(device.getId().getId().toString(), inEvent.getBody().get("entityId").asText());
 
-        Event outEvent = events.getData().stream().filter(e -> e.getBody().get("type").asText().equals(DataConstants.OUT)).findFirst().get();
+        Event outEvent = events.stream().filter(e -> e.getBody().get("type").asText().equals(DataConstants.OUT)).findFirst().get();
         Assert.assertEquals(rootRuleChain.getFirstRuleNodeId(), outEvent.getEntityId());
         Assert.assertEquals(device.getId().getId().toString(), outEvent.getBody().get("entityId").asText());
 
@@ -291,15 +295,17 @@ public abstract class AbstractRuleEngineFlowIntegrationTest extends AbstractRule
         RuleChain finalRuleChain = rootRuleChain;
         RuleNode lastRuleNode = secondaryMetaData.getNodes().stream().filter(node -> !node.getId().equals(finalRuleChain.getFirstRuleNodeId())).findFirst().get();
 
-        events = getDebugEvents(savedTenant.getId(), lastRuleNode.getId(), 1000);
+        eventsPage = getDebugEvents(savedTenant.getId(), lastRuleNode.getId(), 1000);
+        events = eventsPage.getData().stream().filter(filterByCustomEvent()).collect(Collectors.toList());
+
 
-        Assert.assertEquals(2, events.getData().size());
+        Assert.assertEquals(2, events.size());
 
-        inEvent = events.getData().stream().filter(e -> e.getBody().get("type").asText().equals(DataConstants.IN)).findFirst().get();
+        inEvent = events.stream().filter(e -> e.getBody().get("type").asText().equals(DataConstants.IN)).findFirst().get();
         Assert.assertEquals(lastRuleNode.getId(), inEvent.getEntityId());
         Assert.assertEquals(device.getId().getId().toString(), inEvent.getBody().get("entityId").asText());
 
-        outEvent = events.getData().stream().filter(e -> e.getBody().get("type").asText().equals(DataConstants.OUT)).findFirst().get();
+        outEvent = events.stream().filter(e -> e.getBody().get("type").asText().equals(DataConstants.OUT)).findFirst().get();
         Assert.assertEquals(lastRuleNode.getId(), outEvent.getEntityId());
         Assert.assertEquals(device.getId().getId().toString(), outEvent.getBody().get("entityId").asText());
 

application/src/test/java/org/thingsboard/server/rules/lifecycle/AbstractRuleEngineLifecycleIntegrationTest.java

@@ -45,6 +45,8 @@ import org.thingsboard.server.dao.attributes.AttributesService;
 
 import java.io.IOException;
 import java.util.Collections;
+import java.util.List;
+import java.util.stream.Collectors;
 
 import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status;
 
@@ -144,15 +146,16 @@ public abstract class AbstractRuleEngineLifecycleIntegrationTest extends Abstrac
 
         Thread.sleep(3000);
 
-        TimePageData<Event> events = getDebugEvents(savedTenant.getId(), ruleChain.getFirstRuleNodeId(), 1000);
+        TimePageData<Event> eventsPage = getDebugEvents(savedTenant.getId(), ruleChain.getFirstRuleNodeId(), 1000);
+        List<Event> events = eventsPage.getData().stream().filter(filterByCustomEvent()).collect(Collectors.toList());
 
-        Assert.assertEquals(2, events.getData().size());
+        Assert.assertEquals(2, events.size());
 
-        Event inEvent = events.getData().stream().filter(e -> e.getBody().get("type").asText().equals(DataConstants.IN)).findFirst().get();
+        Event inEvent = events.stream().filter(e -> e.getBody().get("type").asText().equals(DataConstants.IN)).findFirst().get();
         Assert.assertEquals(ruleChain.getFirstRuleNodeId(), inEvent.getEntityId());
         Assert.assertEquals(device.getId().getId().toString(), inEvent.getBody().get("entityId").asText());
 
-        Event outEvent = events.getData().stream().filter(e -> e.getBody().get("type").asText().equals(DataConstants.OUT)).findFirst().get();
+        Event outEvent = events.stream().filter(e -> e.getBody().get("type").asText().equals(DataConstants.OUT)).findFirst().get();
         Assert.assertEquals(ruleChain.getFirstRuleNodeId(), outEvent.getEntityId());
         Assert.assertEquals(device.getId().getId().toString(), outEvent.getBody().get("entityId").asText());
 
@@ -212,15 +215,16 @@ public abstract class AbstractRuleEngineLifecycleIntegrationTest extends Abstrac
 
         Thread.sleep(3000);
 
-        TimePageData<Event> events = getDebugEvents(savedTenant.getId(), ruleChain.getFirstRuleNodeId(), 1000);
+        TimePageData<Event> eventsPage = getDebugEvents(savedTenant.getId(), ruleChain.getFirstRuleNodeId(), 1000);
+        List<Event> events = eventsPage.getData().stream().filter(filterByCustomEvent()).collect(Collectors.toList());
 
-        Assert.assertEquals(2, events.getData().size());
+        Assert.assertEquals(2, events.size());
 
-        Event inEvent = events.getData().stream().filter(e -> e.getBody().get("type").asText().equals(DataConstants.IN)).findFirst().get();
+        Event inEvent = events.stream().filter(e -> e.getBody().get("type").asText().equals(DataConstants.IN)).findFirst().get();
         Assert.assertEquals(ruleChain.getFirstRuleNodeId(), inEvent.getEntityId());
         Assert.assertEquals(device.getId().getId().toString(), inEvent.getBody().get("entityId").asText());
 
-        Event outEvent = events.getData().stream().filter(e -> e.getBody().get("type").asText().equals(DataConstants.OUT)).findFirst().get();
+        Event outEvent = events.stream().filter(e -> e.getBody().get("type").asText().equals(DataConstants.OUT)).findFirst().get();
         Assert.assertEquals(ruleChain.getFirstRuleNodeId(), outEvent.getEntityId());
         Assert.assertEquals(device.getId().getId().toString(), outEvent.getBody().get("entityId").asText());
 

dao/src/main/java/org/thingsboard/server/dao/queue/db/MsgAck.java

@@ -1,32 +0,0 @@
-/**
- * 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.dao.queue.db;
-
-import lombok.Data;
-import lombok.EqualsAndHashCode;
-
-import java.util.UUID;
-
-@Data
-@EqualsAndHashCode
-public class MsgAck {
-
-    private final UUID msgId;
-    private final UUID nodeId;
-    private final long clusteredPartition;
-    private final long tsPartition;
-
-}

dao/src/main/java/org/thingsboard/server/dao/queue/db/UnprocessedMsgFilter.java

@@ -1,35 +0,0 @@
-/**
- * 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.dao.queue.db;
-
-import org.springframework.stereotype.Component;
-import org.thingsboard.server.common.msg.TbMsg;
-import org.thingsboard.server.dao.queue.db.MsgAck;
-
-import java.util.Collection;
-import java.util.List;
-import java.util.Set;
-import java.util.UUID;
-import java.util.stream.Collectors;
-
-@Component
-public class UnprocessedMsgFilter {
-
-    public Collection<TbMsg> filter(List<TbMsg> msgs, List<MsgAck> acks) {
-        Set<UUID> processedIds = acks.stream().map(MsgAck::getMsgId).collect(Collectors.toSet());
-        return msgs.stream().filter(i -> !processedIds.contains(i.getId())).collect(Collectors.toList());
-    }
-}

dao/src/main/java/org/thingsboard/server/dao/queue/db/nosql/CassandraMsgQueue.java

@@ -1,87 +0,0 @@
-/**
- * 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.dao.queue.db.nosql;
-
-import com.datastax.driver.core.utils.UUIDs;
-import com.google.common.collect.Lists;
-import com.google.common.util.concurrent.Futures;
-import com.google.common.util.concurrent.ListenableFuture;
-import lombok.extern.slf4j.Slf4j;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
-import org.springframework.stereotype.Component;
-import org.thingsboard.server.common.data.id.TenantId;
-import org.thingsboard.server.common.msg.TbMsg;
-import org.thingsboard.server.dao.queue.MsgQueue;
-import org.thingsboard.server.dao.queue.db.MsgAck;
-import org.thingsboard.server.dao.queue.db.UnprocessedMsgFilter;
-import org.thingsboard.server.dao.queue.db.repository.AckRepository;
-import org.thingsboard.server.dao.queue.db.repository.MsgRepository;
-import org.thingsboard.server.dao.util.NoSqlDao;
-
-import java.util.List;
-import java.util.UUID;
-
-@Component
-@ConditionalOnProperty(prefix = "rule.queue", value = "type", havingValue = "db")
-@Slf4j
-@NoSqlDao
-public class CassandraMsgQueue implements MsgQueue {
-
-    @Autowired
-    private MsgRepository msgRepository;
-    @Autowired
-    private AckRepository ackRepository;
-    @Autowired
-    private UnprocessedMsgFilter unprocessedMsgFilter;
-    @Autowired
-    private QueuePartitioner queuePartitioner;
-
-    @Override
-    public ListenableFuture<Void> put(TenantId tenantId, TbMsg msg, UUID nodeId, long clusterPartition) {
-        long msgTime = getMsgTime(msg);
-        long tsPartition = queuePartitioner.getPartition(msgTime);
-        return msgRepository.save(msg, nodeId, clusterPartition, tsPartition, msgTime);
-    }
-
-    @Override
-    public ListenableFuture<Void> ack(TenantId tenantId, TbMsg msg, UUID nodeId, long clusterPartition) {
-        long tsPartition = queuePartitioner.getPartition(getMsgTime(msg));
-        MsgAck ack = new MsgAck(msg.getId(), nodeId, clusterPartition, tsPartition);
-        return ackRepository.ack(ack);
-    }
-
-    @Override
-    public Iterable<TbMsg> findUnprocessed(TenantId tenantId, UUID nodeId, long clusterPartition) {
-        List<TbMsg> unprocessedMsgs = Lists.newArrayList();
-        for (Long tsPartition : queuePartitioner.findUnprocessedPartitions(nodeId, clusterPartition)) {
-            List<TbMsg> msgs = msgRepository.findMsgs(nodeId, clusterPartition, tsPartition);
-            List<MsgAck> acks = ackRepository.findAcks(nodeId, clusterPartition, tsPartition);
-            unprocessedMsgs.addAll(unprocessedMsgFilter.filter(msgs, acks));
-        }
-        return unprocessedMsgs;
-    }
-
-    @Override
-    public ListenableFuture<Void> cleanUp(TenantId tenantId) {
-        return Futures.immediateFuture(null);
-    }
-
-    private long getMsgTime(TbMsg msg) {
-        return UUIDs.unixTimestamp(msg.getId());
-    }
-
-}

dao/src/main/java/org/thingsboard/server/dao/queue/db/nosql/QueuePartitioner.java

@@ -1,86 +0,0 @@
-/**
- * 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.dao.queue.db.nosql;
-
-import com.google.common.collect.Lists;
-import lombok.extern.slf4j.Slf4j;
-import org.springframework.beans.factory.annotation.Value;
-import org.springframework.stereotype.Component;
-import org.thingsboard.server.dao.queue.db.repository.ProcessedPartitionRepository;
-import org.thingsboard.server.dao.timeseries.TsPartitionDate;
-import org.thingsboard.server.dao.util.NoSqlDao;
-
-import java.time.Clock;
-import java.time.Instant;
-import java.time.LocalDateTime;
-import java.time.ZoneOffset;
-import java.util.List;
-import java.util.Optional;
-import java.util.UUID;
-import java.util.concurrent.TimeUnit;
-
-@Component
-@Slf4j
-@NoSqlDao
-public class QueuePartitioner {
-
-    private final TsPartitionDate tsFormat;
-    private ProcessedPartitionRepository processedPartitionRepository;
-    private Clock clock = Clock.systemUTC();
-
-    public QueuePartitioner(@Value("${cassandra.queue.partitioning}") String partitioning,
-                            ProcessedPartitionRepository processedPartitionRepository) {
-        this.processedPartitionRepository = processedPartitionRepository;
-        Optional<TsPartitionDate> partition = TsPartitionDate.parse(partitioning);
-        if (partition.isPresent()) {
-            tsFormat = partition.get();
-        } else {
-            log.warn("Incorrect configuration of partitioning {}", partitioning);
-            throw new RuntimeException("Failed to parse partitioning property: " + partitioning + "!");
-        }
-    }
-
-    public long getPartition(long ts) {
-        //TODO: use TsPartitionDate.truncateTo?
-        LocalDateTime time = LocalDateTime.ofInstant(Instant.ofEpochMilli(ts), ZoneOffset.UTC);
-        return tsFormat.truncatedTo(time).toInstant(ZoneOffset.UTC).toEpochMilli();
-    }
-
-    public List<Long> findUnprocessedPartitions(UUID nodeId, long clusteredHash) {
-        Optional<Long> lastPartitionOption = processedPartitionRepository.findLastProcessedPartition(nodeId, clusteredHash);
-        long lastPartition = lastPartitionOption.orElse(System.currentTimeMillis() - TimeUnit.DAYS.toMillis(7));
-        List<Long> unprocessedPartitions = Lists.newArrayList();
-
-        LocalDateTime current = LocalDateTime.ofInstant(Instant.ofEpochMilli(lastPartition), ZoneOffset.UTC);
-        LocalDateTime end = LocalDateTime.ofInstant(Instant.now(clock), ZoneOffset.UTC)
-                .plus(1L, tsFormat.getTruncateUnit());
-
-        while (current.isBefore(end)) {
-            current = current.plus(1L, tsFormat.getTruncateUnit());
-            unprocessedPartitions.add(tsFormat.truncatedTo(current).toInstant(ZoneOffset.UTC).toEpochMilli());
-        }
-
-        return unprocessedPartitions;
-    }
-
-    public void setClock(Clock clock) {
-        this.clock = clock;
-    }
-
-    public void checkProcessedPartitions() {
-        //todo-vp: we need to implement this
-    }
-}

dao/src/main/java/org/thingsboard/server/dao/queue/db/nosql/repository/CassandraAckRepository.java

@@ -1,68 +0,0 @@
-/**
- * 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.dao.queue.db.nosql.repository;
-
-import com.datastax.driver.core.BoundStatement;
-import com.datastax.driver.core.PreparedStatement;
-import com.datastax.driver.core.ResultSet;
-import com.datastax.driver.core.ResultSetFuture;
-import com.datastax.driver.core.Row;
-import com.google.common.base.Function;
-import com.google.common.util.concurrent.Futures;
-import com.google.common.util.concurrent.ListenableFuture;
-import org.springframework.beans.factory.annotation.Value;
-import org.springframework.stereotype.Component;
-import org.thingsboard.server.dao.nosql.CassandraAbstractDao;
-import org.thingsboard.server.dao.queue.db.MsgAck;
-import org.thingsboard.server.dao.queue.db.repository.AckRepository;
-import org.thingsboard.server.dao.util.NoSqlDao;
-
-import java.util.ArrayList;
-import java.util.List;
-import java.util.UUID;
-
-@Component
-@NoSqlDao
-public class CassandraAckRepository extends CassandraAbstractDao implements AckRepository {
-
-    @Value("${cassandra.queue.ack.ttl}")
-    private int ackQueueTtl;
-
-    @Override
-    public ListenableFuture<Void> ack(MsgAck msgAck) {
-        String insert = "INSERT INTO msg_ack_queue (node_id, cluster_partition, ts_partition, msg_id) VALUES (?, ?, ?, ?) USING TTL ?";
-        PreparedStatement statement = prepare(insert);
-        BoundStatement boundStatement = statement.bind(msgAck.getNodeId(), msgAck.getClusteredPartition(),
-                msgAck.getTsPartition(), msgAck.getMsgId(), ackQueueTtl);
-        ResultSetFuture resultSetFuture = executeAsyncWrite(boundStatement);
-        return Futures.transform(resultSetFuture, (Function<ResultSet, Void>) input -> null);
-    }
-
-    @Override
-    public List<MsgAck> findAcks(UUID nodeId, long clusterPartition, long tsPartition) {
-        String select = "SELECT msg_id FROM msg_ack_queue WHERE " +
-                "node_id = ? AND cluster_partition = ? AND ts_partition = ?";
-        PreparedStatement statement = prepare(select);
-        BoundStatement boundStatement = statement.bind(nodeId, clusterPartition, tsPartition);
-        ResultSet rows = executeRead(boundStatement);
-        List<MsgAck> msgs = new ArrayList<>();
-        for (Row row : rows) {
-            msgs.add(new MsgAck(row.getUUID("msg_id"), nodeId, clusterPartition, tsPartition));
-        }
-        return msgs;
-    }
-
-}

dao/src/main/java/org/thingsboard/server/dao/queue/db/nosql/repository/CassandraMsgRepository.java

@@ -1,67 +0,0 @@
-/**
- * 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.dao.queue.db.nosql.repository;
-
-import com.datastax.driver.core.BoundStatement;
-import com.datastax.driver.core.PreparedStatement;
-import com.datastax.driver.core.ResultSet;
-import com.datastax.driver.core.ResultSetFuture;
-import com.datastax.driver.core.Row;
-import com.google.common.base.Function;
-import com.google.common.util.concurrent.Futures;
-import com.google.common.util.concurrent.ListenableFuture;
-import org.springframework.beans.factory.annotation.Value;
-import org.springframework.stereotype.Component;
-import org.thingsboard.server.common.msg.TbMsg;
-import org.thingsboard.server.dao.nosql.CassandraAbstractDao;
-import org.thingsboard.server.dao.queue.db.repository.MsgRepository;
-import org.thingsboard.server.dao.util.NoSqlDao;
-
-import java.util.ArrayList;
-import java.util.List;
-import java.util.UUID;
-
-@Component
-@NoSqlDao
-public class CassandraMsgRepository extends CassandraAbstractDao implements MsgRepository {
-
-    @Value("${cassandra.queue.msg.ttl}")
-    private int msqQueueTtl;
-
-    @Override
-    public ListenableFuture<Void> save(TbMsg msg, UUID nodeId, long clusterPartition, long tsPartition, long msgTs) {
-        String insert = "INSERT INTO msg_queue (node_id, cluster_partition, ts_partition, ts, msg) VALUES (?, ?, ?, ?, ?) USING TTL ?";
-        PreparedStatement statement = prepare(insert);
-        BoundStatement boundStatement = statement.bind(nodeId, clusterPartition, tsPartition, msgTs, TbMsg.toBytes(msg), msqQueueTtl);
-        ResultSetFuture resultSetFuture = executeAsyncWrite(boundStatement);
-        return Futures.transform(resultSetFuture, (Function<ResultSet, Void>) input -> null);
-    }
-
-    @Override
-    public List<TbMsg> findMsgs(UUID nodeId, long clusterPartition, long tsPartition) {
-        String select = "SELECT node_id, cluster_partition, ts_partition, ts, msg FROM msg_queue WHERE " +
-                "node_id = ? AND cluster_partition = ? AND ts_partition = ?";
-        PreparedStatement statement = prepare(select);
-        BoundStatement boundStatement = statement.bind(nodeId, clusterPartition, tsPartition);
-        ResultSet rows = executeRead(boundStatement);
-        List<TbMsg> msgs = new ArrayList<>();
-        for (Row row : rows) {
-            msgs.add(TbMsg.fromBytes(row.getBytes("msg")));
-        }
-        return msgs;
-    }
-
-}

dao/src/main/java/org/thingsboard/server/dao/queue/db/nosql/repository/CassandraProcessedPartitionRepository.java

@@ -1,64 +0,0 @@
-/**
- * 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.dao.queue.db.nosql.repository;
-
-import com.datastax.driver.core.BoundStatement;
-import com.datastax.driver.core.PreparedStatement;
-import com.datastax.driver.core.ResultSet;
-import com.datastax.driver.core.ResultSetFuture;
-import com.datastax.driver.core.Row;
-import com.google.common.base.Function;
-import com.google.common.util.concurrent.Futures;
-import com.google.common.util.concurrent.ListenableFuture;
-import org.springframework.beans.factory.annotation.Value;
-import org.springframework.stereotype.Component;
-import org.thingsboard.server.dao.nosql.CassandraAbstractDao;
-import org.thingsboard.server.dao.queue.db.repository.ProcessedPartitionRepository;
-import org.thingsboard.server.dao.util.NoSqlDao;
-
-import java.util.Optional;
-import java.util.UUID;
-
-@Component
-@NoSqlDao
-public class CassandraProcessedPartitionRepository extends CassandraAbstractDao implements ProcessedPartitionRepository {
-
-    @Value("${cassandra.queue.partitions.ttl}")
-    private int partitionsTtl;
-
-    @Override
-    public ListenableFuture<Void> partitionProcessed(UUID nodeId, long clusterPartition, long tsPartition) {
-        String insert = "INSERT INTO processed_msg_partitions (node_id, cluster_partition, ts_partition) VALUES (?, ?, ?) USING TTL ?";
-        PreparedStatement prepared = prepare(insert);
-        BoundStatement boundStatement = prepared.bind(nodeId, clusterPartition, tsPartition, partitionsTtl);
-        ResultSetFuture resultSetFuture = executeAsyncWrite(boundStatement);
-        return Futures.transform(resultSetFuture, (Function<ResultSet, Void>) input -> null);
-    }
-
-    @Override
-    public Optional<Long> findLastProcessedPartition(UUID nodeId, long clusteredHash) {
-        String select = "SELECT ts_partition FROM processed_msg_partitions WHERE " +
-                "node_id = ? AND cluster_partition = ?";
-        PreparedStatement prepared = prepare(select);
-        BoundStatement boundStatement = prepared.bind(nodeId, clusteredHash);
-        Row row = executeRead(boundStatement).one();
-        if (row == null) {
-            return Optional.empty();
-        }
-
-        return Optional.of(row.getLong("ts_partition"));
-    }
-}

dao/src/main/java/org/thingsboard/server/dao/queue/db/repository/AckRepository.java

@@ -1,29 +0,0 @@
-/**
- * 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.dao.queue.db.repository;
-
-import com.google.common.util.concurrent.ListenableFuture;
-import org.thingsboard.server.dao.queue.db.MsgAck;
-
-import java.util.List;
-import java.util.UUID;
-
-public interface AckRepository {
-
-    ListenableFuture<Void> ack(MsgAck msgAck);
-
-    List<MsgAck> findAcks(UUID nodeId, long clusterPartition, long tsPartition);
-}

dao/src/main/java/org/thingsboard/server/dao/queue/db/repository/MsgRepository.java

@@ -1,30 +0,0 @@
-/**
- * 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.dao.queue.db.repository;
-
-import com.google.common.util.concurrent.ListenableFuture;
-import org.thingsboard.server.common.msg.TbMsg;
-
-import java.util.List;
-import java.util.UUID;
-
-public interface MsgRepository {
-
-    ListenableFuture<Void> save(TbMsg msg, UUID nodeId, long clusterPartition, long tsPartition, long msgTs);
-
-    List<TbMsg> findMsgs(UUID nodeId, long clusterPartition, long tsPartition);
-
-}

dao/src/main/java/org/thingsboard/server/dao/queue/db/repository/ProcessedPartitionRepository.java

@@ -1,29 +0,0 @@
-/**
- * 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.dao.queue.db.repository;
-
-import com.google.common.util.concurrent.ListenableFuture;
-
-import java.util.Optional;
-import java.util.UUID;
-
-public interface ProcessedPartitionRepository {
-
-    ListenableFuture<Void> partitionProcessed(UUID nodeId, long clusteredHash, long partition);
-
-    Optional<Long> findLastProcessedPartition(UUID nodeId, long clusteredHash);
-
-}

dao/src/main/java/org/thingsboard/server/dao/queue/db/sql/SqlMsgQueue.java

@@ -1,20 +0,0 @@
-/**
- * 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.dao.queue.db.sql;
-
-//@todo-vp: implement
-public class SqlMsgQueue {
-}

dao/src/main/java/org/thingsboard/server/dao/queue/memory/InMemoryMsgQueue.java

@@ -40,7 +40,7 @@ import java.util.concurrent.Executors;
  * Created by ashvayka on 27.04.18.
  */
 @Component
-@ConditionalOnProperty(prefix = "rule.queue", value = "type", havingValue = "memory", matchIfMissing = true)
+@ConditionalOnProperty(prefix = "actors.rule.queue", value = "type", havingValue = "memory", matchIfMissing = true)
 @Slf4j
 public class InMemoryMsgQueue implements MsgQueue {
 

dao/src/main/java/org/thingsboard/server/dao/timeseries/CassandraBaseTimeseriesDao.java

@@ -82,6 +82,9 @@ public class CassandraBaseTimeseriesDao extends CassandraAbstractAsyncDao implem
     @Value("${cassandra.query.ts_key_value_partitioning}")
     private String partitioning;
 
+    @Value("${cassandra.query.ts_key_value_ttl}")
+    private long systemTtl;
+
     private TsPartitionDate tsFormat;
 
     private PreparedStatement partitionInsertStmt;
@@ -287,6 +290,7 @@ public class CassandraBaseTimeseriesDao extends CassandraAbstractAsyncDao implem
 
     @Override
     public ListenableFuture<Void> save(EntityId entityId, TsKvEntry tsKvEntry, long ttl) {
+        ttl = computeTtl(ttl);
         long partition = toPartitionTs(tsKvEntry.getTs());
         DataType type = tsKvEntry.getDataType();
         BoundStatement stmt = (ttl == 0 ? getSaveStmt(type) : getSaveTtlStmt(type)).bind();
@@ -304,6 +308,7 @@ public class CassandraBaseTimeseriesDao extends CassandraAbstractAsyncDao implem
 
     @Override
     public ListenableFuture<Void> savePartition(EntityId entityId, long tsKvEntryTs, String key, long ttl) {
+        ttl = computeTtl(ttl);
         long partition = toPartitionTs(tsKvEntryTs);
         log.debug("Saving partition {} for the entity [{}-{}] and key {}", partition, entityId.getEntityType(), entityId.getId(), key);
         BoundStatement stmt = (ttl == 0 ? getPartitionInsertStmt() : getPartitionInsertTtlStmt()).bind();
@@ -317,6 +322,17 @@ public class CassandraBaseTimeseriesDao extends CassandraAbstractAsyncDao implem
         return getFuture(executeAsyncWrite(stmt), rs -> null);
     }
 
+    private long computeTtl(long ttl) {
+        if (systemTtl > 0) {
+            if (ttl == 0) {
+                ttl = systemTtl;
+            } else {
+                ttl = Math.min(systemTtl, ttl);
+            }
+        }
+        return ttl;
+    }
+
     @Override
     public ListenableFuture<Void> saveLatest(EntityId entityId, TsKvEntry tsKvEntry) {
         BoundStatement stmt = getLatestStmt().bind()

dao/src/test/java/org/thingsboard/server/dao/queue/db/nosql/QueuePartitionerTest.java

@@ -1,81 +0,0 @@
-/**
- * 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.dao.queue.db.nosql;
-
-
-import org.junit.Before;
-import org.junit.Test;
-import org.junit.runner.RunWith;
-import org.mockito.Mock;
-import org.mockito.runners.MockitoJUnitRunner;
-import org.thingsboard.server.dao.queue.db.repository.ProcessedPartitionRepository;
-
-import java.time.Clock;
-import java.time.Instant;
-import java.time.ZoneOffset;
-import java.time.temporal.ChronoUnit;
-import java.util.List;
-import java.util.Optional;
-import java.util.UUID;
-
-import static org.junit.Assert.assertEquals;
-import static org.mockito.Mockito.when;
-
-@RunWith(MockitoJUnitRunner.class)
-public class QueuePartitionerTest {
-
-    private QueuePartitioner queuePartitioner;
-
-    @Mock
-    private ProcessedPartitionRepository partitionRepo;
-
-    private Instant startInstant;
-    private Instant endInstant;
-
-    @Before
-    public void init() {
-        queuePartitioner = new QueuePartitioner("MINUTES", partitionRepo);
-        startInstant = Instant.now();
-        endInstant = startInstant.plus(2, ChronoUnit.MINUTES);
-        queuePartitioner.setClock(Clock.fixed(endInstant, ZoneOffset.UTC));
-    }
-
-    @Test
-    public void partitionCalculated() {
-        long time = 1519390191425L;
-        long partition = queuePartitioner.getPartition(time);
-        assertEquals(1519390140000L, partition);
-    }
-
-    @Test
-    public void unprocessedPartitionsReturned() {
-        UUID nodeId = UUID.randomUUID();
-        long clusteredHash = 101L;
-        when(partitionRepo.findLastProcessedPartition(nodeId, clusteredHash)).thenReturn(Optional.of(startInstant.toEpochMilli()));
-        List<Long> actual = queuePartitioner.findUnprocessedPartitions(nodeId, clusteredHash);
-        assertEquals(3, actual.size());
-    }
-
-    @Test
-    public void defaultShiftUsedIfNoPartitionWasProcessed() {
-        UUID nodeId = UUID.randomUUID();
-        long clusteredHash = 101L;
-        when(partitionRepo.findLastProcessedPartition(nodeId, clusteredHash)).thenReturn(Optional.empty());
-        List<Long> actual = queuePartitioner.findUnprocessedPartitions(nodeId, clusteredHash);
-        assertEquals(10083, actual.size());
-    }
-
-}

dao/src/test/java/org/thingsboard/server/dao/queue/db/nosql/UnprocessedMsgFilterTest.java

@@ -1,47 +0,0 @@
-/**
- * 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.dao.queue.db.nosql;
-
-import com.google.common.collect.Lists;
-import org.junit.Test;
-import org.thingsboard.server.common.msg.TbMsg;
-import org.thingsboard.server.dao.queue.db.MsgAck;
-import org.thingsboard.server.dao.queue.db.UnprocessedMsgFilter;
-
-import java.util.Collection;
-import java.util.List;
-import java.util.UUID;
-
-import static org.junit.Assert.assertEquals;
-
-public class UnprocessedMsgFilterTest {
-
-    private UnprocessedMsgFilter msgFilter = new UnprocessedMsgFilter();
-
-    @Test
-    public void acknowledgedMsgsAreFilteredOut() {
-        UUID id1 = UUID.randomUUID();
-        UUID id2 = UUID.randomUUID();
-        TbMsg msg1 = new TbMsg(id1, "T", null, null, null, null, null, null, 0L);
-        TbMsg msg2 = new TbMsg(id2, "T", null, null, null, null, null, null, 0L);
-        List<TbMsg> msgs = Lists.newArrayList(msg1, msg2);
-        List<MsgAck> acks = Lists.newArrayList(new MsgAck(id2, UUID.randomUUID(), 1L, 1L));
-        Collection<TbMsg> actual = msgFilter.filter(msgs, acks);
-        assertEquals(1, actual.size());
-        assertEquals(msg1, actual.iterator().next());
-    }
-
-}

dao/src/test/java/org/thingsboard/server/dao/queue/db/nosql/repository/CassandraAckRepositoryTest.java

@@ -1,82 +0,0 @@
-/**
- * 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.dao.queue.db.nosql.repository;
-
-import com.datastax.driver.core.utils.UUIDs;
-import com.google.common.collect.Lists;
-import com.google.common.util.concurrent.ListenableFuture;
-import org.junit.Test;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.test.util.ReflectionTestUtils;
-import org.thingsboard.server.dao.service.AbstractServiceTest;
-import org.thingsboard.server.dao.service.DaoNoSqlTest;
-import org.thingsboard.server.dao.queue.db.MsgAck;
-
-import java.util.List;
-import java.util.UUID;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.TimeUnit;
-
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertTrue;
-
-@DaoNoSqlTest
-public class CassandraAckRepositoryTest extends AbstractServiceTest {
-
-    @Autowired
-    private CassandraAckRepository ackRepository;
-
-    @Test
-    public void acksInPartitionCouldBeFound() {
-        UUID nodeId = UUID.fromString("055eee50-1883-11e8-b380-65b5d5335ba9");
-
-        List<MsgAck> extectedAcks = Lists.newArrayList(
-                new MsgAck(UUID.fromString("bebaeb60-1888-11e8-bf21-65b5d5335ba9"), nodeId, 101L, 300L),
-                new MsgAck(UUID.fromString("12baeb60-1888-11e8-bf21-65b5d5335ba9"), nodeId, 101L, 300L)
-        );
-
-        List<MsgAck> actualAcks = ackRepository.findAcks(nodeId, 101L, 300L);
-        assertEquals(extectedAcks, actualAcks);
-    }
-
-    @Test
-    public void ackCanBeSavedAndRead() throws ExecutionException, InterruptedException {
-        UUID msgId = UUIDs.timeBased();
-        UUID nodeId = UUIDs.timeBased();
-        MsgAck ack = new MsgAck(msgId, nodeId, 10L, 20L);
-        ListenableFuture<Void> future = ackRepository.ack(ack);
-        future.get();
-        List<MsgAck> actualAcks = ackRepository.findAcks(nodeId, 10L, 20L);
-        assertEquals(1, actualAcks.size());
-        assertEquals(ack, actualAcks.get(0));
-    }
-
-    @Test
-    public void expiredAcksAreNotReturned() throws ExecutionException, InterruptedException {
-        ReflectionTestUtils.setField(ackRepository, "ackQueueTtl", 1);
-        UUID msgId = UUIDs.timeBased();
-        UUID nodeId = UUIDs.timeBased();
-        MsgAck ack = new MsgAck(msgId, nodeId, 30L, 40L);
-        ListenableFuture<Void> future = ackRepository.ack(ack);
-        future.get();
-        List<MsgAck> actualAcks = ackRepository.findAcks(nodeId, 30L, 40L);
-        assertEquals(1, actualAcks.size());
-        TimeUnit.SECONDS.sleep(2);
-        assertTrue(ackRepository.findAcks(nodeId, 30L, 40L).isEmpty());
-    }
-
-
-}

dao/src/test/java/org/thingsboard/server/dao/queue/db/nosql/repository/CassandraMsgRepositoryTest.java

@@ -1,87 +0,0 @@
-/**
- * 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.dao.queue.db.nosql.repository;
-
-//import static org.junit.jupiter.api.Assertions.*;
-
-import com.datastax.driver.core.utils.UUIDs;
-import com.google.common.util.concurrent.ListenableFuture;
-import org.junit.Test;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.test.util.ReflectionTestUtils;
-import org.thingsboard.server.common.data.id.DeviceId;
-import org.thingsboard.server.common.data.id.RuleChainId;
-import org.thingsboard.server.common.data.id.RuleNodeId;
-import org.thingsboard.server.common.msg.TbMsg;
-import org.thingsboard.server.common.msg.TbMsgDataType;
-import org.thingsboard.server.common.msg.TbMsgMetaData;
-import org.thingsboard.server.dao.service.AbstractServiceTest;
-import org.thingsboard.server.dao.service.DaoNoSqlTest;
-
-import java.util.List;
-import java.util.UUID;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.TimeUnit;
-
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertTrue;
-
-@DaoNoSqlTest
-public class CassandraMsgRepositoryTest extends AbstractServiceTest {
-
-    @Autowired
-    private CassandraMsgRepository msgRepository;
-
-    @Test
-    public void msgCanBeSavedAndRead() throws ExecutionException, InterruptedException {
-        TbMsg msg = new TbMsg(UUIDs.timeBased(), "type", new DeviceId(UUIDs.timeBased()), null, TbMsgDataType.JSON, "0000",
-                new RuleChainId(UUIDs.timeBased()), new RuleNodeId(UUIDs.timeBased()), 0L);
-        UUID nodeId = UUIDs.timeBased();
-        ListenableFuture<Void> future = msgRepository.save(msg, nodeId, 1L, 1L, 1L);
-        future.get();
-        List<TbMsg> msgs = msgRepository.findMsgs(nodeId, 1L, 1L);
-        assertEquals(1, msgs.size());
-    }
-
-    @Test
-    public void expiredMsgsAreNotReturned() throws ExecutionException, InterruptedException {
-        ReflectionTestUtils.setField(msgRepository, "msqQueueTtl", 1);
-        TbMsg msg = new TbMsg(UUIDs.timeBased(), "type", new DeviceId(UUIDs.timeBased()), null, TbMsgDataType.JSON, "0000",
-                new RuleChainId(UUIDs.timeBased()), new RuleNodeId(UUIDs.timeBased()), 0L);
-        UUID nodeId = UUIDs.timeBased();
-        ListenableFuture<Void> future = msgRepository.save(msg, nodeId, 2L, 2L, 2L);
-        future.get();
-        TimeUnit.SECONDS.sleep(2);
-        assertTrue(msgRepository.findMsgs(nodeId, 2L, 2L).isEmpty());
-    }
-
-    @Test
-    public void protoBufConverterWorkAsExpected() throws ExecutionException, InterruptedException {
-        TbMsgMetaData metaData = new TbMsgMetaData();
-        metaData.putValue("key", "value");
-        String dataStr = "someContent";
-        TbMsg msg = new TbMsg(UUIDs.timeBased(), "type", new DeviceId(UUIDs.timeBased()), metaData, TbMsgDataType.JSON, dataStr,
-                new RuleChainId(UUIDs.timeBased()), new RuleNodeId(UUIDs.timeBased()), 0L);
-        UUID nodeId = UUIDs.timeBased();
-        ListenableFuture<Void> future = msgRepository.save(msg, nodeId, 1L, 1L, 1L);
-        future.get();
-        List<TbMsg> msgs = msgRepository.findMsgs(nodeId, 1L, 1L);
-        assertEquals(1, msgs.size());
-        assertEquals(msg, msgs.get(0));
-    }
-
-
-}

dao/src/test/java/org/thingsboard/server/dao/queue/db/nosql/repository/CassandraProcessedPartitionRepositoryTest.java

@@ -1,83 +0,0 @@
-/**
- * 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.dao.queue.db.nosql.repository;
-
-import com.datastax.driver.core.utils.UUIDs;
-import com.google.common.util.concurrent.Futures;
-import com.google.common.util.concurrent.ListenableFuture;
-import org.junit.Test;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.test.util.ReflectionTestUtils;
-import org.thingsboard.server.dao.service.AbstractServiceTest;
-import org.thingsboard.server.dao.service.DaoNoSqlTest;
-
-import java.util.List;
-import java.util.Optional;
-import java.util.UUID;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.TimeUnit;
-
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertTrue;
-
-@DaoNoSqlTest
-public class CassandraProcessedPartitionRepositoryTest extends AbstractServiceTest {
-
-    @Autowired
-    private CassandraProcessedPartitionRepository partitionRepository;
-
-    @Test
-    public void lastProcessedPartitionCouldBeFound() {
-        UUID nodeId = UUID.fromString("055eee50-1883-11e8-b380-65b5d5335ba9");
-        Optional<Long> lastProcessedPartition = partitionRepository.findLastProcessedPartition(nodeId, 101L);
-        assertTrue(lastProcessedPartition.isPresent());
-        assertEquals((Long) 777L, lastProcessedPartition.get());
-    }
-
-    @Test
-    public void highestProcessedPartitionReturned() throws ExecutionException, InterruptedException {
-        UUID nodeId = UUIDs.timeBased();
-        ListenableFuture<Void> future1 = partitionRepository.partitionProcessed(nodeId, 303L, 100L);
-        ListenableFuture<Void> future2 = partitionRepository.partitionProcessed(nodeId, 303L, 200L);
-        ListenableFuture<Void> future3 = partitionRepository.partitionProcessed(nodeId, 303L, 10L);
-        ListenableFuture<List<Void>> allFutures = Futures.allAsList(future1, future2, future3);
-        allFutures.get();
-        Optional<Long> actual = partitionRepository.findLastProcessedPartition(nodeId, 303L);
-        assertTrue(actual.isPresent());
-        assertEquals((Long) 200L, actual.get());
-    }
-
-    @Test
-    public void expiredPartitionsAreNotReturned() throws ExecutionException, InterruptedException {
-        ReflectionTestUtils.setField(partitionRepository, "partitionsTtl", 1);
-        UUID nodeId = UUIDs.timeBased();
-        ListenableFuture<Void> future = partitionRepository.partitionProcessed(nodeId, 404L, 10L);
-        future.get();
-        Optional<Long> actual = partitionRepository.findLastProcessedPartition(nodeId, 404L);
-        assertEquals((Long) 10L, actual.get());
-        TimeUnit.SECONDS.sleep(2);
-        assertFalse(partitionRepository.findLastProcessedPartition(nodeId, 404L).isPresent());
-    }
-
-    @Test
-    public void ifNoPartitionsWereProcessedEmptyResultReturned() {
-        UUID nodeId = UUIDs.timeBased();
-        Optional<Long> actual = partitionRepository.findLastProcessedPartition(nodeId, 505L);
-        assertFalse(actual.isPresent());
-    }
-
-}

dao/src/test/resources/cassandra-test.properties

@@ -46,6 +46,8 @@ cassandra.query.default_fetch_size=2000
 
 cassandra.query.ts_key_value_partitioning=HOURS
 
+cassandra.query.ts_key_value_ttl=0
+
 cassandra.query.max_limit_per_request=1000
 cassandra.query.buffer_size=100000
 cassandra.query.concurrent_limit=1000

docker/tb/run-application.sh

@@ -18,6 +18,11 @@
 
 dpkg -i /thingsboard.deb
 
+# Copying env variables into conf files
+printenv | awk -F "=" '{print "export " $1 "='\''" $2 "'\''"}' >> /usr/share/thingsboard/conf/thingsboard.conf
+
+cat /usr/share/thingsboard/conf/thingsboard.conf
+
 if [ "$DATABASE_TYPE" == "cassandra" ]; then
     until nmap $CASSANDRA_HOST -p $CASSANDRA_PORT | grep "$CASSANDRA_PORT/tcp open\|filtered"
     do
@@ -46,12 +51,6 @@ if [ "$ADD_SCHEMA_AND_SYSTEM_DATA" == "true" ]; then
     fi
 fi
 
-
-# Copying env variables into conf files
-printenv | awk -F "=" '{print "export " $1 "='\''" $2 "'\''"}' >> /usr/share/thingsboard/conf/thingsboard.conf
-
-cat /usr/share/thingsboard/conf/thingsboard.conf
-
 echo "Starting 'Thingsboard' service..."
 service thingsboard start
 

rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/metadata/TbAbstractGetAttributesNode.java

@@ -15,7 +15,6 @@
  */
 package org.thingsboard.rule.engine.metadata;
 
-import com.google.common.base.Function;
 import com.google.common.util.concurrent.Futures;
 import com.google.common.util.concurrent.ListenableFuture;
 import org.apache.commons.collections.CollectionUtils;
@@ -33,9 +32,11 @@ import java.util.List;
 import static org.thingsboard.rule.engine.DonAsynchron.withCallback;
 import static org.thingsboard.rule.engine.api.TbRelationTypes.FAILURE;
 import static org.thingsboard.rule.engine.api.TbRelationTypes.SUCCESS;
-import static org.thingsboard.server.common.data.DataConstants.*;
+import static org.thingsboard.server.common.data.DataConstants.CLIENT_SCOPE;
+import static org.thingsboard.server.common.data.DataConstants.SERVER_SCOPE;
+import static org.thingsboard.server.common.data.DataConstants.SHARED_SCOPE;
 
-public abstract class TbAbstractGetAttributesNode<C extends TbGetAttributesNodeConfiguration, T extends EntityId>  implements TbNode {
+public abstract class TbAbstractGetAttributesNode<C extends TbGetAttributesNodeConfiguration, T extends EntityId> implements TbNode {
 
     protected C config;
 
@@ -59,7 +60,7 @@ public abstract class TbAbstractGetAttributesNode<C extends TbGetAttributesNodeC
     }
 
     private void safePutAttributes(TbContext ctx, TbMsg msg, T entityId) {
-        if(entityId == null || entityId.isNullUid()) {
+        if (entityId == null || entityId.isNullUid()) {
             ctx.tellNext(msg, FAILURE);
             return;
         }
@@ -78,7 +79,13 @@ public abstract class TbAbstractGetAttributesNode<C extends TbGetAttributesNodeC
         }
         ListenableFuture<List<AttributeKvEntry>> latest = ctx.getAttributesService().find(entityId, scope, keys);
         return Futures.transform(latest, l -> {
-            l.forEach(r -> msg.getMetaData().putValue(prefix + r.getKey(), r.getValueAsString()));
+            l.forEach(r -> {
+                if (r.getValue() != null) {
+                    msg.getMetaData().putValue(prefix + r.getKey(), r.getValueAsString());
+                } else {
+                    throw new RuntimeException("[" + scope + "][" + r.getKey() + "] attribute value is not present in the DB!");
+                }
+            });
             return null;
         });
     }
@@ -89,7 +96,13 @@ public abstract class TbAbstractGetAttributesNode<C extends TbGetAttributesNodeC
         }
         ListenableFuture<List<TsKvEntry>> latest = ctx.getTimeseriesService().findLatest(entityId, keys);
         return Futures.transform(latest, l -> {
-            l.forEach(r -> msg.getMetaData().putValue(r.getKey(), r.getValueAsString()));
+            l.forEach(r -> {
+                if (r.getValue() != null) {
+                    msg.getMetaData().putValue(r.getKey(), r.getValueAsString());
+                } else {
+                    throw new RuntimeException("[" + r.getKey() + "] telemetry value is not present in the DB!");
+                }
+            });
             return null;
         });
     }

rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgTimeseriesNode.java

@@ -69,10 +69,8 @@ public class TbMsgTimeseriesNode implements TbNode {
             try {
                 ts = Long.parseLong(tsStr);
             } catch (NumberFormatException e) {}
-        }
+        } else {
-        if (ts == -1) {
+            ts = System.currentTimeMillis();
-            ctx.tellFailure(msg, new IllegalArgumentException("Msg metadata doesn't contain valid ts value: " + msg.getMetaData()));
-            return;
         }
         String src = msg.getData();
         TelemetryUploadRequest telemetryUploadRequest = JsonConverter.convertToTelemetry(new JsonParser().parse(src), ts);

transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportService.java

@@ -119,8 +119,8 @@ public class MqttTransportService {
         try {
             serverChannel.close().sync();
         } finally {
-            bossGroup.shutdownGracefully();
             workerGroup.shutdownGracefully();
+            bossGroup.shutdownGracefully();
         }
         log.info("MQTT transport stopped!");
     }

ui/src/app/locale/locale.constant-zh.js

@@ -280,6 +280,38 @@ export default function addLocaleChinese(locales) {
             "selected-attributes": "{ count, select, 1 {1 属性} other {# 属性} } 被选中",
             "selected-telemetry": "{ count, select, 1 {1 遥测} other {# 遥测} } 被选中"
         },
+        "audit-log": {
+            "audit": "审计",
+            "audit-logs": "审计日志",
+            "timestamp": "时间戳",
+            "entity-type": "实体类型",
+            "entity-name": "实体名称",
+            "user": "用户",
+            "type": "类型",
+            "status": "状态",
+            "details": "详情",
+            "type-added": "添加",
+            "type-deleted": "删除",
+            "type-updated": "更新",
+            "type-attributes-updated": "更新属性",
+            "type-attributes-deleted": "删除属性",
+            "type-rpc-call": "RPC调用",
+            "type-credentials-updated": "更新凭证",
+            "type-assigned-to-customer": "分配给客户",
+            "type-unassigned-from-customer": "未分配给客户",
+            "type-activated": "激活",
+            "type-suspended": "暂停",
+            "type-credentials-read": "读取凭证",
+            "type-attributes-read": "读取属性",
+            "status-success": "成功",
+            "status-failure": "失败",
+            "audit-log-details": "审计日志详情",
+            "no-audit-logs-prompt": "找不到日志",
+            "action-data": "活动数据",
+            "failure-details": "失败详情",
+            "search": "查找审计日志",
+            "clear-search": "清空查找"
+        },
         "confirm-on-exit": {
             "message": "您有未保存的更改。确定要离开此页吗？",
             "html-message": "您有未保存的更改。<br/> 确定要离开此页面吗？",

ui/src/app/widget/widget-editor.controller.js

@@ -20,6 +20,7 @@ import 'brace/mode/javascript';
 import 'brace/mode/html';
 import 'brace/mode/css';
 import 'brace/mode/json';
+import 'ace-builds/src-min-noconflict/ace';
 import 'ace-builds/src-min-noconflict/snippets/javascript';
 import 'ace-builds/src-min-noconflict/snippets/text';
 import 'ace-builds/src-min-noconflict/snippets/html';
@@ -662,4 +663,4 @@ export default function WidgetEditorController(widgetService, userService, types
 
 }
 
-/* eslint-enable angular/angularelement */
+/* eslint-enable angular/angularelement */