RuleEngine Cassandra queue API

@@ -548,3 +548,55 @@ CREATE MATERIALIZED VIEW IF NOT EXISTS thingsboard.event_by_id AS
     AND event_type IS NOT NULL AND event_uid IS NOT NULL
     AND event_type IS NOT NULL AND event_uid IS NOT NULL
     PRIMARY KEY ((tenant_id, entity_type, entity_id), id, event_type, event_uid)
     PRIMARY KEY ((tenant_id, entity_type, entity_id), id, event_type, event_uid)
     WITH CLUSTERING ORDER BY (id ASC, event_type ASC, event_uid ASC);
     WITH CLUSTERING ORDER BY (id ASC, event_type ASC, event_uid ASC);
+
+CREATE TABLE IF NOT EXISTS thingsboard.msg_queue (
+    node_id         timeuuid,
+    clustered_hash    bigint,
+    partition       bigint,
+    ts              bigint,
+    msg             blob,
+	PRIMARY KEY ((node_id, cluster_hash, partition), ts)
+	WITH CLUSTERING ORDER BY (ts DESC)
+	AND compaction = {
+        'class': 'org.apache.cassandra.db.compaction.DateTieredCompactionStrategy',
+        'min_threshold': '5',
+        'base_time_seconds': '43200',
+        'max_window_size_seconds': '43200'
+        'tombstone_threshold': '0.9',
+        'unchecked_tombstone_compaction': 'true',
+    };
+);
+
+CREATE TABLE IF NOT EXISTS thingsboard.msg_ack_queue (
+    node_id         timeuuid,
+    clustered_hash    bigint,
+    partition       bigint,
+    ts              bigint,
+    msg_id              timeuuid,
+	PRIMARY KEY ((node_id, cluster_hash, partition), ts)
+	WITH CLUSTERING ORDER BY (ts DESC)
+	AND compaction = {
+        'class': 'org.apache.cassandra.db.compaction.DateTieredCompactionStrategy',
+        'min_threshold': '5',
+        'base_time_seconds': '43200',
+        'max_window_size_seconds': '43200'
+        'tombstone_threshold': '0.9',
+        'unchecked_tombstone_compaction': 'true',
+    };
+);
+
+CREATE TABLE IF NOT EXISTS thingsboard.processed_msg_partitions (
+    node_id         timeuuid,
+    clustered_hash    bigint,
+    partition       bigint,
+	PRIMARY KEY ((node_id, cluster_hash), partition)
+	WITH CLUSTERING ORDER BY (partition DESC)
+	AND compaction = {
+        'class': 'org.apache.cassandra.db.compaction.DateTieredCompactionStrategy',
+        'min_threshold': '5',
+        'base_time_seconds': '43200',
+        'max_window_size_seconds': '43200'
+        'tombstone_threshold': '0.9',
+        'unchecked_tombstone_compaction': 'true',
+    };
+);

