zbeacon
4 years ago
75 changed files with 1729 additions and 144 deletions
@ -0,0 +1,107 @@ |
|||
## Install ThingsBoard Edge and connect to cloud instructions |
|||
|
|||
Here is the list of commands, that can be used to quickly install and connect ThingsBoard Edge to the cloud using docker compose. |
|||
|
|||
### Prerequisites |
|||
|
|||
Install <a href="https://docs.docker.com/engine/install/" target="_blank"> Docker CE</a> and <a href="https://docs.docker.com/compose/install/" target="_blank"> Docker Compose</a>. |
|||
|
|||
### Create data and logs folders |
|||
|
|||
Run following commands, before starting docker container(s), to create folders for storing data and logs. |
|||
These commands additionally will change owner of newly created folders to docker container user. |
|||
To do this (to change user) **chown** command is used, and this command requires *sudo* permissions (command will request password for a *sudo* access): |
|||
|
|||
```bash |
|||
mkdir -p ~/.mytb-edge-data && sudo chown -R 799:799 ~/.mytb-edge-data |
|||
mkdir -p ~/.mytb-edge-logs && sudo chown -R 799:799 ~/.mytb-edge-logs |
|||
{:copy-code} |
|||
``` |
|||
|
|||
### Running ThingsBoard Edge as docker service |
|||
|
|||
${LOCALHOST_WARNING} |
|||
|
|||
Create docker compose file for ThingsBoard Edge service: |
|||
|
|||
```bash |
|||
nano docker-compose.yml |
|||
{:copy-code} |
|||
``` |
|||
|
|||
Add the following lines to the yml file: |
|||
|
|||
```bash |
|||
version: '3.0' |
|||
services: |
|||
mytbedge: |
|||
restart: always |
|||
image: "thingsboard/tb-edge:${TB_EDGE_VERSION}" |
|||
ports: |
|||
- "8080:8080" |
|||
- "1883:1883" |
|||
- "5683-5688:5683-5688/udp" |
|||
environment: |
|||
SPRING_DATASOURCE_URL: jdbc:postgresql://postgres:5432/tb-edge |
|||
CLOUD_ROUTING_KEY: ${CLOUD_ROUTING_KEY} |
|||
CLOUD_ROUTING_SECRET: ${CLOUD_ROUTING_SECRET} |
|||
CLOUD_RPC_HOST: ${BASE_URL} |
|||
CLOUD_RPC_PORT: ${CLOUD_RPC_PORT} |
|||
CLOUD_RPC_SSL_ENABLED: ${CLOUD_RPC_SSL_ENABLED} |
|||
volumes: |
|||
- ~/.mytb-edge-data:/data |
|||
- ~/.mytb-edge-logs:/var/log/tb-edge |
|||
postgres: |
|||
restart: always |
|||
image: "postgres:12" |
|||
ports: |
|||
- "5432" |
|||
environment: |
|||
POSTGRES_DB: tb-edge |
|||
POSTGRES_PASSWORD: postgres |
|||
volumes: |
|||
- ~/.mytb-edge-data/db:/var/lib/postgresql/data |
|||
{:copy-code} |
|||
``` |
|||
|
|||
#### [Optional] Update bind ports |
|||
If ThingsBoard Edge is going to be running on the same machine where ThingsBoard server (cloud) is running, you'll need to update docker compose port mapping to avoid port collision between ThingsBoard server and ThingsBoard Edge. |
|||
|
|||
Please update next lines of `docker-compose.yml` file: |
|||
|
|||
```bash |
|||
ports: |
|||
- "18080:8080" |
|||
- "11883:1883" |
|||
- "15683-15688:5683-5688/udp" |
|||
``` |
|||
Make sure that ports above (18080, 11883, 15683-15688) are not used by any other application. |
|||
|
|||
#### Start ThingsBoard Edge |
|||
Set the terminal in the directory which contains the `docker-compose.yml` file and execute the following commands to up this docker compose directly: |
|||
|
|||
```bash |
|||
docker compose up -d |
|||
docker compose logs -f mytbedge |
|||
{:copy-code} |
|||
``` |
|||
|
|||
###### NOTE: Docker Compose V2 vs docker-compose (with a hyphen) |
|||
|
|||
ThingsBoard supports Docker Compose V2 (Docker Desktop or Compose plugin) starting from **3.4.2** release, because **docker-compose** as standalone setup is no longer supported by Docker. |
|||
We **strongly** recommend to update to Docker Compose V2 and use it. |
|||
If you still rely on using Docker Compose as docker-compose (with a hyphen), then please execute the following commands to start ThingsBoard Edge: |
|||
|
|||
```bash |
|||
docker-compose up -d |
|||
docker-compose logs -f mytbedge |
|||
``` |
|||
|
|||
#### Open ThingsBoard Edge UI |
|||
|
|||
Once started, you will be able to open **ThingsBoard Edge UI** using the following link http://localhost:8080. |
|||
|
|||
###### NOTE: Edge HTTP bind port update |
|||
|
|||
Use next **ThingsBoard Edge UI** link **http://localhost:18080** if you updated HTTP 8080 bind port to **18080**. |
|||
|
|||
@ -0,0 +1,3 @@ |
|||
###### WARNING NOTE: 'localhost' can not be used as CLOUD_RPC_HOST |
|||
|
|||
Please note that your ThingsBoard base URL is **'localhost'** at the moment. **'localhost'** cannot be used for docker containers - please update **CLOUD_RPC_HOST** environment variable below to the IP address of your machine (*docker **host** machine*). IP address must be `192.168.1.XX` or similar format. In other case - ThingsBoard Edge service, that is running in docker container, will not be able to connect to the cloud. |
|||
@ -0,0 +1,94 @@ |
|||
/** |
|||
* Copyright © 2016-2022 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.service.edge.instructions; |
|||
|
|||
import lombok.RequiredArgsConstructor; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.springframework.beans.factory.annotation.Value; |
|||
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; |
|||
import org.springframework.stereotype.Service; |
|||
import org.thingsboard.server.common.data.edge.Edge; |
|||
import org.thingsboard.server.common.data.edge.EdgeInstallInstructions; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
import org.thingsboard.server.queue.util.TbCoreComponent; |
|||
import org.thingsboard.server.service.install.InstallScripts; |
|||
|
|||
import javax.servlet.http.HttpServletRequest; |
|||
import java.io.IOException; |
|||
import java.nio.charset.StandardCharsets; |
|||
import java.nio.file.Files; |
|||
import java.nio.file.Path; |
|||
import java.nio.file.Paths; |
|||
|
|||
@Service |
|||
@Slf4j |
|||
@RequiredArgsConstructor |
|||
@ConditionalOnProperty(prefix = "edges", value = "enabled", havingValue = "true") |
|||
@TbCoreComponent |
|||
public class DefaultEdgeInstallService implements EdgeInstallService { |
|||
|
|||
private static final String EDGE_DIR = "edge"; |
|||
|
|||
private static final String EDGE_INSTALL_INSTRUCTIONS_DIR = "install_instructions"; |
|||
|
|||
private final InstallScripts installScripts; |
|||
|
|||
@Value("${edges.rpc.port}") |
|||
private int rpcPort; |
|||
|
|||
@Value("${edges.rpc.ssl.enabled}") |
|||
private boolean sslEnabled; |
|||
|
|||
@Value("${app.version:unknown}") |
|||
private String appVersion; |
|||
|
|||
@Override |
|||
public EdgeInstallInstructions getDockerInstallInstructions(TenantId tenantId, Edge edge, HttpServletRequest request) { |
|||
String dockerInstallInstructions = readFile(resolveFile("docker", "instructions.md")); |
|||
String baseUrl = request.getServerName(); |
|||
if (baseUrl.contains("localhost") || baseUrl.contains("127.0.0.1")) { |
|||
String localhostWarning = readFile(resolveFile("docker", "localhost_warning.md")); |
|||
dockerInstallInstructions = dockerInstallInstructions.replace("${LOCALHOST_WARNING}", localhostWarning); |
|||
dockerInstallInstructions = dockerInstallInstructions.replace("${BASE_URL}", "!!!REPLACE_ME_TO_HOST_IP_ADDRESS!!!"); |
|||
} else { |
|||
dockerInstallInstructions = dockerInstallInstructions.replace("${LOCALHOST_WARNING}", ""); |
|||
dockerInstallInstructions = dockerInstallInstructions.replace("${BASE_URL}", baseUrl); |
|||
} |
|||
dockerInstallInstructions = dockerInstallInstructions.replace("${TB_EDGE_VERSION}", appVersion + "EDGE"); |
|||
dockerInstallInstructions = dockerInstallInstructions.replace("${CLOUD_ROUTING_KEY}", edge.getRoutingKey()); |
|||
dockerInstallInstructions = dockerInstallInstructions.replace("${CLOUD_ROUTING_SECRET}", edge.getSecret()); |
|||
dockerInstallInstructions = dockerInstallInstructions.replace("${CLOUD_RPC_PORT}", Integer.toString(rpcPort)); |
|||
dockerInstallInstructions = dockerInstallInstructions.replace("${CLOUD_RPC_SSL_ENABLED}", Boolean.toString(sslEnabled)); |
|||
return new EdgeInstallInstructions(dockerInstallInstructions); |
|||
} |
|||
|
|||
private String readFile(Path file) { |
|||
try { |
|||
return new String(Files.readAllBytes(file), StandardCharsets.UTF_8); |
|||
} catch (IOException e) { |
|||
log.warn("Failed to read file: {}", file, e); |
|||
throw new RuntimeException(e); |
|||
} |
|||
} |
|||
|
|||
private Path resolveFile(String subDir, String... subDirs) { |
|||
return getEdgeInstallInstructionsDir().resolve(Paths.get(subDir, subDirs)); |
|||
} |
|||
|
|||
private Path getEdgeInstallInstructionsDir() { |
|||
return Paths.get(installScripts.getDataDir(), InstallScripts.JSON_DIR, EDGE_DIR, EDGE_INSTALL_INSTRUCTIONS_DIR); |
|||
} |
|||
} |
|||
@ -0,0 +1,28 @@ |
|||
/** |
|||
* Copyright © 2016-2022 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.service.edge.instructions; |
|||
|
|||
import org.thingsboard.server.common.data.edge.Edge; |
|||
import org.thingsboard.server.common.data.edge.EdgeInstallInstructions; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
|
|||
import javax.servlet.http.HttpServletRequest; |
|||
|
|||
public interface EdgeInstallService { |
|||
|
|||
EdgeInstallInstructions getDockerInstallInstructions(TenantId tenantId, Edge edge, HttpServletRequest request); |
|||
|
|||
} |
|||
@ -0,0 +1,32 @@ |
|||
/** |
|||
* Copyright © 2016-2022 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.data.edge; |
|||
|
|||
import io.swagger.annotations.ApiModel; |
|||
import io.swagger.annotations.ApiModelProperty; |
|||
import lombok.AllArgsConstructor; |
|||
import lombok.Data; |
|||
import lombok.NoArgsConstructor; |
|||
|
|||
@ApiModel |
|||
@Data |
|||
@AllArgsConstructor |
|||
@NoArgsConstructor |
|||
public class EdgeInstallInstructions { |
|||
|
|||
@ApiModelProperty(position = 1, value = "Markdown with docker install instructions") |
|||
private String dockerInstallInstructions; |
|||
} |
|||
@ -0,0 +1,22 @@ |
|||
/** |
|||
* Copyright © 2016-2022 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.deduplication; |
|||
|
|||
public enum DeduplicationId { |
|||
|
|||
ORIGINATOR, TENANT, CUSTOMER |
|||
|
|||
} |
|||
@ -0,0 +1,22 @@ |
|||
/** |
|||
* Copyright © 2016-2022 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.deduplication; |
|||
|
|||
public enum DeduplicationStrategy { |
|||
|
|||
FIRST, LAST, ALL |
|||
|
|||
} |
|||
@ -0,0 +1,237 @@ |
|||
/** |
|||
* Copyright © 2016-2022 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.deduplication; |
|||
|
|||
import com.fasterxml.jackson.databind.node.ArrayNode; |
|||
import com.fasterxml.jackson.databind.node.ObjectNode; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.springframework.data.util.Pair; |
|||
import org.thingsboard.common.util.JacksonUtil; |
|||
import org.thingsboard.rule.engine.api.RuleNode; |
|||
import org.thingsboard.rule.engine.api.TbContext; |
|||
import org.thingsboard.rule.engine.api.TbNode; |
|||
import org.thingsboard.rule.engine.api.TbNodeConfiguration; |
|||
import org.thingsboard.rule.engine.api.TbNodeException; |
|||
import org.thingsboard.rule.engine.api.TbRelationTypes; |
|||
import org.thingsboard.rule.engine.api.util.TbNodeUtils; |
|||
import org.thingsboard.server.common.data.id.EntityId; |
|||
import org.thingsboard.server.common.data.plugin.ComponentType; |
|||
import org.thingsboard.server.common.data.util.TbPair; |
|||
import org.thingsboard.server.common.msg.TbMsg; |
|||
import org.thingsboard.server.common.msg.TbMsgMetaData; |
|||
|
|||
import java.util.ArrayList; |
|||
import java.util.Comparator; |
|||
import java.util.HashMap; |
|||
import java.util.Iterator; |
|||
import java.util.LinkedList; |
|||
import java.util.List; |
|||
import java.util.Map; |
|||
import java.util.Optional; |
|||
import java.util.concurrent.ExecutionException; |
|||
import java.util.concurrent.TimeUnit; |
|||
|
|||
@RuleNode( |
|||
type = ComponentType.ACTION, |
|||
name = "deduplication", |
|||
configClazz = TbMsgDeduplicationNodeConfiguration.class, |
|||
nodeDescription = "Deduplicate messages for a configurable period based on a specified deduplication strategy.", |
|||
nodeDetails = "Rule node allows you to select one of the following strategy to deduplicate messages: <br></br>" + |
|||
"<b>FIRST</b> - return first message that arrived during deduplication period.<br></br>" + |
|||
"<b>LAST</b> - return last message that arrived during deduplication period.<br></br>" + |
|||
"<b>ALL</b> - return all messages as a single JSON array message. Where each element represents object with <b>msg</b> and <b>metadata</b> inner properties.<br></br>", |
|||
icon = "content_copy", |
|||
uiResources = {"static/rulenode/rulenode-core-config.js"}, |
|||
configDirective = "tbActionNodeMsgDeduplicationConfig" |
|||
) |
|||
@Slf4j |
|||
public class TbMsgDeduplicationNode implements TbNode { |
|||
|
|||
private static final String TB_MSG_DEDUPLICATION_TIMEOUT_MSG = "TbMsgDeduplicationNodeMsg"; |
|||
private static final int TB_MSG_DEDUPLICATION_TIMEOUT = 5000; |
|||
public static final int TB_MSG_DEDUPLICATION_RETRY_DELAY = 10; |
|||
|
|||
private TbMsgDeduplicationNodeConfiguration config; |
|||
|
|||
private final Map<EntityId, List<TbMsg>> deduplicationMap; |
|||
private long deduplicationInterval; |
|||
private long lastScheduledTs; |
|||
private DeduplicationId deduplicationId; |
|||
|
|||
public TbMsgDeduplicationNode() { |
|||
this.deduplicationMap = new HashMap<>(); |
|||
} |
|||
|
|||
@Override |
|||
public void init(TbContext ctx, TbNodeConfiguration configuration) throws TbNodeException { |
|||
this.config = TbNodeUtils.convert(configuration, TbMsgDeduplicationNodeConfiguration.class); |
|||
this.deduplicationInterval = TimeUnit.SECONDS.toMillis(config.getInterval()); |
|||
this.deduplicationId = config.getId(); |
|||
scheduleTickMsg(ctx); |
|||
} |
|||
|
|||
@Override |
|||
public void onMsg(TbContext ctx, TbMsg msg) throws ExecutionException, InterruptedException, TbNodeException { |
|||
if (TB_MSG_DEDUPLICATION_TIMEOUT_MSG.equals(msg.getType())) { |
|||
try { |
|||
processDeduplication(ctx); |
|||
} finally { |
|||
scheduleTickMsg(ctx); |
|||
} |
|||
} else { |
|||
processOnRegularMsg(ctx, msg); |
|||
} |
|||
} |
|||
|
|||
@Override |
|||
public void destroy() { |
|||
deduplicationMap.clear(); |
|||
} |
|||
|
|||
private void processOnRegularMsg(TbContext ctx, TbMsg msg) { |
|||
EntityId id = getDeduplicationId(ctx, msg); |
|||
List<TbMsg> deduplicationMsgs = deduplicationMap.computeIfAbsent(id, k -> new LinkedList<>()); |
|||
if (deduplicationMsgs.size() < config.getMaxPendingMsgs()) { |
|||
log.trace("[{}][{}] Adding msg: [{}][{}] to the pending msgs map ...", ctx.getSelfId(), id, msg.getId(), msg.getMetaDataTs()); |
|||
deduplicationMsgs.add(msg); |
|||
ctx.ack(msg); |
|||
} else { |
|||
log.trace("[{}] Max limit of pending messages reached for deduplication id: [{}]", ctx.getSelfId(), id); |
|||
ctx.tellFailure(msg, new RuntimeException("[" + ctx.getSelfId() + "] Max limit of pending messages reached for deduplication id: [" + id + "]")); |
|||
} |
|||
} |
|||
|
|||
private EntityId getDeduplicationId(TbContext ctx, TbMsg msg) { |
|||
switch (deduplicationId) { |
|||
case ORIGINATOR: |
|||
return msg.getOriginator(); |
|||
case TENANT: |
|||
return ctx.getTenantId(); |
|||
case CUSTOMER: |
|||
return msg.getCustomerId(); |
|||
default: |
|||
throw new IllegalStateException("Unsupported deduplication id: " + deduplicationId); |
|||
} |
|||
} |
|||
|
|||
private void processDeduplication(TbContext ctx) { |
|||
if (deduplicationMap.isEmpty()) { |
|||
return; |
|||
} |
|||
List<TbMsg> deduplicationResults = new ArrayList<>(); |
|||
long deduplicationTimeoutMs = System.currentTimeMillis(); |
|||
deduplicationMap.forEach((entityId, tbMsgs) -> { |
|||
if (tbMsgs.isEmpty()) { |
|||
return; |
|||
} |
|||
Optional<TbPair<Long, Long>> packBoundsOpt = findValidPack(tbMsgs, deduplicationTimeoutMs); |
|||
while (packBoundsOpt.isPresent()) { |
|||
TbPair<Long, Long> packBounds = packBoundsOpt.get(); |
|||
if (DeduplicationStrategy.ALL.equals(config.getStrategy())) { |
|||
List<TbMsg> pack = new ArrayList<>(); |
|||
for (Iterator<TbMsg> iterator = tbMsgs.iterator(); iterator.hasNext(); ) { |
|||
TbMsg msg = iterator.next(); |
|||
long msgTs = msg.getMetaDataTs(); |
|||
if (msgTs >= packBounds.getFirst() && msgTs < packBounds.getSecond()) { |
|||
pack.add(msg); |
|||
iterator.remove(); |
|||
} |
|||
} |
|||
deduplicationResults.add(TbMsg.newMsg( |
|||
config.getQueueName(), |
|||
config.getOutMsgType(), |
|||
entityId, |
|||
getMetadata(), |
|||
getMergedData(pack))); |
|||
} else { |
|||
TbMsg resultMsg = null; |
|||
boolean searchMin = DeduplicationStrategy.FIRST.equals(config.getStrategy()); |
|||
for (Iterator<TbMsg> iterator = tbMsgs.iterator(); iterator.hasNext(); ) { |
|||
TbMsg msg = iterator.next(); |
|||
long msgTs = msg.getMetaDataTs(); |
|||
if (msgTs >= packBounds.getFirst() && msgTs < packBounds.getSecond()) { |
|||
iterator.remove(); |
|||
if (resultMsg == null |
|||
|| (searchMin && msg.getMetaDataTs() < resultMsg.getMetaDataTs()) |
|||
|| (!searchMin && msg.getMetaDataTs() > resultMsg.getMetaDataTs())) { |
|||
resultMsg = msg; |
|||
} |
|||
} |
|||
} |
|||
deduplicationResults.add(resultMsg); |
|||
} |
|||
packBoundsOpt = findValidPack(tbMsgs, deduplicationTimeoutMs); |
|||
} |
|||
}); |
|||
deduplicationResults.forEach(outMsg -> enqueueForTellNextWithRetry(ctx, outMsg, 0)); |
|||
} |
|||
|
|||
private Optional<TbPair<Long, Long>> findValidPack(List<TbMsg> msgs, long deduplicationTimeoutMs) { |
|||
Optional<TbMsg> min = msgs.stream().min(Comparator.comparing(TbMsg::getMetaDataTs)); |
|||
return min.map(minTsMsg -> { |
|||
long packStartTs = minTsMsg.getMetaDataTs(); |
|||
long packEndTs = packStartTs + deduplicationInterval; |
|||
if (packEndTs <= deduplicationTimeoutMs) { |
|||
return new TbPair<>(packStartTs, packEndTs); |
|||
} |
|||
return null; |
|||
}); |
|||
} |
|||
|
|||
private void enqueueForTellNextWithRetry(TbContext ctx, TbMsg msg, int retryAttempt) { |
|||
if (config.getMaxRetries() > retryAttempt) { |
|||
ctx.enqueueForTellNext(msg, TbRelationTypes.SUCCESS, |
|||
() -> { |
|||
log.trace("[{}][{}][{}] Successfully enqueue deduplication result message!", ctx.getSelfId(), msg.getOriginator(), retryAttempt); |
|||
}, |
|||
throwable -> { |
|||
log.trace("[{}][{}][{}] Failed to enqueue deduplication output message due to: ", ctx.getSelfId(), msg.getOriginator(), retryAttempt, throwable); |
|||
ctx.schedule(() -> { |
|||
enqueueForTellNextWithRetry(ctx, msg, retryAttempt + 1); |
|||
}, TB_MSG_DEDUPLICATION_RETRY_DELAY, TimeUnit.SECONDS); |
|||
}); |
|||
} |
|||
} |
|||
|
|||
private void scheduleTickMsg(TbContext ctx) { |
|||
long curTs = System.currentTimeMillis(); |
|||
if (lastScheduledTs == 0L) { |
|||
lastScheduledTs = curTs; |
|||
} |
|||
lastScheduledTs += TB_MSG_DEDUPLICATION_TIMEOUT; |
|||
long curDelay = Math.max(0L, (lastScheduledTs - curTs)); |
|||
TbMsg tickMsg = ctx.newMsg(null, TB_MSG_DEDUPLICATION_TIMEOUT_MSG, ctx.getSelfId(), new TbMsgMetaData(), ""); |
|||
ctx.tellSelf(tickMsg, curDelay); |
|||
} |
|||
|
|||
private String getMergedData(List<TbMsg> msgs) { |
|||
ArrayNode mergedData = JacksonUtil.OBJECT_MAPPER.createArrayNode(); |
|||
msgs.forEach(msg -> { |
|||
ObjectNode msgNode = JacksonUtil.newObjectNode(); |
|||
msgNode.set("msg", JacksonUtil.toJsonNode(msg.getData())); |
|||
msgNode.set("metadata", JacksonUtil.valueToTree(msg.getMetaData().getData())); |
|||
mergedData.add(msgNode); |
|||
}); |
|||
return JacksonUtil.toString(mergedData); |
|||
} |
|||
|
|||
private TbMsgMetaData getMetadata() { |
|||
TbMsgMetaData metaData = new TbMsgMetaData(); |
|||
metaData.putValue("ts", String.valueOf(System.currentTimeMillis())); |
|||
return metaData; |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,46 @@ |
|||
/** |
|||
* Copyright © 2016-2022 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.deduplication; |
|||
|
|||
import lombok.Data; |
|||
import org.thingsboard.rule.engine.api.NodeConfiguration; |
|||
|
|||
@Data |
|||
public class TbMsgDeduplicationNodeConfiguration implements NodeConfiguration<TbMsgDeduplicationNodeConfiguration> { |
|||
|
|||
private int interval; |
|||
private DeduplicationId id; |
|||
private DeduplicationStrategy strategy; |
|||
|
|||
// Advanced settings:
|
|||
private int maxPendingMsgs; |
|||
private int maxRetries; |
|||
|
|||
// only for DeduplicationStrategy.ALL:
|
|||
private String outMsgType; |
|||
private String queueName; |
|||
|
|||
@Override |
|||
public TbMsgDeduplicationNodeConfiguration defaultConfiguration() { |
|||
TbMsgDeduplicationNodeConfiguration configuration = new TbMsgDeduplicationNodeConfiguration(); |
|||
configuration.setInterval(60); |
|||
configuration.setId(DeduplicationId.ORIGINATOR); |
|||
configuration.setStrategy(DeduplicationStrategy.FIRST); |
|||
configuration.setMaxPendingMsgs(100); |
|||
configuration.setMaxRetries(3); |
|||
return configuration; |
|||
} |
|||
} |
|||
File diff suppressed because one or more lines are too long
@ -0,0 +1,402 @@ |
|||
/** |
|||
* Copyright © 2016-2022 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.action; |
|||
|
|||
import com.fasterxml.jackson.databind.node.ArrayNode; |
|||
import com.fasterxml.jackson.databind.node.ObjectNode; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.junit.jupiter.api.AfterEach; |
|||
import org.junit.jupiter.api.Assertions; |
|||
import org.junit.jupiter.api.BeforeEach; |
|||
import org.junit.jupiter.api.Test; |
|||
import org.mockito.ArgumentCaptor; |
|||
import org.mockito.ArgumentMatchers; |
|||
import org.mockito.stubbing.Answer; |
|||
import org.thingsboard.common.util.JacksonUtil; |
|||
import org.thingsboard.common.util.ThingsBoardThreadFactory; |
|||
import org.thingsboard.rule.engine.api.TbContext; |
|||
import org.thingsboard.rule.engine.api.TbNodeConfiguration; |
|||
import org.thingsboard.rule.engine.api.TbNodeException; |
|||
import org.thingsboard.rule.engine.api.TbRelationTypes; |
|||
import org.thingsboard.rule.engine.deduplication.DeduplicationStrategy; |
|||
import org.thingsboard.rule.engine.deduplication.TbMsgDeduplicationNode; |
|||
import org.thingsboard.rule.engine.deduplication.TbMsgDeduplicationNodeConfiguration; |
|||
import org.thingsboard.server.common.data.id.DeviceId; |
|||
import org.thingsboard.server.common.data.id.EntityId; |
|||
import org.thingsboard.server.common.data.id.RuleNodeId; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
import org.thingsboard.server.common.msg.TbMsg; |
|||
import org.thingsboard.server.common.msg.TbMsgMetaData; |
|||
import org.thingsboard.server.common.msg.session.SessionMsgType; |
|||
|
|||
import java.util.ArrayList; |
|||
import java.util.List; |
|||
import java.util.Random; |
|||
import java.util.UUID; |
|||
import java.util.concurrent.CountDownLatch; |
|||
import java.util.concurrent.ExecutionException; |
|||
import java.util.concurrent.Executors; |
|||
import java.util.concurrent.ScheduledExecutorService; |
|||
import java.util.concurrent.TimeUnit; |
|||
import java.util.concurrent.atomic.AtomicInteger; |
|||
import java.util.concurrent.atomic.AtomicLong; |
|||
import java.util.function.Consumer; |
|||
|
|||
import static org.mockito.ArgumentMatchers.any; |
|||
import static org.mockito.ArgumentMatchers.eq; |
|||
import static org.mockito.ArgumentMatchers.isNull; |
|||
import static org.mockito.ArgumentMatchers.nullable; |
|||
import static org.mockito.Mockito.doAnswer; |
|||
import static org.mockito.Mockito.mock; |
|||
import static org.mockito.Mockito.spy; |
|||
import static org.mockito.Mockito.times; |
|||
import static org.mockito.Mockito.verify; |
|||
import static org.mockito.Mockito.when; |
|||
|
|||
@Slf4j |
|||
public class TbMsgDeduplicationNodeTest { |
|||
|
|||
private static final String MAIN_QUEUE_NAME = "Main"; |
|||
private static final String HIGH_PRIORITY_QUEUE_NAME = "HighPriority"; |
|||
private static final String TB_MSG_DEDUPLICATION_TIMEOUT_MSG = "TbMsgDeduplicationNodeMsg"; |
|||
|
|||
private TbContext ctx; |
|||
|
|||
private final ThingsBoardThreadFactory factory = ThingsBoardThreadFactory.forName("de-duplication-node-test"); |
|||
private final ScheduledExecutorService executorService = Executors.newSingleThreadScheduledExecutor(factory); |
|||
private final int deduplicationInterval = 1; |
|||
|
|||
private TenantId tenantId; |
|||
|
|||
private TbMsgDeduplicationNode node; |
|||
private TbMsgDeduplicationNodeConfiguration config; |
|||
private TbNodeConfiguration nodeConfiguration; |
|||
|
|||
private CountDownLatch awaitTellSelfLatch; |
|||
|
|||
@BeforeEach |
|||
public void init() throws TbNodeException { |
|||
ctx = mock(TbContext.class); |
|||
|
|||
tenantId = TenantId.fromUUID(UUID.randomUUID()); |
|||
RuleNodeId ruleNodeId = new RuleNodeId(UUID.randomUUID()); |
|||
|
|||
when(ctx.getSelfId()).thenReturn(ruleNodeId); |
|||
when(ctx.getTenantId()).thenReturn(tenantId); |
|||
|
|||
doAnswer((Answer<TbMsg>) invocationOnMock -> { |
|||
String type = (String) (invocationOnMock.getArguments())[1]; |
|||
EntityId originator = (EntityId) (invocationOnMock.getArguments())[2]; |
|||
TbMsgMetaData metaData = (TbMsgMetaData) (invocationOnMock.getArguments())[3]; |
|||
String data = (String) (invocationOnMock.getArguments())[4]; |
|||
return TbMsg.newMsg(type, originator, metaData.copy(), data); |
|||
}).when(ctx).newMsg(isNull(), eq(TB_MSG_DEDUPLICATION_TIMEOUT_MSG), nullable(EntityId.class), any(TbMsgMetaData.class), any(String.class)); |
|||
node = spy(new TbMsgDeduplicationNode()); |
|||
config = new TbMsgDeduplicationNodeConfiguration().defaultConfiguration(); |
|||
} |
|||
|
|||
private void invokeTellSelf(int maxNumberOfInvocation) { |
|||
invokeTellSelf(maxNumberOfInvocation, false, 0); |
|||
} |
|||
|
|||
private void invokeTellSelf(int maxNumberOfInvocation, boolean delayScheduleTimeout, int delayMultiplier) { |
|||
AtomicLong scheduleTimeout = new AtomicLong(deduplicationInterval); |
|||
AtomicInteger scheduleCount = new AtomicInteger(0); |
|||
doAnswer((Answer<Void>) invocationOnMock -> { |
|||
scheduleCount.getAndIncrement(); |
|||
if (scheduleCount.get() <= maxNumberOfInvocation) { |
|||
TbMsg msg = (TbMsg) (invocationOnMock.getArguments())[0]; |
|||
executorService.schedule(() -> { |
|||
try { |
|||
node.onMsg(ctx, msg); |
|||
awaitTellSelfLatch.countDown(); |
|||
} catch (ExecutionException | InterruptedException | TbNodeException e) { |
|||
log.error("Failed to execute tellSelf method call due to: ", e); |
|||
} |
|||
}, scheduleTimeout.get(), TimeUnit.SECONDS); |
|||
if (delayScheduleTimeout) { |
|||
scheduleTimeout.set(scheduleTimeout.get() * delayMultiplier); |
|||
} |
|||
} |
|||
|
|||
return null; |
|||
}).when(ctx).tellSelf(ArgumentMatchers.any(TbMsg.class), ArgumentMatchers.anyLong()); |
|||
} |
|||
|
|||
@AfterEach |
|||
public void destroy() { |
|||
executorService.shutdown(); |
|||
node.destroy(); |
|||
} |
|||
|
|||
@Test |
|||
public void given_100_messages_strategy_first_then_verifyOutput() throws TbNodeException, ExecutionException, InterruptedException { |
|||
int wantedNumberOfTellSelfInvocation = 2; |
|||
int msgCount = 100; |
|||
awaitTellSelfLatch = new CountDownLatch(wantedNumberOfTellSelfInvocation); |
|||
invokeTellSelf(wantedNumberOfTellSelfInvocation); |
|||
|
|||
config.setInterval(deduplicationInterval); |
|||
config.setMaxPendingMsgs(msgCount); |
|||
nodeConfiguration = new TbNodeConfiguration(JacksonUtil.valueToTree(config)); |
|||
node.init(ctx, nodeConfiguration); |
|||
|
|||
DeviceId deviceId = new DeviceId(UUID.randomUUID()); |
|||
long currentTimeMillis = System.currentTimeMillis(); |
|||
|
|||
List<TbMsg> inputMsgs = getTbMsgs(deviceId, msgCount, currentTimeMillis, 500); |
|||
for (TbMsg msg : inputMsgs) { |
|||
node.onMsg(ctx, msg); |
|||
} |
|||
|
|||
TbMsg msgToReject = createMsg(deviceId, inputMsgs.get(inputMsgs.size() - 1).getMetaDataTs() + 2); |
|||
node.onMsg(ctx, msgToReject); |
|||
|
|||
awaitTellSelfLatch.await(); |
|||
|
|||
ArgumentCaptor<TbMsg> newMsgCaptor = ArgumentCaptor.forClass(TbMsg.class); |
|||
ArgumentCaptor<Runnable> successCaptor = ArgumentCaptor.forClass(Runnable.class); |
|||
ArgumentCaptor<Consumer<Throwable>> failureCaptor = ArgumentCaptor.forClass(Consumer.class); |
|||
|
|||
verify(ctx, times(msgCount)).ack(any()); |
|||
verify(ctx, times(1)).tellFailure(eq(msgToReject), any()); |
|||
verify(node, times(msgCount + wantedNumberOfTellSelfInvocation + 1)).onMsg(eq(ctx), any()); |
|||
verify(ctx, times(1)).enqueueForTellNext(newMsgCaptor.capture(), eq(TbRelationTypes.SUCCESS), successCaptor.capture(), failureCaptor.capture()); |
|||
Assertions.assertEquals(inputMsgs.get(0), newMsgCaptor.getValue()); |
|||
} |
|||
|
|||
@Test |
|||
public void given_100_messages_strategy_last_then_verifyOutput() throws TbNodeException, ExecutionException, InterruptedException { |
|||
int wantedNumberOfTellSelfInvocation = 2; |
|||
int msgCount = 100; |
|||
awaitTellSelfLatch = new CountDownLatch(wantedNumberOfTellSelfInvocation); |
|||
invokeTellSelf(wantedNumberOfTellSelfInvocation); |
|||
|
|||
config.setStrategy(DeduplicationStrategy.LAST); |
|||
config.setInterval(deduplicationInterval); |
|||
config.setMaxPendingMsgs(msgCount); |
|||
nodeConfiguration = new TbNodeConfiguration(JacksonUtil.valueToTree(config)); |
|||
node.init(ctx, nodeConfiguration); |
|||
|
|||
DeviceId deviceId = new DeviceId(UUID.randomUUID()); |
|||
long currentTimeMillis = System.currentTimeMillis(); |
|||
|
|||
List<TbMsg> inputMsgs = getTbMsgs(deviceId, msgCount, currentTimeMillis, 500); |
|||
TbMsg msgWithLatestTs = getMsgWithLatestTs(inputMsgs); |
|||
|
|||
for (TbMsg msg : inputMsgs) { |
|||
node.onMsg(ctx, msg); |
|||
} |
|||
|
|||
TbMsg msgToReject = createMsg(deviceId, inputMsgs.get(inputMsgs.size() - 1).getMetaDataTs() + 2); |
|||
node.onMsg(ctx, msgToReject); |
|||
|
|||
awaitTellSelfLatch.await(); |
|||
|
|||
ArgumentCaptor<TbMsg> newMsgCaptor = ArgumentCaptor.forClass(TbMsg.class); |
|||
ArgumentCaptor<Runnable> successCaptor = ArgumentCaptor.forClass(Runnable.class); |
|||
ArgumentCaptor<Consumer<Throwable>> failureCaptor = ArgumentCaptor.forClass(Consumer.class); |
|||
|
|||
verify(ctx, times(msgCount)).ack(any()); |
|||
verify(ctx, times(1)).tellFailure(eq(msgToReject), any()); |
|||
verify(node, times(msgCount + wantedNumberOfTellSelfInvocation + 1)).onMsg(eq(ctx), any()); |
|||
verify(ctx, times(1)).enqueueForTellNext(newMsgCaptor.capture(), eq(TbRelationTypes.SUCCESS), successCaptor.capture(), failureCaptor.capture()); |
|||
Assertions.assertEquals(msgWithLatestTs, newMsgCaptor.getValue()); |
|||
} |
|||
|
|||
@Test |
|||
public void given_100_messages_strategy_all_then_verifyOutput() throws TbNodeException, ExecutionException, InterruptedException { |
|||
int wantedNumberOfTellSelfInvocation = 2; |
|||
int msgCount = 100; |
|||
awaitTellSelfLatch = new CountDownLatch(wantedNumberOfTellSelfInvocation); |
|||
invokeTellSelf(wantedNumberOfTellSelfInvocation); |
|||
|
|||
config.setInterval(deduplicationInterval); |
|||
config.setStrategy(DeduplicationStrategy.ALL); |
|||
config.setOutMsgType(SessionMsgType.POST_ATTRIBUTES_REQUEST.name()); |
|||
config.setQueueName(HIGH_PRIORITY_QUEUE_NAME); |
|||
nodeConfiguration = new TbNodeConfiguration(JacksonUtil.valueToTree(config)); |
|||
node.init(ctx, nodeConfiguration); |
|||
|
|||
DeviceId deviceId = new DeviceId(UUID.randomUUID()); |
|||
long currentTimeMillis = System.currentTimeMillis(); |
|||
|
|||
List<TbMsg> inputMsgs = getTbMsgs(deviceId, msgCount, currentTimeMillis, 500); |
|||
for (TbMsg msg : inputMsgs) { |
|||
node.onMsg(ctx, msg); |
|||
} |
|||
|
|||
awaitTellSelfLatch.await(); |
|||
|
|||
ArgumentCaptor<TbMsg> newMsgCaptor = ArgumentCaptor.forClass(TbMsg.class); |
|||
ArgumentCaptor<Runnable> successCaptor = ArgumentCaptor.forClass(Runnable.class); |
|||
ArgumentCaptor<Consumer<Throwable>> failureCaptor = ArgumentCaptor.forClass(Consumer.class); |
|||
|
|||
verify(ctx, times(msgCount)).ack(any()); |
|||
verify(node, times(msgCount + wantedNumberOfTellSelfInvocation)).onMsg(eq(ctx), any()); |
|||
verify(ctx, times(1)).enqueueForTellNext(newMsgCaptor.capture(), eq(TbRelationTypes.SUCCESS), successCaptor.capture(), failureCaptor.capture()); |
|||
|
|||
Assertions.assertEquals(1, newMsgCaptor.getAllValues().size()); |
|||
TbMsg outMessage = newMsgCaptor.getAllValues().get(0); |
|||
Assertions.assertEquals(getMergedData(inputMsgs), outMessage.getData()); |
|||
Assertions.assertEquals(deviceId, outMessage.getOriginator()); |
|||
Assertions.assertEquals(config.getOutMsgType(), outMessage.getType()); |
|||
Assertions.assertEquals(config.getQueueName(), outMessage.getQueueName()); |
|||
} |
|||
|
|||
@Test |
|||
public void given_100_messages_strategy_all_then_verifyOutput_2_packs() throws TbNodeException, ExecutionException, InterruptedException { |
|||
int wantedNumberOfTellSelfInvocation = 2; |
|||
int msgCount = 100; |
|||
awaitTellSelfLatch = new CountDownLatch(wantedNumberOfTellSelfInvocation); |
|||
invokeTellSelf(wantedNumberOfTellSelfInvocation, true, 3); |
|||
|
|||
config.setInterval(deduplicationInterval); |
|||
config.setStrategy(DeduplicationStrategy.ALL); |
|||
config.setOutMsgType(SessionMsgType.POST_ATTRIBUTES_REQUEST.name()); |
|||
config.setQueueName(HIGH_PRIORITY_QUEUE_NAME); |
|||
nodeConfiguration = new TbNodeConfiguration(JacksonUtil.valueToTree(config)); |
|||
node.init(ctx, nodeConfiguration); |
|||
|
|||
DeviceId deviceId = new DeviceId(UUID.randomUUID()); |
|||
long currentTimeMillis = System.currentTimeMillis(); |
|||
|
|||
List<TbMsg> firstMsgPack = getTbMsgs(deviceId, msgCount / 2, currentTimeMillis, 500); |
|||
for (TbMsg msg : firstMsgPack) { |
|||
node.onMsg(ctx, msg); |
|||
} |
|||
long firstPackDeduplicationPackEndTs = firstMsgPack.get(0).getMetaDataTs() + TimeUnit.SECONDS.toMillis(deduplicationInterval); |
|||
|
|||
List<TbMsg> secondMsgPack = getTbMsgs(deviceId, msgCount / 2, firstPackDeduplicationPackEndTs, 500); |
|||
for (TbMsg msg : secondMsgPack) { |
|||
node.onMsg(ctx, msg); |
|||
} |
|||
|
|||
awaitTellSelfLatch.await(); |
|||
|
|||
ArgumentCaptor<TbMsg> newMsgCaptor = ArgumentCaptor.forClass(TbMsg.class); |
|||
ArgumentCaptor<Runnable> successCaptor = ArgumentCaptor.forClass(Runnable.class); |
|||
ArgumentCaptor<Consumer<Throwable>> failureCaptor = ArgumentCaptor.forClass(Consumer.class); |
|||
|
|||
verify(ctx, times(msgCount)).ack(any()); |
|||
verify(node, times(msgCount + wantedNumberOfTellSelfInvocation)).onMsg(eq(ctx), any()); |
|||
verify(ctx, times(2)).enqueueForTellNext(newMsgCaptor.capture(), eq(TbRelationTypes.SUCCESS), successCaptor.capture(), failureCaptor.capture()); |
|||
|
|||
List<TbMsg> resultMsgs = newMsgCaptor.getAllValues(); |
|||
Assertions.assertEquals(2, resultMsgs.size()); |
|||
|
|||
TbMsg firstMsg = resultMsgs.get(0); |
|||
Assertions.assertEquals(getMergedData(firstMsgPack), firstMsg.getData()); |
|||
Assertions.assertEquals(deviceId, firstMsg.getOriginator()); |
|||
Assertions.assertEquals(config.getOutMsgType(), firstMsg.getType()); |
|||
Assertions.assertEquals(config.getQueueName(), firstMsg.getQueueName()); |
|||
|
|||
TbMsg secondMsg = resultMsgs.get(1); |
|||
Assertions.assertEquals(getMergedData(secondMsgPack), secondMsg.getData()); |
|||
Assertions.assertEquals(deviceId, secondMsg.getOriginator()); |
|||
Assertions.assertEquals(config.getOutMsgType(), secondMsg.getType()); |
|||
Assertions.assertEquals(config.getQueueName(), secondMsg.getQueueName()); |
|||
} |
|||
|
|||
@Test |
|||
public void given_100_messages_strategy_last_then_verifyOutput_2_packs() throws TbNodeException, ExecutionException, InterruptedException { |
|||
int wantedNumberOfTellSelfInvocation = 2; |
|||
int msgCount = 100; |
|||
awaitTellSelfLatch = new CountDownLatch(wantedNumberOfTellSelfInvocation); |
|||
invokeTellSelf(wantedNumberOfTellSelfInvocation, true, 3); |
|||
|
|||
config.setInterval(deduplicationInterval); |
|||
config.setStrategy(DeduplicationStrategy.LAST); |
|||
nodeConfiguration = new TbNodeConfiguration(JacksonUtil.valueToTree(config)); |
|||
node.init(ctx, nodeConfiguration); |
|||
|
|||
DeviceId deviceId = new DeviceId(UUID.randomUUID()); |
|||
long currentTimeMillis = System.currentTimeMillis(); |
|||
|
|||
List<TbMsg> firstMsgPack = getTbMsgs(deviceId, msgCount / 2, currentTimeMillis, 500); |
|||
for (TbMsg msg : firstMsgPack) { |
|||
node.onMsg(ctx, msg); |
|||
} |
|||
long firstPackDeduplicationPackEndTs = firstMsgPack.get(0).getMetaDataTs() + TimeUnit.SECONDS.toMillis(deduplicationInterval); |
|||
TbMsg msgWithLatestTsInFirstPack = getMsgWithLatestTs(firstMsgPack); |
|||
|
|||
List<TbMsg> secondMsgPack = getTbMsgs(deviceId, msgCount / 2, firstPackDeduplicationPackEndTs, 500); |
|||
for (TbMsg msg : secondMsgPack) { |
|||
node.onMsg(ctx, msg); |
|||
} |
|||
TbMsg msgWithLatestTsInSecondPack = getMsgWithLatestTs(secondMsgPack); |
|||
|
|||
awaitTellSelfLatch.await(); |
|||
|
|||
ArgumentCaptor<TbMsg> newMsgCaptor = ArgumentCaptor.forClass(TbMsg.class); |
|||
ArgumentCaptor<Runnable> successCaptor = ArgumentCaptor.forClass(Runnable.class); |
|||
ArgumentCaptor<Consumer<Throwable>> failureCaptor = ArgumentCaptor.forClass(Consumer.class); |
|||
|
|||
verify(ctx, times(msgCount)).ack(any()); |
|||
verify(node, times(msgCount + wantedNumberOfTellSelfInvocation)).onMsg(eq(ctx), any()); |
|||
verify(ctx, times(2)).enqueueForTellNext(newMsgCaptor.capture(), eq(TbRelationTypes.SUCCESS), successCaptor.capture(), failureCaptor.capture()); |
|||
|
|||
List<TbMsg> resultMsgs = newMsgCaptor.getAllValues(); |
|||
Assertions.assertEquals(2, resultMsgs.size()); |
|||
Assertions.assertTrue(resultMsgs.contains(msgWithLatestTsInFirstPack)); |
|||
Assertions.assertTrue(resultMsgs.contains(msgWithLatestTsInSecondPack)); |
|||
} |
|||
|
|||
private TbMsg getMsgWithLatestTs(List<TbMsg> firstMsgPack) { |
|||
int indexOfLastMsgInArray = firstMsgPack.size() - 1; |
|||
int indexToSetMaxTs = new Random().nextInt(indexOfLastMsgInArray) + 1; |
|||
TbMsg currentMaxTsMsg = firstMsgPack.get(indexOfLastMsgInArray); |
|||
TbMsg newLastMsgOfArray = firstMsgPack.get(indexToSetMaxTs); |
|||
firstMsgPack.set(indexOfLastMsgInArray, newLastMsgOfArray); |
|||
firstMsgPack.set(indexToSetMaxTs, currentMaxTsMsg); |
|||
return currentMaxTsMsg; |
|||
} |
|||
|
|||
private List<TbMsg> getTbMsgs(DeviceId deviceId, int msgCount, long currentTimeMillis, int initTsStep) { |
|||
List<TbMsg> inputMsgs = new ArrayList<>(); |
|||
var ts = currentTimeMillis + initTsStep; |
|||
for (int i = 0; i < msgCount; i++) { |
|||
inputMsgs.add(createMsg(deviceId, ts)); |
|||
ts += 2; |
|||
} |
|||
return inputMsgs; |
|||
} |
|||
|
|||
private TbMsg createMsg(DeviceId deviceId, long ts) { |
|||
ObjectNode dataNode = JacksonUtil.newObjectNode(); |
|||
dataNode.put("deviceId", deviceId.getId().toString()); |
|||
TbMsgMetaData metaData = new TbMsgMetaData(); |
|||
metaData.putValue("ts", String.valueOf(ts)); |
|||
return TbMsg.newMsg( |
|||
MAIN_QUEUE_NAME, |
|||
SessionMsgType.POST_TELEMETRY_REQUEST.name(), |
|||
deviceId, |
|||
metaData, |
|||
JacksonUtil.toString(dataNode)); |
|||
} |
|||
|
|||
private String getMergedData(List<TbMsg> msgs) { |
|||
ArrayNode mergedData = JacksonUtil.OBJECT_MAPPER.createArrayNode(); |
|||
msgs.forEach(msg -> { |
|||
ObjectNode msgNode = JacksonUtil.newObjectNode(); |
|||
msgNode.set("msg", JacksonUtil.toJsonNode(msg.getData())); |
|||
msgNode.set("metadata", JacksonUtil.valueToTree(msg.getMetaData().getData())); |
|||
mergedData.add(msgNode); |
|||
}); |
|||
return JacksonUtil.toString(mergedData); |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,43 @@ |
|||
<!-- |
|||
|
|||
Copyright © 2016-2022 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. |
|||
|
|||
--> |
|||
<div style="min-width: 800px;"> |
|||
<mat-toolbar color="primary"> |
|||
<h2><mat-icon>info_outline</mat-icon> |
|||
{{ 'edge.install-connect-instructions' | translate }}</h2> |
|||
<span fxFlex></span> |
|||
<button mat-button mat-icon-button |
|||
(click)="cancel()" |
|||
type="button"> |
|||
<mat-icon class="material-icons">close</mat-icon> |
|||
</button> |
|||
</mat-toolbar> |
|||
<mat-progress-bar color="warn" mode="indeterminate" *ngIf="isLoading$ | async"> |
|||
</mat-progress-bar> |
|||
<div style="height: 4px;" *ngIf="!(isLoading$ | async)"></div> |
|||
<div mat-dialog-content> |
|||
<tb-markdown [data]="instructions" lineNumbers fallbackToPlainMarkdown></tb-markdown> |
|||
</div> |
|||
<div mat-dialog-actions fxLayout="row" fxLayoutAlign="end center"> |
|||
<button mat-button color="primary" |
|||
type="button" |
|||
[disabled]="(isLoading$ | async)" |
|||
(click)="cancel()" cdkFocusInitial> |
|||
{{ 'action.close' | translate }} |
|||
</button> |
|||
</div> |
|||
</div> |
|||
@ -0,0 +1,46 @@ |
|||
///
|
|||
/// Copyright © 2016-2022 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.
|
|||
///
|
|||
|
|||
import { Component, Inject } from '@angular/core'; |
|||
import { MAT_DIALOG_DATA, MatDialogRef } from "@angular/material/dialog"; |
|||
import { DialogComponent } from "@shared/components/dialog.component"; |
|||
import { Store } from "@ngrx/store"; |
|||
import { AppState } from "@core/core.state"; |
|||
import { Router } from "@angular/router"; |
|||
|
|||
export interface EdgeInstructionsData { |
|||
instructions: string; |
|||
} |
|||
|
|||
@Component({ |
|||
selector: 'tb-edge-instructions', |
|||
templateUrl: './edge-instructions-dialog.component.html' |
|||
}) |
|||
export class EdgeInstructionsDialogComponent extends DialogComponent<EdgeInstructionsDialogComponent, EdgeInstructionsData> { |
|||
|
|||
instructions: string = this.data.instructions; |
|||
|
|||
constructor(protected store: Store<AppState>, |
|||
protected router: Router, |
|||
public dialogRef: MatDialogRef<EdgeInstructionsDialogComponent, EdgeInstructionsData>, |
|||
@Inject(MAT_DIALOG_DATA) public data: EdgeInstructionsData) { |
|||
super(store, router, dialogRef); |
|||
} |
|||
|
|||
cancel(): void { |
|||
this.dialogRef.close(null); |
|||
} |
|||
} |
|||
Loading…
Reference in new issue