Andrii Shvaika
3 years ago
26 changed files with 593 additions and 35 deletions
@ -0,0 +1,57 @@ |
|||
/** |
|||
* Copyright © 2016-2023 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 org.springframework.stereotype.Component; |
|||
import org.thingsboard.server.common.data.TbResource; |
|||
import org.thingsboard.server.common.data.id.TbResourceId; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
import org.thingsboard.server.gen.edge.v1.ResourceUpdateMsg; |
|||
import org.thingsboard.server.gen.edge.v1.UpdateMsgType; |
|||
import org.thingsboard.server.queue.util.TbCoreComponent; |
|||
|
|||
@Component |
|||
@TbCoreComponent |
|||
public class ResourceMsgConstructor { |
|||
|
|||
public ResourceUpdateMsg constructResourceUpdatedMsg(UpdateMsgType msgType, TbResource tbResource) { |
|||
ResourceUpdateMsg.Builder builder = ResourceUpdateMsg.newBuilder() |
|||
.setMsgType(msgType) |
|||
.setIdMSB(tbResource.getId().getId().getMostSignificantBits()) |
|||
.setIdLSB(tbResource.getId().getId().getLeastSignificantBits()) |
|||
.setTitle(tbResource.getTitle()) |
|||
.setResourceKey(tbResource.getResourceKey()) |
|||
.setResourceType(tbResource.getResourceType().name()) |
|||
.setFileName(tbResource.getFileName()); |
|||
if (tbResource.getData() != null) { |
|||
builder.setData(tbResource.getData()); |
|||
} |
|||
if (tbResource.getEtag() != null) { |
|||
builder.setEtag(tbResource.getEtag()); |
|||
} |
|||
if (tbResource.getTenantId().equals(TenantId.SYS_TENANT_ID)) { |
|||
builder.setIsSystem(true); |
|||
} |
|||
return builder.build(); |
|||
} |
|||
|
|||
public ResourceUpdateMsg constructResourceDeleteMsg(TbResourceId tbResourceId) { |
|||
return ResourceUpdateMsg.newBuilder() |
|||
.setMsgType(UpdateMsgType.ENTITY_DELETED_RPC_MESSAGE) |
|||
.setIdMSB(tbResourceId.getId().getMostSignificantBits()) |
|||
.setIdLSB(tbResourceId.getId().getLeastSignificantBits()).build(); |
|||
} |
|||
} |
|||
@ -0,0 +1,49 @@ |
|||
/** |
|||
* Copyright © 2016-2023 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.fetch; |
|||
|
|||
import lombok.AllArgsConstructor; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.thingsboard.server.common.data.EdgeUtils; |
|||
import org.thingsboard.server.common.data.TbResource; |
|||
import org.thingsboard.server.common.data.edge.Edge; |
|||
import org.thingsboard.server.common.data.edge.EdgeEvent; |
|||
import org.thingsboard.server.common.data.edge.EdgeEventActionType; |
|||
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.PageLink; |
|||
import org.thingsboard.server.dao.resource.ResourceService; |
|||
|
|||
@Slf4j |
|||
@AllArgsConstructor |
|||
public abstract class BaseResourceEdgeEventFetcher extends BasePageableEdgeEventFetcher<TbResource> { |
|||
|
|||
protected final ResourceService resourceService; |
|||
|
|||
@Override |
|||
PageData<TbResource> fetchPageData(TenantId tenantId, Edge edge, PageLink pageLink) { |
|||
return findTenantResources(tenantId, pageLink); |
|||
} |
|||
|
|||
@Override |
|||
EdgeEvent constructEdgeEvent(TenantId tenantId, Edge edge, TbResource tbResource) { |
|||
return EdgeUtils.constructEdgeEvent(tenantId, edge.getId(), EdgeEventType.TB_RESOURCE, |
|||
EdgeEventActionType.ADDED, tbResource.getId(), null); |
|||
} |
|||
|
|||
protected abstract PageData<TbResource> findTenantResources(TenantId tenantId, PageLink pageLink); |
|||
} |
|||
@ -0,0 +1,34 @@ |
|||
/** |
|||
* Copyright © 2016-2023 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.fetch; |
|||
|
|||
import org.thingsboard.server.common.data.TbResource; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
import org.thingsboard.server.common.data.page.PageData; |
|||
import org.thingsboard.server.common.data.page.PageLink; |
|||
import org.thingsboard.server.dao.resource.ResourceService; |
|||
|
|||
public class SystemResourcesEdgeEventFetcher extends BaseResourceEdgeEventFetcher { |
|||
|
|||
public SystemResourcesEdgeEventFetcher(ResourceService resourceService) { |
|||
super(resourceService); |
|||
} |
|||
|
|||
@Override |
|||
protected PageData<TbResource> findTenantResources(TenantId tenantId, PageLink pageLink) { |
|||
return resourceService.findAllTenantResources(TenantId.SYS_TENANT_ID, pageLink); |
|||
} |
|||
} |
|||
@ -0,0 +1,36 @@ |
|||
/** |
|||
* Copyright © 2016-2023 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.fetch; |
|||
|
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.thingsboard.server.common.data.TbResource; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
import org.thingsboard.server.common.data.page.PageData; |
|||
import org.thingsboard.server.common.data.page.PageLink; |
|||
import org.thingsboard.server.dao.resource.ResourceService; |
|||
|
|||
@Slf4j |
|||
public class TenantResourcesEdgeEventFetcher extends BaseResourceEdgeEventFetcher { |
|||
|
|||
public TenantResourcesEdgeEventFetcher(ResourceService resourceService) { |
|||
super(resourceService); |
|||
} |
|||
|
|||
@Override |
|||
protected PageData<TbResource> findTenantResources(TenantId tenantId, PageLink pageLink) { |
|||
return resourceService.findAllTenantResources(tenantId, pageLink); |
|||
} |
|||
} |
|||
@ -0,0 +1,61 @@ |
|||
/** |
|||
* Copyright © 2016-2023 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.resource; |
|||
|
|||
import com.datastax.oss.driver.api.core.uuid.Uuids; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.thingsboard.server.common.data.ResourceType; |
|||
import org.thingsboard.server.common.data.TbResource; |
|||
import org.thingsboard.server.common.data.TbResourceInfo; |
|||
import org.thingsboard.server.common.data.id.TbResourceId; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
import org.thingsboard.server.gen.edge.v1.ResourceUpdateMsg; |
|||
import org.thingsboard.server.service.edge.rpc.processor.BaseEdgeProcessor; |
|||
|
|||
@Slf4j |
|||
public abstract class BaseResourceProcessor extends BaseEdgeProcessor { |
|||
|
|||
protected void saveOrUpdateTbResource(TenantId tenantId, TbResourceId tbResourceId, ResourceUpdateMsg resourceUpdateMsg) { |
|||
try { |
|||
boolean created = false; |
|||
TbResource resource = resourceService.findResourceById(tenantId, tbResourceId); |
|||
if (resource == null) { |
|||
resource = new TbResource(); |
|||
if (resourceUpdateMsg.getIsSystem()) { |
|||
resource.setTenantId(TenantId.SYS_TENANT_ID); |
|||
} else { |
|||
resource.setTenantId(tenantId); |
|||
} |
|||
resource.setCreatedTime(Uuids.unixTimestamp(tbResourceId.getId())); |
|||
created = true; |
|||
} |
|||
resource.setTitle(resourceUpdateMsg.getTitle()); |
|||
resource.setResourceKey(resourceUpdateMsg.getResourceKey()); |
|||
resource.setResourceType(ResourceType.valueOf(resourceUpdateMsg.getResourceType())); |
|||
resource.setFileName(resourceUpdateMsg.getFileName()); |
|||
resource.setData(resourceUpdateMsg.hasData() ? resourceUpdateMsg.getData() : null); |
|||
resource.setEtag(resourceUpdateMsg.hasEtag() ? resourceUpdateMsg.getEtag() : null); |
|||
resourceValidator.validate(resource, TbResourceInfo::getTenantId); |
|||
if (created) { |
|||
resource.setId(tbResourceId); |
|||
} |
|||
resourceService.saveResource(resource, false); |
|||
} catch (Exception e) { |
|||
log.error("[{}] Failed to process resource update msg [{}]", tenantId, resourceUpdateMsg, e); |
|||
throw e; |
|||
} |
|||
} |
|||
} |
|||
@ -0,0 +1,101 @@ |
|||
/** |
|||
* Copyright © 2016-2023 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.resource; |
|||
|
|||
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.EdgeUtils; |
|||
import org.thingsboard.server.common.data.TbResource; |
|||
import org.thingsboard.server.common.data.edge.Edge; |
|||
import org.thingsboard.server.common.data.edge.EdgeEvent; |
|||
import org.thingsboard.server.common.data.id.TbResourceId; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
import org.thingsboard.server.dao.exception.DataValidationException; |
|||
import org.thingsboard.server.gen.edge.v1.DownlinkMsg; |
|||
import org.thingsboard.server.gen.edge.v1.ResourceUpdateMsg; |
|||
import org.thingsboard.server.gen.edge.v1.UpdateMsgType; |
|||
import org.thingsboard.server.queue.util.TbCoreComponent; |
|||
|
|||
import java.util.UUID; |
|||
|
|||
@Component |
|||
@Slf4j |
|||
@TbCoreComponent |
|||
public class ResourceEdgeProcessor extends BaseResourceProcessor { |
|||
|
|||
public ListenableFuture<Void> processResourceMsgFromEdge(TenantId tenantId, Edge edge, ResourceUpdateMsg resourceUpdateMsg) { |
|||
TbResourceId tbResourceId = new TbResourceId(new UUID(resourceUpdateMsg.getIdMSB(), resourceUpdateMsg.getIdLSB())); |
|||
try { |
|||
edgeSynchronizationManager.getEdgeId().set(edge.getId()); |
|||
|
|||
switch (resourceUpdateMsg.getMsgType()) { |
|||
case ENTITY_CREATED_RPC_MESSAGE: |
|||
case ENTITY_UPDATED_RPC_MESSAGE: |
|||
super.saveOrUpdateTbResource(tenantId, tbResourceId, resourceUpdateMsg); |
|||
break; |
|||
case ENTITY_DELETED_RPC_MESSAGE: |
|||
TbResource tbResourceToDelete = resourceService.findResourceById(tenantId, tbResourceId); |
|||
if (tbResourceToDelete != null) { |
|||
resourceService.deleteResource(tenantId, tbResourceId); |
|||
} |
|||
break; |
|||
case UNRECOGNIZED: |
|||
return handleUnsupportedMsgType(resourceUpdateMsg.getMsgType()); |
|||
} |
|||
} catch (DataValidationException e) { |
|||
if (e.getMessage().contains("files size limit is exhausted")) { |
|||
log.warn("[{}] Resource data size has been exhausted {}", tenantId, resourceUpdateMsg, e); |
|||
return Futures.immediateFuture(null); |
|||
} else { |
|||
return Futures.immediateFailedFuture(e); |
|||
} |
|||
} finally { |
|||
edgeSynchronizationManager.getEdgeId().remove(); |
|||
} |
|||
return Futures.immediateFuture(null); |
|||
} |
|||
|
|||
public DownlinkMsg convertResourceEventToDownlink(EdgeEvent edgeEvent) { |
|||
TbResourceId tbResourceId = new TbResourceId(edgeEvent.getEntityId()); |
|||
DownlinkMsg downlinkMsg = null; |
|||
switch (edgeEvent.getAction()) { |
|||
case ADDED: |
|||
case UPDATED: |
|||
TbResource tbResource = resourceService.findResourceById(edgeEvent.getTenantId(), tbResourceId); |
|||
if (tbResource != null) { |
|||
UpdateMsgType msgType = getUpdateMsgType(edgeEvent.getAction()); |
|||
ResourceUpdateMsg resourceUpdateMsg = |
|||
resourceMsgConstructor.constructResourceUpdatedMsg(msgType, tbResource); |
|||
downlinkMsg = DownlinkMsg.newBuilder() |
|||
.setDownlinkMsgId(EdgeUtils.nextPositiveInt()) |
|||
.addResourceUpdateMsg(resourceUpdateMsg) |
|||
.build(); |
|||
} |
|||
break; |
|||
case DELETED: |
|||
ResourceUpdateMsg resourceUpdateMsg = |
|||
resourceMsgConstructor.constructResourceDeleteMsg(tbResourceId); |
|||
downlinkMsg = DownlinkMsg.newBuilder() |
|||
.setDownlinkMsgId(EdgeUtils.nextPositiveInt()) |
|||
.addResourceUpdateMsg(resourceUpdateMsg) |
|||
.build(); |
|||
break; |
|||
} |
|||
return downlinkMsg; |
|||
} |
|||
} |
|||
File diff suppressed because one or more lines are too long
Loading…
Reference in new issue