+/**
+ * Copyright © 2016-2017 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.rule.engine.api;
+
+import com.google.common.util.concurrent.ListenableFuture;
+
+import java.util.UUID;
+
+public interface MsqQueue {
+
+    ListenableFuture<Void> put(TbMsg msg, UUID nodeId, long clusteredHash);
+
+    ListenableFuture<Void> ack(TbMsg msg, UUID nodeId, long clusteredHash);
+
+    Iterable<TbMsg> findUnprocessed(UUID nodeId, long clusteredHash);
+}

+/**
+ * Copyright © 2016-2017 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.rule.engine.api;
 package org.thingsboard.rule.engine.api;
 
 
 import org.thingsboard.server.common.msg.cluster.ServerAddress;
 import org.thingsboard.server.common.msg.cluster.ServerAddress;

+/**
+ * Copyright © 2016-2017 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.rule.engine.api;
 package org.thingsboard.rule.engine.api;
 
 
 import lombok.Data;
 import lombok.Data;

+/**
+ * Copyright © 2016-2017 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.rule.engine.api;
 package org.thingsboard.rule.engine.api;
 
 
 import lombok.Data;
 import lombok.Data;

+/**
+ * Copyright © 2016-2017 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.rule.engine.api;
 package org.thingsboard.rule.engine.api;
 
 
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutionException;

+/**
+ * Copyright © 2016-2017 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.rule.engine.api;
 package org.thingsboard.rule.engine.api;
 
 
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.JsonNode;

+/**
+ * Copyright © 2016-2017 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.rule.engine.api;
 package org.thingsboard.rule.engine.api;
 
 
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.core.JsonProcessingException;

+/**
+ * Copyright © 2016-2017 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.rule.engine.api;
 package org.thingsboard.rule.engine.api;
 
 
 /**
 /**

+/**
+ * Copyright © 2016-2017 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.rule.engine;
 package org.thingsboard.rule.engine;
 
 
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.core.JsonProcessingException;

+/**
+ * Copyright © 2016-2017 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.rule.engine.filter;
 package org.thingsboard.rule.engine.filter;
 
 
 import lombok.extern.slf4j.Slf4j;
 import lombok.extern.slf4j.Slf4j;

+/**
+ * Copyright © 2016-2017 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.rule.engine.filter;
 package org.thingsboard.rule.engine.filter;
 
 
 import lombok.Data;
 import lombok.Data;

+/**
+ * Copyright © 2016-2017 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.rule.engine.metadata;
 package org.thingsboard.rule.engine.metadata;
 
 
 import lombok.extern.slf4j.Slf4j;
 import lombok.extern.slf4j.Slf4j;

+/**
+ * Copyright © 2016-2017 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.rule.engine.metadata;
 package org.thingsboard.rule.engine.metadata;
 
 
 import lombok.Data;
 import lombok.Data;

+/**
+ * Copyright © 2016-2017 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.rule.engine.queue.cassandra;
+
+import org.springframework.stereotype.Component;
+import org.thingsboard.rule.engine.api.TbMsg;
+
+import java.util.UUID;
+
+@Component
+public class AckBuilder {
+
+    public MsgAck build(TbMsg msg, UUID nodeId, long clusteredHash) {
+        return null;
+    }
+}

+/**
+ * Copyright © 2016-2017 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.rule.engine.queue.cassandra;
+
+import com.google.common.collect.Lists;
+import com.google.common.util.concurrent.ListenableFuture;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+import org.thingsboard.rule.engine.api.MsqQueue;
+import org.thingsboard.rule.engine.api.TbMsg;
+import org.thingsboard.rule.engine.queue.cassandra.repository.AckRepository;
+import org.thingsboard.rule.engine.queue.cassandra.repository.MsgRepository;
+import org.thingsboard.rule.engine.queue.cassandra.repository.ProcessedPartitionRepository;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Optional;
+import java.util.UUID;
+
+@Component
+public class CassandraMsqQueue implements MsqQueue {
+
+    @Autowired
+    private MsgRepository msgRepository;
+
+    @Autowired
+    private AckRepository ackRepository;
+
+    @Autowired
+    private AckBuilder ackBuilder;
+
+    @Autowired
+    private UnprocessedMsgFilter unprocessedMsgFilter;
+
+    @Autowired
+    private ProcessedPartitionRepository processedPartitionRepository;
+
+    @Override
+    public ListenableFuture<Void> put(TbMsg msg, UUID nodeId, long clusteredHash) {
+        return msgRepository.save(msg, nodeId, clusteredHash, getPartition(msg));
+    }
+
+    @Override
+    public ListenableFuture<Void> ack(TbMsg msg, UUID nodeId, long clusteredHash) {
+        MsgAck ack = ackBuilder.build(msg, nodeId, clusteredHash);
+        return ackRepository.ack(ack);
+    }
+
+    @Override
+    public Iterable<TbMsg> findUnprocessed(UUID nodeId, long clusteredHash) {
+        List<TbMsg> unprocessedMsgs = Lists.newArrayList();
+        for (Long partition : findUnprocessedPartitions(nodeId, clusteredHash)) {
+            Iterable<TbMsg> msgs = msgRepository.findMsgs(nodeId, clusteredHash, partition);
+            Iterable<MsgAck> acks = ackRepository.findAcks(nodeId, clusteredHash, partition);
+            unprocessedMsgs.addAll(unprocessedMsgFilter.filter(msgs, acks));
+        }
+        return unprocessedMsgs;
+    }
+
+    private List<Long> findUnprocessedPartitions(UUID nodeId, long clusteredHash) {
+        Optional<Long> lastPartition = processedPartitionRepository.findLastProcessedPartition(nodeId, clusteredHash);
+        return Collections.emptyList();
+    }
+
+    private long getPartition(TbMsg msg) {
+        return Long.MIN_VALUE;
+    }
+
+}

+/**
+ * Copyright © 2016-2017 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.rule.engine.queue.cassandra;
+
+import lombok.Data;
+
+import java.util.UUID;
+
+@Data
+public class MsgAck {
+
+    private final UUID msgId;
+    private final UUID nodeId;
+    private final long clusteredHash;
+    private final long partition;
+    private final long ts;
+
+}

+/**
+ * Copyright © 2016-2017 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.rule.engine.queue.cassandra;
+
+import org.thingsboard.rule.engine.api.TbMsg;
+
+import java.util.Collection;
+import java.util.Collections;
+
+public class UnprocessedMsgFilter {
+
+    public Collection<TbMsg> filter(Iterable<TbMsg> msgs, Iterable<MsgAck> acks) {
+        return Collections.emptyList();
+    }
+}

+/**
+ * Copyright © 2016-2017 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.rule.engine.queue.cassandra.repository;
+
+import com.google.common.util.concurrent.ListenableFuture;
+import org.thingsboard.rule.engine.queue.cassandra.MsgAck;
+
+import java.util.UUID;
+
+public interface AckRepository {
+
+    ListenableFuture<Void> ack(MsgAck msgAck);
+
+    Iterable<MsgAck> findAcks(UUID nodeId, long clusteredHash, long partition);
+}

+/**
+ * Copyright © 2016-2017 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.rule.engine.queue.cassandra.repository;
+
+import com.google.common.util.concurrent.ListenableFuture;
+import org.thingsboard.rule.engine.api.TbMsg;
+
+import java.util.UUID;
+
+public interface MsgRepository {
+
+    ListenableFuture<Void> save(TbMsg msg, UUID nodeId, long clusteredHash, long partition);
+
+    Iterable<TbMsg> findMsgs(UUID nodeId, long clusteredHash, long partition);
+
+}

+package org.thingsboard.rule.engine.queue.cassandra.repository;
+
+import java.util.Optional;
+import java.util.UUID;
+
+public interface ProcessedPartitionRepository {
+
+    void partitionProcessed(UUID nodeId, long clusteredHash, long partition);
+
+    Optional<Long> findLastProcessedPartition(UUID nodeId, long clusteredHash);
+
+}

+/**
+ * Copyright © 2016-2017 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.rule.engine.queue.jpa;
+
+//@todo-vp: implement
+public class SqlMsgQueue {
+}

+/**
+ * Copyright © 2016-2017 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.rule.engine.transform;
 package org.thingsboard.rule.engine.transform;
 
 
 import lombok.extern.slf4j.Slf4j;
 import lombok.extern.slf4j.Slf4j;