196 changed files with 6871 additions and 2002 deletions
@ -0,0 +1,32 @@ |
|||||
|
-- |
||||
|
-- Copyright © 2016-2020 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. |
||||
|
-- |
||||
|
|
||||
|
CREATE OR REPLACE PROCEDURE cleanup_edge_events_by_ttl(IN ttl bigint, INOUT deleted bigint) |
||||
|
LANGUAGE plpgsql AS |
||||
|
$$ |
||||
|
DECLARE |
||||
|
ttl_ts bigint; |
||||
|
ttl_deleted_count bigint DEFAULT 0; |
||||
|
BEGIN |
||||
|
IF ttl > 0 THEN |
||||
|
ttl_ts := (EXTRACT(EPOCH FROM current_timestamp) * 1000 - ttl::bigint * 1000)::bigint; |
||||
|
EXECUTE format( |
||||
|
'WITH deleted AS (DELETE FROM edge_event WHERE ts < %L::bigint RETURNING *) SELECT count(*) FROM deleted', ttl_ts) into ttl_deleted_count; |
||||
|
END IF; |
||||
|
RAISE NOTICE 'Edge events removed by ttl: %', ttl_deleted_count; |
||||
|
deleted := ttl_deleted_count; |
||||
|
END |
||||
|
$$; |
||||
@ -0,0 +1,71 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2020 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 lombok.extern.slf4j.Slf4j; |
||||
|
import org.springframework.beans.factory.annotation.Autowired; |
||||
|
import org.springframework.security.access.prepost.PreAuthorize; |
||||
|
import org.springframework.web.bind.annotation.PathVariable; |
||||
|
import org.springframework.web.bind.annotation.RequestMapping; |
||||
|
import org.springframework.web.bind.annotation.RequestMethod; |
||||
|
import org.springframework.web.bind.annotation.RequestParam; |
||||
|
import org.springframework.web.bind.annotation.ResponseBody; |
||||
|
import org.springframework.web.bind.annotation.RestController; |
||||
|
import org.thingsboard.server.common.data.edge.EdgeEvent; |
||||
|
import org.thingsboard.server.common.data.exception.ThingsboardException; |
||||
|
import org.thingsboard.server.common.data.id.EdgeId; |
||||
|
import org.thingsboard.server.common.data.id.TenantId; |
||||
|
import org.thingsboard.server.common.data.page.PageData; |
||||
|
import org.thingsboard.server.common.data.page.TimePageLink; |
||||
|
import org.thingsboard.server.dao.edge.EdgeEventService; |
||||
|
import org.thingsboard.server.queue.util.TbCoreComponent; |
||||
|
import org.thingsboard.server.service.security.permission.Operation; |
||||
|
|
||||
|
@Slf4j |
||||
|
@RestController |
||||
|
@TbCoreComponent |
||||
|
@RequestMapping("/api") |
||||
|
public class EdgeEventController extends BaseController { |
||||
|
|
||||
|
@Autowired |
||||
|
private EdgeEventService edgeEventService; |
||||
|
|
||||
|
public static final String EDGE_ID = "edgeId"; |
||||
|
|
||||
|
@PreAuthorize("hasAnyAuthority('SYS_ADMIN', 'TENANT_ADMIN', 'CUSTOMER_USER')") |
||||
|
@RequestMapping(value = "/edge/{edgeId}/events", method = RequestMethod.GET) |
||||
|
@ResponseBody |
||||
|
public PageData<EdgeEvent> getEdgeEvents( |
||||
|
@PathVariable(EDGE_ID) String strEdgeId, |
||||
|
@RequestParam int pageSize, |
||||
|
@RequestParam int page, |
||||
|
@RequestParam(required = false) String textSearch, |
||||
|
@RequestParam(required = false) String sortProperty, |
||||
|
@RequestParam(required = false) String sortOrder, |
||||
|
@RequestParam(required = false) Long startTime, |
||||
|
@RequestParam(required = false) Long endTime) throws ThingsboardException { |
||||
|
checkParameter(EDGE_ID, strEdgeId); |
||||
|
try { |
||||
|
TenantId tenantId = getCurrentUser().getTenantId(); |
||||
|
EdgeId edgeId = new EdgeId(toUUID(strEdgeId)); |
||||
|
checkEdgeId(edgeId, Operation.READ); |
||||
|
TimePageLink pageLink = createTimePageLink(pageSize, page, textSearch, sortProperty, sortOrder, startTime, endTime); |
||||
|
return checkNotNull(edgeEventService.findEdgeEvents(tenantId, edgeId, pageLink, false)); |
||||
|
} catch (Exception e) { |
||||
|
throw handleException(e); |
||||
|
} |
||||
|
} |
||||
|
} |
||||
File diff suppressed because it is too large
@ -0,0 +1,26 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2020 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 org.thingsboard.server.common.data.edge.Edge; |
||||
|
import org.thingsboard.server.common.data.id.EdgeId; |
||||
|
|
||||
|
public interface EdgeRpcService { |
||||
|
|
||||
|
void updateEdge(Edge edge); |
||||
|
|
||||
|
void deleteEdge(EdgeId edgeId); |
||||
|
} |
||||
@ -0,0 +1,38 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2020 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.constructor; |
||||
|
|
||||
|
import lombok.extern.slf4j.Slf4j; |
||||
|
import org.springframework.stereotype.Component; |
||||
|
import org.thingsboard.server.common.data.AdminSettings; |
||||
|
import org.thingsboard.server.dao.util.mapping.JacksonUtil; |
||||
|
import org.thingsboard.server.gen.edge.AdminSettingsUpdateMsg; |
||||
|
|
||||
|
@Slf4j |
||||
|
@Component |
||||
|
public class AdminSettingsMsgConstructor { |
||||
|
|
||||
|
public AdminSettingsUpdateMsg constructAdminSettingsUpdateMsg(AdminSettings adminSettings) { |
||||
|
AdminSettingsUpdateMsg.Builder builder = AdminSettingsUpdateMsg.newBuilder() |
||||
|
.setKey(adminSettings.getKey()) |
||||
|
.setJsonValue(JacksonUtil.toString(adminSettings.getJsonValue())); |
||||
|
if (adminSettings.getId() != null) { |
||||
|
builder.setIsSystem(true); |
||||
|
} |
||||
|
return builder.build(); |
||||
|
} |
||||
|
|
||||
|
} |
||||
@ -0,0 +1,72 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2020 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.constructor; |
||||
|
|
||||
|
import lombok.extern.slf4j.Slf4j; |
||||
|
import org.springframework.stereotype.Component; |
||||
|
import org.thingsboard.server.common.data.Customer; |
||||
|
import org.thingsboard.server.common.data.id.CustomerId; |
||||
|
import org.thingsboard.server.dao.util.mapping.JacksonUtil; |
||||
|
import org.thingsboard.server.gen.edge.CustomerUpdateMsg; |
||||
|
import org.thingsboard.server.gen.edge.UpdateMsgType; |
||||
|
|
||||
|
@Component |
||||
|
@Slf4j |
||||
|
public class CustomerMsgConstructor { |
||||
|
|
||||
|
public CustomerUpdateMsg constructCustomerUpdatedMsg(UpdateMsgType msgType, Customer customer) { |
||||
|
CustomerUpdateMsg.Builder builder = CustomerUpdateMsg.newBuilder() |
||||
|
.setMsgType(msgType) |
||||
|
.setIdMSB(customer.getId().getId().getMostSignificantBits()) |
||||
|
.setIdLSB(customer.getId().getId().getLeastSignificantBits()) |
||||
|
.setTitle(customer.getTitle()); |
||||
|
if (customer.getCountry() != null) { |
||||
|
builder.setCountry(customer.getCountry()); |
||||
|
} |
||||
|
if (customer.getState() != null) { |
||||
|
builder.setState(customer.getState()); |
||||
|
} |
||||
|
if (customer.getCity() != null) { |
||||
|
builder.setCity(customer.getCity()); |
||||
|
} |
||||
|
if (customer.getAddress() != null) { |
||||
|
builder.setAddress(customer.getAddress()); |
||||
|
} |
||||
|
if (customer.getAddress2() != null) { |
||||
|
builder.setAddress2(customer.getAddress2()); |
||||
|
} |
||||
|
if (customer.getZip() != null) { |
||||
|
builder.setZip(customer.getZip()); |
||||
|
} |
||||
|
if (customer.getPhone() != null) { |
||||
|
builder.setPhone(customer.getPhone()); |
||||
|
} |
||||
|
if (customer.getEmail() != null) { |
||||
|
builder.setEmail(customer.getEmail()); |
||||
|
} |
||||
|
if (customer.getAdditionalInfo() != null) { |
||||
|
builder.setAdditionalInfo(JacksonUtil.toString(customer.getAdditionalInfo())); |
||||
|
} |
||||
|
return builder.build(); |
||||
|
} |
||||
|
|
||||
|
public CustomerUpdateMsg constructCustomerDeleteMsg(CustomerId customerId) { |
||||
|
return CustomerUpdateMsg.newBuilder() |
||||
|
.setMsgType(UpdateMsgType.ENTITY_DELETED_RPC_MESSAGE) |
||||
|
.setIdMSB(customerId.getId().getMostSignificantBits()) |
||||
|
.setIdLSB(customerId.getId().getLeastSignificantBits()).build(); |
||||
|
} |
||||
|
} |
||||
@ -0,0 +1,97 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2020 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.constructor; |
||||
|
|
||||
|
import com.fasterxml.jackson.databind.JsonNode; |
||||
|
import com.fasterxml.jackson.databind.ObjectMapper; |
||||
|
import lombok.extern.slf4j.Slf4j; |
||||
|
import org.springframework.stereotype.Component; |
||||
|
import org.thingsboard.rule.engine.api.RuleEngineDeviceRpcRequest; |
||||
|
import org.thingsboard.server.common.data.Device; |
||||
|
import org.thingsboard.server.common.data.id.CustomerId; |
||||
|
import org.thingsboard.server.common.data.id.DeviceId; |
||||
|
import org.thingsboard.server.common.data.security.DeviceCredentials; |
||||
|
import org.thingsboard.server.dao.util.mapping.JacksonUtil; |
||||
|
import org.thingsboard.server.gen.edge.DeviceCredentialsUpdateMsg; |
||||
|
import org.thingsboard.server.gen.edge.DeviceRpcCallMsg; |
||||
|
import org.thingsboard.server.gen.edge.DeviceUpdateMsg; |
||||
|
import org.thingsboard.server.gen.edge.RpcRequestMsg; |
||||
|
import org.thingsboard.server.gen.edge.UpdateMsgType; |
||||
|
|
||||
|
@Component |
||||
|
@Slf4j |
||||
|
public class DeviceMsgConstructor { |
||||
|
|
||||
|
protected static final ObjectMapper mapper = new ObjectMapper(); |
||||
|
|
||||
|
public DeviceUpdateMsg constructDeviceUpdatedMsg(UpdateMsgType msgType, Device device, CustomerId customerId) { |
||||
|
DeviceUpdateMsg.Builder builder = DeviceUpdateMsg.newBuilder() |
||||
|
.setMsgType(msgType) |
||||
|
.setIdMSB(device.getId().getId().getMostSignificantBits()) |
||||
|
.setIdLSB(device.getId().getId().getLeastSignificantBits()) |
||||
|
.setName(device.getName()) |
||||
|
.setType(device.getType()); |
||||
|
if (device.getLabel() != null) { |
||||
|
builder.setLabel(device.getLabel()); |
||||
|
} |
||||
|
if (customerId != null) { |
||||
|
builder.setCustomerIdMSB(customerId.getId().getMostSignificantBits()); |
||||
|
builder.setCustomerIdLSB(customerId.getId().getLeastSignificantBits()); |
||||
|
} |
||||
|
if (device.getAdditionalInfo() != null) { |
||||
|
builder.setAdditionalInfo(JacksonUtil.toString(device.getAdditionalInfo())); |
||||
|
} |
||||
|
return builder.build(); |
||||
|
} |
||||
|
|
||||
|
public DeviceCredentialsUpdateMsg constructDeviceCredentialsUpdatedMsg(DeviceCredentials deviceCredentials) { |
||||
|
DeviceCredentialsUpdateMsg.Builder builder = DeviceCredentialsUpdateMsg.newBuilder() |
||||
|
.setDeviceIdMSB(deviceCredentials.getDeviceId().getId().getMostSignificantBits()) |
||||
|
.setDeviceIdLSB(deviceCredentials.getDeviceId().getId().getLeastSignificantBits()); |
||||
|
if (deviceCredentials.getCredentialsType() != null) { |
||||
|
builder.setCredentialsType(deviceCredentials.getCredentialsType().name()) |
||||
|
.setCredentialsId(deviceCredentials.getCredentialsId()); |
||||
|
} |
||||
|
if (deviceCredentials.getCredentialsValue() != null) { |
||||
|
builder.setCredentialsValue(deviceCredentials.getCredentialsValue()); |
||||
|
} |
||||
|
return builder.build(); |
||||
|
} |
||||
|
|
||||
|
public DeviceUpdateMsg constructDeviceDeleteMsg(DeviceId deviceId) { |
||||
|
return DeviceUpdateMsg.newBuilder() |
||||
|
.setMsgType(UpdateMsgType.ENTITY_DELETED_RPC_MESSAGE) |
||||
|
.setIdMSB(deviceId.getId().getMostSignificantBits()) |
||||
|
.setIdLSB(deviceId.getId().getLeastSignificantBits()).build(); |
||||
|
} |
||||
|
|
||||
|
public DeviceRpcCallMsg constructDeviceRpcCallMsg(JsonNode body) { |
||||
|
RuleEngineDeviceRpcRequest request = mapper.convertValue(body, RuleEngineDeviceRpcRequest.class); |
||||
|
RpcRequestMsg.Builder requestBuilder = RpcRequestMsg.newBuilder(); |
||||
|
requestBuilder.setMethod(request.getMethod()); |
||||
|
requestBuilder.setParams(request.getBody()); |
||||
|
DeviceRpcCallMsg.Builder builder = DeviceRpcCallMsg.newBuilder() |
||||
|
.setDeviceIdMSB(request.getDeviceId().getId().getMostSignificantBits()) |
||||
|
.setDeviceIdLSB(request.getDeviceId().getId().getLeastSignificantBits()) |
||||
|
.setRequestIdMSB(request.getRequestUUID().getMostSignificantBits()) |
||||
|
.setRequestIdLSB(request.getRequestUUID().getLeastSignificantBits()) |
||||
|
.setExpirationTime(request.getExpirationTime()) |
||||
|
.setOriginServiceId(request.getOriginServiceId()) |
||||
|
.setOneway(request.isOneway()) |
||||
|
.setRequestMsg(requestBuilder.build()); |
||||
|
return builder.build(); |
||||
|
} |
||||
|
} |
||||
@ -1,69 +0,0 @@ |
|||||
/** |
|
||||
* Copyright © 2016-2020 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.constructor; |
|
||||
|
|
||||
import lombok.extern.slf4j.Slf4j; |
|
||||
import org.springframework.beans.factory.annotation.Autowired; |
|
||||
import org.springframework.stereotype.Component; |
|
||||
import org.thingsboard.server.common.data.Device; |
|
||||
import org.thingsboard.server.common.data.id.DeviceId; |
|
||||
import org.thingsboard.server.common.data.security.DeviceCredentials; |
|
||||
import org.thingsboard.server.dao.device.DeviceCredentialsService; |
|
||||
import org.thingsboard.server.gen.edge.DeviceUpdateMsg; |
|
||||
import org.thingsboard.server.gen.edge.UpdateMsgType; |
|
||||
|
|
||||
@Component |
|
||||
@Slf4j |
|
||||
public class DeviceUpdateMsgConstructor { |
|
||||
|
|
||||
@Autowired |
|
||||
private DeviceCredentialsService deviceCredentialsService; |
|
||||
|
|
||||
public DeviceUpdateMsg constructDeviceUpdatedMsg(UpdateMsgType msgType, Device device) { |
|
||||
DeviceUpdateMsg.Builder builder = DeviceUpdateMsg.newBuilder() |
|
||||
.setMsgType(msgType) |
|
||||
.setIdMSB(device.getId().getId().getMostSignificantBits()) |
|
||||
.setIdLSB(device.getId().getId().getLeastSignificantBits()) |
|
||||
.setName(device.getName()) |
|
||||
.setType(device.getType()); |
|
||||
if (device.getLabel() != null) { |
|
||||
builder.setLabel(device.getLabel()); |
|
||||
} |
|
||||
if (msgType.equals(UpdateMsgType.ENTITY_CREATED_RPC_MESSAGE) || |
|
||||
msgType.equals(UpdateMsgType.ENTITY_UPDATED_RPC_MESSAGE) || |
|
||||
msgType.equals(UpdateMsgType.DEVICE_CONFLICT_RPC_MESSAGE)) { |
|
||||
DeviceCredentials deviceCredentials |
|
||||
= deviceCredentialsService.findDeviceCredentialsByDeviceId(device.getTenantId(), device.getId()); |
|
||||
if (deviceCredentials != null) { |
|
||||
if (deviceCredentials.getCredentialsType() != null) { |
|
||||
builder.setCredentialsType(deviceCredentials.getCredentialsType().name()) |
|
||||
.setCredentialsId(deviceCredentials.getCredentialsId()); |
|
||||
} |
|
||||
if (deviceCredentials.getCredentialsValue() != null) { |
|
||||
builder.setCredentialsValue(deviceCredentials.getCredentialsValue()); |
|
||||
} |
|
||||
} |
|
||||
} |
|
||||
return builder.build(); |
|
||||
} |
|
||||
|
|
||||
public DeviceUpdateMsg constructDeviceDeleteMsg(DeviceId deviceId) { |
|
||||
return DeviceUpdateMsg.newBuilder() |
|
||||
.setMsgType(UpdateMsgType.ENTITY_DELETED_RPC_MESSAGE) |
|
||||
.setIdMSB(deviceId.getId().getMostSignificantBits()) |
|
||||
.setIdLSB(deviceId.getId().getLeastSignificantBits()).build(); |
|
||||
} |
|
||||
} |
|
||||
@ -0,0 +1,61 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2020 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.constructor; |
||||
|
|
||||
|
import lombok.extern.slf4j.Slf4j; |
||||
|
import org.springframework.stereotype.Component; |
||||
|
import org.thingsboard.server.common.data.id.TenantId; |
||||
|
import org.thingsboard.server.common.data.id.WidgetTypeId; |
||||
|
import org.thingsboard.server.common.data.widget.WidgetType; |
||||
|
import org.thingsboard.server.dao.util.mapping.JacksonUtil; |
||||
|
import org.thingsboard.server.gen.edge.UpdateMsgType; |
||||
|
import org.thingsboard.server.gen.edge.WidgetTypeUpdateMsg; |
||||
|
|
||||
|
@Component |
||||
|
@Slf4j |
||||
|
public class WidgetTypeMsgConstructor { |
||||
|
|
||||
|
public WidgetTypeUpdateMsg constructWidgetTypeUpdateMsg(UpdateMsgType msgType, WidgetType widgetType) { |
||||
|
WidgetTypeUpdateMsg.Builder builder = WidgetTypeUpdateMsg.newBuilder() |
||||
|
.setMsgType(msgType) |
||||
|
.setIdMSB(widgetType.getId().getId().getMostSignificantBits()) |
||||
|
.setIdLSB(widgetType.getId().getId().getLeastSignificantBits()); |
||||
|
if (widgetType.getBundleAlias() != null) { |
||||
|
builder.setBundleAlias(widgetType.getBundleAlias()); |
||||
|
} |
||||
|
if (widgetType.getAlias() != null) { |
||||
|
builder.setAlias(widgetType.getAlias()); |
||||
|
} |
||||
|
if (widgetType.getName() != null) { |
||||
|
builder.setName(widgetType.getName()); |
||||
|
} |
||||
|
if (widgetType.getDescriptor() != null) { |
||||
|
builder.setDescriptorJson(JacksonUtil.toString(widgetType.getDescriptor())); |
||||
|
} |
||||
|
if (widgetType.getTenantId().equals(TenantId.SYS_TENANT_ID)) { |
||||
|
builder.setIsSystem(true); |
||||
|
} |
||||
|
return builder.build(); |
||||
|
} |
||||
|
|
||||
|
public WidgetTypeUpdateMsg constructWidgetTypeDeleteMsg(WidgetTypeId widgetTypeId) { |
||||
|
return WidgetTypeUpdateMsg.newBuilder() |
||||
|
.setMsgType(UpdateMsgType.ENTITY_DELETED_RPC_MESSAGE) |
||||
|
.setIdMSB(widgetTypeId.getId().getMostSignificantBits()) |
||||
|
.setIdLSB(widgetTypeId.getId().getLeastSignificantBits()) |
||||
|
.build(); |
||||
|
} |
||||
|
} |
||||
@ -0,0 +1,54 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2020 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.constructor; |
||||
|
|
||||
|
import com.google.protobuf.ByteString; |
||||
|
import lombok.extern.slf4j.Slf4j; |
||||
|
import org.springframework.stereotype.Component; |
||||
|
import org.thingsboard.server.common.data.id.TenantId; |
||||
|
import org.thingsboard.server.common.data.id.WidgetsBundleId; |
||||
|
import org.thingsboard.server.common.data.widget.WidgetsBundle; |
||||
|
import org.thingsboard.server.gen.edge.UpdateMsgType; |
||||
|
import org.thingsboard.server.gen.edge.WidgetsBundleUpdateMsg; |
||||
|
|
||||
|
@Component |
||||
|
@Slf4j |
||||
|
public class WidgetsBundleMsgConstructor { |
||||
|
|
||||
|
public WidgetsBundleUpdateMsg constructWidgetsBundleUpdateMsg(UpdateMsgType msgType, WidgetsBundle widgetsBundle) { |
||||
|
WidgetsBundleUpdateMsg.Builder builder = WidgetsBundleUpdateMsg.newBuilder() |
||||
|
.setMsgType(msgType) |
||||
|
.setIdMSB(widgetsBundle.getId().getId().getMostSignificantBits()) |
||||
|
.setIdLSB(widgetsBundle.getId().getId().getLeastSignificantBits()) |
||||
|
.setTitle(widgetsBundle.getTitle()) |
||||
|
.setAlias(widgetsBundle.getAlias()); |
||||
|
if (widgetsBundle.getImage() != null) { |
||||
|
builder.setImage(ByteString.copyFrom(widgetsBundle.getImage())); |
||||
|
} |
||||
|
if (widgetsBundle.getTenantId().equals(TenantId.SYS_TENANT_ID)) { |
||||
|
builder.setIsSystem(true); |
||||
|
} |
||||
|
return builder.build(); |
||||
|
} |
||||
|
|
||||
|
public WidgetsBundleUpdateMsg constructWidgetsBundleDeleteMsg(WidgetsBundleId widgetsBundleId) { |
||||
|
return WidgetsBundleUpdateMsg.newBuilder() |
||||
|
.setMsgType(UpdateMsgType.ENTITY_DELETED_RPC_MESSAGE) |
||||
|
.setIdMSB(widgetsBundleId.getId().getMostSignificantBits()) |
||||
|
.setIdLSB(widgetsBundleId.getId().getLeastSignificantBits()) |
||||
|
.build(); |
||||
|
} |
||||
|
} |
||||
@ -0,0 +1,98 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2020 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.processor; |
||||
|
|
||||
|
import com.google.common.util.concurrent.Futures; |
||||
|
import com.google.common.util.concurrent.ListenableFuture; |
||||
|
import lombok.extern.slf4j.Slf4j; |
||||
|
import org.springframework.stereotype.Component; |
||||
|
import org.thingsboard.server.common.data.EntityType; |
||||
|
import org.thingsboard.server.common.data.alarm.Alarm; |
||||
|
import org.thingsboard.server.common.data.alarm.AlarmSeverity; |
||||
|
import org.thingsboard.server.common.data.alarm.AlarmStatus; |
||||
|
import org.thingsboard.server.common.data.id.EntityId; |
||||
|
import org.thingsboard.server.common.data.id.TenantId; |
||||
|
import org.thingsboard.server.gen.edge.AlarmUpdateMsg; |
||||
|
import org.thingsboard.server.queue.util.TbCoreComponent; |
||||
|
|
||||
|
@Component |
||||
|
@Slf4j |
||||
|
@TbCoreComponent |
||||
|
public class AlarmProcessor extends BaseProcessor { |
||||
|
|
||||
|
public ListenableFuture<Void> onAlarmUpdate(TenantId tenantId, AlarmUpdateMsg alarmUpdateMsg) { |
||||
|
EntityId originatorId = getAlarmOriginator(tenantId, alarmUpdateMsg.getOriginatorName(), |
||||
|
EntityType.valueOf(alarmUpdateMsg.getOriginatorType())); |
||||
|
if (originatorId == null) { |
||||
|
return Futures.immediateFuture(null); |
||||
|
} |
||||
|
try { |
||||
|
Alarm existentAlarm = alarmService.findLatestByOriginatorAndType(tenantId, originatorId, alarmUpdateMsg.getType()).get(); |
||||
|
switch (alarmUpdateMsg.getMsgType()) { |
||||
|
case ENTITY_CREATED_RPC_MESSAGE: |
||||
|
case ENTITY_UPDATED_RPC_MESSAGE: |
||||
|
if (existentAlarm == null || existentAlarm.getStatus().isCleared()) { |
||||
|
existentAlarm = new Alarm(); |
||||
|
existentAlarm.setTenantId(tenantId); |
||||
|
existentAlarm.setType(alarmUpdateMsg.getName()); |
||||
|
existentAlarm.setOriginator(originatorId); |
||||
|
existentAlarm.setSeverity(AlarmSeverity.valueOf(alarmUpdateMsg.getSeverity())); |
||||
|
existentAlarm.setStartTs(alarmUpdateMsg.getStartTs()); |
||||
|
existentAlarm.setClearTs(alarmUpdateMsg.getClearTs()); |
||||
|
existentAlarm.setPropagate(alarmUpdateMsg.getPropagate()); |
||||
|
} |
||||
|
existentAlarm.setStatus(AlarmStatus.valueOf(alarmUpdateMsg.getStatus())); |
||||
|
existentAlarm.setAckTs(alarmUpdateMsg.getAckTs()); |
||||
|
existentAlarm.setEndTs(alarmUpdateMsg.getEndTs()); |
||||
|
existentAlarm.setDetails(mapper.readTree(alarmUpdateMsg.getDetails())); |
||||
|
alarmService.createOrUpdateAlarm(existentAlarm); |
||||
|
break; |
||||
|
case ALARM_ACK_RPC_MESSAGE: |
||||
|
if (existentAlarm != null) { |
||||
|
alarmService.ackAlarm(tenantId, existentAlarm.getId(), alarmUpdateMsg.getAckTs()); |
||||
|
} |
||||
|
break; |
||||
|
case ALARM_CLEAR_RPC_MESSAGE: |
||||
|
if (existentAlarm != null) { |
||||
|
alarmService.clearAlarm(tenantId, existentAlarm.getId(), mapper.readTree(alarmUpdateMsg.getDetails()), alarmUpdateMsg.getAckTs()); |
||||
|
} |
||||
|
break; |
||||
|
case ENTITY_DELETED_RPC_MESSAGE: |
||||
|
if (existentAlarm != null) { |
||||
|
alarmService.deleteAlarm(tenantId, existentAlarm.getId()); |
||||
|
} |
||||
|
break; |
||||
|
} |
||||
|
return Futures.immediateFuture(null); |
||||
|
} catch (Exception e) { |
||||
|
log.error("Failed to process alarm update msg [{}]", alarmUpdateMsg, e); |
||||
|
return Futures.immediateFailedFuture(new RuntimeException("Failed to process alarm update msg", e)); |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
private EntityId getAlarmOriginator(TenantId tenantId, String entityName, EntityType entityType) { |
||||
|
switch (entityType) { |
||||
|
case DEVICE: |
||||
|
return deviceService.findDeviceByTenantIdAndName(tenantId, entityName).getId(); |
||||
|
case ASSET: |
||||
|
return assetService.findAssetByTenantIdAndName(tenantId, entityName).getId(); |
||||
|
case ENTITY_VIEW: |
||||
|
return entityViewService.findEntityViewByTenantIdAndName(tenantId, entityName).getId(); |
||||
|
default: |
||||
|
return null; |
||||
|
} |
||||
|
} |
||||
|
} |
||||
@ -0,0 +1,113 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2020 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.processor; |
||||
|
|
||||
|
import com.fasterxml.jackson.databind.JsonNode; |
||||
|
import com.fasterxml.jackson.databind.ObjectMapper; |
||||
|
import com.google.common.util.concurrent.ListenableFuture; |
||||
|
import lombok.extern.slf4j.Slf4j; |
||||
|
import org.springframework.beans.factory.annotation.Autowired; |
||||
|
import org.thingsboard.server.common.data.audit.ActionType; |
||||
|
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.EntityId; |
||||
|
import org.thingsboard.server.common.data.id.TenantId; |
||||
|
import org.thingsboard.server.dao.alarm.AlarmService; |
||||
|
import org.thingsboard.server.dao.asset.AssetService; |
||||
|
import org.thingsboard.server.dao.attributes.AttributesService; |
||||
|
import org.thingsboard.server.dao.customer.CustomerService; |
||||
|
import org.thingsboard.server.dao.dashboard.DashboardService; |
||||
|
import org.thingsboard.server.dao.device.DeviceCredentialsService; |
||||
|
import org.thingsboard.server.dao.device.DeviceService; |
||||
|
import org.thingsboard.server.dao.edge.EdgeEventService; |
||||
|
import org.thingsboard.server.dao.entityview.EntityViewService; |
||||
|
import org.thingsboard.server.dao.relation.RelationService; |
||||
|
import org.thingsboard.server.dao.user.UserService; |
||||
|
import org.thingsboard.server.service.executors.DbCallbackExecutorService; |
||||
|
import org.thingsboard.server.service.queue.TbClusterService; |
||||
|
import org.thingsboard.server.service.rpc.TbRuleEngineDeviceRpcService; |
||||
|
import org.thingsboard.server.service.state.DeviceStateService; |
||||
|
|
||||
|
@Slf4j |
||||
|
public abstract class BaseProcessor { |
||||
|
|
||||
|
protected static final ObjectMapper mapper = new ObjectMapper(); |
||||
|
|
||||
|
@Autowired |
||||
|
protected AlarmService alarmService; |
||||
|
|
||||
|
@Autowired |
||||
|
protected DeviceService deviceService; |
||||
|
|
||||
|
@Autowired |
||||
|
protected DashboardService dashboardService; |
||||
|
|
||||
|
@Autowired |
||||
|
protected AssetService assetService; |
||||
|
|
||||
|
@Autowired |
||||
|
protected EntityViewService entityViewService; |
||||
|
|
||||
|
@Autowired |
||||
|
protected CustomerService customerService; |
||||
|
|
||||
|
@Autowired |
||||
|
protected UserService userService; |
||||
|
|
||||
|
@Autowired |
||||
|
protected RelationService relationService; |
||||
|
|
||||
|
@Autowired |
||||
|
protected DeviceCredentialsService deviceCredentialsService; |
||||
|
|
||||
|
@Autowired |
||||
|
protected AttributesService attributesService; |
||||
|
|
||||
|
@Autowired |
||||
|
protected TbClusterService tbClusterService; |
||||
|
|
||||
|
@Autowired |
||||
|
protected DeviceStateService deviceStateService; |
||||
|
|
||||
|
@Autowired |
||||
|
protected EdgeEventService edgeEventService; |
||||
|
|
||||
|
@Autowired |
||||
|
protected DbCallbackExecutorService dbCallbackExecutorService; |
||||
|
|
||||
|
protected ListenableFuture<EdgeEvent> saveEdgeEvent(TenantId tenantId, |
||||
|
EdgeId edgeId, |
||||
|
EdgeEventType type, |
||||
|
ActionType action, |
||||
|
EntityId entityId, |
||||
|
JsonNode body) { |
||||
|
log.debug("Pushing event to edge queue. tenantId [{}], edgeId [{}], type[{}], " + |
||||
|
"action [{}], entityId [{}], body [{}]", |
||||
|
tenantId, edgeId, type, action, entityId, body); |
||||
|
|
||||
|
EdgeEvent edgeEvent = new EdgeEvent(); |
||||
|
edgeEvent.setTenantId(tenantId); |
||||
|
edgeEvent.setEdgeId(edgeId); |
||||
|
edgeEvent.setType(type); |
||||
|
edgeEvent.setAction(action.name()); |
||||
|
if (entityId != null) { |
||||
|
edgeEvent.setEntityId(entityId.getId()); |
||||
|
} |
||||
|
edgeEvent.setBody(body); |
||||
|
return edgeEventService.saveAsync(edgeEvent); |
||||
|
} |
||||
|
} |
||||
@ -0,0 +1,252 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2020 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.processor; |
||||
|
|
||||
|
import com.fasterxml.jackson.core.JsonProcessingException; |
||||
|
import com.fasterxml.jackson.databind.node.ObjectNode; |
||||
|
import com.google.common.util.concurrent.Futures; |
||||
|
import com.google.common.util.concurrent.ListenableFuture; |
||||
|
import com.google.common.util.concurrent.SettableFuture; |
||||
|
import lombok.extern.slf4j.Slf4j; |
||||
|
import org.apache.commons.lang.RandomStringUtils; |
||||
|
import org.apache.commons.lang.StringUtils; |
||||
|
import org.springframework.stereotype.Component; |
||||
|
import org.thingsboard.rule.engine.api.RpcError; |
||||
|
import org.thingsboard.server.common.data.DataConstants; |
||||
|
import org.thingsboard.server.common.data.Device; |
||||
|
import org.thingsboard.server.common.data.audit.ActionType; |
||||
|
import org.thingsboard.server.common.data.edge.Edge; |
||||
|
import org.thingsboard.server.common.data.edge.EdgeEventType; |
||||
|
import org.thingsboard.server.common.data.id.CustomerId; |
||||
|
import org.thingsboard.server.common.data.id.DeviceId; |
||||
|
import org.thingsboard.server.common.data.id.EdgeId; |
||||
|
import org.thingsboard.server.common.data.id.EntityId; |
||||
|
import org.thingsboard.server.common.data.id.TenantId; |
||||
|
import org.thingsboard.server.common.data.relation.EntityRelation; |
||||
|
import org.thingsboard.server.common.data.relation.RelationTypeGroup; |
||||
|
import org.thingsboard.server.common.data.security.DeviceCredentials; |
||||
|
import org.thingsboard.server.common.data.security.DeviceCredentialsType; |
||||
|
import org.thingsboard.server.common.msg.TbMsg; |
||||
|
import org.thingsboard.server.common.msg.TbMsgDataType; |
||||
|
import org.thingsboard.server.common.msg.TbMsgMetaData; |
||||
|
import org.thingsboard.server.dao.util.mapping.JacksonUtil; |
||||
|
import org.thingsboard.server.gen.edge.DeviceCredentialsUpdateMsg; |
||||
|
import org.thingsboard.server.gen.edge.DeviceRpcCallMsg; |
||||
|
import org.thingsboard.server.gen.edge.DeviceUpdateMsg; |
||||
|
import org.thingsboard.server.queue.TbQueueCallback; |
||||
|
import org.thingsboard.server.queue.TbQueueMsgMetadata; |
||||
|
import org.thingsboard.server.queue.util.TbCoreComponent; |
||||
|
import org.thingsboard.server.service.rpc.FromDeviceRpcResponse; |
||||
|
|
||||
|
import java.util.UUID; |
||||
|
import java.util.concurrent.locks.ReentrantLock; |
||||
|
|
||||
|
@Component |
||||
|
@Slf4j |
||||
|
@TbCoreComponent |
||||
|
public class DeviceProcessor extends BaseProcessor { |
||||
|
|
||||
|
private static final ReentrantLock deviceCreationLock = new ReentrantLock(); |
||||
|
|
||||
|
public ListenableFuture<Void> onDeviceUpdate(TenantId tenantId, Edge edge, DeviceUpdateMsg deviceUpdateMsg) { |
||||
|
DeviceId edgeDeviceId = new DeviceId(new UUID(deviceUpdateMsg.getIdMSB(), deviceUpdateMsg.getIdLSB())); |
||||
|
switch (deviceUpdateMsg.getMsgType()) { |
||||
|
case ENTITY_CREATED_RPC_MESSAGE: |
||||
|
String deviceName = deviceUpdateMsg.getName(); |
||||
|
Device device = deviceService.findDeviceByTenantIdAndName(tenantId, deviceName); |
||||
|
if (device != null) { |
||||
|
// device with this name already exists on the cloud - update ID on the edge
|
||||
|
if (!device.getId().equals(edgeDeviceId)) { |
||||
|
saveEdgeEvent(tenantId, edge.getId(), EdgeEventType.DEVICE, ActionType.ENTITY_EXISTS_REQUEST, device.getId(), null); |
||||
|
} |
||||
|
} else { |
||||
|
Device deviceById = deviceService.findDeviceById(edge.getTenantId(), edgeDeviceId); |
||||
|
if (deviceById != null) { |
||||
|
// this ID already used by other device - create new device and update ID on the edge
|
||||
|
device = createDevice(tenantId, edge, deviceUpdateMsg); |
||||
|
saveEdgeEvent(tenantId, edge.getId(), EdgeEventType.DEVICE, ActionType.ENTITY_EXISTS_REQUEST, device.getId(), null); |
||||
|
} else { |
||||
|
device = createDevice(tenantId, edge, deviceUpdateMsg); |
||||
|
} |
||||
|
} |
||||
|
// TODO: voba - assign device only in case device is not assigned yet. Missing functionality to check this relation prior assignment
|
||||
|
deviceService.assignDeviceToEdge(edge.getTenantId(), device.getId(), edge.getId()); |
||||
|
break; |
||||
|
case ENTITY_UPDATED_RPC_MESSAGE: |
||||
|
updateDevice(tenantId, edge, deviceUpdateMsg); |
||||
|
break; |
||||
|
case ENTITY_DELETED_RPC_MESSAGE: |
||||
|
Device deviceToDelete = deviceService.findDeviceById(tenantId, edgeDeviceId); |
||||
|
if (deviceToDelete != null) { |
||||
|
deviceService.unassignDeviceFromEdge(tenantId, edgeDeviceId, edge.getId()); |
||||
|
} |
||||
|
break; |
||||
|
case UNRECOGNIZED: |
||||
|
log.error("Unsupported msg type {}", deviceUpdateMsg.getMsgType()); |
||||
|
return Futures.immediateFailedFuture(new RuntimeException("Unsupported msg type " + deviceUpdateMsg.getMsgType())); |
||||
|
} |
||||
|
return Futures.immediateFuture(null); |
||||
|
} |
||||
|
|
||||
|
|
||||
|
public ListenableFuture<Void> onDeviceCredentialsUpdate(TenantId tenantId, DeviceCredentialsUpdateMsg deviceCredentialsUpdateMsg) { |
||||
|
log.debug("Executing onDeviceCredentialsUpdate, deviceCredentialsUpdateMsg [{}]", deviceCredentialsUpdateMsg); |
||||
|
DeviceId deviceId = new DeviceId(new UUID(deviceCredentialsUpdateMsg.getDeviceIdMSB(), deviceCredentialsUpdateMsg.getDeviceIdLSB())); |
||||
|
ListenableFuture<Device> deviceFuture = deviceService.findDeviceByIdAsync(tenantId, deviceId); |
||||
|
return Futures.transform(deviceFuture, device -> { |
||||
|
if (device != null) { |
||||
|
log.debug("Updating device credentials for device [{}]. New device credentials Id [{}], value [{}]", |
||||
|
device.getName(), deviceCredentialsUpdateMsg.getCredentialsId(), deviceCredentialsUpdateMsg.getCredentialsValue()); |
||||
|
try { |
||||
|
DeviceCredentials deviceCredentials = deviceCredentialsService.findDeviceCredentialsByDeviceId(tenantId, device.getId()); |
||||
|
deviceCredentials.setCredentialsType(DeviceCredentialsType.valueOf(deviceCredentialsUpdateMsg.getCredentialsType())); |
||||
|
deviceCredentials.setCredentialsId(deviceCredentialsUpdateMsg.getCredentialsId()); |
||||
|
deviceCredentials.setCredentialsValue(deviceCredentialsUpdateMsg.getCredentialsValue()); |
||||
|
deviceCredentialsService.updateDeviceCredentials(tenantId, deviceCredentials); |
||||
|
} catch (Exception e) { |
||||
|
log.error("Can't update device credentials for device [{}], deviceCredentialsUpdateMsg [{}]", device.getName(), deviceCredentialsUpdateMsg, e); |
||||
|
throw new RuntimeException(e); |
||||
|
} |
||||
|
} |
||||
|
return null; |
||||
|
}, dbCallbackExecutorService); |
||||
|
} |
||||
|
|
||||
|
|
||||
|
private void updateDevice(TenantId tenantId, Edge edge, DeviceUpdateMsg deviceUpdateMsg) { |
||||
|
DeviceId deviceId = new DeviceId(new UUID(deviceUpdateMsg.getIdMSB(), deviceUpdateMsg.getIdLSB())); |
||||
|
Device device = deviceService.findDeviceById(tenantId, deviceId); |
||||
|
device.setName(deviceUpdateMsg.getName()); |
||||
|
device.setType(deviceUpdateMsg.getType()); |
||||
|
device.setLabel(deviceUpdateMsg.getLabel()); |
||||
|
device.setAdditionalInfo(JacksonUtil.toJsonNode(deviceUpdateMsg.getAdditionalInfo())); |
||||
|
deviceService.saveDevice(device); |
||||
|
|
||||
|
saveEdgeEvent(tenantId, edge.getId(), EdgeEventType.DEVICE, ActionType.CREDENTIALS_REQUEST, deviceId, null); |
||||
|
} |
||||
|
|
||||
|
private Device createDevice(TenantId tenantId, Edge edge, DeviceUpdateMsg deviceUpdateMsg) { |
||||
|
Device device; |
||||
|
try { |
||||
|
deviceCreationLock.lock(); |
||||
|
DeviceId deviceId = new DeviceId(new UUID(deviceUpdateMsg.getIdMSB(), deviceUpdateMsg.getIdLSB())); |
||||
|
device = new Device(); |
||||
|
device.setTenantId(edge.getTenantId()); |
||||
|
device.setCustomerId(edge.getCustomerId()); |
||||
|
device.setId(deviceId); |
||||
|
device.setName(deviceUpdateMsg.getName()); |
||||
|
device.setType(deviceUpdateMsg.getType()); |
||||
|
device.setLabel(deviceUpdateMsg.getLabel()); |
||||
|
device.setAdditionalInfo(JacksonUtil.toJsonNode(deviceUpdateMsg.getAdditionalInfo())); |
||||
|
device = deviceService.saveDevice(device); |
||||
|
createDeviceCredentials(device); |
||||
|
createRelationFromEdge(tenantId, edge.getId(), device.getId()); |
||||
|
deviceStateService.onDeviceAdded(device); |
||||
|
pushDeviceCreatedEventToRuleEngine(tenantId, edge, device); |
||||
|
|
||||
|
saveEdgeEvent(tenantId, edge.getId(), EdgeEventType.DEVICE, ActionType.CREDENTIALS_REQUEST, deviceId, null); |
||||
|
} finally { |
||||
|
deviceCreationLock.unlock(); |
||||
|
} |
||||
|
return device; |
||||
|
} |
||||
|
|
||||
|
private void createRelationFromEdge(TenantId tenantId, EdgeId edgeId, EntityId entityId) { |
||||
|
EntityRelation relation = new EntityRelation(); |
||||
|
relation.setFrom(edgeId); |
||||
|
relation.setTo(entityId); |
||||
|
relation.setTypeGroup(RelationTypeGroup.COMMON); |
||||
|
relation.setType(EntityRelation.EDGE_TYPE); |
||||
|
relationService.saveRelation(tenantId, relation); |
||||
|
} |
||||
|
|
||||
|
private void createDeviceCredentials(Device device) { |
||||
|
DeviceCredentials deviceCredentials = new DeviceCredentials(); |
||||
|
deviceCredentials.setDeviceId(device.getId()); |
||||
|
deviceCredentials.setCredentialsType(DeviceCredentialsType.ACCESS_TOKEN); |
||||
|
deviceCredentials.setCredentialsId(RandomStringUtils.randomAlphanumeric(20)); |
||||
|
deviceCredentialsService.createDeviceCredentials(device.getTenantId(), deviceCredentials); |
||||
|
} |
||||
|
|
||||
|
private void pushDeviceCreatedEventToRuleEngine(TenantId tenantId, Edge edge, Device device) { |
||||
|
try { |
||||
|
DeviceId deviceId = device.getId(); |
||||
|
ObjectNode entityNode = mapper.valueToTree(device); |
||||
|
TbMsg tbMsg = TbMsg.newMsg(DataConstants.ENTITY_CREATED, deviceId, |
||||
|
getActionTbMsgMetaData(edge, device.getCustomerId()), TbMsgDataType.JSON, mapper.writeValueAsString(entityNode)); |
||||
|
tbClusterService.pushMsgToRuleEngine(tenantId, deviceId, tbMsg, new TbQueueCallback() { |
||||
|
@Override |
||||
|
public void onSuccess(TbQueueMsgMetadata metadata) { |
||||
|
// TODO: voba - handle success
|
||||
|
log.debug("Successfully send ENTITY_CREATED EVENT to rule engine [{}]", device); |
||||
|
} |
||||
|
|
||||
|
@Override |
||||
|
public void onFailure(Throwable t) { |
||||
|
// TODO: voba - handle failure
|
||||
|
log.debug("Failed to send ENTITY_CREATED EVENT to rule engine [{}]", device, t); |
||||
|
} |
||||
|
|
||||
|
; |
||||
|
}); |
||||
|
} catch (JsonProcessingException | IllegalArgumentException e) { |
||||
|
log.warn("[{}] Failed to push device action to rule engine: {}", device.getId(), DataConstants.ENTITY_CREATED, e); |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
|
||||
|
private TbMsgMetaData getActionTbMsgMetaData(Edge edge, CustomerId customerId) { |
||||
|
TbMsgMetaData metaData = getTbMsgMetaData(edge); |
||||
|
if (customerId != null && !customerId.isNullUid()) { |
||||
|
metaData.putValue("customerId", customerId.toString()); |
||||
|
} |
||||
|
return metaData; |
||||
|
} |
||||
|
|
||||
|
|
||||
|
private TbMsgMetaData getTbMsgMetaData(Edge edge) { |
||||
|
TbMsgMetaData metaData = new TbMsgMetaData(); |
||||
|
metaData.putValue("edgeId", edge.getId().toString()); |
||||
|
metaData.putValue("edgeName", edge.getName()); |
||||
|
return metaData; |
||||
|
} |
||||
|
|
||||
|
public ListenableFuture<Void> processDeviceRpcCallResponseMsg(TenantId tenantId, DeviceRpcCallMsg deviceRpcCallMsg) { |
||||
|
SettableFuture<Void> futureToSet = SettableFuture.create(); |
||||
|
UUID uuid = new UUID(deviceRpcCallMsg.getRequestIdMSB(), deviceRpcCallMsg.getRequestIdLSB()); |
||||
|
FromDeviceRpcResponse response; |
||||
|
if (!StringUtils.isEmpty(deviceRpcCallMsg.getResponseMsg().getError())) { |
||||
|
response = new FromDeviceRpcResponse(uuid, null, RpcError.valueOf(deviceRpcCallMsg.getResponseMsg().getError())); |
||||
|
} else { |
||||
|
response = new FromDeviceRpcResponse(uuid, deviceRpcCallMsg.getResponseMsg().getResponse(), null); |
||||
|
} |
||||
|
TbQueueCallback callback = new TbQueueCallback() { |
||||
|
@Override |
||||
|
public void onSuccess(TbQueueMsgMetadata metadata) { |
||||
|
futureToSet.set(null); |
||||
|
} |
||||
|
|
||||
|
@Override |
||||
|
public void onFailure(Throwable t) { |
||||
|
log.error("Can't process push notification to core [{}]", deviceRpcCallMsg, t); |
||||
|
futureToSet.setException(t); |
||||
|
} |
||||
|
}; |
||||
|
tbClusterService.pushNotificationToCore(deviceRpcCallMsg.getOriginServiceId(), response, callback); |
||||
|
return futureToSet; |
||||
|
} |
||||
|
|
||||
|
} |
||||
@ -0,0 +1,104 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2020 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.processor; |
||||
|
|
||||
|
import com.google.common.util.concurrent.Futures; |
||||
|
import com.google.common.util.concurrent.ListenableFuture; |
||||
|
import lombok.extern.slf4j.Slf4j; |
||||
|
import org.springframework.stereotype.Component; |
||||
|
import org.thingsboard.server.common.data.EntityType; |
||||
|
import org.thingsboard.server.common.data.exception.ThingsboardErrorCode; |
||||
|
import org.thingsboard.server.common.data.exception.ThingsboardException; |
||||
|
import org.thingsboard.server.common.data.id.AssetId; |
||||
|
import org.thingsboard.server.common.data.id.CustomerId; |
||||
|
import org.thingsboard.server.common.data.id.DashboardId; |
||||
|
import org.thingsboard.server.common.data.id.DeviceId; |
||||
|
import org.thingsboard.server.common.data.id.EntityId; |
||||
|
import org.thingsboard.server.common.data.id.EntityIdFactory; |
||||
|
import org.thingsboard.server.common.data.id.EntityViewId; |
||||
|
import org.thingsboard.server.common.data.id.TenantId; |
||||
|
import org.thingsboard.server.common.data.id.UserId; |
||||
|
import org.thingsboard.server.common.data.relation.EntityRelation; |
||||
|
import org.thingsboard.server.common.data.relation.RelationTypeGroup; |
||||
|
import org.thingsboard.server.gen.edge.RelationUpdateMsg; |
||||
|
import org.thingsboard.server.queue.util.TbCoreComponent; |
||||
|
|
||||
|
import java.util.UUID; |
||||
|
|
||||
|
@Component |
||||
|
@Slf4j |
||||
|
@TbCoreComponent |
||||
|
public class RelationProcessor extends BaseProcessor { |
||||
|
|
||||
|
public ListenableFuture<Void> onRelationUpdate(TenantId tenantId, RelationUpdateMsg relationUpdateMsg) { |
||||
|
log.info("onRelationUpdate {}", relationUpdateMsg); |
||||
|
try { |
||||
|
EntityRelation entityRelation = new EntityRelation(); |
||||
|
|
||||
|
UUID fromUUID = new UUID(relationUpdateMsg.getFromIdMSB(), relationUpdateMsg.getFromIdLSB()); |
||||
|
EntityId fromId = EntityIdFactory.getByTypeAndUuid(EntityType.valueOf(relationUpdateMsg.getFromEntityType()), fromUUID); |
||||
|
entityRelation.setFrom(fromId); |
||||
|
|
||||
|
UUID toUUID = new UUID(relationUpdateMsg.getToIdMSB(), relationUpdateMsg.getToIdLSB()); |
||||
|
EntityId toId = EntityIdFactory.getByTypeAndUuid(EntityType.valueOf(relationUpdateMsg.getToEntityType()), toUUID); |
||||
|
entityRelation.setTo(toId); |
||||
|
|
||||
|
entityRelation.setType(relationUpdateMsg.getType()); |
||||
|
entityRelation.setTypeGroup(RelationTypeGroup.valueOf(relationUpdateMsg.getTypeGroup())); |
||||
|
entityRelation.setAdditionalInfo(mapper.readTree(relationUpdateMsg.getAdditionalInfo())); |
||||
|
switch (relationUpdateMsg.getMsgType()) { |
||||
|
case ENTITY_CREATED_RPC_MESSAGE: |
||||
|
case ENTITY_UPDATED_RPC_MESSAGE: |
||||
|
if (isEntityExists(tenantId, entityRelation.getTo()) |
||||
|
&& isEntityExists(tenantId, entityRelation.getFrom())) { |
||||
|
relationService.saveRelationAsync(tenantId, entityRelation); |
||||
|
} |
||||
|
break; |
||||
|
case ENTITY_DELETED_RPC_MESSAGE: |
||||
|
relationService.deleteRelation(tenantId, entityRelation); |
||||
|
break; |
||||
|
case UNRECOGNIZED: |
||||
|
log.error("Unsupported msg type"); |
||||
|
} |
||||
|
return Futures.immediateFuture(null); |
||||
|
} catch (Exception e) { |
||||
|
log.error("Failed to process relation update msg [{}]", relationUpdateMsg, e); |
||||
|
return Futures.immediateFailedFuture(new RuntimeException("Failed to process relation update msg", e)); |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
|
||||
|
private boolean isEntityExists(TenantId tenantId, EntityId entityId) throws ThingsboardException { |
||||
|
switch (entityId.getEntityType()) { |
||||
|
case DEVICE: |
||||
|
return deviceService.findDeviceById(tenantId, new DeviceId(entityId.getId())) != null; |
||||
|
case ASSET: |
||||
|
return assetService.findAssetById(tenantId, new AssetId(entityId.getId())) != null; |
||||
|
case ENTITY_VIEW: |
||||
|
return entityViewService.findEntityViewById(tenantId, new EntityViewId(entityId.getId())) != null; |
||||
|
case CUSTOMER: |
||||
|
return customerService.findCustomerById(tenantId, new CustomerId(entityId.getId())) != null; |
||||
|
case USER: |
||||
|
return userService.findUserById(tenantId, new UserId(entityId.getId())) != null; |
||||
|
case DASHBOARD: |
||||
|
return dashboardService.findDashboardById(tenantId, new DashboardId(entityId.getId())) != null; |
||||
|
default: |
||||
|
throw new ThingsboardException("Unsupported entity type " + entityId.getEntityType(), ThingsboardErrorCode.INVALID_ARGUMENTS); |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
|
||||
|
} |
||||
@ -0,0 +1,244 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2020 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.processor; |
||||
|
|
||||
|
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 com.google.gson.Gson; |
||||
|
import com.google.gson.JsonObject; |
||||
|
import lombok.extern.slf4j.Slf4j; |
||||
|
import org.springframework.stereotype.Component; |
||||
|
import org.thingsboard.rule.engine.api.msg.DeviceAttributesEventNotificationMsg; |
||||
|
import org.thingsboard.server.common.data.DataConstants; |
||||
|
import org.thingsboard.server.common.data.Device; |
||||
|
import org.thingsboard.server.common.data.EntityType; |
||||
|
import org.thingsboard.server.common.data.EntityView; |
||||
|
import org.thingsboard.server.common.data.asset.Asset; |
||||
|
import org.thingsboard.server.common.data.id.AssetId; |
||||
|
import org.thingsboard.server.common.data.id.CustomerId; |
||||
|
import org.thingsboard.server.common.data.id.DashboardId; |
||||
|
import org.thingsboard.server.common.data.id.DeviceId; |
||||
|
import org.thingsboard.server.common.data.id.EntityId; |
||||
|
import org.thingsboard.server.common.data.id.EntityViewId; |
||||
|
import org.thingsboard.server.common.data.id.TenantId; |
||||
|
import org.thingsboard.server.common.data.id.UserId; |
||||
|
import org.thingsboard.server.common.data.kv.AttributeKey; |
||||
|
import org.thingsboard.server.common.data.kv.AttributeKvEntry; |
||||
|
import org.thingsboard.server.common.msg.TbMsg; |
||||
|
import org.thingsboard.server.common.msg.TbMsgMetaData; |
||||
|
import org.thingsboard.server.common.msg.session.SessionMsgType; |
||||
|
import org.thingsboard.server.common.transport.adaptor.JsonConverter; |
||||
|
import org.thingsboard.server.common.transport.util.JsonUtils; |
||||
|
import org.thingsboard.server.gen.edge.AttributeDeleteMsg; |
||||
|
import org.thingsboard.server.gen.edge.EntityDataProto; |
||||
|
import org.thingsboard.server.gen.transport.TransportProtos; |
||||
|
import org.thingsboard.server.queue.TbQueueCallback; |
||||
|
import org.thingsboard.server.queue.TbQueueMsgMetadata; |
||||
|
import org.thingsboard.server.queue.util.TbCoreComponent; |
||||
|
|
||||
|
import javax.annotation.Nullable; |
||||
|
import java.util.ArrayList; |
||||
|
import java.util.HashSet; |
||||
|
import java.util.List; |
||||
|
import java.util.Set; |
||||
|
import java.util.UUID; |
||||
|
|
||||
|
@Component |
||||
|
@Slf4j |
||||
|
@TbCoreComponent |
||||
|
public class TelemetryProcessor extends BaseProcessor { |
||||
|
|
||||
|
private final Gson gson = new Gson(); |
||||
|
|
||||
|
public List<ListenableFuture<Void>> onTelemetryUpdate(TenantId tenantId, EntityDataProto entityData) { |
||||
|
List<ListenableFuture<Void>> result = new ArrayList<>(); |
||||
|
EntityId entityId = constructEntityId(entityData); |
||||
|
if ((entityData.hasPostAttributesMsg() || entityData.hasPostTelemetryMsg() || entityData.hasAttributesUpdatedMsg()) && entityId != null) { |
||||
|
TbMsgMetaData metaData = constructBaseMsgMetadata(tenantId, entityId); |
||||
|
metaData.putValue(DataConstants.MSG_SOURCE_KEY, DataConstants.EDGE_MSG_SOURCE); |
||||
|
if (entityData.hasPostAttributesMsg()) { |
||||
|
result.add(processPostAttributes(tenantId, entityId, entityData.getPostAttributesMsg(), metaData)); |
||||
|
} |
||||
|
if (entityData.hasAttributesUpdatedMsg()) { |
||||
|
metaData.putValue("scope", entityData.getPostAttributeScope()); |
||||
|
result.add(processAttributesUpdate(tenantId, entityId, entityData.getAttributesUpdatedMsg(), metaData)); |
||||
|
} |
||||
|
if (entityData.hasPostTelemetryMsg()) { |
||||
|
result.add(processPostTelemetry(tenantId, entityId, entityData.getPostTelemetryMsg(), metaData)); |
||||
|
} |
||||
|
} |
||||
|
if (entityData.hasAttributeDeleteMsg()) { |
||||
|
result.add(processAttributeDeleteMsg(tenantId, entityId, entityData.getAttributeDeleteMsg(), entityData.getEntityType())); |
||||
|
} |
||||
|
return result; |
||||
|
} |
||||
|
|
||||
|
private TbMsgMetaData constructBaseMsgMetadata(TenantId tenantId, EntityId entityId) { |
||||
|
TbMsgMetaData metaData = new TbMsgMetaData(); |
||||
|
switch (entityId.getEntityType()) { |
||||
|
case DEVICE: |
||||
|
Device device = deviceService.findDeviceById(tenantId, new DeviceId(entityId.getId())); |
||||
|
if (device != null) { |
||||
|
metaData.putValue("deviceName", device.getName()); |
||||
|
metaData.putValue("deviceType", device.getType()); |
||||
|
} |
||||
|
break; |
||||
|
case ASSET: |
||||
|
Asset asset = assetService.findAssetById(tenantId, new AssetId(entityId.getId())); |
||||
|
if (asset != null) { |
||||
|
metaData.putValue("assetName", asset.getName()); |
||||
|
metaData.putValue("assetType", asset.getType()); |
||||
|
} |
||||
|
break; |
||||
|
case ENTITY_VIEW: |
||||
|
EntityView entityView = entityViewService.findEntityViewById(tenantId, new EntityViewId(entityId.getId())); |
||||
|
if (entityView != null) { |
||||
|
metaData.putValue("entityViewName", entityView.getName()); |
||||
|
metaData.putValue("entityViewType", entityView.getType()); |
||||
|
} |
||||
|
break; |
||||
|
default: |
||||
|
log.debug("Using empty metadata for entityId [{}]", entityId); |
||||
|
break; |
||||
|
} |
||||
|
return metaData; |
||||
|
} |
||||
|
|
||||
|
private ListenableFuture<Void> processPostTelemetry(TenantId tenantId, EntityId entityId, TransportProtos.PostTelemetryMsg msg, TbMsgMetaData metaData) { |
||||
|
SettableFuture<Void> futureToSet = SettableFuture.create(); |
||||
|
for (TransportProtos.TsKvListProto tsKv : msg.getTsKvListList()) { |
||||
|
JsonObject json = JsonUtils.getJsonObject(tsKv.getKvList()); |
||||
|
metaData.putValue("ts", tsKv.getTs() + ""); |
||||
|
TbMsg tbMsg = TbMsg.newMsg(SessionMsgType.POST_TELEMETRY_REQUEST.name(), entityId, metaData, gson.toJson(json)); |
||||
|
tbClusterService.pushMsgToRuleEngine(tenantId, tbMsg.getOriginator(), tbMsg, new TbQueueCallback() { |
||||
|
@Override |
||||
|
public void onSuccess(TbQueueMsgMetadata metadata) { |
||||
|
futureToSet.set(null); |
||||
|
} |
||||
|
|
||||
|
@Override |
||||
|
public void onFailure(Throwable t) { |
||||
|
log.error("Can't process post telemetry [{}]", msg, t); |
||||
|
futureToSet.setException(t); |
||||
|
} |
||||
|
}); |
||||
|
} |
||||
|
return futureToSet; |
||||
|
} |
||||
|
|
||||
|
private ListenableFuture<Void> processPostAttributes(TenantId tenantId, EntityId entityId, TransportProtos.PostAttributeMsg msg, TbMsgMetaData metaData) { |
||||
|
SettableFuture<Void> futureToSet = SettableFuture.create(); |
||||
|
JsonObject json = JsonUtils.getJsonObject(msg.getKvList()); |
||||
|
TbMsg tbMsg = TbMsg.newMsg(SessionMsgType.POST_ATTRIBUTES_REQUEST.name(), entityId, metaData, gson.toJson(json)); |
||||
|
tbClusterService.pushMsgToRuleEngine(tenantId, tbMsg.getOriginator(), tbMsg, new TbQueueCallback() { |
||||
|
@Override |
||||
|
public void onSuccess(TbQueueMsgMetadata metadata) { |
||||
|
futureToSet.set(null); |
||||
|
} |
||||
|
|
||||
|
@Override |
||||
|
public void onFailure(Throwable t) { |
||||
|
log.error("Can't process post attributes [{}]", msg, t); |
||||
|
futureToSet.setException(t); |
||||
|
} |
||||
|
}); |
||||
|
return futureToSet; |
||||
|
} |
||||
|
|
||||
|
private ListenableFuture<Void> processAttributesUpdate(TenantId tenantId, EntityId entityId, TransportProtos.PostAttributeMsg msg, TbMsgMetaData metaData) { |
||||
|
SettableFuture<Void> futureToSet = SettableFuture.create(); |
||||
|
JsonObject json = JsonUtils.getJsonObject(msg.getKvList()); |
||||
|
Set<AttributeKvEntry> attributes = JsonConverter.convertToAttributes(json); |
||||
|
ListenableFuture<List<Void>> future = attributesService.save(tenantId, entityId, metaData.getValue("scope"), new ArrayList<>(attributes)); |
||||
|
Futures.addCallback(future, new FutureCallback<List<Void>>() { |
||||
|
@Override |
||||
|
public void onSuccess(@Nullable List<Void> voids) { |
||||
|
TbMsg tbMsg = TbMsg.newMsg(DataConstants.ATTRIBUTES_UPDATED, entityId, metaData, gson.toJson(json)); |
||||
|
tbClusterService.pushMsgToRuleEngine(tenantId, tbMsg.getOriginator(), tbMsg, new TbQueueCallback() { |
||||
|
@Override |
||||
|
public void onSuccess(TbQueueMsgMetadata metadata) { |
||||
|
futureToSet.set(null); |
||||
|
} |
||||
|
|
||||
|
@Override |
||||
|
public void onFailure(Throwable t) { |
||||
|
log.error("Can't process attributes update [{}]", msg, t); |
||||
|
futureToSet.setException(t); |
||||
|
} |
||||
|
}); |
||||
|
} |
||||
|
|
||||
|
@Override |
||||
|
public void onFailure(Throwable t) { |
||||
|
log.error("Can't process attributes update [{}]", msg, t); |
||||
|
futureToSet.setException(t); |
||||
|
} |
||||
|
}, dbCallbackExecutorService); |
||||
|
return futureToSet; |
||||
|
} |
||||
|
|
||||
|
private ListenableFuture<Void> processAttributeDeleteMsg(TenantId tenantId, EntityId entityId, AttributeDeleteMsg attributeDeleteMsg, String entityType) { |
||||
|
SettableFuture<Void> futureToSet = SettableFuture.create(); |
||||
|
String scope = attributeDeleteMsg.getScope(); |
||||
|
List<String> attributeNames = attributeDeleteMsg.getAttributeNamesList(); |
||||
|
attributesService.removeAll(tenantId, entityId, scope, attributeNames); |
||||
|
if (EntityType.DEVICE.name().equals(entityType)) { |
||||
|
Set<AttributeKey> attributeKeys = new HashSet<>(); |
||||
|
for (String attributeName : attributeNames) { |
||||
|
attributeKeys.add(new AttributeKey(scope, attributeName)); |
||||
|
} |
||||
|
tbClusterService.pushMsgToCore(DeviceAttributesEventNotificationMsg.onDelete( |
||||
|
tenantId, (DeviceId) entityId, attributeKeys), new TbQueueCallback() { |
||||
|
@Override |
||||
|
public void onSuccess(TbQueueMsgMetadata metadata) { |
||||
|
futureToSet.set(null); |
||||
|
} |
||||
|
|
||||
|
@Override |
||||
|
public void onFailure(Throwable t) { |
||||
|
log.error("Can't process attribute delete msg [{}]", attributeDeleteMsg, t); |
||||
|
futureToSet.setException(t); |
||||
|
} |
||||
|
}); |
||||
|
} |
||||
|
return futureToSet; |
||||
|
} |
||||
|
|
||||
|
private EntityId constructEntityId(EntityDataProto entityData) { |
||||
|
EntityType entityType = EntityType.valueOf(entityData.getEntityType()); |
||||
|
switch (entityType) { |
||||
|
case DEVICE: |
||||
|
return new DeviceId(new UUID(entityData.getEntityIdMSB(), entityData.getEntityIdLSB())); |
||||
|
case ASSET: |
||||
|
return new AssetId(new UUID(entityData.getEntityIdMSB(), entityData.getEntityIdLSB())); |
||||
|
case ENTITY_VIEW: |
||||
|
return new EntityViewId(new UUID(entityData.getEntityIdMSB(), entityData.getEntityIdLSB())); |
||||
|
case DASHBOARD: |
||||
|
return new DashboardId(new UUID(entityData.getEntityIdMSB(), entityData.getEntityIdLSB())); |
||||
|
case TENANT: |
||||
|
return new TenantId(new UUID(entityData.getEntityIdMSB(), entityData.getEntityIdLSB())); |
||||
|
case CUSTOMER: |
||||
|
return new CustomerId(new UUID(entityData.getEntityIdMSB(), entityData.getEntityIdLSB())); |
||||
|
case USER: |
||||
|
return new UserId(new UUID(entityData.getEntityIdMSB(), entityData.getEntityIdLSB())); |
||||
|
default: |
||||
|
log.warn("Unsupported entity type [{}] during construct of entity id. EntityDataProto [{}]", entityData.getEntityType(), entityData); |
||||
|
return null; |
||||
|
} |
||||
|
} |
||||
|
} |
||||
@ -0,0 +1,56 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2020 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.ttl.edge; |
||||
|
|
||||
|
import lombok.extern.slf4j.Slf4j; |
||||
|
import org.springframework.beans.factory.annotation.Value; |
||||
|
import org.springframework.scheduling.annotation.Scheduled; |
||||
|
import org.springframework.stereotype.Service; |
||||
|
import org.thingsboard.server.dao.util.PsqlDao; |
||||
|
import org.thingsboard.server.service.ttl.AbstractCleanUpService; |
||||
|
|
||||
|
import java.sql.Connection; |
||||
|
import java.sql.DriverManager; |
||||
|
import java.sql.SQLException; |
||||
|
|
||||
|
@PsqlDao |
||||
|
@Slf4j |
||||
|
@Service |
||||
|
public class EdgeEventsCleanUpService extends AbstractCleanUpService { |
||||
|
|
||||
|
@Value("${sql.ttl.edge_events.edge_events_ttl}") |
||||
|
private long ttl; |
||||
|
|
||||
|
@Value("${sql.ttl.edge_events.enabled}") |
||||
|
private boolean ttlTaskExecutionEnabled; |
||||
|
|
||||
|
@Scheduled(initialDelayString = "${sql.ttl.edge_events.execution_interval_ms}", fixedDelayString = "${sql.ttl.edge_events.execution_interval_ms}") |
||||
|
public void cleanUp() { |
||||
|
if (ttlTaskExecutionEnabled) { |
||||
|
try (Connection conn = DriverManager.getConnection(dbUrl, dbUserName, dbPassword)) { |
||||
|
doCleanUp(conn); |
||||
|
} catch (SQLException e) { |
||||
|
log.error("SQLException occurred during TTL task execution ", e); |
||||
|
} |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
@Override |
||||
|
protected void doCleanUp(Connection connection) throws SQLException { |
||||
|
long totalEdgeEventsRemoved = executeQuery(connection, "call cleanup_edge_events_by_ttl(" + ttl + ", 0);"); |
||||
|
log.info("Total edge events removed by TTL: [{}]", totalEdgeEventsRemoved); |
||||
|
} |
||||
|
} |
||||
@ -0,0 +1,125 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2020 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.core.type.TypeReference; |
||||
|
import lombok.extern.slf4j.Slf4j; |
||||
|
import org.junit.After; |
||||
|
import org.junit.Assert; |
||||
|
import org.junit.Before; |
||||
|
import org.junit.Test; |
||||
|
import org.thingsboard.server.common.data.Device; |
||||
|
import org.thingsboard.server.common.data.Tenant; |
||||
|
import org.thingsboard.server.common.data.User; |
||||
|
import org.thingsboard.server.common.data.asset.Asset; |
||||
|
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.TenantId; |
||||
|
import org.thingsboard.server.common.data.page.PageData; |
||||
|
import org.thingsboard.server.common.data.page.TimePageLink; |
||||
|
import org.thingsboard.server.common.data.relation.EntityRelation; |
||||
|
import org.thingsboard.server.common.data.security.Authority; |
||||
|
|
||||
|
import java.util.List; |
||||
|
|
||||
|
import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status; |
||||
|
|
||||
|
@Slf4j |
||||
|
public class BaseEdgeEventControllerTest extends AbstractControllerTest { |
||||
|
|
||||
|
private Tenant savedTenant; |
||||
|
private TenantId tenantId; |
||||
|
private User tenantAdmin; |
||||
|
|
||||
|
@Before |
||||
|
public void beforeTest() throws Exception { |
||||
|
loginSysAdmin(); |
||||
|
|
||||
|
Tenant tenant = new Tenant(); |
||||
|
tenant.setTitle("My tenant"); |
||||
|
savedTenant = doPost("/api/tenant", tenant, Tenant.class); |
||||
|
tenantId = savedTenant.getId(); |
||||
|
Assert.assertNotNull(savedTenant); |
||||
|
|
||||
|
tenantAdmin = new User(); |
||||
|
tenantAdmin.setAuthority(Authority.TENANT_ADMIN); |
||||
|
tenantAdmin.setTenantId(savedTenant.getId()); |
||||
|
tenantAdmin.setEmail("tenant2@thingsboard.org"); |
||||
|
tenantAdmin.setFirstName("Joe"); |
||||
|
tenantAdmin.setLastName("Downs"); |
||||
|
|
||||
|
tenantAdmin = createUserAndLogin(tenantAdmin, "testPassword1"); |
||||
|
} |
||||
|
|
||||
|
@After |
||||
|
public void afterTest() throws Exception { |
||||
|
loginSysAdmin(); |
||||
|
|
||||
|
doDelete("/api/tenant/" + savedTenant.getId().getId().toString()) |
||||
|
.andExpect(status().isOk()); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
public void testGetEdgeEvents() throws Exception { |
||||
|
Thread.sleep(1000); |
||||
|
Edge edge = constructEdge("TestEdge", "default"); |
||||
|
edge = doPost("/api/edge", edge, Edge.class); |
||||
|
|
||||
|
Device device = constructDevice("TestDevice", "default"); |
||||
|
Device savedDevice = doPost("/api/device", device, Device.class); |
||||
|
|
||||
|
doPost("/api/edge/" + edge.getId().toString() + "/device/" + savedDevice.getId().toString(), Device.class); |
||||
|
Thread.sleep(1000); |
||||
|
|
||||
|
Asset asset = constructAsset("TestAsset", "default"); |
||||
|
Asset savedAsset = doPost("/api/asset", asset, Asset.class); |
||||
|
|
||||
|
doPost("/api/edge/" + edge.getId().toString() + "/asset/" + savedAsset.getId().toString(), Asset.class); |
||||
|
Thread.sleep(1000); |
||||
|
|
||||
|
EntityRelation relation = new EntityRelation(savedAsset.getId(), savedDevice.getId(), EntityRelation.CONTAINS_TYPE); |
||||
|
|
||||
|
doPost("/api/relation", relation); |
||||
|
Thread.sleep(1000); |
||||
|
|
||||
|
List<EdgeEvent> edgeEvents = doGetTypedWithTimePageLink("/api/edge/" + edge.getId().toString() + "/events?", |
||||
|
new TypeReference<PageData<EdgeEvent>>() { |
||||
|
}, new TimePageLink(4)).getData(); |
||||
|
|
||||
|
Assert.assertFalse(edgeEvents.isEmpty()); |
||||
|
Assert.assertEquals(4, edgeEvents.size()); |
||||
|
Assert.assertEquals(EdgeEventType.RELATION, edgeEvents.get(0).getType()); |
||||
|
Assert.assertEquals(EdgeEventType.ASSET, edgeEvents.get(1).getType()); |
||||
|
Assert.assertEquals(EdgeEventType.DEVICE, edgeEvents.get(2).getType()); |
||||
|
Assert.assertEquals(EdgeEventType.RULE_CHAIN, edgeEvents.get(3).getType()); |
||||
|
} |
||||
|
|
||||
|
private Device constructDevice(String name, String type) { |
||||
|
Device device = new Device(); |
||||
|
device.setName(name); |
||||
|
device.setType(type); |
||||
|
return device; |
||||
|
} |
||||
|
|
||||
|
private Asset constructAsset(String name, String type) { |
||||
|
Asset asset = new Asset(); |
||||
|
asset.setName(name); |
||||
|
asset.setType(type); |
||||
|
return asset; |
||||
|
} |
||||
|
|
||||
|
} |
||||
@ -0,0 +1,23 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2020 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.nosql; |
||||
|
|
||||
|
import org.thingsboard.server.controller.BaseEdgeEventControllerTest; |
||||
|
import org.thingsboard.server.dao.service.DaoNoSqlTest; |
||||
|
|
||||
|
@DaoNoSqlTest |
||||
|
public class EdgeEventControllerNoSqlTest extends BaseEdgeEventControllerTest { |
||||
|
} |
||||
@ -0,0 +1,23 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2020 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.sql; |
||||
|
|
||||
|
import org.thingsboard.server.controller.BaseEdgeEventControllerTest; |
||||
|
import org.thingsboard.server.dao.service.DaoSqlTest; |
||||
|
|
||||
|
@DaoSqlTest |
||||
|
public class EdgeEventControllerSqlTest extends BaseEdgeEventControllerTest { |
||||
|
} |
||||
File diff suppressed because it is too large
@ -0,0 +1,47 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2020 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.edge; |
||||
|
|
||||
|
import org.cassandraunit.dataset.cql.ClassPathCQLDataSet; |
||||
|
import org.junit.BeforeClass; |
||||
|
import org.junit.ClassRule; |
||||
|
import org.junit.extensions.cpsuite.ClasspathSuite; |
||||
|
import org.junit.runner.RunWith; |
||||
|
import org.thingsboard.server.dao.CustomCassandraCQLUnit; |
||||
|
import org.thingsboard.server.queue.memory.InMemoryStorage; |
||||
|
|
||||
|
import java.util.Arrays; |
||||
|
|
||||
|
@RunWith(ClasspathSuite.class) |
||||
|
@ClasspathSuite.ClassnameFilters({ |
||||
|
"org.thingsboard.server.edge.nosql.*Test"}) |
||||
|
public class EdgeNoSqlTestSuite { |
||||
|
|
||||
|
@ClassRule |
||||
|
public static CustomCassandraCQLUnit cassandraUnit = |
||||
|
new CustomCassandraCQLUnit( |
||||
|
Arrays.asList( |
||||
|
new ClassPathCQLDataSet("cassandra/schema-ts.cql", false, false), |
||||
|
new ClassPathCQLDataSet("cassandra/schema-entities.cql", false, false), |
||||
|
new ClassPathCQLDataSet("cassandra/system-data.cql", false, false), |
||||
|
new ClassPathCQLDataSet("cassandra/system-test.cql", false, false)), |
||||
|
"cassandra-test.yaml", 30000l); |
||||
|
|
||||
|
@BeforeClass |
||||
|
public static void cleanupInMemStorage(){ |
||||
|
InMemoryStorage.getInstance().cleanup(); |
||||
|
} |
||||
|
} |
||||
@ -0,0 +1,41 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2020 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.edge; |
||||
|
|
||||
|
import org.junit.BeforeClass; |
||||
|
import org.junit.ClassRule; |
||||
|
import org.junit.extensions.cpsuite.ClasspathSuite; |
||||
|
import org.junit.runner.RunWith; |
||||
|
import org.thingsboard.server.dao.CustomSqlUnit; |
||||
|
import org.thingsboard.server.queue.memory.InMemoryStorage; |
||||
|
|
||||
|
import java.util.Arrays; |
||||
|
|
||||
|
@RunWith(ClasspathSuite.class) |
||||
|
@ClasspathSuite.ClassnameFilters({"org.thingsboard.server.edge.sql.*Test"}) |
||||
|
public class EdgeSqlTestSuite { |
||||
|
|
||||
|
@ClassRule |
||||
|
public static CustomSqlUnit sqlUnit = new CustomSqlUnit( |
||||
|
Arrays.asList("sql/schema-ts-hsql.sql", "sql/schema-entities-hsql.sql", "sql/schema-entities-idx.sql", "sql/system-data.sql"), |
||||
|
"sql/hsql/drop-all-tables.sql", |
||||
|
"sql-test.properties"); |
||||
|
|
||||
|
@BeforeClass |
||||
|
public static void cleanupInMemStorage(){ |
||||
|
InMemoryStorage.getInstance().cleanup(); |
||||
|
} |
||||
|
} |
||||
@ -0,0 +1,268 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2020 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.edge.imitator; |
||||
|
|
||||
|
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.MoreExecutors; |
||||
|
import com.google.protobuf.AbstractMessage; |
||||
|
import lombok.Getter; |
||||
|
import lombok.extern.slf4j.Slf4j; |
||||
|
import org.checkerframework.checker.nullness.qual.Nullable; |
||||
|
import org.thingsboard.edge.rpc.EdgeGrpcClient; |
||||
|
import org.thingsboard.edge.rpc.EdgeRpcClient; |
||||
|
import org.thingsboard.server.common.data.id.UserId; |
||||
|
import org.thingsboard.server.gen.edge.AlarmUpdateMsg; |
||||
|
import org.thingsboard.server.gen.edge.AssetUpdateMsg; |
||||
|
import org.thingsboard.server.gen.edge.CustomerUpdateMsg; |
||||
|
import org.thingsboard.server.gen.edge.DashboardUpdateMsg; |
||||
|
import org.thingsboard.server.gen.edge.DeviceCredentialsUpdateMsg; |
||||
|
import org.thingsboard.server.gen.edge.DeviceUpdateMsg; |
||||
|
import org.thingsboard.server.gen.edge.DownlinkMsg; |
||||
|
import org.thingsboard.server.gen.edge.DownlinkResponseMsg; |
||||
|
import org.thingsboard.server.gen.edge.EdgeConfiguration; |
||||
|
import org.thingsboard.server.gen.edge.EntityDataProto; |
||||
|
import org.thingsboard.server.gen.edge.EntityViewUpdateMsg; |
||||
|
import org.thingsboard.server.gen.edge.RelationUpdateMsg; |
||||
|
import org.thingsboard.server.gen.edge.RuleChainMetadataUpdateMsg; |
||||
|
import org.thingsboard.server.gen.edge.RuleChainUpdateMsg; |
||||
|
import org.thingsboard.server.gen.edge.UplinkMsg; |
||||
|
import org.thingsboard.server.gen.edge.UplinkResponseMsg; |
||||
|
import org.thingsboard.server.gen.edge.UserCredentialsUpdateMsg; |
||||
|
import org.thingsboard.server.gen.edge.UserUpdateMsg; |
||||
|
import org.thingsboard.server.gen.edge.WidgetTypeUpdateMsg; |
||||
|
import org.thingsboard.server.gen.edge.WidgetsBundleUpdateMsg; |
||||
|
|
||||
|
import java.lang.reflect.Field; |
||||
|
import java.util.ArrayList; |
||||
|
import java.util.List; |
||||
|
import java.util.Optional; |
||||
|
import java.util.UUID; |
||||
|
import java.util.concurrent.CountDownLatch; |
||||
|
import java.util.concurrent.TimeUnit; |
||||
|
|
||||
|
@Slf4j |
||||
|
public class EdgeImitator { |
||||
|
|
||||
|
private String routingKey; |
||||
|
private String routingSecret; |
||||
|
|
||||
|
private EdgeRpcClient edgeRpcClient; |
||||
|
|
||||
|
private CountDownLatch messagesLatch; |
||||
|
private CountDownLatch responsesLatch; |
||||
|
private List<Class<? extends AbstractMessage>> ignoredTypes; |
||||
|
|
||||
|
@Getter |
||||
|
private EdgeConfiguration configuration; |
||||
|
@Getter |
||||
|
private UserId userId; |
||||
|
@Getter |
||||
|
private List<AbstractMessage> downlinkMsgs; |
||||
|
|
||||
|
public EdgeImitator(String host, int port, String routingKey, String routingSecret) throws NoSuchFieldException, IllegalAccessException { |
||||
|
edgeRpcClient = new EdgeGrpcClient(); |
||||
|
messagesLatch = new CountDownLatch(0); |
||||
|
responsesLatch = new CountDownLatch(0); |
||||
|
downlinkMsgs = new ArrayList<>(); |
||||
|
ignoredTypes = new ArrayList<>(); |
||||
|
this.routingKey = routingKey; |
||||
|
this.routingSecret = routingSecret; |
||||
|
setEdgeCredentials("rpcHost", host); |
||||
|
setEdgeCredentials("rpcPort", port); |
||||
|
setEdgeCredentials("keepAliveTimeSec", 300); |
||||
|
} |
||||
|
|
||||
|
private void setEdgeCredentials(String fieldName, Object value) throws NoSuchFieldException, IllegalAccessException { |
||||
|
Field fieldToSet = edgeRpcClient.getClass().getDeclaredField(fieldName); |
||||
|
fieldToSet.setAccessible(true); |
||||
|
fieldToSet.set(edgeRpcClient, value); |
||||
|
fieldToSet.setAccessible(false); |
||||
|
} |
||||
|
|
||||
|
public void connect() { |
||||
|
edgeRpcClient.connect(routingKey, routingSecret, |
||||
|
this::onUplinkResponse, |
||||
|
this::onEdgeUpdate, |
||||
|
this::onDownlink, |
||||
|
this::onClose); |
||||
|
|
||||
|
edgeRpcClient.sendSyncRequestMsg(); |
||||
|
} |
||||
|
|
||||
|
public void disconnect() throws InterruptedException { |
||||
|
edgeRpcClient.disconnect(false); |
||||
|
} |
||||
|
|
||||
|
public void sendUplinkMsg(UplinkMsg uplinkMsg) { |
||||
|
edgeRpcClient.sendUplinkMsg(uplinkMsg); |
||||
|
} |
||||
|
|
||||
|
private void onUplinkResponse(UplinkResponseMsg msg) { |
||||
|
log.info("onUplinkResponse: {}", msg); |
||||
|
responsesLatch.countDown(); |
||||
|
} |
||||
|
|
||||
|
private void onEdgeUpdate(EdgeConfiguration edgeConfiguration) { |
||||
|
this.configuration = edgeConfiguration; |
||||
|
} |
||||
|
|
||||
|
private void onUserUpdate(UserUpdateMsg userUpdateMsg) { |
||||
|
this.userId = new UserId(new UUID(userUpdateMsg.getIdMSB(), userUpdateMsg.getIdLSB())); |
||||
|
} |
||||
|
|
||||
|
private void onDownlink(DownlinkMsg downlinkMsg) { |
||||
|
ListenableFuture<List<Void>> future = processDownlinkMsg(downlinkMsg); |
||||
|
Futures.addCallback(future, new FutureCallback<List<Void>>() { |
||||
|
@Override |
||||
|
public void onSuccess(@Nullable List<Void> result) { |
||||
|
DownlinkResponseMsg downlinkResponseMsg = DownlinkResponseMsg.newBuilder().setSuccess(true).build(); |
||||
|
edgeRpcClient.sendDownlinkResponseMsg(downlinkResponseMsg); |
||||
|
} |
||||
|
|
||||
|
@Override |
||||
|
public void onFailure(Throwable t) { |
||||
|
DownlinkResponseMsg downlinkResponseMsg = DownlinkResponseMsg.newBuilder().setSuccess(false).setErrorMsg(t.getMessage()).build(); |
||||
|
edgeRpcClient.sendDownlinkResponseMsg(downlinkResponseMsg); |
||||
|
} |
||||
|
}, MoreExecutors.directExecutor()); |
||||
|
} |
||||
|
|
||||
|
private void onClose(Exception e) { |
||||
|
log.info("onClose: {}", e.getMessage()); |
||||
|
} |
||||
|
|
||||
|
private ListenableFuture<List<Void>> processDownlinkMsg(DownlinkMsg downlinkMsg) { |
||||
|
List<ListenableFuture<Void>> result = new ArrayList<>(); |
||||
|
if (downlinkMsg.getDeviceUpdateMsgList() != null && !downlinkMsg.getDeviceUpdateMsgList().isEmpty()) { |
||||
|
for (DeviceUpdateMsg deviceUpdateMsg: downlinkMsg.getDeviceUpdateMsgList()) { |
||||
|
result.add(saveDownlinkMsg(deviceUpdateMsg)); |
||||
|
} |
||||
|
} |
||||
|
if (downlinkMsg.getDeviceCredentialsUpdateMsgList() != null && !downlinkMsg.getDeviceCredentialsUpdateMsgList().isEmpty()) { |
||||
|
for (DeviceCredentialsUpdateMsg deviceCredentialsUpdateMsg: downlinkMsg.getDeviceCredentialsUpdateMsgList()) { |
||||
|
result.add(saveDownlinkMsg(deviceCredentialsUpdateMsg)); |
||||
|
} |
||||
|
} |
||||
|
if (downlinkMsg.getAssetUpdateMsgList() != null && !downlinkMsg.getAssetUpdateMsgList().isEmpty()) { |
||||
|
for (AssetUpdateMsg assetUpdateMsg: downlinkMsg.getAssetUpdateMsgList()) { |
||||
|
result.add(saveDownlinkMsg(assetUpdateMsg)); |
||||
|
} |
||||
|
} |
||||
|
if (downlinkMsg.getRuleChainUpdateMsgList() != null && !downlinkMsg.getRuleChainUpdateMsgList().isEmpty()) { |
||||
|
for (RuleChainUpdateMsg ruleChainUpdateMsg: downlinkMsg.getRuleChainUpdateMsgList()) { |
||||
|
result.add(saveDownlinkMsg(ruleChainUpdateMsg)); |
||||
|
} |
||||
|
} |
||||
|
if (downlinkMsg.getRuleChainMetadataUpdateMsgList() != null && !downlinkMsg.getRuleChainMetadataUpdateMsgList().isEmpty()) { |
||||
|
for (RuleChainMetadataUpdateMsg ruleChainMetadataUpdateMsg: downlinkMsg.getRuleChainMetadataUpdateMsgList()) { |
||||
|
result.add(saveDownlinkMsg(ruleChainMetadataUpdateMsg)); |
||||
|
} |
||||
|
} |
||||
|
if (downlinkMsg.getDashboardUpdateMsgList() != null && !downlinkMsg.getDashboardUpdateMsgList().isEmpty()) { |
||||
|
for (DashboardUpdateMsg dashboardUpdateMsg: downlinkMsg.getDashboardUpdateMsgList()) { |
||||
|
result.add(saveDownlinkMsg(dashboardUpdateMsg)); |
||||
|
} |
||||
|
} |
||||
|
if (downlinkMsg.getRelationUpdateMsgList() != null && !downlinkMsg.getRelationUpdateMsgList().isEmpty()) { |
||||
|
for (RelationUpdateMsg relationUpdateMsg: downlinkMsg.getRelationUpdateMsgList()) { |
||||
|
result.add(saveDownlinkMsg(relationUpdateMsg)); |
||||
|
} |
||||
|
} |
||||
|
if (downlinkMsg.getAlarmUpdateMsgList() != null && !downlinkMsg.getAlarmUpdateMsgList().isEmpty()) { |
||||
|
for (AlarmUpdateMsg alarmUpdateMsg: downlinkMsg.getAlarmUpdateMsgList()) { |
||||
|
result.add(saveDownlinkMsg(alarmUpdateMsg)); |
||||
|
} |
||||
|
} |
||||
|
if (downlinkMsg.getEntityDataList() != null && !downlinkMsg.getEntityDataList().isEmpty()) { |
||||
|
for (EntityDataProto entityData: downlinkMsg.getEntityDataList()) { |
||||
|
result.add(saveDownlinkMsg(entityData)); |
||||
|
} |
||||
|
} |
||||
|
if (downlinkMsg.getEntityViewUpdateMsgList() != null && !downlinkMsg.getEntityViewUpdateMsgList().isEmpty()) { |
||||
|
for (EntityViewUpdateMsg entityViewUpdateMsg: downlinkMsg.getEntityViewUpdateMsgList()) { |
||||
|
result.add(saveDownlinkMsg(entityViewUpdateMsg)); |
||||
|
} |
||||
|
} |
||||
|
if (downlinkMsg.getCustomerUpdateMsgList() != null && !downlinkMsg.getCustomerUpdateMsgList().isEmpty()) { |
||||
|
for (CustomerUpdateMsg customerUpdateMsg: downlinkMsg.getCustomerUpdateMsgList()) { |
||||
|
result.add(saveDownlinkMsg(customerUpdateMsg)); |
||||
|
} |
||||
|
} |
||||
|
if (downlinkMsg.getWidgetsBundleUpdateMsgList() != null && !downlinkMsg.getWidgetsBundleUpdateMsgList().isEmpty()) { |
||||
|
for (WidgetsBundleUpdateMsg widgetsBundleUpdateMsg: downlinkMsg.getWidgetsBundleUpdateMsgList()) { |
||||
|
result.add(saveDownlinkMsg(widgetsBundleUpdateMsg)); |
||||
|
} |
||||
|
} |
||||
|
if (downlinkMsg.getWidgetTypeUpdateMsgList() != null && !downlinkMsg.getWidgetTypeUpdateMsgList().isEmpty()) { |
||||
|
for (WidgetTypeUpdateMsg widgetTypeUpdateMsg: downlinkMsg.getWidgetTypeUpdateMsgList()) { |
||||
|
result.add(saveDownlinkMsg(widgetTypeUpdateMsg)); |
||||
|
} |
||||
|
} |
||||
|
if (downlinkMsg.getUserUpdateMsgList() != null && !downlinkMsg.getUserUpdateMsgList().isEmpty()) { |
||||
|
for (UserUpdateMsg userUpdateMsg: downlinkMsg.getUserUpdateMsgList()) { |
||||
|
onUserUpdate(userUpdateMsg); |
||||
|
result.add(saveDownlinkMsg(userUpdateMsg)); |
||||
|
} |
||||
|
} |
||||
|
if (downlinkMsg.getUserCredentialsUpdateMsgList() != null && !downlinkMsg.getUserCredentialsUpdateMsgList().isEmpty()) { |
||||
|
for (UserCredentialsUpdateMsg userCredentialsUpdateMsg: downlinkMsg.getUserCredentialsUpdateMsgList()) { |
||||
|
result.add(saveDownlinkMsg(userCredentialsUpdateMsg)); |
||||
|
} |
||||
|
} |
||||
|
return Futures.allAsList(result); |
||||
|
} |
||||
|
|
||||
|
private ListenableFuture<Void> saveDownlinkMsg(AbstractMessage message) { |
||||
|
if (!ignoredTypes.contains(message.getClass())) { |
||||
|
downlinkMsgs.add(message); |
||||
|
messagesLatch.countDown(); |
||||
|
} |
||||
|
return Futures.immediateFuture(null); |
||||
|
} |
||||
|
|
||||
|
public void waitForMessages() throws InterruptedException { |
||||
|
messagesLatch.await(5, TimeUnit.SECONDS); |
||||
|
} |
||||
|
|
||||
|
public void expectMessageAmount(int messageAmount) { |
||||
|
messagesLatch = new CountDownLatch(messageAmount); |
||||
|
} |
||||
|
|
||||
|
public void waitForResponses() throws InterruptedException { responsesLatch.await(5, TimeUnit.SECONDS); } |
||||
|
|
||||
|
public void expectResponsesAmount(int messageAmount) { |
||||
|
responsesLatch = new CountDownLatch(messageAmount); |
||||
|
} |
||||
|
|
||||
|
public <T> Optional<T> findMessageByType(Class<T> tClass) { |
||||
|
return (Optional<T>) downlinkMsgs.stream().filter(downlinkMsg -> downlinkMsg.getClass().isAssignableFrom(tClass)).findAny(); |
||||
|
} |
||||
|
|
||||
|
public AbstractMessage getLatestMessage() { |
||||
|
return downlinkMsgs.get(downlinkMsgs.size() - 1); |
||||
|
} |
||||
|
|
||||
|
public void ignoreType(Class<? extends AbstractMessage> type) { |
||||
|
ignoredTypes.add(type); |
||||
|
} |
||||
|
|
||||
|
public void allowIgnoredTypes() { |
||||
|
ignoredTypes.clear(); |
||||
|
} |
||||
|
|
||||
|
} |
||||
@ -0,0 +1,23 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2020 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.edge.nosql; |
||||
|
|
||||
|
import org.thingsboard.server.dao.service.DaoNoSqlTest; |
||||
|
import org.thingsboard.server.edge.BaseEdgeTest; |
||||
|
|
||||
|
@DaoNoSqlTest |
||||
|
public class EdgeNoSqlTest extends BaseEdgeTest { |
||||
|
} |
||||
@ -0,0 +1,23 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2020 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.edge.sql; |
||||
|
|
||||
|
import org.thingsboard.server.dao.service.DaoSqlTest; |
||||
|
import org.thingsboard.server.edge.BaseEdgeTest; |
||||
|
|
||||
|
@DaoSqlTest |
||||
|
public class EdgeSqlTest extends BaseEdgeTest { |
||||
|
} |
||||
@ -1,50 +0,0 @@ |
|||||
/** |
|
||||
* Copyright © 2016-2020 The Thingsboard Authors |
|
||||
* |
|
||||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|
||||
* you may not use this file except in compliance with the License. |
|
||||
* You may obtain a copy of the License at |
|
||||
* |
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
|
||||
* |
|
||||
* Unless required by applicable law or agreed to in writing, software |
|
||||
* distributed under the License is distributed on an "AS IS" BASIS, |
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|
||||
* See the License for the specific language governing permissions and |
|
||||
* limitations under the License. |
|
||||
*/ |
|
||||
package org.thingsboard.server.common.data; |
|
||||
|
|
||||
import lombok.AllArgsConstructor; |
|
||||
import lombok.Getter; |
|
||||
import lombok.Setter; |
|
||||
import org.thingsboard.server.common.data.id.EdgeId; |
|
||||
import org.thingsboard.server.common.data.id.RuleChainId; |
|
||||
|
|
||||
@AllArgsConstructor |
|
||||
public class ShortEdgeInfo { |
|
||||
|
|
||||
@Getter @Setter |
|
||||
private EdgeId edgeId; |
|
||||
|
|
||||
@Getter @Setter |
|
||||
private String title; |
|
||||
|
|
||||
@Getter @Setter |
|
||||
private RuleChainId rootRuleChainId; |
|
||||
|
|
||||
@Override |
|
||||
public boolean equals(Object o) { |
|
||||
if (this == o) return true; |
|
||||
if (o == null || getClass() != o.getClass()) return false; |
|
||||
|
|
||||
ShortEdgeInfo that = (ShortEdgeInfo) o; |
|
||||
|
|
||||
return edgeId.equals(that.edgeId); |
|
||||
} |
|
||||
|
|
||||
@Override |
|
||||
public int hashCode() { |
|
||||
return edgeId.hashCode(); |
|
||||
} |
|
||||
} |
|
||||
Some files were not shown because too many files changed in this diff
Loading…
Reference in new issue