nickAS21
5 months ago
2903 changed files with 126206 additions and 54841 deletions
@ -0,0 +1,66 @@ |
|||
|
|||
## Running tests in parallel with a reasonable memory usage |
|||
|
|||
```bash |
|||
export MAVEN_OPTS="-Xmx1024m" |
|||
export NODE_OPTIONS="--max_old_space_size=4096" |
|||
export SUREFIRE_JAVA_OPTS="-Xmx1200m -Xss256k -XX:+ExitOnOutOfMemoryError" |
|||
|
|||
# Compile and install all modules, skip packaging artifacts not needed for tests |
|||
mvn clean install -T6 -DskipTests -Dpkg.skip=true |
|||
|
|||
mvn test -pl='!application,!dao,!ui-ngx,!msa/js-executor,!msa/web-ui' -T4 |
|||
mvn test -pl dao -Dparallel=packages -DforkCount=4 |
|||
|
|||
mvn test -pl application -Dtest='!**/nosql/**,org.thingsboard.server.controller.**' -DforkCount=6 -Dparallel=classes -Dsurefire.rerunFailingTestsCount=2 -Dsurefire.failOnFlakeCount=5 |
|||
mvn test -pl application -Dtest='!**/nosql/**,org.thingsboard.server.edge.**' -DforkCount=4 -Dparallel=packages -Dsurefire.rerunFailingTestsCount=2 -Dsurefire.failOnFlakeCount=5 |
|||
mvn test -pl application -Dtest='!**/nosql/**,org.thingsboard.server.service.**' -DforkCount=6 -Dparallel=packages -Dsurefire.rerunFailingTestsCount=2 -Dsurefire.failOnFlakeCount=5 |
|||
mvn test -pl application -Dtest='!**/nosql/**,org.thingsboard.server.transport.mqtt.**' -DforkCount=6 -Dparallel=classes -Dsurefire.rerunFailingTestsCount=2 -Dsurefire.failOnFlakeCount=5 |
|||
mvn test -pl application -Dtest='!**/nosql/**,org.thingsboard.server.transport.coap.**' -DforkCount=6 -Dparallel=classes -Dsurefire.rerunFailingTestsCount=2 -Dsurefire.failOnFlakeCount=5 |
|||
mvn test -pl application -Dtest='!**/nosql/**,org.thingsboard.server.transport.lwm2m.**' -DforkCount=6 -Dparallel=packages -Dsurefire.rerunFailingTestsCount=2 -Dsurefire.failOnFlakeCount=5 |
|||
mvn test -pl application -Dtest='**/*TestSuite.java' -DforkCount=4 -Dparallel=classes -Dsurefire.rerunFailingTestsCount=2 -Dsurefire.failOnFlakeCount=5 |
|||
|
|||
#the rest of application tests |
|||
mvn test -pl application -Dtest=' |
|||
!**/nosql/**, |
|||
!org.thingsboard.server.controller.**, |
|||
!org.thingsboard.server.edge.**, |
|||
!org.thingsboard.server.service.**, |
|||
!org.thingsboard.server.transport.mqtt.**, |
|||
!org.thingsboard.server.transport.coap.**, |
|||
!org.thingsboard.server.transport.lwm2m.**, |
|||
!**/*TestSuite.java |
|||
' -DforkCount=6 -Dparallel=packages -Dsurefire.rerunFailingTestsCount=2 -Dsurefire.failOnFlakeCount=5 |
|||
``` |
|||
|
|||
## pkg.skip.* flags reference |
|||
|
|||
Use `-Dpkg.skip=true` to skip all packaging at once (equivalent to all four flags below). |
|||
|
|||
| Flag | Skips | Safe to skip for tests? | |
|||
|----------------------------|-------------------------------------------|--------------------------------------------------------------| |
|||
| `-Dpkg.skip=true` | All of the below (bootjar + deb + rpm + zip) | Yes | |
|||
| `-Dpkg.skip.bootjar=true` | `spring-boot:repackage` (`*-boot.jar`) | Yes — tests use the regular `.jar`, not the fat boot jar | |
|||
| `-Dpkg.skip.deb=true` | Gradle `buildDeb` + Maven `attach-artifact` | Yes — MSA docker modules copy the DEB from `target/` directly | |
|||
| `-Dpkg.skip.rpm=true` | Gradle `buildRpm` | Yes — no test depends on the RPM | |
|||
| `-Dpkg.skip.zip=true` | `maven-assembly-plugin` Windows ZIP | Yes — no test depends on the ZIP | |
|||
|
|||
## Testcontainers compatibility with the Docker API workaround |
|||
|
|||
In case your tests failed to run testcontainers due to unsupported Docker API version |
|||
|
|||
:coffee: testcontainers (Docker API 1.32) + :whale: docker 29 (min API 1.44) workaround |
|||
|
|||
Add to /etc/docker/daemon.json and restart docker |
|||
```json |
|||
{ |
|||
"min-api-version": "1.32" |
|||
} |
|||
``` |
|||
|
|||
Same works on Mac, except `daemon.json` are located in another folder and required to be edited from Docker Desktop UI. |
|||
|
|||
Tip: If your testcontainer are struggling to find any Docker. You can try to remove the testcontainers property file. It will be recreated on the next testcontainers run. |
|||
```bash |
|||
rm ~/.testcontainers.properties |
|||
``` |
|||
File diff suppressed because one or more lines are too long
File diff suppressed because one or more lines are too long
@ -0,0 +1,25 @@ |
|||
-- |
|||
-- Copyright © 2016-2026 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. |
|||
-- |
|||
|
|||
-- LTS cumulative schema update file. |
|||
-- All statements must be idempotent (use IF NOT EXISTS, ADD COLUMN IF NOT EXISTS, DO $$ ... END $$ guards, etc.). |
|||
-- This file is executed by SystemPatchApplier on every version increase within the LTS family. |
|||
|
|||
-- CALCULATED FIELD ADDITIONAL INFO ADDITION START |
|||
|
|||
ALTER TABLE calculated_field ADD COLUMN IF NOT EXISTS additional_info varchar; |
|||
|
|||
-- CALCULATED FIELD ADDITIONAL INFO ADDITION END |
|||
@ -0,0 +1,64 @@ |
|||
/** |
|||
* Copyright © 2016-2026 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.config; |
|||
|
|||
import lombok.RequiredArgsConstructor; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.springframework.security.config.annotation.web.configurers.HeadersConfigurer; |
|||
import org.springframework.security.web.header.writers.StaticHeadersWriter; |
|||
import org.springframework.stereotype.Component; |
|||
import org.springframework.util.StringUtils; |
|||
|
|||
@Slf4j |
|||
@Component |
|||
@RequiredArgsConstructor |
|||
public class HttpSecurityHeadersCustomizer { |
|||
|
|||
private final HttpSecurityHeadersProperties properties; |
|||
|
|||
public void customize(HeadersConfigurer<?> headers) { |
|||
if (properties.getXContentTypeOptions().isEnabled()) { |
|||
headers.contentTypeOptions(config -> {}); |
|||
} |
|||
|
|||
if (properties.getReferrerPolicy().isEnabled()) { |
|||
headers.addHeaderWriter(new StaticHeadersWriter("Referrer-Policy", properties.getReferrerPolicy().getValue())); |
|||
} |
|||
|
|||
if (properties.getXFrameOptions().isEnabled()) { |
|||
String value = properties.getXFrameOptions().getValue(); |
|||
if ("DENY".equalsIgnoreCase(value)) { |
|||
headers.frameOptions(HeadersConfigurer.FrameOptionsConfig::deny); |
|||
} else { |
|||
if (!"SAMEORIGIN".equalsIgnoreCase(value)) { |
|||
log.warn("Unrecognized X-Frame-Options value '{}', falling back to SAMEORIGIN. Valid values: DENY, SAMEORIGIN", value); |
|||
} |
|||
headers.frameOptions(HeadersConfigurer.FrameOptionsConfig::sameOrigin); |
|||
} |
|||
} |
|||
|
|||
if (properties.getContentSecurityPolicy().isEnabled() && StringUtils.hasText(properties.getContentSecurityPolicy().getValue())) { |
|||
headers.contentSecurityPolicy(csp -> { |
|||
csp.policyDirectives(properties.getContentSecurityPolicy().getValue()); |
|||
if (properties.getContentSecurityPolicy().isReportOnly()) { |
|||
csp.reportOnly(); |
|||
} |
|||
}); |
|||
} |
|||
|
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,56 @@ |
|||
/** |
|||
* Copyright © 2016-2026 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.config; |
|||
|
|||
import lombok.Data; |
|||
import org.springframework.boot.context.properties.ConfigurationProperties; |
|||
import org.springframework.stereotype.Component; |
|||
|
|||
@Component |
|||
@ConfigurationProperties(prefix = "security.headers") |
|||
@Data |
|||
public class HttpSecurityHeadersProperties { |
|||
|
|||
private XContentTypeOptions xContentTypeOptions = new XContentTypeOptions(); |
|||
private ReferrerPolicy referrerPolicy = new ReferrerPolicy(); |
|||
private XFrameOptions xFrameOptions = new XFrameOptions(); |
|||
private ContentSecurityPolicy contentSecurityPolicy = new ContentSecurityPolicy(); |
|||
|
|||
@Data |
|||
public static class XContentTypeOptions { |
|||
private boolean enabled = true; |
|||
} |
|||
|
|||
@Data |
|||
public static class ReferrerPolicy { |
|||
private boolean enabled = true; |
|||
private String value = "strict-origin-when-cross-origin"; |
|||
} |
|||
|
|||
@Data |
|||
public static class XFrameOptions { |
|||
private boolean enabled = false; |
|||
private String value = "SAMEORIGIN"; |
|||
} |
|||
|
|||
@Data |
|||
public static class ContentSecurityPolicy { |
|||
private boolean enabled = false; |
|||
private String value = ""; |
|||
private boolean reportOnly = false; |
|||
} |
|||
|
|||
} |
|||
File diff suppressed because it is too large
@ -0,0 +1,51 @@ |
|||
/** |
|||
* Copyright © 2016-2026 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.config; |
|||
|
|||
import org.springframework.beans.factory.annotation.Value; |
|||
import org.springframework.stereotype.Component; |
|||
import org.thingsboard.rule.engine.api.TbHttpClientSettings; |
|||
import org.thingsboard.server.queue.util.TbRuleEngineComponent; |
|||
|
|||
@TbRuleEngineComponent |
|||
@Component |
|||
public class TbHttpClientSettingsComponent implements TbHttpClientSettings { |
|||
|
|||
@Value("${actors.rule.external.http_client.max_parallel_requests:0}") |
|||
private int maxParallelRequests; |
|||
|
|||
@Value("${actors.rule.external.http_client.max_pending_requests:0}") |
|||
private int maxPendingRequests; |
|||
|
|||
@Value("${actors.rule.external.http_client.pool_max_connections:0}") |
|||
private int poolMaxConnections; |
|||
|
|||
@Override |
|||
public int getMaxParallelRequests() { |
|||
return maxParallelRequests; |
|||
} |
|||
|
|||
@Override |
|||
public int getMaxPendingRequests() { |
|||
return maxPendingRequests; |
|||
} |
|||
|
|||
@Override |
|||
public int getPoolMaxConnections() { |
|||
return poolMaxConnections; |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,259 @@ |
|||
/** |
|||
* Copyright © 2016-2026 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.controller; |
|||
|
|||
import com.fasterxml.jackson.databind.JsonNode; |
|||
import io.swagger.v3.oas.annotations.Parameter; |
|||
import io.swagger.v3.oas.annotations.media.Schema; |
|||
import lombok.RequiredArgsConstructor; |
|||
import org.springframework.http.HttpStatus; |
|||
import org.springframework.security.access.prepost.PreAuthorize; |
|||
import org.springframework.web.bind.annotation.DeleteMapping; |
|||
import org.springframework.web.bind.annotation.GetMapping; |
|||
import org.springframework.web.bind.annotation.PathVariable; |
|||
import org.springframework.web.bind.annotation.PostMapping; |
|||
import org.springframework.web.bind.annotation.RequestBody; |
|||
import org.springframework.web.bind.annotation.RequestMapping; |
|||
import org.springframework.web.bind.annotation.RequestParam; |
|||
import org.springframework.web.bind.annotation.ResponseStatus; |
|||
import org.springframework.web.bind.annotation.RestController; |
|||
import org.thingsboard.server.common.data.EntityType; |
|||
import org.thingsboard.server.common.data.EventInfo; |
|||
import org.thingsboard.server.common.data.cf.AlarmRuleDefinition; |
|||
import org.thingsboard.server.common.data.exception.ThingsboardErrorCode; |
|||
import org.thingsboard.server.common.data.cf.AlarmRuleDefinitionInfo; |
|||
import org.thingsboard.server.common.data.cf.CalculatedField; |
|||
import org.thingsboard.server.common.data.cf.CalculatedFieldFilter; |
|||
import org.thingsboard.server.common.data.cf.CalculatedFieldInfo; |
|||
import org.thingsboard.server.common.data.cf.CalculatedFieldType; |
|||
import org.thingsboard.server.common.data.event.EventType; |
|||
import org.thingsboard.server.common.data.exception.ThingsboardException; |
|||
import org.thingsboard.server.common.data.id.CalculatedFieldId; |
|||
import org.thingsboard.server.common.data.id.EntityId; |
|||
import org.thingsboard.server.common.data.id.EntityIdFactory; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
import org.thingsboard.server.common.data.page.PageData; |
|||
import org.thingsboard.server.common.data.page.PageLink; |
|||
import org.thingsboard.server.config.annotations.ApiOperation; |
|||
import org.thingsboard.server.dao.event.EventService; |
|||
import org.thingsboard.server.queue.util.TbCoreComponent; |
|||
import org.thingsboard.server.service.entitiy.cf.TbCalculatedFieldService; |
|||
import org.thingsboard.server.service.security.model.SecurityUser; |
|||
import org.thingsboard.server.service.security.permission.Operation; |
|||
|
|||
import java.util.EnumSet; |
|||
import java.util.Map; |
|||
import java.util.Optional; |
|||
import java.util.Set; |
|||
import java.util.UUID; |
|||
import java.util.stream.Collectors; |
|||
|
|||
import static org.thingsboard.server.controller.ControllerConstants.CF_TEXT_SEARCH_DESCRIPTION; |
|||
import static org.thingsboard.server.controller.ControllerConstants.ENTITY_ID_PARAM_DESCRIPTION; |
|||
import static org.thingsboard.server.controller.ControllerConstants.ENTITY_TYPE_PARAM_DESCRIPTION; |
|||
import static org.thingsboard.server.controller.ControllerConstants.MARKDOWN_CODE_BLOCK_END; |
|||
import static org.thingsboard.server.controller.ControllerConstants.MARKDOWN_CODE_BLOCK_START; |
|||
import static org.thingsboard.server.controller.ControllerConstants.PAGE_NUMBER_DESCRIPTION; |
|||
import static org.thingsboard.server.controller.ControllerConstants.PAGE_SIZE_DESCRIPTION; |
|||
import static org.thingsboard.server.controller.ControllerConstants.SORT_ORDER_DESCRIPTION; |
|||
import static org.thingsboard.server.controller.ControllerConstants.SORT_PROPERTY_DESCRIPTION; |
|||
import static org.thingsboard.server.controller.ControllerConstants.TENANT_AUTHORITY_PARAGRAPH; |
|||
import static org.thingsboard.server.controller.ControllerConstants.UUID_WIKI_LINK; |
|||
|
|||
@RestController |
|||
@TbCoreComponent |
|||
@RequestMapping("/api") |
|||
@RequiredArgsConstructor |
|||
public class AlarmRuleController extends BaseController { |
|||
|
|||
private final TbCalculatedFieldService tbCalculatedFieldService; |
|||
private final EventService eventService; |
|||
|
|||
public static final String ALARM_RULE_ID = "alarmRuleId"; |
|||
|
|||
private static final String TEST_SCRIPT_EXPRESSION = |
|||
"Execute the alarm rule TBEL condition expression and return the result. " + |
|||
"Alarm rule expressions must return a boolean value. The format of request: \n\n" |
|||
+ MARKDOWN_CODE_BLOCK_START |
|||
+ "{\n" + |
|||
" \"expression\": \"return temperature > 50;\",\n" + |
|||
" \"arguments\": {\n" + |
|||
" \"temperature\": { \"type\": \"SINGLE_VALUE\", \"ts\": 1739776478057, \"value\": 55 }\n" + |
|||
" }\n" + |
|||
"}" |
|||
+ MARKDOWN_CODE_BLOCK_END |
|||
+ "\n\n Expected result JSON contains \"output\" and \"error\"."; |
|||
|
|||
@ApiOperation(value = "Create Or Update Alarm Rule (saveAlarmRule)", |
|||
notes = "Creates or Updates the Alarm Rule. When creating alarm rule, platform generates Alarm Rule Id as " + UUID_WIKI_LINK + |
|||
"The newly created Alarm Rule Id will be present in the response. " + |
|||
"Specify existing Alarm Rule Id to update the alarm rule. " + |
|||
"Referencing non-existing Alarm Rule Id will cause 'Not Found' error. " + |
|||
"Remove 'id', 'tenantId' from the request body example (below) to create new Alarm Rule entity. " |
|||
+ TENANT_AUTHORITY_PARAGRAPH) |
|||
@PreAuthorize("hasAuthority('TENANT_ADMIN')") |
|||
@PostMapping("/alarm/rule") |
|||
public AlarmRuleDefinition saveAlarmRule(@io.swagger.v3.oas.annotations.parameters.RequestBody(description = "A JSON value representing the alarm rule.") |
|||
@RequestBody AlarmRuleDefinition alarmRuleDefinition) throws Exception { |
|||
alarmRuleDefinition.setTenantId(getTenantId()); |
|||
checkEntityId(alarmRuleDefinition.getEntityId(), Operation.WRITE_CALCULATED_FIELD); |
|||
if (alarmRuleDefinition.getId() != null) { |
|||
checkAlarmRule(alarmRuleDefinition.getId()); |
|||
} |
|||
CalculatedField calculatedField = alarmRuleDefinition.toCalculatedField(); |
|||
checkReferencedEntities(calculatedField.getConfiguration()); |
|||
CalculatedField saved = tbCalculatedFieldService.save(calculatedField, getCurrentUser()); |
|||
return AlarmRuleDefinition.fromCalculatedField(saved); |
|||
} |
|||
|
|||
@ApiOperation(value = "Get Alarm Rule (getAlarmRuleById)", |
|||
notes = "Fetch the Alarm Rule object based on the provided Alarm Rule Id." + TENANT_AUTHORITY_PARAGRAPH) |
|||
@PreAuthorize("hasAuthority('TENANT_ADMIN')") |
|||
@GetMapping("/alarm/rule/{alarmRuleId}") |
|||
public AlarmRuleDefinition getAlarmRuleById(@Parameter @PathVariable(ALARM_RULE_ID) String strAlarmRuleId) throws ThingsboardException { |
|||
checkParameter(ALARM_RULE_ID, strAlarmRuleId); |
|||
CalculatedFieldId calculatedFieldId = new CalculatedFieldId(toUUID(strAlarmRuleId)); |
|||
CalculatedField calculatedField = checkAlarmRule(calculatedFieldId); |
|||
checkEntityId(calculatedField.getEntityId(), Operation.READ_CALCULATED_FIELD); |
|||
return AlarmRuleDefinition.fromCalculatedField(calculatedField); |
|||
} |
|||
|
|||
@ApiOperation(value = "Get Alarm Rules by Entity Id (getAlarmRulesByEntityId)", |
|||
notes = "Fetch the Alarm Rules based on the provided Entity Id." + TENANT_AUTHORITY_PARAGRAPH) |
|||
@PreAuthorize("hasAuthority('TENANT_ADMIN')") |
|||
@GetMapping(value = "/alarm/rules/{entityType}/{entityId}") |
|||
public PageData<AlarmRuleDefinition> getAlarmRulesByEntityId( |
|||
@Parameter(description = ENTITY_TYPE_PARAM_DESCRIPTION, required = true, schema = @Schema(defaultValue = "DEVICE")) @PathVariable("entityType") String entityType, |
|||
@Parameter(description = ENTITY_ID_PARAM_DESCRIPTION, required = true) @PathVariable("entityId") String entityIdStr, |
|||
@Parameter(description = PAGE_SIZE_DESCRIPTION, required = true) @RequestParam int pageSize, |
|||
@Parameter(description = PAGE_NUMBER_DESCRIPTION, required = true) @RequestParam int page, |
|||
@Parameter(description = CF_TEXT_SEARCH_DESCRIPTION) @RequestParam(required = false) String textSearch, |
|||
@Parameter(description = SORT_PROPERTY_DESCRIPTION, schema = @Schema(allowableValues = {"createdTime", "name"})) @RequestParam(required = false) String sortProperty, |
|||
@Parameter(description = SORT_ORDER_DESCRIPTION, schema = @Schema(allowableValues = {"ASC", "DESC"})) @RequestParam(required = false) String sortOrder) throws ThingsboardException { |
|||
PageLink pageLink = createPageLink(pageSize, page, textSearch, sortProperty, sortOrder); |
|||
checkParameter("entityId", entityIdStr); |
|||
EntityId entityId = EntityIdFactory.getByTypeAndUuid(entityType, entityIdStr); |
|||
checkEntityId(entityId, Operation.READ_CALCULATED_FIELD); |
|||
PageData<CalculatedField> result = checkNotNull(tbCalculatedFieldService.findByTenantIdAndEntityId(getTenantId(), entityId, CalculatedFieldType.ALARM, pageLink)); |
|||
return result.mapData(AlarmRuleDefinition::fromCalculatedField); |
|||
} |
|||
|
|||
@ApiOperation(value = "Get alarm rules (getAlarmRules)", |
|||
notes = "Fetch tenant alarm rules based on the filter." + TENANT_AUTHORITY_PARAGRAPH) |
|||
@PreAuthorize("hasAuthority('TENANT_ADMIN')") |
|||
@GetMapping(value = "/alarm/rules") |
|||
public PageData<AlarmRuleDefinitionInfo> getAlarmRules(@Parameter(description = PAGE_SIZE_DESCRIPTION, required = true) |
|||
@RequestParam int pageSize, |
|||
@Parameter(description = PAGE_NUMBER_DESCRIPTION, required = true) |
|||
@RequestParam int page, |
|||
@Parameter(description = "Entity type filter. If not specified, alarm rules for all supported entity types will be returned.") |
|||
@RequestParam(required = false) EntityType entityType, |
|||
@Parameter(description = "Entities filter. If not specified, alarm rules for entity type filter will be returned.") |
|||
@RequestParam(required = false) Set<UUID> entities, |
|||
@Parameter(description = CF_TEXT_SEARCH_DESCRIPTION) |
|||
@RequestParam(required = false) String textSearch, |
|||
@Parameter(description = SORT_PROPERTY_DESCRIPTION, schema = @Schema(allowableValues = {"createdTime", "name"})) |
|||
@RequestParam(required = false) String sortProperty, |
|||
@Parameter(description = SORT_ORDER_DESCRIPTION, schema = @Schema(allowableValues = {"ASC", "DESC"})) |
|||
@RequestParam(required = false) String sortOrder) throws ThingsboardException { |
|||
PageLink pageLink = createPageLink(pageSize, page, textSearch, sortProperty, sortOrder); |
|||
SecurityUser user = getCurrentUser(); |
|||
|
|||
Set<EntityType> entityTypes; |
|||
if (entityType == null) { |
|||
entityTypes = CalculatedField.SUPPORTED_ENTITIES.entrySet().stream() |
|||
.filter(entry -> entry.getValue().contains(CalculatedFieldType.ALARM)) |
|||
.map(Map.Entry::getKey) |
|||
.collect(Collectors.toSet()); |
|||
} else { |
|||
entityTypes = EnumSet.of(entityType); |
|||
} |
|||
|
|||
CalculatedFieldFilter filter = CalculatedFieldFilter.builder() |
|||
.types(EnumSet.of(CalculatedFieldType.ALARM)) |
|||
.entityTypes(entityTypes) |
|||
.entityIds(entities) |
|||
.build(); |
|||
PageData<CalculatedFieldInfo> result = calculatedFieldService.findCalculatedFieldsByTenantIdAndFilter(user.getTenantId(), filter, pageLink); |
|||
return result.mapData(AlarmRuleDefinitionInfo::fromCalculatedFieldInfo); |
|||
} |
|||
|
|||
@ApiOperation(value = "Get alarm rule names (getAlarmRuleNames)", |
|||
notes = "Fetch the list of alarm rule names." + TENANT_AUTHORITY_PARAGRAPH) |
|||
@PreAuthorize("hasAuthority('TENANT_ADMIN')") |
|||
@GetMapping(value = "/alarm/rules/names") |
|||
public PageData<String> getAlarmRuleNames(@Parameter(description = PAGE_SIZE_DESCRIPTION, required = true) |
|||
@RequestParam int pageSize, |
|||
@Parameter(description = PAGE_NUMBER_DESCRIPTION, required = true) |
|||
@RequestParam int page, |
|||
@Parameter(description = CF_TEXT_SEARCH_DESCRIPTION) |
|||
@RequestParam(required = false) String textSearch, |
|||
@Parameter(description = SORT_ORDER_DESCRIPTION, schema = @Schema(allowableValues = {"ASC", "DESC"})) |
|||
@RequestParam(required = false) String sortOrder) throws ThingsboardException { |
|||
PageLink pageLink = createPageLink(pageSize, page, textSearch, "name", sortOrder); |
|||
return calculatedFieldService.findCalculatedFieldNamesByTenantIdAndType(getTenantId(), CalculatedFieldType.ALARM, pageLink); |
|||
} |
|||
|
|||
@ApiOperation(value = "Delete Alarm Rule (deleteAlarmRule)", |
|||
notes = "Deletes the alarm rule. Referencing non-existing Alarm Rule Id will cause an error." + TENANT_AUTHORITY_PARAGRAPH) |
|||
@PreAuthorize("hasAuthority('TENANT_ADMIN')") |
|||
@DeleteMapping("/alarm/rule/{alarmRuleId}") |
|||
@ResponseStatus(HttpStatus.OK) |
|||
public void deleteAlarmRule(@PathVariable(ALARM_RULE_ID) String strAlarmRuleId) throws Exception { |
|||
checkParameter(ALARM_RULE_ID, strAlarmRuleId); |
|||
CalculatedFieldId calculatedFieldId = new CalculatedFieldId(toUUID(strAlarmRuleId)); |
|||
CalculatedField calculatedField = checkAlarmRule(calculatedFieldId); |
|||
checkEntityId(calculatedField.getEntityId(), Operation.WRITE_CALCULATED_FIELD); |
|||
tbCalculatedFieldService.delete(calculatedField, getCurrentUser()); |
|||
} |
|||
|
|||
@ApiOperation(value = "Get latest alarm rule debug event (getLatestAlarmRuleDebugEvent)", |
|||
notes = "Gets latest alarm rule debug event for specified alarm rule id. " + |
|||
"Referencing non-existing alarm rule id will cause an error. " + TENANT_AUTHORITY_PARAGRAPH) |
|||
@PreAuthorize("hasAuthority('TENANT_ADMIN')") |
|||
@GetMapping("/alarm/rule/{alarmRuleId}/debug") |
|||
public JsonNode getLatestAlarmRuleDebugEvent(@Parameter @PathVariable(ALARM_RULE_ID) String strAlarmRuleId) throws ThingsboardException { |
|||
checkParameter(ALARM_RULE_ID, strAlarmRuleId); |
|||
CalculatedFieldId calculatedFieldId = new CalculatedFieldId(toUUID(strAlarmRuleId)); |
|||
CalculatedField calculatedField = checkAlarmRule(calculatedFieldId); |
|||
checkEntityId(calculatedField.getEntityId(), Operation.READ_CALCULATED_FIELD); |
|||
TenantId tenantId = getCurrentUser().getTenantId(); |
|||
return Optional.ofNullable(eventService.findLatestEvents(tenantId, calculatedFieldId, EventType.DEBUG_CALCULATED_FIELD, 1)) |
|||
.flatMap(events -> events.stream().map(EventInfo::getBody).findFirst()) |
|||
.orElse(null); |
|||
} |
|||
|
|||
@ApiOperation(value = "Test alarm rule TBEL expression (testAlarmRuleScript)", |
|||
notes = TEST_SCRIPT_EXPRESSION + TENANT_AUTHORITY_PARAGRAPH) |
|||
@PreAuthorize("hasAuthority('TENANT_ADMIN')") |
|||
@PostMapping("/alarm/rule/testScript") |
|||
public JsonNode testAlarmRuleScript( |
|||
@io.swagger.v3.oas.annotations.parameters.RequestBody(description = "Test alarm rule TBEL condition expression. The expression must return a boolean value.") |
|||
@RequestBody JsonNode inputParams) throws ThingsboardException { |
|||
checkParameter("expression", inputParams.has("expression") ? inputParams.get("expression").asText() : null); |
|||
return tbCalculatedFieldService.executeTestScript(getTenantId(), inputParams); |
|||
} |
|||
|
|||
private CalculatedField checkAlarmRule(CalculatedFieldId calculatedFieldId) throws ThingsboardException { |
|||
CalculatedField calculatedField = tbCalculatedFieldService.findById(calculatedFieldId, getCurrentUser()); |
|||
checkNotNull(calculatedField); |
|||
if (calculatedField.getType() != CalculatedFieldType.ALARM) { |
|||
throw new ThingsboardException("Alarm rule not found", ThingsboardErrorCode.ITEM_NOT_FOUND); |
|||
} |
|||
return calculatedField; |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,85 @@ |
|||
/** |
|||
* Copyright © 2016-2026 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.rpc; |
|||
|
|||
import lombok.RequiredArgsConstructor; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; |
|||
import org.springframework.stereotype.Component; |
|||
import org.thingsboard.server.common.data.edge.EdgeEvent; |
|||
import org.thingsboard.server.common.data.edge.EdgeEventType; |
|||
import org.thingsboard.server.gen.edge.v1.DownlinkMsg; |
|||
import org.thingsboard.server.gen.edge.v1.EdgeVersion; |
|||
import org.thingsboard.server.queue.util.TbCoreComponent; |
|||
import org.thingsboard.server.service.edge.EdgeContextComponent; |
|||
import org.thingsboard.server.service.edge.EdgeMsgConstructorUtils; |
|||
import org.thingsboard.server.service.edge.rpc.utils.EdgeVersionUtils; |
|||
|
|||
import java.util.ArrayList; |
|||
import java.util.List; |
|||
|
|||
@Component |
|||
@Slf4j |
|||
@RequiredArgsConstructor |
|||
@ConditionalOnProperty(prefix = "edges", value = "enabled", havingValue = "true") |
|||
@TbCoreComponent |
|||
public class DownlinkMessageMapper { |
|||
|
|||
private final EdgeContextComponent ctx; |
|||
|
|||
public List<DownlinkMsg> convertToDownlinkMsgsPack(EdgeSessionState state, List<EdgeEvent> edgeEvents) { |
|||
List<DownlinkMsg> result = new ArrayList<>(); |
|||
List<EdgeEvent> filtered = EdgeMsgConstructorUtils.mergeAndFilterDownlinkDuplicates(edgeEvents); |
|||
for (EdgeEvent edgeEvent : filtered) { |
|||
log.trace("[{}][{}] converting edge event to downlink msg [{}]", state.getTenantId(), state.getEdgeId(), edgeEvent); |
|||
DownlinkMsg downlinkMsg = null; |
|||
try { |
|||
switch (edgeEvent.getAction()) { |
|||
case UPDATED, ADDED, DELETED, ASSIGNED_TO_EDGE, UNASSIGNED_FROM_EDGE, ALARM_ACK, ALARM_CLEAR, |
|||
ALARM_DELETE, CREDENTIALS_UPDATED, RELATION_ADD_OR_UPDATE, RELATION_DELETED, RPC_CALL, |
|||
ASSIGNED_TO_CUSTOMER, UNASSIGNED_FROM_CUSTOMER, ADDED_COMMENT, UPDATED_COMMENT, DELETED_COMMENT -> { |
|||
downlinkMsg = convertEntityEventToDownlink(state, edgeEvent); |
|||
if (downlinkMsg != null && downlinkMsg.getWidgetTypeUpdateMsgCount() > 0) { |
|||
log.trace("[{}][{}] widgetTypeUpdateMsg message processed, downlinkMsgId = {}", |
|||
state.getTenantId(), state.getEdgeId(), downlinkMsg.getDownlinkMsgId()); |
|||
} else { |
|||
log.trace("[{}][{}] entity message processed [{}]", state.getTenantId(), state.getEdgeId(), downlinkMsg); |
|||
} |
|||
} |
|||
case ATTRIBUTES_UPDATED, POST_ATTRIBUTES, ATTRIBUTES_DELETED, TIMESERIES_UPDATED -> |
|||
downlinkMsg = ctx.getTelemetryProcessor().convertTelemetryEventToDownlink(state.getEdge(), edgeEvent); |
|||
default -> log.warn("[{}][{}] Unsupported action type [{}]", state.getTenantId(), state.getEdgeId(), edgeEvent.getAction()); |
|||
} |
|||
} catch (Exception e) { |
|||
log.trace("[{}][{}] Exception during converting edge event to downlink msg", state.getTenantId(), state.getEdgeId(), e); |
|||
} |
|||
if (downlinkMsg != null) { |
|||
result.add(downlinkMsg); |
|||
} |
|||
} |
|||
return result; |
|||
} |
|||
|
|||
protected DownlinkMsg convertEntityEventToDownlink(EdgeSessionState state, EdgeEvent edgeEvent) { |
|||
log.trace("[{}] Executing convertEntityEventToDownlink, edgeEvent [{}], action [{}]", edgeEvent.getTenantId(), edgeEvent, edgeEvent.getAction()); |
|||
if ((EdgeEventType.OAUTH2_CLIENT.equals(edgeEvent.getType()) || EdgeEventType.DOMAIN.equals(edgeEvent.getType())) && |
|||
(EdgeVersionUtils.isEdgeVersionOlderThan(state.getEdgeVersion(), EdgeVersion.V_3_8_0))) { |
|||
return null; |
|||
} |
|||
|
|||
return ctx.getProcessor(edgeEvent.getType()).convertEdgeEventToDownlink(edgeEvent, state.getEdgeVersion()); |
|||
} |
|||
} |
|||
@ -1,711 +0,0 @@ |
|||
/** |
|||
* Copyright © 2016-2026 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.rpc; |
|||
|
|||
import com.fasterxml.jackson.databind.node.ObjectNode; |
|||
import com.google.common.util.concurrent.FutureCallback; |
|||
import com.google.common.util.concurrent.Futures; |
|||
import io.grpc.Server; |
|||
import io.grpc.netty.shaded.io.grpc.netty.NettyServerBuilder; |
|||
import io.grpc.stub.StreamObserver; |
|||
import jakarta.annotation.Nullable; |
|||
import jakarta.annotation.PreDestroy; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.springframework.beans.factory.annotation.Autowired; |
|||
import org.springframework.beans.factory.annotation.Value; |
|||
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; |
|||
import org.springframework.context.annotation.Lazy; |
|||
import org.springframework.stereotype.Service; |
|||
import org.thingsboard.common.util.JacksonUtil; |
|||
import org.thingsboard.common.util.ThingsBoardExecutors; |
|||
import org.thingsboard.rule.engine.api.AttributesSaveRequest; |
|||
import org.thingsboard.rule.engine.api.TimeseriesSaveRequest; |
|||
import org.thingsboard.server.cache.TbTransactionalCache; |
|||
import org.thingsboard.server.cluster.TbClusterService; |
|||
import org.thingsboard.server.common.data.AttributeScope; |
|||
import org.thingsboard.server.common.data.DataConstants; |
|||
import org.thingsboard.server.common.data.ResourceUtils; |
|||
import org.thingsboard.server.common.data.edge.Edge; |
|||
import org.thingsboard.server.common.data.edge.EdgeEvent; |
|||
import org.thingsboard.server.common.data.id.EdgeId; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
import org.thingsboard.server.common.data.kv.BooleanDataEntry; |
|||
import org.thingsboard.server.common.data.kv.LongDataEntry; |
|||
import org.thingsboard.server.common.data.msg.TbMsgType; |
|||
import org.thingsboard.server.common.data.notification.rule.trigger.EdgeConnectionTrigger; |
|||
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.common.msg.edge.EdgeEventUpdateMsg; |
|||
import org.thingsboard.server.common.msg.edge.EdgeHighPriorityMsg; |
|||
import org.thingsboard.server.common.msg.edge.EdgeSessionMsg; |
|||
import org.thingsboard.server.common.msg.edge.FromEdgeSyncResponse; |
|||
import org.thingsboard.server.common.msg.edge.ToEdgeSyncRequest; |
|||
import org.thingsboard.server.gen.edge.v1.EdgeRpcServiceGrpc; |
|||
import org.thingsboard.server.gen.edge.v1.RequestMsg; |
|||
import org.thingsboard.server.gen.edge.v1.ResponseMsg; |
|||
import org.thingsboard.server.queue.discovery.TbServiceInfoProvider; |
|||
import org.thingsboard.server.queue.discovery.TopicService; |
|||
import org.thingsboard.server.queue.kafka.KafkaAdmin; |
|||
import org.thingsboard.server.queue.provider.TbCoreQueueFactory; |
|||
import org.thingsboard.server.queue.util.AfterStartUp; |
|||
import org.thingsboard.server.queue.util.TbCoreComponent; |
|||
import org.thingsboard.server.service.edge.EdgeContextComponent; |
|||
import org.thingsboard.server.service.telemetry.TelemetrySubscriptionService; |
|||
|
|||
import java.io.IOException; |
|||
import java.io.InputStream; |
|||
import java.util.ArrayList; |
|||
import java.util.Collection; |
|||
import java.util.HashMap; |
|||
import java.util.List; |
|||
import java.util.Map; |
|||
import java.util.Optional; |
|||
import java.util.UUID; |
|||
import java.util.concurrent.ConcurrentHashMap; |
|||
import java.util.concurrent.ConcurrentMap; |
|||
import java.util.concurrent.ScheduledExecutorService; |
|||
import java.util.concurrent.ScheduledFuture; |
|||
import java.util.concurrent.TimeUnit; |
|||
import java.util.concurrent.locks.Lock; |
|||
import java.util.concurrent.locks.ReentrantLock; |
|||
import java.util.function.Consumer; |
|||
import java.util.function.Function; |
|||
|
|||
import static org.thingsboard.server.service.state.DefaultDeviceStateService.ACTIVITY_STATE; |
|||
import static org.thingsboard.server.service.state.DefaultDeviceStateService.LAST_CONNECT_TIME; |
|||
import static org.thingsboard.server.service.state.DefaultDeviceStateService.LAST_DISCONNECT_TIME; |
|||
|
|||
@Service |
|||
@Slf4j |
|||
@ConditionalOnProperty(prefix = "edges", value = "enabled", havingValue = "true") |
|||
@TbCoreComponent |
|||
public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase implements EdgeRpcService { |
|||
|
|||
private final ConcurrentMap<EdgeId, EdgeGrpcSession> sessions = new ConcurrentHashMap<>(); |
|||
private final ConcurrentMap<UUID, EdgeGrpcSession> sessionsById = new ConcurrentHashMap<>(); |
|||
private final ConcurrentMap<EdgeId, Lock> sessionNewEventsLocks = new ConcurrentHashMap<>(); |
|||
private final Map<EdgeId, Boolean> sessionNewEvents = new HashMap<>(); |
|||
private final ConcurrentMap<EdgeId, ScheduledFuture<?>> sessionEdgeEventChecks = new ConcurrentHashMap<>(); |
|||
private final ConcurrentMap<UUID, Consumer<FromEdgeSyncResponse>> localSyncEdgeRequests = new ConcurrentHashMap<>(); |
|||
private final ConcurrentMap<EdgeId, Boolean> edgeEventsMigrationProcessed = new ConcurrentHashMap<>(); |
|||
private final List<EdgeGrpcSession> zombieSessions = new ArrayList<>(); |
|||
|
|||
@Value("${edges.rpc.port}") |
|||
private int rpcPort; |
|||
@Value("${edges.rpc.ssl.enabled}") |
|||
private boolean sslEnabled; |
|||
@Value("${edges.rpc.ssl.cert}") |
|||
private String certFileResource; |
|||
@Value("${edges.rpc.ssl.private_key}") |
|||
private String privateKeyResource; |
|||
@Value("${edges.state.persistToTelemetry:false}") |
|||
private boolean persistToTelemetry; |
|||
@Value("${edges.rpc.client_max_keep_alive_time_sec:1}") |
|||
private int clientMaxKeepAliveTimeSec; |
|||
@Value("${edges.rpc.max_inbound_message_size:4194304}") |
|||
private int maxInboundMessageSize; |
|||
@Value("${edges.rpc.keep_alive_time_sec:10}") |
|||
private int keepAliveTimeSec; |
|||
@Value("${edges.rpc.keep_alive_timeout_sec:5}") |
|||
private int keepAliveTimeoutSec; |
|||
@Value("${edges.scheduler_pool_size}") |
|||
private int schedulerPoolSize; |
|||
|
|||
@Value("${edges.send_scheduler_pool_size}") |
|||
private int sendSchedulerPoolSize; |
|||
|
|||
@Value("${edges.max_high_priority_queue_size_per_session:10000}") |
|||
private int maxHighPriorityQueueSizePerSession; |
|||
|
|||
@Autowired |
|||
@Lazy |
|||
private EdgeContextComponent ctx; |
|||
|
|||
@Autowired |
|||
private TelemetrySubscriptionService tsSubService; |
|||
|
|||
@Autowired |
|||
private TbClusterService clusterService; |
|||
|
|||
@Autowired |
|||
private TbServiceInfoProvider serviceInfoProvider; |
|||
|
|||
@Autowired |
|||
private TbTransactionalCache<EdgeId, String> edgeIdServiceIdCache; |
|||
|
|||
@Autowired |
|||
private TopicService topicService; |
|||
|
|||
@Autowired |
|||
private TbCoreQueueFactory tbCoreQueueFactory; |
|||
|
|||
@Autowired |
|||
private Optional<KafkaAdmin> kafkaAdmin; |
|||
|
|||
private Server server; |
|||
|
|||
private ScheduledExecutorService edgeEventProcessingExecutorService; |
|||
|
|||
private ScheduledExecutorService sendDownlinkExecutorService; |
|||
|
|||
private ScheduledExecutorService executorService; |
|||
|
|||
@AfterStartUp(order = AfterStartUp.REGULAR_SERVICE) |
|||
public void onStartUp() { |
|||
log.info("Initializing Edge RPC service!"); |
|||
NettyServerBuilder builder = NettyServerBuilder.forPort(rpcPort) |
|||
.permitKeepAliveTime(clientMaxKeepAliveTimeSec, TimeUnit.SECONDS) |
|||
.keepAliveTime(keepAliveTimeSec, TimeUnit.SECONDS) |
|||
.keepAliveTimeout(keepAliveTimeoutSec, TimeUnit.SECONDS) |
|||
.permitKeepAliveWithoutCalls(true) |
|||
.maxInboundMessageSize(maxInboundMessageSize) |
|||
.addService(this); |
|||
if (sslEnabled) { |
|||
try { |
|||
InputStream certFileIs = ResourceUtils.getInputStream(this, certFileResource); |
|||
InputStream privateKeyFileIs = ResourceUtils.getInputStream(this, privateKeyResource); |
|||
builder.useTransportSecurity(certFileIs, privateKeyFileIs); |
|||
} catch (Exception e) { |
|||
log.error("Unable to set up SSL context. Reason: " + e.getMessage(), e); |
|||
throw new RuntimeException("Unable to set up SSL context!", e); |
|||
} |
|||
} |
|||
server = builder.build(); |
|||
log.info("Going to start Edge RPC server using port: {}", rpcPort); |
|||
try { |
|||
server.start(); |
|||
} catch (IOException e) { |
|||
log.error("Failed to start Edge RPC server!", e); |
|||
throw new RuntimeException("Failed to start Edge RPC server!"); |
|||
} |
|||
this.edgeEventProcessingExecutorService = ThingsBoardExecutors.newScheduledThreadPool(schedulerPoolSize, "edge-event-check-scheduler"); |
|||
this.sendDownlinkExecutorService = ThingsBoardExecutors.newScheduledThreadPool(sendSchedulerPoolSize, "edge-send-scheduler"); |
|||
this.executorService = ThingsBoardExecutors.newSingleThreadScheduledExecutor("edge-service"); |
|||
this.executorService.scheduleAtFixedRate(this::cleanupZombieSessions, 60, 60, TimeUnit.SECONDS); |
|||
log.info("Edge RPC service initialized!"); |
|||
} |
|||
|
|||
@PreDestroy |
|||
public void destroy() { |
|||
if (server != null) { |
|||
server.shutdownNow(); |
|||
} |
|||
for (Map.Entry<EdgeId, ScheduledFuture<?>> entry : sessionEdgeEventChecks.entrySet()) { |
|||
EdgeId edgeId = entry.getKey(); |
|||
ScheduledFuture<?> sessionEdgeEventCheck = entry.getValue(); |
|||
if (sessionEdgeEventCheck != null && !sessionEdgeEventCheck.isCancelled() && !sessionEdgeEventCheck.isDone()) { |
|||
sessionEdgeEventCheck.cancel(true); |
|||
sessionEdgeEventChecks.remove(edgeId); |
|||
} |
|||
} |
|||
if (edgeEventProcessingExecutorService != null) { |
|||
edgeEventProcessingExecutorService.shutdownNow(); |
|||
} |
|||
if (sendDownlinkExecutorService != null) { |
|||
sendDownlinkExecutorService.shutdownNow(); |
|||
} |
|||
if (executorService != null) { |
|||
executorService.shutdownNow(); |
|||
} |
|||
} |
|||
|
|||
@Override |
|||
public StreamObserver<RequestMsg> handleMsgs(StreamObserver<ResponseMsg> outputStream) { |
|||
EdgeGrpcSession session = createEdgeGrpcSession(outputStream); |
|||
return session.getInputStream(); |
|||
} |
|||
|
|||
private EdgeGrpcSession createEdgeGrpcSession(StreamObserver<ResponseMsg> outputStream) { |
|||
return kafkaAdmin.isPresent() |
|||
? new KafkaEdgeGrpcSession(ctx, topicService, tbCoreQueueFactory, kafkaAdmin.get(), outputStream, this::onEdgeConnect, this::onEdgeDisconnect, |
|||
sendDownlinkExecutorService, maxInboundMessageSize, maxHighPriorityQueueSizePerSession) |
|||
: new PostgresEdgeGrpcSession(ctx, outputStream, this::onEdgeConnect, this::onEdgeDisconnect, |
|||
sendDownlinkExecutorService, maxInboundMessageSize, maxHighPriorityQueueSizePerSession); |
|||
} |
|||
|
|||
@Override |
|||
public void onToEdgeSessionMsg(TenantId tenantId, EdgeSessionMsg msg) { |
|||
switch (msg.getMsgType()) { |
|||
case EDGE_HIGH_PRIORITY_TO_EDGE_SESSION_MSG -> { |
|||
EdgeHighPriorityMsg edgeHighPriorityMsg = (EdgeHighPriorityMsg) msg; |
|||
log.trace("[{}] edgeEventMsg [{}]", tenantId, msg); |
|||
onEdgeHighPriorityEvent(edgeHighPriorityMsg); |
|||
} |
|||
case EDGE_EVENT_UPDATE_TO_EDGE_SESSION_MSG -> { |
|||
EdgeEventUpdateMsg edgeEventUpdateMsg = (EdgeEventUpdateMsg) msg; |
|||
log.trace("[{}] onToEdgeEventUpdateMsg [{}]", tenantId, msg); |
|||
onEdgeEventUpdate(tenantId, edgeEventUpdateMsg.getEdgeId()); |
|||
} |
|||
case EDGE_SYNC_REQUEST_TO_EDGE_SESSION_MSG -> { |
|||
ToEdgeSyncRequest toEdgeSyncRequest = (ToEdgeSyncRequest) msg; |
|||
log.trace("[{}] toEdgeSyncRequest [{}]", tenantId, msg); |
|||
startSyncProcess(tenantId, toEdgeSyncRequest.getEdgeId(), toEdgeSyncRequest.getId(), toEdgeSyncRequest.getServiceId()); |
|||
} |
|||
case EDGE_SYNC_RESPONSE_FROM_EDGE_SESSION_MSG -> { |
|||
FromEdgeSyncResponse fromEdgeSyncResponse = (FromEdgeSyncResponse) msg; |
|||
log.trace("[{}] fromEdgeSyncResponse [{}]", tenantId, msg); |
|||
processSyncResponse(fromEdgeSyncResponse); |
|||
} |
|||
} |
|||
} |
|||
|
|||
@Override |
|||
public void updateEdge(TenantId tenantId, Edge edge) { |
|||
if (edge == null) { |
|||
log.warn("[{}] Edge is null - edge is removed and outdated notification is in process!", tenantId); |
|||
return; |
|||
} |
|||
EdgeGrpcSession session = sessions.get(edge.getId()); |
|||
if (session != null && session.isConnected()) { |
|||
log.debug("[{}] Updating configuration for edge [{}] [{}]", tenantId, edge.getName(), edge.getId()); |
|||
session.onConfigurationUpdate(edge); |
|||
} else { |
|||
log.debug("[{}] Session doesn't exist for edge [{}] [{}]", tenantId, edge.getName(), edge.getId()); |
|||
} |
|||
} |
|||
|
|||
@Override |
|||
public void deleteEdge(TenantId tenantId, EdgeId edgeId) { |
|||
EdgeGrpcSession session = sessions.get(edgeId); |
|||
if (session != null && session.isConnected()) { |
|||
log.info("[{}] Closing and removing session for edge [{}]", tenantId, edgeId); |
|||
destroySession(session); |
|||
session.cleanUp(); |
|||
sessions.remove(edgeId); |
|||
sessionsById.remove(session.getSessionId()); |
|||
final Lock newEventLock = sessionNewEventsLocks.computeIfAbsent(edgeId, id -> new ReentrantLock()); |
|||
newEventLock.lock(); |
|||
try { |
|||
sessionNewEvents.remove(edgeId); |
|||
} finally { |
|||
newEventLock.unlock(); |
|||
} |
|||
cancelScheduleEdgeEventsCheck(edgeId); |
|||
} |
|||
} |
|||
|
|||
private void onEdgeEventUpdate(TenantId tenantId, EdgeId edgeId) { |
|||
EdgeGrpcSession session = sessions.get(edgeId); |
|||
if (session != null && session.isConnected()) { |
|||
log.trace("[{}] onEdgeEventUpdate [{}]", tenantId, edgeId.getId()); |
|||
updateSessionEventsFlag(tenantId, edgeId); |
|||
} |
|||
} |
|||
|
|||
private void onEdgeHighPriorityEvent(EdgeHighPriorityMsg msg) { |
|||
TenantId tenantId = msg.getTenantId(); |
|||
EdgeEvent edgeEvent = msg.getEdgeEvent(); |
|||
EdgeId edgeId = edgeEvent.getEdgeId(); |
|||
EdgeGrpcSession session = sessions.get(edgeId); |
|||
if (session != null && session.isConnected()) { |
|||
log.trace("[{}] onEdgeEvent [{}]", tenantId, edgeId); |
|||
session.addEventToHighPriorityQueue(edgeEvent); |
|||
updateSessionEventsFlag(tenantId, edgeId); |
|||
} |
|||
} |
|||
|
|||
private void updateSessionEventsFlag(TenantId tenantId, EdgeId edgeId) { |
|||
final Lock newEventLock = sessionNewEventsLocks.computeIfAbsent(edgeId, id -> new ReentrantLock()); |
|||
newEventLock.lock(); |
|||
try { |
|||
if (Boolean.FALSE.equals(sessionNewEvents.get(edgeId))) { |
|||
log.trace("[{}] set session new events flag to true [{}]", tenantId, edgeId.getId()); |
|||
sessionNewEvents.put(edgeId, true); |
|||
} |
|||
} finally { |
|||
newEventLock.unlock(); |
|||
} |
|||
} |
|||
|
|||
private void onEdgeConnect(EdgeId edgeId, EdgeGrpcSession edgeGrpcSession) { |
|||
Edge edge = edgeGrpcSession.getEdge(); |
|||
TenantId tenantId = edge.getTenantId(); |
|||
log.info("[{}][{}] edge [{}] connected successfully.", tenantId, edgeGrpcSession.getSessionId(), edgeId); |
|||
if (sessions.containsKey(edgeId)) { |
|||
EdgeGrpcSession existing = sessions.get(edgeId); |
|||
if (existing != null) { |
|||
log.info("[{}][{}] Replacing existing session [{}] for edge [{}]", tenantId, edgeGrpcSession.getSessionId(), existing.getSessionId(), edgeId); |
|||
destroySession(existing); |
|||
sessionsById.remove(existing.getSessionId()); |
|||
} |
|||
} |
|||
sessions.put(edgeId, edgeGrpcSession); |
|||
sessionsById.put(edgeGrpcSession.getSessionId(), edgeGrpcSession); |
|||
final Lock newEventLock = sessionNewEventsLocks.computeIfAbsent(edgeId, id -> new ReentrantLock()); |
|||
newEventLock.lock(); |
|||
try { |
|||
sessionNewEvents.put(edgeId, true); |
|||
} finally { |
|||
newEventLock.unlock(); |
|||
} |
|||
save(tenantId, edgeId, ACTIVITY_STATE, true); |
|||
long lastConnectTs = System.currentTimeMillis(); |
|||
save(tenantId, edgeId, LAST_CONNECT_TIME, lastConnectTs); |
|||
edgeIdServiceIdCache.put(edgeId, serviceInfoProvider.getServiceId()); |
|||
pushRuleEngineMessage(tenantId, edge, lastConnectTs, TbMsgType.CONNECT_EVENT); |
|||
cancelScheduleEdgeEventsCheck(edgeId); |
|||
edgeEventsMigrationProcessed.putIfAbsent(edgeId, Boolean.FALSE); |
|||
scheduleEdgeEventsCheck(edgeGrpcSession); |
|||
} |
|||
|
|||
private void startSyncProcess(TenantId tenantId, EdgeId edgeId, UUID requestId, String requestServiceId) { |
|||
EdgeGrpcSession session = sessions.get(edgeId); |
|||
if (session != null) { |
|||
if (session.isSyncInProgress()) { |
|||
clusterService.pushEdgeSyncResponseToCore(new FromEdgeSyncResponse(requestId, tenantId, edgeId, false, "Sync process is active at the moment"), requestServiceId); |
|||
} else { |
|||
boolean success = false; |
|||
if (session.isConnected()) { |
|||
session.startSyncProcess(true); |
|||
success = true; |
|||
} |
|||
clusterService.pushEdgeSyncResponseToCore(new FromEdgeSyncResponse(requestId, tenantId, edgeId, success, ""), requestServiceId); |
|||
} |
|||
} |
|||
} |
|||
|
|||
@Override |
|||
public void processSyncRequest(TenantId tenantId, EdgeId edgeId, Consumer<FromEdgeSyncResponse> responseConsumer) { |
|||
ToEdgeSyncRequest request = new ToEdgeSyncRequest(UUID.randomUUID(), tenantId, edgeId, serviceInfoProvider.getServiceId()); |
|||
|
|||
UUID requestId = request.getId(); |
|||
EdgeGrpcSession session = sessions.get(request.getEdgeId()); |
|||
if (session != null && session.isSyncInProgress()) { |
|||
responseConsumer.accept(new FromEdgeSyncResponse(requestId, request.getTenantId(), request.getEdgeId(), false, "Sync process is active at the moment")); |
|||
} else { |
|||
log.trace("[{}][{}] Processing sync edge request [{}], serviceId [{}]", request.getTenantId(), request.getId(), request.getEdgeId(), request.getServiceId()); |
|||
localSyncEdgeRequests.put(requestId, responseConsumer); |
|||
clusterService.pushEdgeSyncRequestToEdge(request); |
|||
scheduleSyncRequestTimeout(request, requestId); |
|||
} |
|||
} |
|||
|
|||
private void scheduleSyncRequestTimeout(ToEdgeSyncRequest request, UUID requestId) { |
|||
log.trace("[{}] scheduling sync edge request", requestId); |
|||
executorService.schedule(() -> { |
|||
log.trace("[{}] checking if sync edge request is not processed...", requestId); |
|||
Consumer<FromEdgeSyncResponse> consumer = localSyncEdgeRequests.remove(requestId); |
|||
if (consumer != null) { |
|||
log.trace("[{}] timeout for processing sync edge request.", requestId); |
|||
consumer.accept(new FromEdgeSyncResponse(requestId, request.getTenantId(), request.getEdgeId(), false, "Edge is not connected")); |
|||
} |
|||
}, 20, TimeUnit.SECONDS); |
|||
} |
|||
|
|||
private void processSyncResponse(FromEdgeSyncResponse response) { |
|||
log.trace("[{}] Received response from sync service: [{}]", response.getId(), response); |
|||
UUID requestId = response.getId(); |
|||
Consumer<FromEdgeSyncResponse> consumer = localSyncEdgeRequests.remove(requestId); |
|||
if (consumer != null) { |
|||
consumer.accept(response); |
|||
} else { |
|||
log.trace("[{}] Unknown or stale sync response received [{}]", requestId, response); |
|||
} |
|||
} |
|||
|
|||
private void scheduleEdgeEventsCheck(EdgeGrpcSession session) { |
|||
EdgeId edgeId = session.getEdge().getId(); |
|||
TenantId tenantId = session.getEdge().getTenantId(); |
|||
|
|||
cancelScheduleEdgeEventsCheck(edgeId); |
|||
|
|||
if (sessions.containsKey(edgeId)) { |
|||
ScheduledFuture<?> edgeEventCheckTask = edgeEventProcessingExecutorService.schedule(() -> { |
|||
try { |
|||
final Lock newEventLock = sessionNewEventsLocks.computeIfAbsent(edgeId, id -> new ReentrantLock()); |
|||
newEventLock.lock(); |
|||
try { |
|||
if (Boolean.TRUE.equals(sessionNewEvents.get(edgeId))) { |
|||
log.trace("[{}][{}] set session new events flag to false", tenantId, edgeId.getId()); |
|||
sessionNewEvents.put(edgeId, false); |
|||
session.processHighPriorityEvents(); |
|||
processEdgeEventMigrationIfNeeded(session, edgeId); |
|||
if (Boolean.TRUE.equals(edgeEventsMigrationProcessed.get(edgeId))) { |
|||
Futures.addCallback(session.processEdgeEvents(), new FutureCallback<>() { |
|||
@Override |
|||
public void onSuccess(Boolean newEventsAdded) { |
|||
if (Boolean.TRUE.equals(newEventsAdded)) { |
|||
log.trace("[{}][{}] new events added. set session new events flag to true", tenantId, edgeId.getId()); |
|||
sessionNewEvents.put(edgeId, true); |
|||
} |
|||
scheduleEdgeEventsCheck(session); |
|||
} |
|||
|
|||
@Override |
|||
public void onFailure(Throwable t) { |
|||
log.warn("[{}] Failed to process edge events for edge [{}]!", tenantId, session.getEdge().getId().getId(), t); |
|||
scheduleEdgeEventsCheck(session); |
|||
} |
|||
}, ctx.getGrpcCallbackExecutorService()); |
|||
} else { |
|||
scheduleEdgeEventsCheck(session); |
|||
} |
|||
} else { |
|||
scheduleEdgeEventsCheck(session); |
|||
} |
|||
} finally { |
|||
newEventLock.unlock(); |
|||
} |
|||
} catch (Exception e) { |
|||
log.warn("[{}] Failed to process edge events for edge [{}]!", tenantId, session.getEdge().getId().getId(), e); |
|||
} |
|||
}, ctx.getEdgeEventStorageSettings().getNoRecordsSleepInterval(), TimeUnit.MILLISECONDS); |
|||
sessionEdgeEventChecks.put(edgeId, edgeEventCheckTask); |
|||
log.trace("[{}] Check edge event scheduled for edge [{}]", tenantId, edgeId.getId()); |
|||
} else { |
|||
log.debug("[{}] Session was removed and edge event check schedule must not be started [{}]", |
|||
tenantId, edgeId.getId()); |
|||
} |
|||
} |
|||
|
|||
private void processEdgeEventMigrationIfNeeded(EdgeGrpcSession session, EdgeId edgeId) throws Exception { |
|||
boolean isMigrationProcessed = edgeEventsMigrationProcessed.getOrDefault(edgeId, Boolean.FALSE); |
|||
if (!isMigrationProcessed) { |
|||
log.info("Starting edge event migration for edge [{}]", edgeId.getId()); |
|||
Boolean eventsExist = session.migrateEdgeEvents().get(); |
|||
if (Boolean.TRUE.equals(eventsExist)) { |
|||
log.info("Migration still in progress for edge [{}]", edgeId.getId()); |
|||
sessionNewEvents.put(edgeId, true); |
|||
scheduleEdgeEventsCheck(session); |
|||
} else if (Boolean.FALSE.equals(eventsExist)) { |
|||
log.info("Migration completed for edge [{}]", edgeId.getId()); |
|||
edgeEventsMigrationProcessed.put(edgeId, true); |
|||
} |
|||
} |
|||
} |
|||
|
|||
private void cancelScheduleEdgeEventsCheck(EdgeId edgeId) { |
|||
log.trace("[{}] cancelling edge event check for edge", edgeId); |
|||
if (sessionEdgeEventChecks.containsKey(edgeId)) { |
|||
ScheduledFuture<?> sessionEdgeEventCheck = sessionEdgeEventChecks.get(edgeId); |
|||
if (sessionEdgeEventCheck != null && !sessionEdgeEventCheck.isCancelled() && !sessionEdgeEventCheck.isDone()) { |
|||
sessionEdgeEventCheck.cancel(true); |
|||
sessionEdgeEventChecks.remove(edgeId); |
|||
} |
|||
} |
|||
} |
|||
|
|||
private void onEdgeDisconnect(Edge edge, UUID sessionId) { |
|||
EdgeId edgeId = edge.getId(); |
|||
log.info("[{}][{}] edge disconnected!", edgeId, sessionId); |
|||
EdgeGrpcSession current = sessions.get(edgeId); |
|||
if (current != null && current.getSessionId().equals(sessionId)) { |
|||
EdgeGrpcSession toRemove = sessions.remove(edgeId); |
|||
final Lock newEventLock = sessionNewEventsLocks.computeIfAbsent(edgeId, id -> new ReentrantLock()); |
|||
newEventLock.lock(); |
|||
try { |
|||
sessionNewEvents.remove(edgeId); |
|||
} finally { |
|||
newEventLock.unlock(); |
|||
} |
|||
destroySession(toRemove); |
|||
sessionsById.remove(sessionId); |
|||
TenantId tenantId = toRemove.getEdge().getTenantId(); |
|||
save(tenantId, edgeId, ACTIVITY_STATE, false); |
|||
long lastDisconnectTs = System.currentTimeMillis(); |
|||
save(tenantId, edgeId, LAST_DISCONNECT_TIME, lastDisconnectTs); |
|||
pushRuleEngineMessage(toRemove.getEdge().getTenantId(), edge, lastDisconnectTs, TbMsgType.DISCONNECT_EVENT); |
|||
cancelScheduleEdgeEventsCheck(edgeId); |
|||
} else { |
|||
log.info("[{}] edge session [{}] is not current anymore. Attempting to destroy it by sessionId.", edgeId, sessionId); |
|||
EdgeGrpcSession stale = sessionsById.remove(sessionId); |
|||
if (stale != null) { |
|||
try { |
|||
destroySession(stale); |
|||
log.info("[{}][{}] Successfully destroyed stale session for edge [{}]", stale.getTenantId(), sessionId, edgeId); |
|||
} catch (Exception e) { |
|||
log.warn("[{}][{}] Failed to destroy stale session for edge [{}]", stale.getTenantId(), sessionId, edgeId, e); |
|||
} |
|||
} else { |
|||
log.debug("[{}] No session found by sessionId [{}] to destroy", edgeId, sessionId); |
|||
} |
|||
} |
|||
edgeIdServiceIdCache.evict(edgeId); |
|||
} |
|||
|
|||
private void destroySession(EdgeGrpcSession session) { |
|||
try (session) { |
|||
if (!session.destroy()) { |
|||
log.warn("[{}][{}] Session destroy failed for edge [{}] with session id [{}]. Adding to zombie queue for later cleanup.", |
|||
session.getTenantId(), session.getEdge().getId(), session.getEdge().getName(), session.getSessionId()); |
|||
zombieSessions.add(session); |
|||
} |
|||
} catch (Exception e) { |
|||
log.warn("[{}][{}] Exception during session destroy for edge [{}] with session id [{}]", |
|||
session.getTenantId(), session.getEdge().getId(), session.getEdge().getName(), session.getSessionId(), e); |
|||
} |
|||
} |
|||
|
|||
private void save(TenantId tenantId, EdgeId edgeId, String key, long value) { |
|||
log.debug("[{}][{}] Updating long edge telemetry [{}] [{}]", tenantId, edgeId, key, value); |
|||
if (persistToTelemetry) { |
|||
tsSubService.saveTimeseries(TimeseriesSaveRequest.builder() |
|||
.tenantId(tenantId) |
|||
.entityId(edgeId) |
|||
.entry(new LongDataEntry(key, value)) |
|||
.callback(new AttributeSaveCallback(tenantId, edgeId, key, value)) |
|||
.build()); |
|||
} else { |
|||
tsSubService.saveAttributes(AttributesSaveRequest.builder() |
|||
.tenantId(tenantId) |
|||
.entityId(edgeId) |
|||
.scope(AttributeScope.SERVER_SCOPE) |
|||
.entry(new LongDataEntry(key, value)) |
|||
.callback(new AttributeSaveCallback(tenantId, edgeId, key, value)) |
|||
.build()); |
|||
} |
|||
} |
|||
|
|||
private void save(TenantId tenantId, EdgeId edgeId, String key, boolean value) { |
|||
log.debug("[{}][{}] Updating boolean edge telemetry [{}] [{}]", tenantId, edgeId, key, value); |
|||
if (persistToTelemetry) { |
|||
tsSubService.saveTimeseries(TimeseriesSaveRequest.builder() |
|||
.tenantId(tenantId) |
|||
.entityId(edgeId) |
|||
.entry(new BooleanDataEntry(key, value)) |
|||
.callback(new AttributeSaveCallback(tenantId, edgeId, key, value)) |
|||
.build()); |
|||
} else { |
|||
tsSubService.saveAttributes(AttributesSaveRequest.builder() |
|||
.tenantId(tenantId) |
|||
.entityId(edgeId) |
|||
.scope(AttributeScope.SERVER_SCOPE) |
|||
.entry(new BooleanDataEntry(key, value)) |
|||
.callback(new AttributeSaveCallback(tenantId, edgeId, key, value)) |
|||
.build()); |
|||
} |
|||
} |
|||
|
|||
private static class AttributeSaveCallback implements FutureCallback<Void> { |
|||
|
|||
private final TenantId tenantId; |
|||
private final EdgeId edgeId; |
|||
private final String key; |
|||
private final Object value; |
|||
|
|||
AttributeSaveCallback(TenantId tenantId, EdgeId edgeId, String key, Object value) { |
|||
this.tenantId = tenantId; |
|||
this.edgeId = edgeId; |
|||
this.key = key; |
|||
this.value = value; |
|||
} |
|||
|
|||
@Override |
|||
public void onSuccess(@Nullable Void result) { |
|||
log.trace("[{}][{}] Successfully updated attribute [{}] with value [{}]", tenantId, edgeId, key, value); |
|||
} |
|||
|
|||
@Override |
|||
public void onFailure(Throwable t) { |
|||
log.warn("[{}][{}] Failed to update attribute [{}] with value [{}]", tenantId, edgeId, key, value, t); |
|||
} |
|||
|
|||
} |
|||
|
|||
private void pushRuleEngineMessage(TenantId tenantId, Edge edge, long ts, TbMsgType msgType) { |
|||
try { |
|||
EdgeId edgeId = edge.getId(); |
|||
ObjectNode edgeState = JacksonUtil.newObjectNode(); |
|||
boolean isConnected = TbMsgType.CONNECT_EVENT.equals(msgType); |
|||
if (isConnected) { |
|||
edgeState.put(ACTIVITY_STATE, true); |
|||
edgeState.put(LAST_CONNECT_TIME, ts); |
|||
} else { |
|||
edgeState.put(ACTIVITY_STATE, false); |
|||
edgeState.put(LAST_DISCONNECT_TIME, ts); |
|||
} |
|||
ctx.getRuleProcessor().process(EdgeConnectionTrigger.builder() |
|||
.tenantId(tenantId) |
|||
.customerId(edge.getCustomerId()) |
|||
.edgeId(edgeId) |
|||
.edgeName(edge.getName()) |
|||
.connected(isConnected).build()); |
|||
String data = JacksonUtil.toString(edgeState); |
|||
TbMsgMetaData md = new TbMsgMetaData(); |
|||
if (!persistToTelemetry) { |
|||
md.putValue(DataConstants.SCOPE, DataConstants.SERVER_SCOPE); |
|||
md.putValue("edgeName", edge.getName()); |
|||
md.putValue("edgeType", edge.getType()); |
|||
} |
|||
TbMsg tbMsg = TbMsg.newMsg() |
|||
.type(msgType) |
|||
.originator(edgeId) |
|||
.copyMetaData(md) |
|||
.dataType(TbMsgDataType.JSON) |
|||
.data(data) |
|||
.build(); |
|||
clusterService.pushMsgToRuleEngine(tenantId, edgeId, tbMsg, null); |
|||
} catch (Exception e) { |
|||
log.warn("[{}][{}] Failed to push {}", tenantId, edge.getId(), msgType, e); |
|||
} |
|||
} |
|||
|
|||
private void cleanupZombieSessions() { |
|||
try { |
|||
tryToDestroyZombieSessions(getZombieSessions(sessions.values()), s -> sessions.remove(s.getEdge().getId())); |
|||
tryToDestroyZombieSessions(getZombieSessions(sessionsById.values()), s -> sessionsById.remove(s.getSessionId())); |
|||
|
|||
zombieSessions.removeIf(zombie -> { |
|||
if (zombie.destroy()) { |
|||
log.info("[{}][{}] Successfully cleaned up zombie session [{}] for edge [{}].", |
|||
zombie.getTenantId(), zombie.getEdge().getId(), zombie.getSessionId(), zombie.getEdge().getName()); |
|||
return true; |
|||
} else { |
|||
log.warn("[{}][{}] Failed to remove zombie session [{}] for edge [{}].", |
|||
zombie.getTenantId(), zombie.getEdge().getId(), zombie.getSessionId(), zombie.getEdge().getName()); |
|||
return false; |
|||
} |
|||
}); |
|||
} catch (Exception e) { |
|||
log.warn("Failed to cleanup kafka sessions", e); |
|||
} |
|||
} |
|||
|
|||
private List<EdgeGrpcSession> getZombieSessions(Collection<EdgeGrpcSession> sessions) { |
|||
List<EdgeGrpcSession> result = new ArrayList<>(); |
|||
for (EdgeGrpcSession session : sessions) { |
|||
if (isKafkaSessionAndZombie(session)) { |
|||
result.add(session); |
|||
} |
|||
} |
|||
return result; |
|||
} |
|||
|
|||
private void tryToDestroyZombieSessions(List<EdgeGrpcSession> sessionsToRemove, Function<EdgeGrpcSession, EdgeGrpcSession> removeFunc) { |
|||
for (EdgeGrpcSession toRemove : sessionsToRemove) { |
|||
log.info("[{}] Destroying session for edge because edge is not connected", toRemove.getEdge().getId()); |
|||
if (toRemove.destroy()) { |
|||
removeFunc.apply(toRemove); |
|||
} |
|||
} |
|||
} |
|||
|
|||
private boolean isKafkaSessionAndZombie(EdgeGrpcSession session) { |
|||
if (session instanceof KafkaEdgeGrpcSession kafkaSession) { |
|||
log.debug("[{}] kafkaSession.isConnected() = {}, kafkaSession.getConsumer().getConsumer().isStopped() = {}", |
|||
kafkaSession.getEdge().getId(), |
|||
kafkaSession.isConnected(), |
|||
kafkaSession.getConsumer() != null ? kafkaSession.getConsumer().getConsumer() != null ? kafkaSession.getConsumer().getConsumer().isStopped() : null : null); |
|||
return !kafkaSession.isConnected() && |
|||
kafkaSession.getConsumer() != null && |
|||
kafkaSession.getConsumer().getConsumer() != null && |
|||
!kafkaSession.getConsumer().getConsumer().isStopped(); |
|||
} |
|||
return false; |
|||
} |
|||
|
|||
} |
|||
@ -1,999 +0,0 @@ |
|||
/** |
|||
* Copyright © 2016-2026 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.rpc; |
|||
|
|||
import com.datastax.oss.driver.api.core.uuid.Uuids; |
|||
import com.google.common.util.concurrent.FutureCallback; |
|||
import com.google.common.util.concurrent.Futures; |
|||
import com.google.common.util.concurrent.ListenableFuture; |
|||
import com.google.common.util.concurrent.SettableFuture; |
|||
import io.grpc.stub.StreamObserver; |
|||
import lombok.Data; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.checkerframework.checker.nullness.qual.Nullable; |
|||
import org.springframework.data.util.Pair; |
|||
import org.thingsboard.server.common.data.AttributeScope; |
|||
import org.thingsboard.server.common.data.DataConstants; |
|||
import org.thingsboard.server.common.data.EdgeUtils; |
|||
import org.thingsboard.server.common.data.StringUtils; |
|||
import org.thingsboard.server.common.data.edge.Edge; |
|||
import org.thingsboard.server.common.data.edge.EdgeEvent; |
|||
import org.thingsboard.server.common.data.edge.EdgeEventType; |
|||
import org.thingsboard.server.common.data.id.EdgeId; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
import org.thingsboard.server.common.data.kv.AttributeKvEntry; |
|||
import org.thingsboard.server.common.data.kv.AttributesSaveResult; |
|||
import org.thingsboard.server.common.data.kv.BaseAttributeKvEntry; |
|||
import org.thingsboard.server.common.data.kv.LongDataEntry; |
|||
import org.thingsboard.server.common.data.kv.StringDataEntry; |
|||
import org.thingsboard.server.common.data.limit.LimitedApi; |
|||
import org.thingsboard.server.common.data.notification.rule.trigger.EdgeCommunicationFailureTrigger; |
|||
import org.thingsboard.server.common.data.page.PageData; |
|||
import org.thingsboard.server.common.data.page.PageLink; |
|||
import org.thingsboard.server.common.data.page.TimePageLink; |
|||
import org.thingsboard.server.common.msg.edge.EdgeEventUpdateMsg; |
|||
import org.thingsboard.server.dao.edge.stats.EdgeStatsKey; |
|||
import org.thingsboard.server.gen.edge.v1.AiModelUpdateMsg; |
|||
import org.thingsboard.server.gen.edge.v1.AlarmCommentUpdateMsg; |
|||
import org.thingsboard.server.gen.edge.v1.AlarmUpdateMsg; |
|||
import org.thingsboard.server.gen.edge.v1.AssetProfileUpdateMsg; |
|||
import org.thingsboard.server.gen.edge.v1.AssetUpdateMsg; |
|||
import org.thingsboard.server.gen.edge.v1.AttributesRequestMsg; |
|||
import org.thingsboard.server.gen.edge.v1.CalculatedFieldRequestMsg; |
|||
import org.thingsboard.server.gen.edge.v1.CalculatedFieldUpdateMsg; |
|||
import org.thingsboard.server.gen.edge.v1.ConnectRequestMsg; |
|||
import org.thingsboard.server.gen.edge.v1.ConnectResponseCode; |
|||
import org.thingsboard.server.gen.edge.v1.ConnectResponseMsg; |
|||
import org.thingsboard.server.gen.edge.v1.DashboardUpdateMsg; |
|||
import org.thingsboard.server.gen.edge.v1.DeviceCredentialsRequestMsg; |
|||
import org.thingsboard.server.gen.edge.v1.DeviceCredentialsUpdateMsg; |
|||
import org.thingsboard.server.gen.edge.v1.DeviceProfileUpdateMsg; |
|||
import org.thingsboard.server.gen.edge.v1.DeviceRpcCallMsg; |
|||
import org.thingsboard.server.gen.edge.v1.DeviceUpdateMsg; |
|||
import org.thingsboard.server.gen.edge.v1.DownlinkMsg; |
|||
import org.thingsboard.server.gen.edge.v1.DownlinkResponseMsg; |
|||
import org.thingsboard.server.gen.edge.v1.EdgeConfiguration; |
|||
import org.thingsboard.server.gen.edge.v1.EdgeUpdateMsg; |
|||
import org.thingsboard.server.gen.edge.v1.EdgeVersion; |
|||
import org.thingsboard.server.gen.edge.v1.EntityDataProto; |
|||
import org.thingsboard.server.gen.edge.v1.EntityViewUpdateMsg; |
|||
import org.thingsboard.server.gen.edge.v1.EntityViewsRequestMsg; |
|||
import org.thingsboard.server.gen.edge.v1.RelationRequestMsg; |
|||
import org.thingsboard.server.gen.edge.v1.RelationUpdateMsg; |
|||
import org.thingsboard.server.gen.edge.v1.RequestMsg; |
|||
import org.thingsboard.server.gen.edge.v1.RequestMsgType; |
|||
import org.thingsboard.server.gen.edge.v1.ResourceUpdateMsg; |
|||
import org.thingsboard.server.gen.edge.v1.ResponseMsg; |
|||
import org.thingsboard.server.gen.edge.v1.RuleChainMetadataRequestMsg; |
|||
import org.thingsboard.server.gen.edge.v1.RuleChainMetadataUpdateMsg; |
|||
import org.thingsboard.server.gen.edge.v1.RuleChainUpdateMsg; |
|||
import org.thingsboard.server.gen.edge.v1.SyncCompletedMsg; |
|||
import org.thingsboard.server.gen.edge.v1.UplinkMsg; |
|||
import org.thingsboard.server.gen.edge.v1.UplinkResponseMsg; |
|||
import org.thingsboard.server.gen.edge.v1.UserCredentialsRequestMsg; |
|||
import org.thingsboard.server.gen.edge.v1.UserCredentialsUpdateMsg; |
|||
import org.thingsboard.server.gen.edge.v1.UserUpdateMsg; |
|||
import org.thingsboard.server.gen.edge.v1.WidgetBundleTypesRequestMsg; |
|||
import org.thingsboard.server.service.edge.EdgeContextComponent; |
|||
import org.thingsboard.server.service.edge.EdgeMsgConstructorUtils; |
|||
import org.thingsboard.server.service.edge.rpc.fetch.EdgeEventFetcher; |
|||
import org.thingsboard.server.service.edge.rpc.fetch.GeneralEdgeEventFetcher; |
|||
import org.thingsboard.server.service.edge.rpc.utils.EdgeVersionUtils; |
|||
|
|||
import java.io.Closeable; |
|||
import java.util.ArrayList; |
|||
import java.util.Arrays; |
|||
import java.util.Collections; |
|||
import java.util.List; |
|||
import java.util.Optional; |
|||
import java.util.UUID; |
|||
import java.util.concurrent.ConcurrentLinkedQueue; |
|||
import java.util.concurrent.ScheduledExecutorService; |
|||
import java.util.concurrent.TimeUnit; |
|||
import java.util.concurrent.locks.Lock; |
|||
import java.util.concurrent.locks.ReentrantLock; |
|||
import java.util.function.BiConsumer; |
|||
|
|||
@Slf4j |
|||
@Data |
|||
public abstract class EdgeGrpcSession implements Closeable { |
|||
|
|||
private static final String QUEUE_START_TS_ATTR_KEY = "queueStartTs"; |
|||
private static final String QUEUE_START_SEQ_ID_ATTR_KEY = "queueStartSeqId"; |
|||
|
|||
private static final int MAX_DOWNLINK_ATTEMPTS = 3; |
|||
private static final String RATE_LIMIT_REACHED = "Rate limit reached"; |
|||
|
|||
protected final ConcurrentLinkedQueue<EdgeEvent> highPriorityQueue = new ConcurrentLinkedQueue<>(); |
|||
|
|||
protected UUID sessionId; |
|||
private BiConsumer<EdgeId, EdgeGrpcSession> sessionOpenListener; |
|||
private BiConsumer<Edge, UUID> sessionCloseListener; |
|||
|
|||
private final EdgeSessionState sessionState = new EdgeSessionState(); |
|||
private final ReentrantLock downlinkMsgLock = new ReentrantLock(); |
|||
private final Lock sequenceDependencyLock = new ReentrantLock(); |
|||
|
|||
protected EdgeContextComponent ctx; |
|||
protected Edge edge; |
|||
protected TenantId tenantId; |
|||
|
|||
private Long newStartTs; |
|||
private Long previousStartTs; |
|||
private Long newStartSeqId; |
|||
private Long previousStartSeqId; |
|||
|
|||
private StreamObserver<RequestMsg> inputStream; |
|||
private StreamObserver<ResponseMsg> outputStream; |
|||
|
|||
private volatile boolean connected; |
|||
private volatile boolean syncInProgress; |
|||
|
|||
private EdgeVersion edgeVersion; |
|||
private int maxInboundMessageSize; |
|||
private int clientMaxInboundMessageSize; |
|||
private int maxHighPriorityQueueSizePerSession; |
|||
|
|||
private ScheduledExecutorService sendDownlinkExecutorService; |
|||
|
|||
public EdgeGrpcSession(EdgeContextComponent ctx, StreamObserver<ResponseMsg> outputStream, |
|||
BiConsumer<EdgeId, EdgeGrpcSession> sessionOpenListener, |
|||
BiConsumer<Edge, UUID> sessionCloseListener, |
|||
ScheduledExecutorService sendDownlinkExecutorService, |
|||
int maxInboundMessageSize, int maxHighPriorityQueueSizePerSession) { |
|||
this.sessionId = UUID.randomUUID(); |
|||
this.ctx = ctx; |
|||
this.outputStream = outputStream; |
|||
this.sessionOpenListener = sessionOpenListener; |
|||
this.sessionCloseListener = sessionCloseListener; |
|||
this.sendDownlinkExecutorService = sendDownlinkExecutorService; |
|||
this.maxInboundMessageSize = maxInboundMessageSize; |
|||
this.maxHighPriorityQueueSizePerSession = maxHighPriorityQueueSizePerSession; |
|||
initInputStream(); |
|||
} |
|||
|
|||
protected abstract ListenableFuture<Boolean> migrateEdgeEvents() throws Exception; |
|||
|
|||
public void initInputStream() { |
|||
inputStream = new StreamObserver<>() { |
|||
@Override |
|||
public void onNext(RequestMsg requestMsg) { |
|||
if (!connected && requestMsg.getMsgType().equals(RequestMsgType.CONNECT_RPC_MESSAGE)) { |
|||
ConnectResponseMsg responseMsg = processConnect(requestMsg.getConnectRequestMsg()); |
|||
outputStream.onNext(ResponseMsg.newBuilder() |
|||
.setConnectResponseMsg(responseMsg) |
|||
.build()); |
|||
if (ConnectResponseCode.ACCEPTED != responseMsg.getResponseCode()) { |
|||
outputStream.onError(new RuntimeException(responseMsg.getErrorMsg())); |
|||
} else { |
|||
if (requestMsg.getConnectRequestMsg().hasMaxInboundMessageSize()) { |
|||
log.debug("[{}][{}] Client max inbound message size: {}", tenantId, sessionId, requestMsg.getConnectRequestMsg().getMaxInboundMessageSize()); |
|||
clientMaxInboundMessageSize = requestMsg.getConnectRequestMsg().getMaxInboundMessageSize(); |
|||
} |
|||
connected = true; |
|||
} |
|||
} |
|||
if (connected) { |
|||
if (requestMsg.getMsgType().equals(RequestMsgType.SYNC_REQUEST_RPC_MESSAGE)) { |
|||
if (requestMsg.hasSyncRequestMsg()) { |
|||
boolean fullSync = false; |
|||
if (requestMsg.getSyncRequestMsg().hasFullSync()) { |
|||
fullSync = requestMsg.getSyncRequestMsg().getFullSync(); |
|||
} |
|||
startSyncProcess(fullSync); |
|||
} else { |
|||
syncInProgress = false; |
|||
} |
|||
} |
|||
if (requestMsg.getMsgType().equals(RequestMsgType.UPLINK_RPC_MESSAGE)) { |
|||
if (requestMsg.hasUplinkMsg()) { |
|||
onUplinkMsg(requestMsg.getUplinkMsg()); |
|||
} |
|||
if (requestMsg.hasDownlinkResponseMsg()) { |
|||
onDownlinkResponse(requestMsg.getDownlinkResponseMsg()); |
|||
} |
|||
} |
|||
} |
|||
} |
|||
|
|||
@Override |
|||
public void onError(Throwable t) { |
|||
log.trace("[{}][{}] Stream was terminated due to error:", tenantId, sessionId, t); |
|||
closeSession(); |
|||
} |
|||
|
|||
@Override |
|||
public void onCompleted() { |
|||
log.info("[{}][{}] Stream was closed and completed successfully!", tenantId, sessionId); |
|||
closeSession(); |
|||
} |
|||
|
|||
private void closeSession() { |
|||
connected = false; |
|||
if (edge != null) { |
|||
try { |
|||
sessionCloseListener.accept(edge, sessionId); |
|||
} catch (Exception ignored) { |
|||
} |
|||
} |
|||
try { |
|||
outputStream.onCompleted(); |
|||
} catch (Exception ignored) { |
|||
} |
|||
} |
|||
}; |
|||
} |
|||
|
|||
public void onConfigurationUpdate(Edge edge) { |
|||
log.debug("[{}] onConfigurationUpdate [{}]", sessionId, edge); |
|||
this.tenantId = edge.getTenantId(); |
|||
this.edge = edge; |
|||
EdgeUpdateMsg edgeConfig = EdgeUpdateMsg.newBuilder() |
|||
.setConfiguration(EdgeMsgConstructorUtils.constructEdgeConfiguration(edge)).build(); |
|||
ResponseMsg edgeConfigMsg = ResponseMsg.newBuilder() |
|||
.setEdgeUpdateMsg(edgeConfig) |
|||
.build(); |
|||
sendDownlinkMsg(edgeConfigMsg); |
|||
} |
|||
|
|||
public void startSyncProcess(boolean fullSync) { |
|||
if (!syncInProgress) { |
|||
log.info("[{}][{}][{}] Staring edge sync process", tenantId, edge.getId(), sessionId); |
|||
syncInProgress = true; |
|||
interruptGeneralProcessingOnSync(); |
|||
doSync(new EdgeSyncCursor(ctx, edge, fullSync)); |
|||
} else { |
|||
log.info("[{}][{}][{}] Sync is already started, skipping starting it now", tenantId, edge.getId(), sessionId); |
|||
} |
|||
} |
|||
|
|||
private void doSync(EdgeSyncCursor cursor) { |
|||
if (cursor.hasNext()) { |
|||
EdgeEventFetcher next = cursor.getNext(); |
|||
log.debug("[{}][{}] starting sync process, cursor current idx = {}, class = {}", |
|||
tenantId, edge.getId(), cursor.getCurrentIdx(), next.getClass().getSimpleName()); |
|||
ListenableFuture<Pair<Long, Long>> future = startProcessingEdgeEvents(next); |
|||
Futures.addCallback(future, new FutureCallback<>() { |
|||
@Override |
|||
public void onSuccess(@Nullable Pair<Long, Long> result) { |
|||
doSync(cursor); |
|||
} |
|||
|
|||
@Override |
|||
public void onFailure(Throwable t) { |
|||
log.error("[{}][{}] Exception during sync process", tenantId, edge.getId(), t); |
|||
} |
|||
}, ctx.getGrpcCallbackExecutorService()); |
|||
} else { |
|||
log.info("[{}][{}] sync process completed", tenantId, edge.getId()); |
|||
DownlinkMsg syncCompleteDownlinkMsg = DownlinkMsg.newBuilder() |
|||
.setDownlinkMsgId(EdgeUtils.nextPositiveInt()) |
|||
.setSyncCompletedMsg(SyncCompletedMsg.newBuilder().build()) |
|||
.build(); |
|||
Futures.addCallback(sendDownlinkMsgsPack(Collections.singletonList(syncCompleteDownlinkMsg)), new FutureCallback<>() { |
|||
@Override |
|||
public void onSuccess(Boolean isInterrupted) { |
|||
markSyncCompletedSendEdgeEventUpdate(); |
|||
} |
|||
|
|||
@Override |
|||
public void onFailure(Throwable t) { |
|||
log.error("[{}][{}] Exception during sending sync complete", tenantId, edge.getId(), t); |
|||
markSyncCompletedSendEdgeEventUpdate(); |
|||
} |
|||
}, ctx.getGrpcCallbackExecutorService()); |
|||
} |
|||
} |
|||
|
|||
protected void processEdgeEvents(EdgeEventFetcher fetcher, PageLink pageLink, SettableFuture<Pair<Long, Long>> result) { |
|||
try { |
|||
log.trace("[{}] Start processing edge events, fetcher = {}, pageLink = {}", edge.getId(), fetcher.getClass().getSimpleName(), pageLink); |
|||
processHighPriorityEvents(); |
|||
PageData<EdgeEvent> pageData = fetcher.fetchEdgeEvents(edge.getTenantId(), edge, pageLink); |
|||
if (isConnected() && !pageData.getData().isEmpty()) { |
|||
if (fetcher instanceof GeneralEdgeEventFetcher) { |
|||
long queueSize = pageData.getTotalElements() - ((long) pageLink.getPageSize() * pageLink.getPage()); |
|||
ctx.getStatsCounterService().ifPresent(statsCounterService -> |
|||
statsCounterService.recordEvent(EdgeStatsKey.DOWNLINK_MSGS_LAG, tenantId, edge.getId(), queueSize)); |
|||
} |
|||
log.trace("[{}][{}][{}] event(s) are going to be processed.", tenantId, edge.getId(), pageData.getData().size()); |
|||
List<DownlinkMsg> downlinkMsgsPack = convertToDownlinkMsgsPack(pageData.getData()); |
|||
Futures.addCallback(sendDownlinkMsgsPack(downlinkMsgsPack), new FutureCallback<>() { |
|||
@Override |
|||
public void onSuccess(@Nullable Boolean isInterrupted) { |
|||
if (Boolean.TRUE.equals(isInterrupted)) { |
|||
log.debug("[{}][{}][{}] Send downlink messages task was interrupted", tenantId, edge.getId(), sessionId); |
|||
result.set(null); |
|||
} else { |
|||
if (isConnected() && pageData.hasNext()) { |
|||
processEdgeEvents(fetcher, pageLink.nextPageLink(), result); |
|||
} else { |
|||
EdgeEvent latestEdgeEvent = pageData.getData().get(pageData.getData().size() - 1); |
|||
UUID idOffset = latestEdgeEvent.getUuidId(); |
|||
if (idOffset != null) { |
|||
Long newStartTs = Uuids.unixTimestamp(idOffset); |
|||
long newStartSeqId = latestEdgeEvent.getSeqId(); |
|||
result.set(Pair.of(newStartTs, newStartSeqId)); |
|||
} else { |
|||
result.set(null); |
|||
} |
|||
} |
|||
} |
|||
} |
|||
|
|||
@Override |
|||
public void onFailure(Throwable t) { |
|||
log.error("[{}] Failed to send downlink msgs pack", edge.getId(), t); |
|||
result.setException(t); |
|||
} |
|||
}, ctx.getGrpcCallbackExecutorService()); |
|||
} else { |
|||
log.trace("[{}] no event(s) found. Stop processing edge events, fetcher = {}, pageLink = {}", edge.getId(), fetcher.getClass().getSimpleName(), pageLink); |
|||
result.set(null); |
|||
} |
|||
} catch (Exception e) { |
|||
log.error("[{}] Failed to fetch edge events", edge.getId(), e); |
|||
result.setException(e); |
|||
} |
|||
} |
|||
|
|||
private ConnectResponseMsg processConnect(ConnectRequestMsg request) { |
|||
log.trace("[{}] processConnect [{}]", sessionId, request); |
|||
Optional<Edge> optional = ctx.getEdgeService().findEdgeByRoutingKey(TenantId.SYS_TENANT_ID, request.getEdgeRoutingKey()); |
|||
if (optional.isPresent()) { |
|||
edge = optional.get(); |
|||
tenantId = edge.getTenantId(); |
|||
try { |
|||
if (edge.getSecret().equals(request.getEdgeSecret())) { |
|||
sessionOpenListener.accept(edge.getId(), this); |
|||
edgeVersion = request.getEdgeVersion(); |
|||
processSaveEdgeVersionAsAttribute(request.getEdgeVersion().name()); |
|||
return ConnectResponseMsg.newBuilder() |
|||
.setResponseCode(ConnectResponseCode.ACCEPTED) |
|||
.setErrorMsg("") |
|||
.setConfiguration(EdgeMsgConstructorUtils.constructEdgeConfiguration(edge)) |
|||
.setMaxInboundMessageSize(maxInboundMessageSize) |
|||
.build(); |
|||
} |
|||
String error = "Failed to validate the edge!"; |
|||
String failureMsg = String.format("%s Provided request secret: %s", error, request.getEdgeSecret()); |
|||
ctx.getRuleProcessor().process(EdgeCommunicationFailureTrigger.builder().tenantId(tenantId).edgeId(edge.getId()) |
|||
.customerId(edge.getCustomerId()).edgeName(edge.getName()).failureMsg(failureMsg).error(error).build()); |
|||
return ConnectResponseMsg.newBuilder() |
|||
.setResponseCode(ConnectResponseCode.BAD_CREDENTIALS) |
|||
.setErrorMsg(failureMsg) |
|||
.setConfiguration(EdgeConfiguration.getDefaultInstance()).build(); |
|||
} catch (Exception e) { |
|||
String failureMsg = "Failed to process edge connection!"; |
|||
ctx.getRuleProcessor().process(EdgeCommunicationFailureTrigger.builder().tenantId(tenantId).edgeId(edge.getId()) |
|||
.customerId(edge.getCustomerId()).edgeName(edge.getName()).failureMsg(failureMsg).error(e.getMessage()).build()); |
|||
log.error(failureMsg, e); |
|||
return ConnectResponseMsg.newBuilder() |
|||
.setResponseCode(ConnectResponseCode.SERVER_UNAVAILABLE) |
|||
.setErrorMsg(failureMsg) |
|||
.setConfiguration(EdgeConfiguration.getDefaultInstance()).build(); |
|||
} |
|||
} |
|||
return ConnectResponseMsg.newBuilder() |
|||
.setResponseCode(ConnectResponseCode.BAD_CREDENTIALS) |
|||
.setErrorMsg("Failed to find the edge! Routing key: " + request.getEdgeRoutingKey()) |
|||
.setConfiguration(EdgeConfiguration.getDefaultInstance()).build(); |
|||
} |
|||
|
|||
private void processSaveEdgeVersionAsAttribute(String edgeVersion) { |
|||
AttributeKvEntry attributeKvEntry = new BaseAttributeKvEntry(new StringDataEntry(DataConstants.EDGE_VERSION_ATTR_KEY, edgeVersion), System.currentTimeMillis()); |
|||
ctx.getAttributesService().save(tenantId, edge.getId(), AttributeScope.SERVER_SCOPE, attributeKvEntry); |
|||
} |
|||
|
|||
private void interruptGeneralProcessingOnSync() { |
|||
log.debug("[{}][{}][{}] Sync process started. General processing interrupted!", tenantId, edge.getId(), sessionId); |
|||
stopCurrentSendDownlinkMsgsTask(true); |
|||
} |
|||
|
|||
protected ListenableFuture<Boolean> sendDownlinkMsgsPack(List<DownlinkMsg> downlinkMsgsPack) { |
|||
interruptPreviousSendDownlinkMsgsTask(); |
|||
|
|||
sessionState.setSendDownlinkMsgsFuture(SettableFuture.create()); |
|||
sessionState.getPendingMsgsMap().clear(); |
|||
|
|||
downlinkMsgsPack.forEach(msg -> sessionState.getPendingMsgsMap().put(msg.getDownlinkMsgId(), msg)); |
|||
scheduleDownlinkMsgsPackSend(1); |
|||
|
|||
return sessionState.getSendDownlinkMsgsFuture(); |
|||
} |
|||
|
|||
private void interruptPreviousSendDownlinkMsgsTask() { |
|||
if (sessionState.getSendDownlinkMsgsFuture() != null && !sessionState.getSendDownlinkMsgsFuture().isDone() |
|||
|| sessionState.getScheduledSendDownlinkTask() != null && !sessionState.getScheduledSendDownlinkTask().isCancelled()) { |
|||
log.debug("[{}][{}][{}] Previous send downlink future was not properly completed, stopping it now!", tenantId, edge.getId(), sessionId); |
|||
stopCurrentSendDownlinkMsgsTask(true); |
|||
} else { |
|||
log.trace("[{}][{}][{}] Previous send downlink future is not active", tenantId, edge.getId(), sessionId); |
|||
} |
|||
} |
|||
|
|||
private void onUplinkMsg(UplinkMsg uplinkMsg) { |
|||
if (isRateLimitViolated(uplinkMsg)) { |
|||
return; |
|||
} |
|||
ListenableFuture<List<Void>> future = processUplinkMsg(uplinkMsg); |
|||
Futures.addCallback(future, new FutureCallback<>() { |
|||
@Override |
|||
public void onSuccess(@Nullable List<Void> result) { |
|||
sendResponseMessage(uplinkMsg.getUplinkMsgId(), true, null); |
|||
} |
|||
|
|||
@Override |
|||
public void onFailure(Throwable t) { |
|||
String errorMsg = EdgeUtils.createErrorMsgFromRootCauseAndStackTrace(t); |
|||
sendResponseMessage(uplinkMsg.getUplinkMsgId(), false, errorMsg); |
|||
} |
|||
}, ctx.getGrpcCallbackExecutorService()); |
|||
} |
|||
|
|||
private boolean isRateLimitViolated(UplinkMsg uplinkMsg) { |
|||
if (!ctx.getRateLimitService().checkRateLimit(LimitedApi.EDGE_UPLINK_MESSAGES, tenantId) || |
|||
!ctx.getRateLimitService().checkRateLimit(LimitedApi.EDGE_UPLINK_MESSAGES_PER_EDGE, tenantId, edge.getId())) { |
|||
String errorMsg = String.format("Failed to process uplink message. %s", RATE_LIMIT_REACHED); |
|||
sendResponseMessage(uplinkMsg.getUplinkMsgId(), false, errorMsg); |
|||
return true; |
|||
} |
|||
return false; |
|||
} |
|||
|
|||
private void scheduleDownlinkMsgsPackSend(int attempt) { |
|||
Runnable sendDownlinkMsgsTask = () -> { |
|||
try { |
|||
if (!isConnected()) { |
|||
stopCurrentSendDownlinkMsgsTask(true); |
|||
return; |
|||
} |
|||
if (!sessionState.getPendingMsgsMap().values().isEmpty()) { |
|||
List<DownlinkMsg> copy = new ArrayList<>(sessionState.getPendingMsgsMap().values()); |
|||
if (attempt > 1) { |
|||
String error = "Failed to deliver the batch"; |
|||
String failureMsg = String.format("{%s} (size: {%s})", error, copy.size()); |
|||
if (attempt == 2) { |
|||
// Send a failure notification only on the second attempt.
|
|||
// This ensures that failure alerts are sent just once to avoid redundant notifications.
|
|||
ctx.getRuleProcessor().process(EdgeCommunicationFailureTrigger.builder().tenantId(tenantId) |
|||
.edgeId(edge.getId()).customerId(edge.getCustomerId()).edgeName(edge.getName()).failureMsg(failureMsg).error(error).build()); |
|||
} |
|||
ctx.getStatsCounterService().ifPresent(statsCounterService -> |
|||
statsCounterService.recordEvent(EdgeStatsKey.DOWNLINK_MSGS_TMP_FAILED, edge.getTenantId(), edge.getId(), 1)); |
|||
log.warn("[{}][{}] {} on attempt {}", tenantId, edge.getId(), failureMsg, attempt); |
|||
log.debug("[{}][{}] entities in failed batch: {}", tenantId, edge.getId(), copy); |
|||
} |
|||
log.trace("[{}][{}][{}] downlink msg(s) are going to be send.", tenantId, edge.getId(), copy.size()); |
|||
for (DownlinkMsg downlinkMsg : copy) { |
|||
if (clientMaxInboundMessageSize != 0 && downlinkMsg.getSerializedSize() > clientMaxInboundMessageSize) { |
|||
String error = String.format("Client max inbound message size %s is exceeded. Please increase value of CLOUD_RPC_MAX_INBOUND_MESSAGE_SIZE " + |
|||
"env variable on the edge and restart it.", clientMaxInboundMessageSize); |
|||
String message = String.format("Downlink msg size %s exceeds client max inbound message size %s. " + |
|||
"Please increase value of CLOUD_RPC_MAX_INBOUND_MESSAGE_SIZE env variable on the edge and restart it.", downlinkMsg.getSerializedSize(), clientMaxInboundMessageSize); |
|||
log.error("[{}][{}][{}] {} Message {}", tenantId, edge.getId(), sessionId, message, downlinkMsg); |
|||
ctx.getRuleProcessor().process(EdgeCommunicationFailureTrigger.builder().tenantId(tenantId) |
|||
.edgeId(edge.getId()).customerId(edge.getCustomerId()).edgeName(edge.getName()).failureMsg(message).error(error).build()); |
|||
ctx.getStatsCounterService().ifPresent(statsCounterService -> |
|||
statsCounterService.recordEvent(EdgeStatsKey.DOWNLINK_MSGS_PERMANENTLY_FAILED, edge.getTenantId(), edge.getId(), 1)); |
|||
sessionState.getPendingMsgsMap().remove(downlinkMsg.getDownlinkMsgId()); |
|||
} else { |
|||
sendDownlinkMsg(ResponseMsg.newBuilder() |
|||
.setDownlinkMsg(downlinkMsg) |
|||
.build()); |
|||
} |
|||
} |
|||
if (attempt < MAX_DOWNLINK_ATTEMPTS) { |
|||
scheduleDownlinkMsgsPackSend(attempt + 1); |
|||
} else { |
|||
String failureMsg = String.format("Failed to deliver messages: %s", copy); |
|||
log.warn("[{}][{}] Failed to deliver the batch after {} attempts. Next messages are going to be discarded {}", |
|||
tenantId, edge.getId(), MAX_DOWNLINK_ATTEMPTS, copy); |
|||
ctx.getRuleProcessor().process(EdgeCommunicationFailureTrigger.builder().tenantId(tenantId).edgeId(edge.getId()) |
|||
.customerId(edge.getCustomerId()).edgeName(edge.getName()).failureMsg(failureMsg) |
|||
.error("Failed to deliver messages after " + MAX_DOWNLINK_ATTEMPTS + " attempts").build()); |
|||
ctx.getStatsCounterService().ifPresent(statsCounterService -> |
|||
statsCounterService.recordEvent(EdgeStatsKey.DOWNLINK_MSGS_PERMANENTLY_FAILED, edge.getTenantId(), edge.getId(), copy.size())); |
|||
stopCurrentSendDownlinkMsgsTask(false); |
|||
} |
|||
} else { |
|||
stopCurrentSendDownlinkMsgsTask(false); |
|||
} |
|||
} catch (Exception e) { |
|||
log.warn("[{}][{}] Failed to send downlink msgs. Error msg {}", tenantId, edge.getId(), e.getMessage(), e); |
|||
stopCurrentSendDownlinkMsgsTask(true); |
|||
} |
|||
}; |
|||
|
|||
if (attempt == 1) { |
|||
sendDownlinkExecutorService.submit(sendDownlinkMsgsTask); |
|||
} else { |
|||
sessionState.setScheduledSendDownlinkTask( |
|||
sendDownlinkExecutorService.schedule( |
|||
sendDownlinkMsgsTask, |
|||
ctx.getEdgeEventStorageSettings().getSleepIntervalBetweenBatches(), |
|||
TimeUnit.MILLISECONDS) |
|||
); |
|||
} |
|||
} |
|||
|
|||
private void sendResponseMessage(int uplinkMsgId, boolean success, String errorMsg) { |
|||
UplinkResponseMsg.Builder responseBuilder = UplinkResponseMsg.newBuilder() |
|||
.setUplinkMsgId(uplinkMsgId) |
|||
.setSuccess(success); |
|||
if (errorMsg != null) { |
|||
responseBuilder.setErrorMsg(errorMsg); |
|||
} |
|||
sendDownlinkMsg(ResponseMsg.newBuilder() |
|||
.setUplinkResponseMsg(responseBuilder.build()) |
|||
.build()); |
|||
} |
|||
|
|||
private void onDownlinkResponse(DownlinkResponseMsg msg) { |
|||
try { |
|||
if (msg.getSuccess()) { |
|||
sessionState.getPendingMsgsMap().remove(msg.getDownlinkMsgId()); |
|||
ctx.getStatsCounterService().ifPresent(statsCounterService -> |
|||
statsCounterService.recordEvent(EdgeStatsKey.DOWNLINK_MSGS_PUSHED, edge.getTenantId(), edge.getId(), 1)); |
|||
log.debug("[{}][{}][{}] Msg has been processed successfully! Msg Id: [{}], Msg: {}", tenantId, edge.getId(), sessionId, msg.getDownlinkMsgId(), msg); |
|||
} else { |
|||
log.debug("[{}][{}][{}] Msg processing failed! Msg Id: [{}], Error msg: {}", tenantId, edge.getId(), sessionId, msg.getDownlinkMsgId(), msg.getErrorMsg()); |
|||
DownlinkMsg downlinkMsg = sessionState.getPendingMsgsMap().get(msg.getDownlinkMsgId()); |
|||
// if NOT timeseries or attributes failures - ack failed downlink
|
|||
if (downlinkMsg.getEntityDataCount() == 0) { |
|||
sessionState.getPendingMsgsMap().remove(msg.getDownlinkMsgId()); |
|||
} |
|||
} |
|||
if (sessionState.getPendingMsgsMap().isEmpty()) { |
|||
log.debug("[{}][{}][{}] Pending msgs map is empty. Stopping current iteration", tenantId, edge.getId(), sessionId); |
|||
stopCurrentSendDownlinkMsgsTask(false); |
|||
} |
|||
} catch (Exception e) { |
|||
log.error("[{}][{}] Can't process downlink response message [{}]", tenantId, edge.getId(), msg, e); |
|||
} |
|||
} |
|||
|
|||
public void processHighPriorityEvents() { |
|||
try { |
|||
if (isConnected() && !isSyncInProgress()) { |
|||
if (highPriorityQueue.isEmpty()) { |
|||
return; |
|||
} |
|||
List<EdgeEvent> highPriorityEvents = new ArrayList<>(); |
|||
EdgeEvent event; |
|||
while ((event = highPriorityQueue.poll()) != null) { |
|||
highPriorityEvents.add(event); |
|||
} |
|||
log.trace("[{}][{}] Sending high priority events {}", tenantId, edge.getId(), highPriorityEvents.size()); |
|||
List<DownlinkMsg> downlinkMsgsPack = convertToDownlinkMsgsPack(highPriorityEvents); |
|||
sendDownlinkMsgsPack(downlinkMsgsPack).get(); |
|||
} |
|||
} catch (Exception e) { |
|||
log.error("[{}] Failed to process high priority events", edge.getId(), e); |
|||
} |
|||
} |
|||
|
|||
public ListenableFuture<Boolean> processEdgeEvents() throws Exception { |
|||
SettableFuture<Boolean> result = SettableFuture.create(); |
|||
if (isConnected() && !isSyncInProgress()) { |
|||
Pair<Long, Long> startTsAndSeqId = getQueueStartTsAndSeqId().get(); |
|||
previousStartTs = startTsAndSeqId.getFirst(); |
|||
previousStartSeqId = startTsAndSeqId.getSecond(); |
|||
GeneralEdgeEventFetcher fetcher = new GeneralEdgeEventFetcher( |
|||
previousStartTs, |
|||
previousStartSeqId, |
|||
false, |
|||
Integer.toUnsignedLong(ctx.getEdgeEventStorageSettings().getMaxReadRecordsCount()), |
|||
ctx.getEdgeEventService(), |
|||
ctx.getEdgeEventStorageSettings().getMisorderingCompensationMillis()); |
|||
log.trace("[{}][{}] starting processing edge events, previousStartTs = {}, previousStartSeqId = {}", |
|||
tenantId, edge.getId(), previousStartTs, previousStartSeqId); |
|||
Futures.addCallback(startProcessingEdgeEvents(fetcher), new FutureCallback<>() { |
|||
@Override |
|||
public void onSuccess(@Nullable Pair<Long, Long> newStartTsAndSeqId) { |
|||
if (newStartTsAndSeqId != null) { |
|||
ListenableFuture<AttributesSaveResult> updateFuture = updateQueueStartTsAndSeqId(newStartTsAndSeqId); |
|||
Futures.addCallback(updateFuture, new FutureCallback<>() { |
|||
@Override |
|||
public void onSuccess(@Nullable AttributesSaveResult saveResult) { |
|||
log.debug("[{}][{}] queue offset was updated [{}]", tenantId, edge.getId(), newStartTsAndSeqId); |
|||
boolean newEventsAvailable; |
|||
if (fetcher.isSeqIdNewCycleStarted()) { |
|||
newEventsAvailable = isNewEdgeEventsAvailable(); |
|||
} else { |
|||
newEventsAvailable = isSeqIdStartedNewCycle(); |
|||
if (!newEventsAvailable) { |
|||
newEventsAvailable = isNewEdgeEventsAvailable(); |
|||
} |
|||
} |
|||
result.set(newEventsAvailable); |
|||
} |
|||
|
|||
@Override |
|||
public void onFailure(Throwable t) { |
|||
log.error("[{}][{}] Failed to update queue offset [{}]", tenantId, edge.getId(), newStartTsAndSeqId, t); |
|||
result.setException(t); |
|||
} |
|||
}, ctx.getGrpcCallbackExecutorService()); |
|||
} else { |
|||
log.trace("[{}][{}] newStartTsAndSeqId is null. Skipping iteration without db update", tenantId, edge.getId()); |
|||
result.set(Boolean.FALSE); |
|||
} |
|||
} |
|||
|
|||
@Override |
|||
public void onFailure(Throwable t) { |
|||
log.error("[{}][{}] Failed to process events", tenantId, edge.getId(), t); |
|||
result.setException(t); |
|||
} |
|||
}, ctx.getGrpcCallbackExecutorService()); |
|||
} else { |
|||
if (isSyncInProgress()) { |
|||
log.trace("[{}][{}] edge sync is not completed yet. Skipping iteration", tenantId, edge.getId()); |
|||
result.set(Boolean.TRUE); |
|||
} else { |
|||
log.trace("[{}][{}] edge is not connected. Skipping iteration", tenantId, edge.getId()); |
|||
result.set(null); |
|||
} |
|||
} |
|||
return result; |
|||
} |
|||
|
|||
protected List<DownlinkMsg> convertToDownlinkMsgsPack(List<EdgeEvent> edgeEvents) { |
|||
List<DownlinkMsg> result = new ArrayList<>(); |
|||
List<EdgeEvent> filtered = EdgeMsgConstructorUtils.mergeAndFilterDownlinkDuplicates(edgeEvents); |
|||
for (EdgeEvent edgeEvent : filtered) { |
|||
log.trace("[{}][{}] converting edge event to downlink msg [{}]", tenantId, edge.getId(), edgeEvent); |
|||
DownlinkMsg downlinkMsg = null; |
|||
try { |
|||
switch (edgeEvent.getAction()) { |
|||
case UPDATED, ADDED, DELETED, ASSIGNED_TO_EDGE, UNASSIGNED_FROM_EDGE, ALARM_ACK, ALARM_CLEAR, |
|||
ALARM_DELETE, CREDENTIALS_UPDATED, RELATION_ADD_OR_UPDATE, RELATION_DELETED, RPC_CALL, |
|||
ASSIGNED_TO_CUSTOMER, UNASSIGNED_FROM_CUSTOMER, ADDED_COMMENT, UPDATED_COMMENT, DELETED_COMMENT -> { |
|||
downlinkMsg = convertEntityEventToDownlink(edgeEvent); |
|||
if (downlinkMsg != null && downlinkMsg.getWidgetTypeUpdateMsgCount() > 0) { |
|||
log.trace("[{}][{}] widgetTypeUpdateMsg message processed, downlinkMsgId = {}", tenantId, edge.getId(), downlinkMsg.getDownlinkMsgId()); |
|||
} else { |
|||
log.trace("[{}][{}] entity message processed [{}]", tenantId, edge.getId(), downlinkMsg); |
|||
} |
|||
} |
|||
case ATTRIBUTES_UPDATED, POST_ATTRIBUTES, ATTRIBUTES_DELETED, TIMESERIES_UPDATED -> |
|||
downlinkMsg = ctx.getTelemetryProcessor().convertTelemetryEventToDownlink(edge, edgeEvent); |
|||
default -> log.warn("[{}][{}] Unsupported action type [{}]", tenantId, edge.getId(), edgeEvent.getAction()); |
|||
} |
|||
} catch (Exception e) { |
|||
log.trace("[{}][{}] Exception during converting edge event to downlink msg", tenantId, edge.getId(), e); |
|||
} |
|||
if (downlinkMsg != null) { |
|||
result.add(downlinkMsg); |
|||
} |
|||
} |
|||
return result; |
|||
} |
|||
|
|||
private ListenableFuture<Pair<Long, Long>> getQueueStartTsAndSeqId() { |
|||
ListenableFuture<List<AttributeKvEntry>> future = |
|||
ctx.getAttributesService().find(edge.getTenantId(), edge.getId(), AttributeScope.SERVER_SCOPE, Arrays.asList(QUEUE_START_TS_ATTR_KEY, QUEUE_START_SEQ_ID_ATTR_KEY)); |
|||
return Futures.transform(future, attributeKvEntries -> { |
|||
long startTs = 0L; |
|||
long startSeqId = 0L; |
|||
for (AttributeKvEntry attributeKvEntry : attributeKvEntries) { |
|||
if (QUEUE_START_TS_ATTR_KEY.equals(attributeKvEntry.getKey())) { |
|||
startTs = attributeKvEntry.getLongValue().isPresent() ? attributeKvEntry.getLongValue().get() : 0L; |
|||
} |
|||
if (QUEUE_START_SEQ_ID_ATTR_KEY.equals(attributeKvEntry.getKey())) { |
|||
startSeqId = attributeKvEntry.getLongValue().isPresent() ? attributeKvEntry.getLongValue().get() : 0L; |
|||
} |
|||
} |
|||
if (startSeqId == 0L) { |
|||
startSeqId = findStartSeqIdFromOldestEventIfAny(); |
|||
} |
|||
return Pair.of(startTs, startSeqId); |
|||
}, ctx.getGrpcCallbackExecutorService()); |
|||
} |
|||
|
|||
private boolean isSeqIdStartedNewCycle() { |
|||
try { |
|||
log.trace("[{}][{}][{}] Checking if seq id started new cycle", tenantId, edge.getId(), sessionId); |
|||
TimePageLink pageLink = new TimePageLink(ctx.getEdgeEventStorageSettings().getMaxReadRecordsCount(), 0, null, null, newStartTs, System.currentTimeMillis()); |
|||
PageData<EdgeEvent> edgeEvents = ctx.getEdgeEventService().findEdgeEvents(edge.getTenantId(), edge.getId(), 0L, previousStartSeqId == 0 ? null : previousStartSeqId - 1, pageLink); |
|||
boolean result = !edgeEvents.getData().isEmpty(); |
|||
log.trace("[{}][{}][{}] Result of check if seq id started new cycle, result = {}", tenantId, edge.getId(), sessionId, result); |
|||
return result; |
|||
} catch (Exception e) { |
|||
log.error("[{}][{}][{}] Failed to execute isSeqIdStartedNewCycle", tenantId, edge.getId(), sessionId, e); |
|||
} |
|||
return false; |
|||
} |
|||
|
|||
private boolean isNewEdgeEventsAvailable() { |
|||
try { |
|||
log.trace("[{}][{}][{}] Checking if new edge events available", tenantId, edge.getId(), sessionId); |
|||
TimePageLink pageLink = new TimePageLink(ctx.getEdgeEventStorageSettings().getMaxReadRecordsCount(), 0, null, null, newStartTs, System.currentTimeMillis()); |
|||
PageData<EdgeEvent> edgeEvents = ctx.getEdgeEventService().findEdgeEvents(edge.getTenantId(), edge.getId(), newStartSeqId, null, pageLink); |
|||
boolean result = !edgeEvents.getData().isEmpty() || !highPriorityQueue.isEmpty(); |
|||
log.trace("[{}][{}][{}] Result of check if new edge events available, result = {}", tenantId, edge.getId(), sessionId, result); |
|||
return result; |
|||
} catch (Exception e) { |
|||
log.error("[{}][{}][{}] Failed to execute isNewEdgeEventsAvailable", tenantId, edge.getId(), sessionId, e); |
|||
} |
|||
return false; |
|||
} |
|||
|
|||
private long findStartSeqIdFromOldestEventIfAny() { |
|||
long startSeqId = 0L; |
|||
try { |
|||
TimePageLink pageLink = new TimePageLink(1, 0, null, null, null, null); |
|||
PageData<EdgeEvent> edgeEvents = ctx.getEdgeEventService().findEdgeEvents(edge.getTenantId(), edge.getId(), null, null, pageLink); |
|||
if (!edgeEvents.getData().isEmpty()) { |
|||
startSeqId = edgeEvents.getData().get(0).getSeqId() - 1; |
|||
} |
|||
} catch (Exception e) { |
|||
log.error("[{}][{}][{}] Failed to execute findStartSeqIdFromOldestEventIfAny", tenantId, edge.getId(), sessionId, e); |
|||
} |
|||
return startSeqId; |
|||
} |
|||
|
|||
private ListenableFuture<AttributesSaveResult> updateQueueStartTsAndSeqId(Pair<Long, Long> pair) { |
|||
newStartTs = pair.getFirst(); |
|||
newStartSeqId = pair.getSecond(); |
|||
log.trace("[{}] updateQueueStartTsAndSeqId [{}][{}][{}]", sessionId, edge.getId(), newStartTs, newStartSeqId); |
|||
List<AttributeKvEntry> attributes = List.of( |
|||
new BaseAttributeKvEntry(new LongDataEntry(QUEUE_START_TS_ATTR_KEY, newStartTs), System.currentTimeMillis()), |
|||
new BaseAttributeKvEntry(new LongDataEntry(QUEUE_START_SEQ_ID_ATTR_KEY, newStartSeqId), System.currentTimeMillis()) |
|||
); |
|||
return ctx.getAttributesService().save(edge.getTenantId(), edge.getId(), AttributeScope.SERVER_SCOPE, attributes); |
|||
} |
|||
|
|||
protected ListenableFuture<Pair<Long, Long>> startProcessingEdgeEvents(EdgeEventFetcher fetcher) { |
|||
SettableFuture<Pair<Long, Long>> result = SettableFuture.create(); |
|||
PageLink pageLink = fetcher.getPageLink(ctx.getEdgeEventStorageSettings().getMaxReadRecordsCount()); |
|||
processEdgeEvents(fetcher, pageLink, result); |
|||
return result; |
|||
} |
|||
|
|||
private void markSyncCompletedSendEdgeEventUpdate() { |
|||
syncInProgress = false; |
|||
ctx.getClusterService().onEdgeEventUpdate(new EdgeEventUpdateMsg(edge.getTenantId(), edge.getId())); |
|||
} |
|||
|
|||
private void stopCurrentSendDownlinkMsgsTask(Boolean isInterrupted) { |
|||
if (sessionState.getSendDownlinkMsgsFuture() != null && !sessionState.getSendDownlinkMsgsFuture().isDone()) { |
|||
sessionState.getSendDownlinkMsgsFuture().set(isInterrupted); |
|||
} |
|||
if (sessionState.getScheduledSendDownlinkTask() != null) { |
|||
sessionState.getScheduledSendDownlinkTask().cancel(true); |
|||
} |
|||
} |
|||
|
|||
private void sendDownlinkMsg(ResponseMsg responseMsg) { |
|||
if (isConnected()) { |
|||
String responseMsgStr = StringUtils.truncate(responseMsg.toString(), 10000); |
|||
log.trace("[{}][{}] Sending downlink msg [{}]", tenantId, edge.getId(), responseMsgStr); |
|||
downlinkMsgLock.lock(); |
|||
String downlinkMsgStr = responseMsg.hasDownlinkMsg() ? String.valueOf(responseMsg.getDownlinkMsg().getDownlinkMsgId()) : responseMsgStr; |
|||
try { |
|||
outputStream.onNext(responseMsg); |
|||
} catch (Exception e) { |
|||
log.trace("[{}][{}] Failed to send downlink message [{}]", tenantId, edge.getId(), downlinkMsgStr, e); |
|||
connected = false; |
|||
sessionCloseListener.accept(edge, sessionId); |
|||
} finally { |
|||
downlinkMsgLock.unlock(); |
|||
} |
|||
log.trace("[{}][{}] downlink msg successfully sent [{}]", tenantId, edge.getId(), downlinkMsgStr); |
|||
} |
|||
} |
|||
|
|||
protected DownlinkMsg convertEntityEventToDownlink(EdgeEvent edgeEvent) { |
|||
log.trace("[{}] Executing convertEntityEventToDownlink, edgeEvent [{}], action [{}]", edgeEvent.getTenantId(), edgeEvent, edgeEvent.getAction()); |
|||
if ((EdgeEventType.OAUTH2_CLIENT.equals(edgeEvent.getType()) || EdgeEventType.DOMAIN.equals(edgeEvent.getType())) && |
|||
(EdgeVersionUtils.isEdgeVersionOlderThan(edgeVersion, EdgeVersion.V_3_8_0))) { |
|||
return null; |
|||
} |
|||
|
|||
return ctx.getProcessor(edgeEvent.getType()).convertEdgeEventToDownlink(edgeEvent, edgeVersion); |
|||
} |
|||
|
|||
public void addEventToHighPriorityQueue(EdgeEvent edgeEvent) { |
|||
while (highPriorityQueue.size() > maxHighPriorityQueueSizePerSession) { |
|||
EdgeEvent oldestHighPriority = highPriorityQueue.poll(); |
|||
if (oldestHighPriority != null) { |
|||
log.warn("[{}][{}][{}] High priority queue is full. Removing oldest high priority event from queue {}", |
|||
tenantId, edge.getId(), sessionId, oldestHighPriority); |
|||
} |
|||
} |
|||
highPriorityQueue.add(edgeEvent); |
|||
ctx.getStatsCounterService().ifPresent(statsCounterService -> |
|||
statsCounterService.recordEvent(EdgeStatsKey.DOWNLINK_MSGS_ADDED, edge.getTenantId(), edgeEvent.getEdgeId(), 1)); |
|||
} |
|||
|
|||
protected ListenableFuture<List<Void>> processUplinkMsg(UplinkMsg uplinkMsg) { |
|||
List<ListenableFuture<Void>> result = new ArrayList<>(); |
|||
try { |
|||
if (uplinkMsg.getDeviceProfileUpdateMsgCount() > 0) { |
|||
for (DeviceProfileUpdateMsg deviceProfileUpdateMsg : uplinkMsg.getDeviceProfileUpdateMsgList()) { |
|||
result.add(ctx.getDeviceProfileProcessor().processDeviceProfileMsgFromEdge(edge.getTenantId(), edge, deviceProfileUpdateMsg)); |
|||
} |
|||
} |
|||
if (uplinkMsg.getDeviceUpdateMsgCount() > 0) { |
|||
for (DeviceUpdateMsg deviceUpdateMsg : uplinkMsg.getDeviceUpdateMsgList()) { |
|||
result.add(ctx.getDeviceProcessor().processDeviceMsgFromEdge(edge.getTenantId(), edge, deviceUpdateMsg)); |
|||
} |
|||
} |
|||
if (uplinkMsg.getDeviceCredentialsUpdateMsgCount() > 0) { |
|||
for (DeviceCredentialsUpdateMsg deviceCredentialsUpdateMsg : uplinkMsg.getDeviceCredentialsUpdateMsgList()) { |
|||
result.add(ctx.getDeviceProcessor().processDeviceCredentialsMsgFromEdge(edge.getTenantId(), edge.getId(), deviceCredentialsUpdateMsg)); |
|||
} |
|||
} |
|||
if (uplinkMsg.getAssetProfileUpdateMsgCount() > 0) { |
|||
for (AssetProfileUpdateMsg assetProfileUpdateMsg : uplinkMsg.getAssetProfileUpdateMsgList()) { |
|||
result.add(ctx.getAssetProfileProcessor().processAssetProfileMsgFromEdge(edge.getTenantId(), edge, assetProfileUpdateMsg)); |
|||
} |
|||
} |
|||
if (uplinkMsg.getAssetUpdateMsgCount() > 0) { |
|||
for (AssetUpdateMsg assetUpdateMsg : uplinkMsg.getAssetUpdateMsgList()) { |
|||
result.add(ctx.getAssetProcessor().processAssetMsgFromEdge(edge.getTenantId(), edge, assetUpdateMsg)); |
|||
} |
|||
} |
|||
if (uplinkMsg.getRuleChainUpdateMsgCount() > 0) { |
|||
for (RuleChainUpdateMsg ruleChainUpdateMsg : uplinkMsg.getRuleChainUpdateMsgList()) { |
|||
result.add(ctx.getRuleChainProcessor().processRuleChainMsgFromEdge(edge.getTenantId(), edge, ruleChainUpdateMsg)); |
|||
} |
|||
} |
|||
if (uplinkMsg.getRuleChainMetadataUpdateMsgCount() > 0) { |
|||
for (RuleChainMetadataUpdateMsg ruleChainMetadataUpdateMsg : uplinkMsg.getRuleChainMetadataUpdateMsgList()) { |
|||
result.add(ctx.getRuleChainProcessor().processRuleChainMetadataMsgFromEdge(edge.getTenantId(), edge, ruleChainMetadataUpdateMsg)); |
|||
} |
|||
} |
|||
if (uplinkMsg.getEntityViewUpdateMsgCount() > 0) { |
|||
for (EntityViewUpdateMsg entityViewUpdateMsg : uplinkMsg.getEntityViewUpdateMsgList()) { |
|||
result.add(ctx.getEntityViewProcessor().processEntityViewMsgFromEdge(edge.getTenantId(), edge, entityViewUpdateMsg)); |
|||
} |
|||
} |
|||
if (uplinkMsg.getEntityDataCount() > 0) { |
|||
for (EntityDataProto entityData : uplinkMsg.getEntityDataList()) { |
|||
result.addAll(ctx.getTelemetryProcessor().processTelemetryMsg(edge.getTenantId(), entityData)); |
|||
} |
|||
} |
|||
if (uplinkMsg.getAlarmUpdateMsgCount() > 0) { |
|||
for (AlarmUpdateMsg alarmUpdateMsg : uplinkMsg.getAlarmUpdateMsgList()) { |
|||
result.add(ctx.getAlarmProcessor().processAlarmMsgFromEdge(edge.getTenantId(), edge.getId(), alarmUpdateMsg)); |
|||
} |
|||
} |
|||
if (uplinkMsg.getAlarmCommentUpdateMsgCount() > 0) { |
|||
for (AlarmCommentUpdateMsg alarmCommentUpdateMsg : uplinkMsg.getAlarmCommentUpdateMsgList()) { |
|||
result.add(ctx.getAlarmCommentProcessor().processAlarmCommentMsgFromEdge(edge.getTenantId(), edge.getId(), alarmCommentUpdateMsg)); |
|||
} |
|||
} |
|||
if (uplinkMsg.getRelationUpdateMsgCount() > 0) { |
|||
for (RelationUpdateMsg relationUpdateMsg : uplinkMsg.getRelationUpdateMsgList()) { |
|||
result.add(ctx.getRelationProcessor().processRelationMsgFromEdge(edge.getTenantId(), edge, relationUpdateMsg)); |
|||
} |
|||
} |
|||
if (uplinkMsg.getDashboardUpdateMsgCount() > 0) { |
|||
for (DashboardUpdateMsg dashboardUpdateMsg : uplinkMsg.getDashboardUpdateMsgList()) { |
|||
result.add(ctx.getDashboardProcessor().processDashboardMsgFromEdge(edge.getTenantId(), edge, dashboardUpdateMsg)); |
|||
} |
|||
} |
|||
if (uplinkMsg.getResourceUpdateMsgCount() > 0) { |
|||
for (ResourceUpdateMsg resourceUpdateMsg : uplinkMsg.getResourceUpdateMsgList()) { |
|||
result.add(ctx.getResourceProcessor().processResourceMsgFromEdge(edge.getTenantId(), edge, resourceUpdateMsg)); |
|||
} |
|||
} |
|||
if (uplinkMsg.getRuleChainMetadataRequestMsgCount() > 0) { |
|||
for (RuleChainMetadataRequestMsg ruleChainMetadataRequestMsg : uplinkMsg.getRuleChainMetadataRequestMsgList()) { |
|||
result.add(ctx.getEdgeRequestsService().processRuleChainMetadataRequestMsg(edge.getTenantId(), edge, ruleChainMetadataRequestMsg)); |
|||
} |
|||
} |
|||
if (uplinkMsg.getAttributesRequestMsgCount() > 0) { |
|||
for (AttributesRequestMsg attributesRequestMsg : uplinkMsg.getAttributesRequestMsgList()) { |
|||
result.add(ctx.getEdgeRequestsService().processAttributesRequestMsg(edge.getTenantId(), edge, attributesRequestMsg)); |
|||
} |
|||
} |
|||
if (uplinkMsg.getRelationRequestMsgCount() > 0) { |
|||
for (RelationRequestMsg relationRequestMsg : uplinkMsg.getRelationRequestMsgList()) { |
|||
result.add(ctx.getEdgeRequestsService().processRelationRequestMsg(edge.getTenantId(), edge, relationRequestMsg)); |
|||
} |
|||
} |
|||
if (uplinkMsg.getCalculatedFieldRequestMsgCount() > 0) { |
|||
for (CalculatedFieldRequestMsg calculatedFieldRequestMsg : uplinkMsg.getCalculatedFieldRequestMsgList()) { |
|||
result.add(ctx.getEdgeRequestsService().processCalculatedFieldRequestMsg(edge.getTenantId(), edge, calculatedFieldRequestMsg)); |
|||
} |
|||
} |
|||
if (uplinkMsg.getUserCredentialsRequestMsgCount() > 0) { |
|||
for (UserCredentialsRequestMsg userCredentialsRequestMsg : uplinkMsg.getUserCredentialsRequestMsgList()) { |
|||
result.add(ctx.getEdgeRequestsService().processUserCredentialsRequestMsg(edge.getTenantId(), edge, userCredentialsRequestMsg)); |
|||
} |
|||
} |
|||
if (uplinkMsg.getDeviceCredentialsRequestMsgCount() > 0) { |
|||
for (DeviceCredentialsRequestMsg deviceCredentialsRequestMsg : uplinkMsg.getDeviceCredentialsRequestMsgList()) { |
|||
result.add(ctx.getEdgeRequestsService().processDeviceCredentialsRequestMsg(edge.getTenantId(), edge, deviceCredentialsRequestMsg)); |
|||
} |
|||
} |
|||
if (uplinkMsg.getDeviceRpcCallMsgCount() > 0) { |
|||
for (DeviceRpcCallMsg deviceRpcCallMsg : uplinkMsg.getDeviceRpcCallMsgList()) { |
|||
result.add(ctx.getDeviceProcessor().processDeviceRpcCallFromEdge(edge.getTenantId(), edge, deviceRpcCallMsg)); |
|||
} |
|||
} |
|||
if (uplinkMsg.getWidgetBundleTypesRequestMsgCount() > 0) { |
|||
for (WidgetBundleTypesRequestMsg widgetBundleTypesRequestMsg : uplinkMsg.getWidgetBundleTypesRequestMsgList()) { |
|||
result.add(ctx.getEdgeRequestsService().processWidgetBundleTypesRequestMsg(edge.getTenantId(), edge, widgetBundleTypesRequestMsg)); |
|||
} |
|||
} |
|||
if (uplinkMsg.getEntityViewsRequestMsgCount() > 0) { |
|||
for (EntityViewsRequestMsg entityViewRequestMsg : uplinkMsg.getEntityViewsRequestMsgList()) { |
|||
result.add(ctx.getEdgeRequestsService().processEntityViewsRequestMsg(edge.getTenantId(), edge, entityViewRequestMsg)); |
|||
} |
|||
} |
|||
if (uplinkMsg.getCalculatedFieldUpdateMsgCount() > 0) { |
|||
for (CalculatedFieldUpdateMsg calculatedFieldUpdateMsg : uplinkMsg.getCalculatedFieldUpdateMsgList()) { |
|||
result.add(ctx.getCalculatedFieldProcessor().processCalculatedFieldMsgFromEdge(edge.getTenantId(), edge, calculatedFieldUpdateMsg)); |
|||
} |
|||
} |
|||
if (uplinkMsg.getAiModelUpdateMsgCount() > 0) { |
|||
for (AiModelUpdateMsg aiModelUpdateMsg : uplinkMsg.getAiModelUpdateMsgList()) { |
|||
result.add(ctx.getAiModelProcessor().processAiModelMsgFromEdge(edge.getTenantId(), edge, aiModelUpdateMsg)); |
|||
} |
|||
} |
|||
if (uplinkMsg.getUserUpdateMsgCount() > 0) { |
|||
for (UserUpdateMsg userUpdateMsg : uplinkMsg.getUserUpdateMsgList()) { |
|||
sequenceDependencyLock.lock(); |
|||
try { |
|||
result.add(ctx.getUserProcessor().processUserMsgFromEdge(edge.getTenantId(), edge, userUpdateMsg)); |
|||
} finally { |
|||
sequenceDependencyLock.unlock(); |
|||
} |
|||
} |
|||
} |
|||
if (uplinkMsg.getUserCredentialsUpdateMsgCount() > 0) { |
|||
for (UserCredentialsUpdateMsg userCredentialsUpdateMsg : uplinkMsg.getUserCredentialsUpdateMsgList()) { |
|||
sequenceDependencyLock.lock(); |
|||
try { |
|||
result.add(ctx.getUserProcessor().processUserCredentialsMsgFromEdge(edge.getTenantId(), edge, userCredentialsUpdateMsg)); |
|||
} finally { |
|||
sequenceDependencyLock.unlock(); |
|||
} |
|||
} |
|||
} |
|||
} catch (Exception e) { |
|||
String failureMsg = String.format("Can't process uplink msg [%s] from edge", uplinkMsg); |
|||
log.trace("[{}][{}] Can't process uplink msg [{}]", tenantId, edge.getId(), uplinkMsg, e); |
|||
ctx.getRuleProcessor().process(EdgeCommunicationFailureTrigger.builder().tenantId(tenantId).edgeId(edge.getId()) |
|||
.customerId(edge.getCustomerId()).edgeName(edge.getName()).failureMsg(failureMsg).error(e.getMessage()).build()); |
|||
return Futures.immediateFailedFuture(e); |
|||
} |
|||
return Futures.allAsList(result); |
|||
} |
|||
|
|||
protected boolean destroy() { |
|||
return true; |
|||
} |
|||
|
|||
protected void cleanUp() {} |
|||
|
|||
@Override |
|||
public void close() { |
|||
log.debug("[{}][{}] Closing session", tenantId, sessionId); |
|||
connected = false; |
|||
try { |
|||
outputStream.onCompleted(); |
|||
} catch (Exception e) { |
|||
log.debug("[{}][{}] Failed to close output stream: {}", tenantId, sessionId, e.getMessage()); |
|||
} |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,238 @@ |
|||
/** |
|||
* Copyright © 2016-2026 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.rpc; |
|||
|
|||
import com.google.common.util.concurrent.Futures; |
|||
import com.google.common.util.concurrent.ListenableFuture; |
|||
import lombok.RequiredArgsConstructor; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
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.notification.rule.trigger.EdgeCommunicationFailureTrigger; |
|||
import org.thingsboard.server.gen.edge.v1.AiModelUpdateMsg; |
|||
import org.thingsboard.server.gen.edge.v1.AlarmCommentUpdateMsg; |
|||
import org.thingsboard.server.gen.edge.v1.AlarmUpdateMsg; |
|||
import org.thingsboard.server.gen.edge.v1.ApiKeyUpdateMsg; |
|||
import org.thingsboard.server.gen.edge.v1.AssetProfileUpdateMsg; |
|||
import org.thingsboard.server.gen.edge.v1.AssetUpdateMsg; |
|||
import org.thingsboard.server.gen.edge.v1.AttributesRequestMsg; |
|||
import org.thingsboard.server.gen.edge.v1.CalculatedFieldRequestMsg; |
|||
import org.thingsboard.server.gen.edge.v1.CalculatedFieldUpdateMsg; |
|||
import org.thingsboard.server.gen.edge.v1.DashboardUpdateMsg; |
|||
import org.thingsboard.server.gen.edge.v1.DeviceCredentialsRequestMsg; |
|||
import org.thingsboard.server.gen.edge.v1.DeviceCredentialsUpdateMsg; |
|||
import org.thingsboard.server.gen.edge.v1.DeviceProfileUpdateMsg; |
|||
import org.thingsboard.server.gen.edge.v1.DeviceRpcCallMsg; |
|||
import org.thingsboard.server.gen.edge.v1.DeviceUpdateMsg; |
|||
import org.thingsboard.server.gen.edge.v1.EntityDataProto; |
|||
import org.thingsboard.server.gen.edge.v1.EntityViewUpdateMsg; |
|||
import org.thingsboard.server.gen.edge.v1.EntityViewsRequestMsg; |
|||
import org.thingsboard.server.gen.edge.v1.RelationRequestMsg; |
|||
import org.thingsboard.server.gen.edge.v1.RelationUpdateMsg; |
|||
import org.thingsboard.server.gen.edge.v1.ResourceUpdateMsg; |
|||
import org.thingsboard.server.gen.edge.v1.RuleChainMetadataRequestMsg; |
|||
import org.thingsboard.server.gen.edge.v1.RuleChainMetadataUpdateMsg; |
|||
import org.thingsboard.server.gen.edge.v1.RuleChainUpdateMsg; |
|||
import org.thingsboard.server.gen.edge.v1.UplinkMsg; |
|||
import org.thingsboard.server.gen.edge.v1.UserCredentialsRequestMsg; |
|||
import org.thingsboard.server.gen.edge.v1.UserCredentialsUpdateMsg; |
|||
import org.thingsboard.server.gen.edge.v1.UserUpdateMsg; |
|||
import org.thingsboard.server.gen.edge.v1.WidgetBundleTypesRequestMsg; |
|||
import org.thingsboard.server.queue.util.TbCoreComponent; |
|||
import org.thingsboard.server.service.edge.EdgeContextComponent; |
|||
|
|||
import java.util.ArrayList; |
|||
import java.util.List; |
|||
|
|||
@Service |
|||
@Slf4j |
|||
@RequiredArgsConstructor |
|||
@ConditionalOnProperty(prefix = "edges", value = "enabled", havingValue = "true") |
|||
@TbCoreComponent |
|||
public class EdgeUplinkMessageDispatcher { |
|||
|
|||
private final EdgeContextComponent ctx; |
|||
|
|||
public ListenableFuture<List<Void>> processUplinkMsg(EdgeSessionState state, UplinkMsg uplinkMsg) { |
|||
Edge edge = state.getEdge(); |
|||
List<ListenableFuture<Void>> result = new ArrayList<>(); |
|||
try { |
|||
if (uplinkMsg.getDeviceProfileUpdateMsgCount() > 0) { |
|||
for (DeviceProfileUpdateMsg deviceProfileUpdateMsg : uplinkMsg.getDeviceProfileUpdateMsgList()) { |
|||
result.add(ctx.getDeviceProfileProcessor().processDeviceProfileMsgFromEdge(edge.getTenantId(), edge, deviceProfileUpdateMsg)); |
|||
} |
|||
} |
|||
if (uplinkMsg.getDeviceUpdateMsgCount() > 0) { |
|||
for (DeviceUpdateMsg deviceUpdateMsg : uplinkMsg.getDeviceUpdateMsgList()) { |
|||
result.add(ctx.getDeviceProcessor().processDeviceMsgFromEdge(edge.getTenantId(), edge, deviceUpdateMsg)); |
|||
} |
|||
} |
|||
if (uplinkMsg.getDeviceCredentialsUpdateMsgCount() > 0) { |
|||
for (DeviceCredentialsUpdateMsg deviceCredentialsUpdateMsg : uplinkMsg.getDeviceCredentialsUpdateMsgList()) { |
|||
result.add(ctx.getDeviceProcessor().processDeviceCredentialsMsgFromEdge(edge.getTenantId(), edge.getId(), deviceCredentialsUpdateMsg)); |
|||
} |
|||
} |
|||
if (uplinkMsg.getAssetProfileUpdateMsgCount() > 0) { |
|||
for (AssetProfileUpdateMsg assetProfileUpdateMsg : uplinkMsg.getAssetProfileUpdateMsgList()) { |
|||
result.add(ctx.getAssetProfileProcessor().processAssetProfileMsgFromEdge(edge.getTenantId(), edge, assetProfileUpdateMsg)); |
|||
} |
|||
} |
|||
if (uplinkMsg.getAssetUpdateMsgCount() > 0) { |
|||
for (AssetUpdateMsg assetUpdateMsg : uplinkMsg.getAssetUpdateMsgList()) { |
|||
result.add(ctx.getAssetProcessor().processAssetMsgFromEdge(edge.getTenantId(), edge, assetUpdateMsg)); |
|||
} |
|||
} |
|||
if (uplinkMsg.getRuleChainUpdateMsgCount() > 0) { |
|||
for (RuleChainUpdateMsg ruleChainUpdateMsg : uplinkMsg.getRuleChainUpdateMsgList()) { |
|||
result.add(ctx.getRuleChainProcessor().processRuleChainMsgFromEdge(edge.getTenantId(), edge, ruleChainUpdateMsg)); |
|||
} |
|||
} |
|||
if (uplinkMsg.getRuleChainMetadataUpdateMsgCount() > 0) { |
|||
for (RuleChainMetadataUpdateMsg ruleChainMetadataUpdateMsg : uplinkMsg.getRuleChainMetadataUpdateMsgList()) { |
|||
result.add(ctx.getRuleChainProcessor().processRuleChainMetadataMsgFromEdge(edge.getTenantId(), edge, ruleChainMetadataUpdateMsg)); |
|||
} |
|||
} |
|||
if (uplinkMsg.getEntityViewUpdateMsgCount() > 0) { |
|||
for (EntityViewUpdateMsg entityViewUpdateMsg : uplinkMsg.getEntityViewUpdateMsgList()) { |
|||
result.add(ctx.getEntityViewProcessor().processEntityViewMsgFromEdge(edge.getTenantId(), edge, entityViewUpdateMsg)); |
|||
} |
|||
} |
|||
if (uplinkMsg.getEntityDataCount() > 0) { |
|||
for (EntityDataProto entityData : uplinkMsg.getEntityDataList()) { |
|||
result.addAll(ctx.getTelemetryProcessor().processTelemetryMsg(edge.getTenantId(), entityData)); |
|||
} |
|||
} |
|||
if (uplinkMsg.getAlarmUpdateMsgCount() > 0) { |
|||
for (AlarmUpdateMsg alarmUpdateMsg : uplinkMsg.getAlarmUpdateMsgList()) { |
|||
result.add(ctx.getAlarmProcessor().processAlarmMsgFromEdge(edge.getTenantId(), edge.getId(), alarmUpdateMsg)); |
|||
} |
|||
} |
|||
if (uplinkMsg.getAlarmCommentUpdateMsgCount() > 0) { |
|||
for (AlarmCommentUpdateMsg alarmCommentUpdateMsg : uplinkMsg.getAlarmCommentUpdateMsgList()) { |
|||
result.add(ctx.getAlarmCommentProcessor().processAlarmCommentMsgFromEdge(edge.getTenantId(), edge.getId(), alarmCommentUpdateMsg)); |
|||
} |
|||
} |
|||
if (uplinkMsg.getRelationUpdateMsgCount() > 0) { |
|||
for (RelationUpdateMsg relationUpdateMsg : uplinkMsg.getRelationUpdateMsgList()) { |
|||
result.add(ctx.getRelationProcessor().processRelationMsgFromEdge(edge.getTenantId(), edge, relationUpdateMsg)); |
|||
} |
|||
} |
|||
if (uplinkMsg.getDashboardUpdateMsgCount() > 0) { |
|||
for (DashboardUpdateMsg dashboardUpdateMsg : uplinkMsg.getDashboardUpdateMsgList()) { |
|||
result.add(ctx.getDashboardProcessor().processDashboardMsgFromEdge(edge.getTenantId(), edge, dashboardUpdateMsg)); |
|||
} |
|||
} |
|||
if (uplinkMsg.getResourceUpdateMsgCount() > 0) { |
|||
for (ResourceUpdateMsg resourceUpdateMsg : uplinkMsg.getResourceUpdateMsgList()) { |
|||
result.add(ctx.getResourceProcessor().processResourceMsgFromEdge(edge.getTenantId(), edge, resourceUpdateMsg)); |
|||
} |
|||
} |
|||
if (uplinkMsg.getRuleChainMetadataRequestMsgCount() > 0) { |
|||
for (RuleChainMetadataRequestMsg ruleChainMetadataRequestMsg : uplinkMsg.getRuleChainMetadataRequestMsgList()) { |
|||
result.add(ctx.getEdgeRequestsService().processRuleChainMetadataRequestMsg(edge.getTenantId(), edge, ruleChainMetadataRequestMsg)); |
|||
} |
|||
} |
|||
if (uplinkMsg.getAttributesRequestMsgCount() > 0) { |
|||
for (AttributesRequestMsg attributesRequestMsg : uplinkMsg.getAttributesRequestMsgList()) { |
|||
result.add(ctx.getEdgeRequestsService().processAttributesRequestMsg(edge.getTenantId(), edge, attributesRequestMsg)); |
|||
} |
|||
} |
|||
if (uplinkMsg.getRelationRequestMsgCount() > 0) { |
|||
for (RelationRequestMsg relationRequestMsg : uplinkMsg.getRelationRequestMsgList()) { |
|||
result.add(ctx.getEdgeRequestsService().processRelationRequestMsg(edge.getTenantId(), edge, relationRequestMsg)); |
|||
} |
|||
} |
|||
if (uplinkMsg.getCalculatedFieldRequestMsgCount() > 0) { |
|||
for (CalculatedFieldRequestMsg calculatedFieldRequestMsg : uplinkMsg.getCalculatedFieldRequestMsgList()) { |
|||
result.add(ctx.getEdgeRequestsService().processCalculatedFieldRequestMsg(edge.getTenantId(), edge, calculatedFieldRequestMsg)); |
|||
} |
|||
} |
|||
if (uplinkMsg.getUserCredentialsRequestMsgCount() > 0) { |
|||
for (UserCredentialsRequestMsg userCredentialsRequestMsg : uplinkMsg.getUserCredentialsRequestMsgList()) { |
|||
result.add(ctx.getEdgeRequestsService().processUserCredentialsRequestMsg(edge.getTenantId(), edge, userCredentialsRequestMsg)); |
|||
} |
|||
} |
|||
if (uplinkMsg.getDeviceCredentialsRequestMsgCount() > 0) { |
|||
for (DeviceCredentialsRequestMsg deviceCredentialsRequestMsg : uplinkMsg.getDeviceCredentialsRequestMsgList()) { |
|||
result.add(ctx.getEdgeRequestsService().processDeviceCredentialsRequestMsg(edge.getTenantId(), edge, deviceCredentialsRequestMsg)); |
|||
} |
|||
} |
|||
if (uplinkMsg.getDeviceRpcCallMsgCount() > 0) { |
|||
for (DeviceRpcCallMsg deviceRpcCallMsg : uplinkMsg.getDeviceRpcCallMsgList()) { |
|||
result.add(ctx.getDeviceProcessor().processDeviceRpcCallFromEdge(edge.getTenantId(), edge, deviceRpcCallMsg)); |
|||
} |
|||
} |
|||
if (uplinkMsg.getWidgetBundleTypesRequestMsgCount() > 0) { |
|||
for (WidgetBundleTypesRequestMsg widgetBundleTypesRequestMsg : uplinkMsg.getWidgetBundleTypesRequestMsgList()) { |
|||
result.add(ctx.getEdgeRequestsService().processWidgetBundleTypesRequestMsg(edge.getTenantId(), edge, widgetBundleTypesRequestMsg)); |
|||
} |
|||
} |
|||
if (uplinkMsg.getEntityViewsRequestMsgCount() > 0) { |
|||
for (EntityViewsRequestMsg entityViewRequestMsg : uplinkMsg.getEntityViewsRequestMsgList()) { |
|||
result.add(ctx.getEdgeRequestsService().processEntityViewsRequestMsg(edge.getTenantId(), edge, entityViewRequestMsg)); |
|||
} |
|||
} |
|||
if (uplinkMsg.getCalculatedFieldUpdateMsgCount() > 0) { |
|||
for (CalculatedFieldUpdateMsg calculatedFieldUpdateMsg : uplinkMsg.getCalculatedFieldUpdateMsgList()) { |
|||
result.add(ctx.getCalculatedFieldProcessor().processCalculatedFieldMsgFromEdge(edge.getTenantId(), edge, calculatedFieldUpdateMsg)); |
|||
} |
|||
} |
|||
if (uplinkMsg.getAiModelUpdateMsgCount() > 0) { |
|||
for (AiModelUpdateMsg aiModelUpdateMsg : uplinkMsg.getAiModelUpdateMsgList()) { |
|||
result.add(ctx.getAiModelProcessor().processAiModelMsgFromEdge(edge.getTenantId(), edge, aiModelUpdateMsg)); |
|||
} |
|||
} |
|||
if (uplinkMsg.getUserUpdateMsgCount() > 0) { |
|||
for (UserUpdateMsg userUpdateMsg : uplinkMsg.getUserUpdateMsgList()) { |
|||
state.lockSequenceDependency(); |
|||
try { |
|||
result.add(ctx.getUserProcessor().processUserMsgFromEdge(edge.getTenantId(), edge, userUpdateMsg)); |
|||
} finally { |
|||
state.unlockSequenceDependency(); |
|||
} |
|||
} |
|||
} |
|||
if (uplinkMsg.getUserCredentialsUpdateMsgCount() > 0) { |
|||
for (UserCredentialsUpdateMsg userCredentialsUpdateMsg : uplinkMsg.getUserCredentialsUpdateMsgList()) { |
|||
state.lockSequenceDependency(); |
|||
try { |
|||
result.add(ctx.getUserProcessor().processUserCredentialsMsgFromEdge(edge.getTenantId(), edge, userCredentialsUpdateMsg)); |
|||
} finally { |
|||
state.unlockSequenceDependency(); |
|||
} |
|||
} |
|||
} |
|||
if (uplinkMsg.getApiKeyUpdateMsgCount() > 0) { |
|||
for (ApiKeyUpdateMsg apiKeyUpdateMsg : uplinkMsg.getApiKeyUpdateMsgList()) { |
|||
state.lockSequenceDependency(); |
|||
try { |
|||
result.add(ctx.getApiKeyProcessor().processApiKeyMsgFromEdge(edge.getTenantId(), edge, apiKeyUpdateMsg)); |
|||
} finally { |
|||
state.unlockSequenceDependency(); |
|||
} |
|||
} |
|||
} |
|||
} catch (Exception e) { |
|||
String failureMsg = String.format("Can't process uplink msg [%s] from edge", uplinkMsg); |
|||
log.trace("[{}][{}] Can't process uplink msg [{}]", edge.getTenantId(), edge.getId(), uplinkMsg, e); |
|||
ctx.getRuleProcessor().process(EdgeCommunicationFailureTrigger.builder().tenantId(edge.getTenantId()).edgeId(edge.getId()) |
|||
.customerId(edge.getCustomerId()).edgeName(edge.getName()).failureMsg(failureMsg).error(e.getMessage()).build()); |
|||
return Futures.immediateFailedFuture(e); |
|||
} |
|||
return Futures.allAsList(result); |
|||
} |
|||
} |
|||
Some files were not shown because too many files changed in this diff
Loading…
Reference in new issue