diff --git a/application/src/main/java/org/thingsboard/server/service/transport/DefaultTransportApiService.java b/application/src/main/java/org/thingsboard/server/service/transport/DefaultTransportApiService.java index 6a14a53161..c371886b5e 100644 --- a/application/src/main/java/org/thingsboard/server/service/transport/DefaultTransportApiService.java +++ b/application/src/main/java/org/thingsboard/server/service/transport/DefaultTransportApiService.java @@ -101,7 +101,9 @@ import org.thingsboard.server.service.executors.DbCallbackExecutorService; import org.thingsboard.server.service.profile.TbDeviceProfileCache; import org.thingsboard.server.service.resource.TbResourceService; +import java.io.BufferedReader; import java.io.IOException; +import java.io.InputStreamReader; import java.util.List; import java.util.Optional; import java.util.UUID; @@ -137,7 +139,6 @@ public class DefaultTransportApiService implements TransportApiService { private final DeviceProvisionService deviceProvisionService; private final TbResourceService resourceService; private final OtaPackageService otaPackageService; - private final OtaPackageDataCache otaPackageDataCache; private final QueueService queueService; private final ConcurrentMap deviceCreationLocks = new ConcurrentHashMap<>(); @@ -181,6 +182,8 @@ public class DefaultTransportApiService implements TransportApiService { result = handle(transportApiRequestMsg.getDeviceCredentialsRequestMsg()); } else if (transportApiRequestMsg.hasOtaPackageRequestMsg()) { result = handle(transportApiRequestMsg.getOtaPackageRequestMsg()); + } else if (transportApiRequestMsg.hasSendOtaPackageBodyRequestMsg()) { + result = handle(transportApiRequestMsg.getSendOtaPackageBodyRequestMsg()); } else if (transportApiRequestMsg.hasGetAllQueueRoutingInfoRequestMsg()) { return Futures.transform(handle(transportApiRequestMsg.getGetAllQueueRoutingInfoRequestMsg()), value -> new TbProtoQueueMsg<>(tbProtoQueueMsg.getKey(), value, tbProtoQueueMsg.getHeaders()), MoreExecutors.directExecutor()); } @@ -600,22 +603,56 @@ public class DefaultTransportApiService implements TransportApiService { builder.setTitle(otaPackageInfo.getTitle()); builder.setVersion(otaPackageInfo.getVersion()); builder.setFileName(otaPackageInfo.getFileName()); + builder.setFileSize(otaPackageInfo.getDataSize()); builder.setContentType(otaPackageInfo.getContentType()); - if (!otaPackageDataCache.has(otaPackageId.toString())) { + } + } + + return Futures.immediateFuture( + TransportApiResponseMsg.newBuilder() + .setOtaPackageResponseMsg(builder.build()) + .build()); + } + + private ListenableFuture handle(TransportProtos.SendOtaPackageBodyRequestMsg requestMsg) { + TenantId tenantId = new TenantId(new UUID(requestMsg.getTenantIdMSB(), requestMsg.getTenantIdLSB())); + OtaPackageId otaPackageId = new OtaPackageId(new UUID(requestMsg.getOtaPackageIdMSB(), requestMsg.getOtaPackageIdLSB())); + OtaPackageInfo otaPackageInfo = otaPackageService.findOtaPackageInfoById(tenantId, otaPackageId); + + TransportProtos.SendOtaPackageBodyResponseMsg.Builder builder = TransportProtos.SendOtaPackageBodyResponseMsg.newBuilder(); + + if (otaPackageInfo == null) { + builder.setResponseStatus(TransportProtos.ResponseStatus.NOT_FOUND); + } else if (otaPackageInfo.hasUrl()) { + builder.setResponseStatus(TransportProtos.ResponseStatus.FAILURE); + log.trace("[{}] Can`t send OtaPackage with URL data!", otaPackageInfo.getId()); + } else { + builder.setResponseStatus(TransportProtos.ResponseStatus.SUCCESS); + + dbCallbackExecutorService.submit(() -> { + try { OtaPackage otaPackage = otaPackageService.findOtaPackageById(tenantId, otaPackageId); - try { - //TODO: Do not put to Redis/InMem Cache and use File system on the Transport service instead. - otaPackageDataCache.put(otaPackageId.toString(), otaPackage.getData().readAllBytes()); - } catch (IOException e) { - log.error("Failed to cache ota package with id {}",otaPackage.getId(), e); - } + var br = new BufferedReader(new InputStreamReader(otaPackage.getData())); + br.read() + } catch (Throwable e) { + log.warn("[{}][{}] Failed to publish ota package to the service [{}] queue: ", tenantId, otaPackageId, ); + } + }); + + if (!otaPackageDataCache.has(otaPackageId.toString())) { + + try { + //TODO: Do not put to Redis/InMem Cache and use File system on the Transport service instead. + otaPackageDataCache.put(otaPackageId.toString(), otaPackage.getData().readAllBytes()); + } catch (IOException e) { + log.error("Failed to cache ota package with id {}", otaPackage.getId(), e); } } } return Futures.immediateFuture( TransportApiResponseMsg.newBuilder() - .setOtaPackageResponseMsg(builder.build()) + .setSendOtaPackageBodyResponseMsg(builder.build()) .build()); } diff --git a/common/cache/src/main/java/org/thingsboard/server/cache/ota/CaffeineOtaPackageCache.java b/common/cache/src/main/java/org/thingsboard/server/cache/ota/CaffeineOtaPackageCache.java deleted file mode 100644 index aaea36abae..0000000000 --- a/common/cache/src/main/java/org/thingsboard/server/cache/ota/CaffeineOtaPackageCache.java +++ /dev/null @@ -1,68 +0,0 @@ -/** - * Copyright © 2016-2022 The Thingsboard Authors - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package org.thingsboard.server.cache.ota; - -import lombok.RequiredArgsConstructor; -import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; -import org.springframework.cache.CacheManager; -import org.springframework.stereotype.Service; - -import static org.thingsboard.server.common.data.CacheConstants.OTA_PACKAGE_DATA_CACHE; - -@Service -@ConditionalOnProperty(prefix = "cache", value = "type", havingValue = "caffeine", matchIfMissing = true) -@RequiredArgsConstructor -public class CaffeineOtaPackageCache implements OtaPackageDataCache { - - private final CacheManager cacheManager; - - @Override - public byte[] get(String key) { - return get(key, 0, 0); - } - - @Override - public byte[] get(String key, int chunkSize, int chunk) { - byte[] data = cacheManager.getCache(OTA_PACKAGE_DATA_CACHE).get(key, byte[].class); - - if (chunkSize < 1) { - return data; - } - - if (data != null && data.length > 0) { - int startIndex = chunkSize * chunk; - - int size = Math.min(data.length - startIndex, chunkSize); - - if (startIndex < data.length && size > 0) { - byte[] result = new byte[size]; - System.arraycopy(data, startIndex, result, 0, size); - return result; - } - } - return new byte[0]; - } - - @Override - public void put(String key, byte[] value) { - cacheManager.getCache(OTA_PACKAGE_DATA_CACHE).putIfAbsent(key, value); - } - - @Override - public void evict(String key) { - cacheManager.getCache(OTA_PACKAGE_DATA_CACHE).evict(key); - } -} diff --git a/common/cache/src/main/java/org/thingsboard/server/cache/ota/RedisOtaPackageDataCache.java b/common/cache/src/main/java/org/thingsboard/server/cache/ota/RedisOtaPackageDataCache.java deleted file mode 100644 index f6ec28d170..0000000000 --- a/common/cache/src/main/java/org/thingsboard/server/cache/ota/RedisOtaPackageDataCache.java +++ /dev/null @@ -1,68 +0,0 @@ -/** - * Copyright © 2016-2022 The Thingsboard Authors - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package org.thingsboard.server.cache.ota; - -import lombok.RequiredArgsConstructor; -import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; -import org.springframework.data.redis.connection.RedisConnection; -import org.springframework.data.redis.connection.RedisConnectionFactory; -import org.springframework.stereotype.Service; - -import static org.thingsboard.server.common.data.CacheConstants.OTA_PACKAGE_DATA_CACHE; - -@Service -@ConditionalOnProperty(prefix = "cache", value = "type", havingValue = "redis") -@RequiredArgsConstructor -public class RedisOtaPackageDataCache implements OtaPackageDataCache { - - private final RedisConnectionFactory redisConnectionFactory; - - @Override - public byte[] get(String key) { - return get(key, 0, 0); - } - - @Override - public byte[] get(String key, int chunkSize, int chunk) { - try (RedisConnection connection = redisConnectionFactory.getConnection()) { - if (chunkSize == 0) { - return connection.get(toOtaPackageCacheKey(key)); - } - - int startIndex = chunkSize * chunk; - int endIndex = startIndex + chunkSize - 1; - return connection.getRange(toOtaPackageCacheKey(key), startIndex, endIndex); - } - } - - @Override - public void put(String key, byte[] value) { - try (RedisConnection connection = redisConnectionFactory.getConnection()) { - connection.set(toOtaPackageCacheKey(key), value); - } - } - - @Override - public void evict(String key) { - try (RedisConnection connection = redisConnectionFactory.getConnection()) { - connection.del(toOtaPackageCacheKey(key)); - } - } - - private byte[] toOtaPackageCacheKey(String key) { - return String.format("%s::%s", OTA_PACKAGE_DATA_CACHE, key).getBytes(); - } -} diff --git a/common/cache/src/main/java/org/thingsboard/server/cache/ota/files/OtaFileState.java b/common/cache/src/main/java/org/thingsboard/server/cache/ota/files/OtaFileState.java index 41e2231479..7507bec793 100644 --- a/common/cache/src/main/java/org/thingsboard/server/cache/ota/files/OtaFileState.java +++ b/common/cache/src/main/java/org/thingsboard/server/cache/ota/files/OtaFileState.java @@ -16,7 +16,6 @@ package org.thingsboard.server.cache.ota.files; import lombok.Data; -import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.common.data.id.OtaPackageId; import java.io.File; diff --git a/common/cluster-api/src/main/proto/queue.proto b/common/cluster-api/src/main/proto/queue.proto index 1714e109a3..6f182eb266 100644 --- a/common/cluster-api/src/main/proto/queue.proto +++ b/common/cluster-api/src/main/proto/queue.proto @@ -462,6 +462,27 @@ message GetOtaPackageResponseMsg { string version = 6; string contentType = 7; string fileName = 8; + int64 fileSize = 9; +} + +message SendOtaPackageBodyRequestMsg { + int64 tenantIdMSB = 1; + int64 tenantIdLSB = 2; + int64 otaPackageIdMSB = 3; + int64 otaPackageIdLSB = 4; + string serviceId = 5; +} + +message SendOtaPackageBodyResponseMsg { + ResponseStatus responseStatus = 1; +} + +message OtaPackageBodyChunkMsg { + int64 otaPackageIdMSB = 1; + int64 otaPackageIdLSB = 2; + int32 chunkIndex = 3; + int32 chunkCount = 4; + bytes data = 5; } message DeviceActivityProto { @@ -896,6 +917,7 @@ message TransportApiRequestMsg { GetDeviceRequestMsg deviceRequestMsg = 12; GetDeviceCredentialsRequestMsg deviceCredentialsRequestMsg = 13; GetAllQueueRoutingInfoRequestMsg getAllQueueRoutingInfoRequestMsg = 14; + SendOtaPackageBodyRequestMsg sendOtaPackageBodyRequestMsg = 15; } /* Response from ThingsBoard Core Service to Transport Service */ @@ -911,6 +933,7 @@ message TransportApiResponseMsg { GetDeviceResponseMsg deviceResponseMsg = 9; GetDeviceCredentialsResponseMsg deviceCredentialsResponseMsg = 10; repeated GetQueueRoutingInfoResponseMsg getQueueRoutingInfoResponseMsgs = 11; + SendOtaPackageBodyResponseMsg sendOtaPackageBodyResponseMsg = 12; } /* Messages that are handled by ThingsBoard Core Service */ diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java index f37715efc8..118a87da3c 100644 --- a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java +++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java @@ -16,6 +16,7 @@ package org.thingsboard.server.transport.mqtt; import com.fasterxml.jackson.databind.JsonNode; +import com.google.common.util.concurrent.ListenableFuture; import com.google.gson.JsonParseException; import io.netty.channel.ChannelFuture; import io.netty.channel.ChannelHandlerContext; @@ -40,8 +41,10 @@ import io.netty.util.CharsetUtil; import io.netty.util.ReferenceCountUtil; import io.netty.util.concurrent.Future; import io.netty.util.concurrent.GenericFutureListener; +import lombok.SneakyThrows; import lombok.extern.slf4j.Slf4j; import org.apache.commons.lang3.StringUtils; +import org.thingsboard.common.util.DonAsynchron; import org.thingsboard.server.common.data.DataConstants; import org.thingsboard.server.common.data.Device; import org.thingsboard.server.common.data.DeviceProfile; @@ -50,6 +53,7 @@ import org.thingsboard.server.common.data.TransportPayloadType; import org.thingsboard.server.common.data.device.profile.MqttTopics; import org.thingsboard.server.common.data.id.DeviceId; import org.thingsboard.server.common.data.id.OtaPackageId; +import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.ota.OtaPackageType; import org.thingsboard.server.common.data.rpc.RpcStatus; import org.thingsboard.server.common.msg.EncryptionUtil; @@ -69,6 +73,7 @@ import org.thingsboard.server.gen.transport.TransportProtos.ProvisionDeviceRespo import org.thingsboard.server.gen.transport.TransportProtos.ValidateDeviceX509CertRequestMsg; import org.thingsboard.server.queue.scheduler.SchedulerComponent; import org.thingsboard.server.transport.mqtt.adaptors.MqttTransportAdaptor; +import org.thingsboard.server.transport.mqtt.ota.OtaPackageShortInfo; import org.thingsboard.server.transport.mqtt.session.DeviceSessionCtx; import org.thingsboard.server.transport.mqtt.session.GatewaySessionHandler; import org.thingsboard.server.transport.mqtt.session.MqttTopicMatcher; @@ -129,7 +134,7 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement volatile InetSocketAddress address; volatile GatewaySessionHandler gatewaySessionHandler; - private final ConcurrentHashMap otaPackSessions; + private final ConcurrentHashMap otaPackSessions; private final ConcurrentHashMap chunkSizes; private final ConcurrentMap rpcAwaitingAck; @@ -482,10 +487,10 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement return; } - String otaPackageId = otaPackSessions.get(requestId); + OtaPackageShortInfo otaPackageInfo = otaPackSessions.get(requestId); - if (otaPackageId != null) { - sendOtaPackage(ctx, mqttMsg.variableHeader().packetId(), otaPackageId, requestId, chunkSize, chunk, type); + if (otaPackageInfo != null) { + sendOtaPackage(ctx, mqttMsg.variableHeader().packetId(), otaPackageInfo, requestId, chunkSize, chunk); } else { TransportProtos.SessionInfoProto sessionInfo = deviceSessionCtx.getSessionInfo(); TransportProtos.GetOtaPackageRequestMsg getOtaPackageRequestMsg = TransportProtos.GetOtaPackageRequestMsg.newBuilder() @@ -576,9 +581,12 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement @Override public void onSuccess(TransportProtos.GetOtaPackageResponseMsg response) { if (TransportProtos.ResponseStatus.SUCCESS.equals(response.getResponseStatus())) { - OtaPackageId firmwareId = new OtaPackageId(new UUID(response.getOtaPackageIdMSB(), response.getOtaPackageIdLSB())); - otaPackSessions.put(requestId, firmwareId.toString()); - sendOtaPackage(ctx, msgId, firmwareId.toString(), requestId, chunkSize, chunk, OtaPackageType.valueOf(response.getType())); + OtaPackageShortInfo info = new OtaPackageShortInfo( + new TenantId(new UUID(msg.getTenantIdMSB(), msg.getTenantIdLSB())), + new OtaPackageId(new UUID(response.getOtaPackageIdMSB(), response.getOtaPackageIdLSB())), + OtaPackageType.valueOf(response.getType()), response.getFileSize()); + otaPackSessions.put(requestId, info); + sendOtaPackage(ctx, msgId, info, requestId, chunkSize, chunk); } else { sendOtaPackageError(ctx, response.getResponseStatus().toString()); } @@ -591,19 +599,28 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement } } - private void sendOtaPackage(ChannelHandlerContext ctx, int msgId, String firmwareId, String requestId, int chunkSize, int chunk, OtaPackageType type) { - log.trace("[{}] Send firmware [{}] to device!", sessionId, firmwareId); + private void sendOtaPackage(ChannelHandlerContext ctx, int msgId, OtaPackageShortInfo otaPackageInfo, String requestId, int chunkSize, int chunk) { + log.trace("[{}] Send firmware [{}] to device!", sessionId, otaPackageInfo.getId()); ack(ctx, msgId); try { - byte[] firmwareChunk = context.getOtaPackageDataCache().get(firmwareId, chunkSize, chunk); - deviceSessionCtx.getPayloadAdaptor() - .convertToPublish(deviceSessionCtx, firmwareChunk, requestId, chunk, type) - .ifPresent(deviceSessionCtx.getChannel()::writeAndFlush); + DonAsynchron.withCallback(context.getOtaPackageService().get(otaPackageInfo.getTenantId(), otaPackageInfo.getId(), chunkSize, chunk), + firmwareChunk -> { + pushChunk(otaPackageInfo, requestId, chunk, firmwareChunk); + }, e -> { + log.trace("[{}] Failed to send firmware response!", sessionId, e); + }, context.getExecutor()); } catch (Exception e) { log.trace("[{}] Failed to send firmware response!", sessionId, e); } } + @SneakyThrows + private void pushChunk(OtaPackageShortInfo otaPackageInfo, String requestId, int chunk, byte[] firmwareChunk) { + deviceSessionCtx.getPayloadAdaptor() + .convertToPublish(deviceSessionCtx, firmwareChunk, requestId, chunk, otaPackageInfo.getType()) + .ifPresent(deviceSessionCtx.getChannel()::writeAndFlush); + } + private void sendOtaPackageError(ChannelHandlerContext ctx, String error) { log.warn("[{}] {}", sessionId, error); deviceSessionCtx.getChannel().writeAndFlush(deviceSessionCtx diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/ota/OtaPackageShortInfo.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/ota/OtaPackageShortInfo.java new file mode 100644 index 0000000000..fa1a4e4454 --- /dev/null +++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/ota/OtaPackageShortInfo.java @@ -0,0 +1,31 @@ +/** + * Copyright © 2016-2022 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.server.transport.mqtt.ota; + +import lombok.Data; +import org.thingsboard.server.common.data.id.OtaPackageId; +import org.thingsboard.server.common.data.id.TenantId; +import org.thingsboard.server.common.data.ota.OtaPackageType; + +@Data +public class OtaPackageShortInfo { + + private final TenantId tenantId; + private final OtaPackageId id; + private final OtaPackageType type; + private final long size; + +} diff --git a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/TransportContext.java b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/TransportContext.java index 7b86ef33c8..205d8fb4d3 100644 --- a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/TransportContext.java +++ b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/TransportContext.java @@ -23,6 +23,7 @@ import org.springframework.beans.factory.annotation.Autowired; import org.thingsboard.common.util.ThingsBoardExecutors; import org.thingsboard.server.cache.ota.OtaPackageDataCache; import org.thingsboard.server.common.transport.limits.TransportRateLimitService; +import org.thingsboard.server.common.transport.ota.TransportOtaPackageService; import org.thingsboard.server.queue.discovery.TbServiceInfoProvider; import org.thingsboard.server.queue.scheduler.SchedulerComponent; @@ -53,7 +54,7 @@ public abstract class TransportContext { @Getter @Autowired - private OtaPackageDataCache otaPackageDataCache; + private TransportOtaPackageService otaPackageService; @Autowired private TransportResourceCache transportResourceCache; diff --git a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/TransportService.java b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/TransportService.java index f1d3f7e26c..436b0f7634 100644 --- a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/TransportService.java +++ b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/TransportService.java @@ -56,6 +56,8 @@ import org.thingsboard.server.gen.transport.TransportProtos.ValidateBasicMqttCre import org.thingsboard.server.gen.transport.TransportProtos.ValidateDeviceLwM2MCredentialsRequestMsg; import org.thingsboard.server.gen.transport.TransportProtos.ValidateDeviceTokenRequestMsg; import org.thingsboard.server.gen.transport.TransportProtos.ValidateDeviceX509CertRequestMsg; +import org.thingsboard.server.gen.transport.TransportProtos.SendOtaPackageBodyRequestMsg; +import org.thingsboard.server.gen.transport.TransportProtos.SendOtaPackageBodyResponseMsg; import java.util.List; import java.util.concurrent.ExecutorService; @@ -127,6 +129,8 @@ public interface TransportService { void process(SessionInfoProto sessionInfoProto, GetOtaPackageRequestMsg msg, TransportServiceCallback callback); + void process(SendOtaPackageBodyRequestMsg msg, TransportServiceCallback callback); + SessionMetaData registerAsyncSession(SessionInfoProto sessionInfo, SessionMsgListener listener); SessionMetaData registerSyncSession(SessionInfoProto sessionInfo, SessionMsgListener listener, long timeout); diff --git a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/ota/DefaultTransportOtaPackageService.java b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/ota/DefaultTransportOtaPackageService.java new file mode 100644 index 0000000000..b2b74e88e0 --- /dev/null +++ b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/ota/DefaultTransportOtaPackageService.java @@ -0,0 +1,178 @@ +/** + * Copyright © 2016-2022 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.server.common.transport.ota; + +import com.google.common.util.concurrent.Futures; +import com.google.common.util.concurrent.ListenableFuture; +import com.google.common.util.concurrent.SettableFuture; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.apache.commons.io.FileUtils; +import org.apache.commons.lang3.StringUtils; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.stereotype.Service; +import org.thingsboard.server.cache.ota.files.OtaFileState; +import org.thingsboard.server.common.data.id.OtaPackageId; +import org.thingsboard.server.common.data.id.TenantId; +import org.thingsboard.server.common.transport.TransportService; +import org.thingsboard.server.common.transport.TransportServiceCallback; +import org.thingsboard.server.gen.transport.TransportProtos; +import org.thingsboard.server.queue.discovery.TbServiceInfoProvider; +import org.thingsboard.server.queue.util.TbTransportComponent; + +import javax.annotation.PostConstruct; +import java.io.File; +import java.io.IOException; +import java.nio.file.Paths; +import java.util.Arrays; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.locks.Lock; + +import static org.thingsboard.server.common.data.CacheConstants.OTA_PACKAGE_DATA_CACHE; + +@Slf4j +@Service +@TbTransportComponent +@RequiredArgsConstructor +public class DefaultTransportOtaPackageService implements TransportOtaPackageService { + + private final static String FILE_NAME_TEMPLATE = "%s.tmp"; + + @Value("${files.temporary_files_directory:}") + private String tmpDir; + @Value("${java.io.tmpdir}") + private String defaultTmpDir; + + private final ConcurrentMap files = new ConcurrentHashMap<>(); + private final TbServiceInfoProvider serviceInfoProvider; + private final TransportService transportService; + + @PostConstruct + public void init() { + if (StringUtils.isEmpty(tmpDir)) { + tmpDir = defaultTmpDir; + } + createTempDirectoryIfNotExist(); + cleanDirectory(); + log.info("Directory {} with temporary ota files cleaned", tmpDir); + } + + @SuppressWarnings("UnstableApiUsage") + @Override + public ListenableFuture get(TenantId tenantId, OtaPackageId otaId, int chunkSize, int chunk) { + var state = files.computeIfAbsent(otaId, tmp -> + new TransportOtaFileState(tenantId, otaId, Paths.get(tmpDir, "tb-transport", "ota", + String.format(FILE_NAME_TEMPLATE, otaId.getId().toString())))); + Lock lock = state.getLock(); + lock.lock(); + try { + if (state.isLoaded() && state.exists()) { + return readChunk(state, chunkSize, chunk); + } + if (state.getLoadFuture() == null) { + state.setLoaded(false); + state.setLoadFuture(requestFile(state)); + } + + state.updateLastActivityTime(); + return Futures.transformAsync(state.getLoadFuture(), tmp -> readChunk(state, chunkSize, chunk), transportService.getCallbackExecutor()); + } finally { + lock.unlock(); + } + } + + private SettableFuture requestFile(TransportOtaFileState state) { + SettableFuture result = SettableFuture.create(); + transportService.process(TransportProtos.SendOtaPackageBodyRequestMsg.newBuilder() + .setTenantIdMSB(state.getTenantId().getId().getMostSignificantBits()) + .setTenantIdLSB(state.getTenantId().getId().getLeastSignificantBits()) + .setOtaPackageIdMSB(state.getOtaId().getId().getMostSignificantBits()) + .setOtaPackageIdMSB(state.getOtaId().getId().getLeastSignificantBits()) + .setServiceId(serviceInfoProvider.getServiceId()) + .build(), new TransportServiceCallback<>() { + @Override + public void onSuccess(TransportProtos.SendOtaPackageBodyResponseMsg msg) { + if (msg.getResponseStatus() == TransportProtos.ResponseStatus.SUCCESS) { + log.debug("[{}] Requested ota file content", state.getOtaId()); + } else { + log.warn("[{}] Failed to request ota file content due to: {}", state.getOtaId(), msg.getResponseStatus()); + state.getLoadFuture().setException(new RuntimeException("Failed to request ota file content due to: " + msg.getResponseStatus())); + } + + } + + @Override + public void onError(Throwable e) { + log.warn("[{}] Failed to request ota file content", state.getOtaId(), e); + state.getLoadFuture().setException(e); + } + }); + return result; + } + + private ListenableFuture readChunk(TransportOtaFileState state, int chunkSize, int chunk) { + //TODO: implement; +// byte[] data = cacheManager.getCache(OTA_PACKAGE_DATA_CACHE).get(key, byte[].class); +// +// if (chunkSize < 1) { +// return data; +// } +// +// if (data != null && data.length > 0) { +// int startIndex = chunkSize * chunk; +// +// int size = Math.min(data.length - startIndex, chunkSize); +// +// if (startIndex < data.length && size > 0) { +// byte[] result = new byte[size]; +// System.arraycopy(data, startIndex, result, 0, size); +// return result; +// } +// } +// return new byte[0]; + + return Futures.immediateFuture(null); + } + + private void createTempDirectoryIfNotExist() { + File directory = Paths.get(tmpDir, "tb-transport", "ota").toFile(); + if (!directory.exists()) { + try { + FileUtils.forceMkdir(directory); + } catch (IOException e) { + log.error("Failed to create directory for temporary files ", e); + } + } + } + + private void cleanDirectory() { + File directory = Paths.get(tmpDir, "tb-transport", "ota").toFile(); + if (directory.isDirectory()) { + File[] files = directory.listFiles(); + if (files == null) return; + Arrays.stream(files).forEach( + file -> { + try { + FileUtils.delete(file); + } catch (Exception e) { + log.error("Failed to delete file {}", file.getName(), e); + } + } + ); + } + } +} diff --git a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/ota/TransportOtaFileState.java b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/ota/TransportOtaFileState.java new file mode 100644 index 0000000000..91caaf6136 --- /dev/null +++ b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/ota/TransportOtaFileState.java @@ -0,0 +1,53 @@ +/** + * Copyright © 2016-2022 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.server.common.transport.ota; + +import com.google.common.util.concurrent.SettableFuture; +import lombok.Data; +import org.thingsboard.server.common.data.id.OtaPackageId; +import org.thingsboard.server.common.data.id.TenantId; + +import java.io.File; +import java.nio.file.Path; +import java.util.concurrent.locks.Lock; +import java.util.concurrent.locks.ReentrantLock; + +@Data +public class TransportOtaFileState { + + private final Lock lock = new ReentrantLock(); + private final TenantId tenantId; + private final OtaPackageId otaId; + private final Path filePath; + + private boolean loaded; + private SettableFuture loadFuture; + + private long lastActivityTime; + + public void updateLastActivityTime() { + lastActivityTime = System.currentTimeMillis(); + } + + public boolean exists() { + return filePath.toFile().exists(); + } + + public File getFile() { + return filePath.toFile(); + } + +} diff --git a/common/cache/src/main/java/org/thingsboard/server/cache/ota/OtaPackageDataCache.java b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/ota/TransportOtaPackageService.java similarity index 62% rename from common/cache/src/main/java/org/thingsboard/server/cache/ota/OtaPackageDataCache.java rename to common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/ota/TransportOtaPackageService.java index 1be24b3b24..cdcc553b67 100644 --- a/common/cache/src/main/java/org/thingsboard/server/cache/ota/OtaPackageDataCache.java +++ b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/ota/TransportOtaPackageService.java @@ -13,20 +13,14 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.thingsboard.server.cache.ota; +package org.thingsboard.server.common.transport.ota; -public interface OtaPackageDataCache { +import com.google.common.util.concurrent.ListenableFuture; +import org.thingsboard.server.common.data.id.OtaPackageId; +import org.thingsboard.server.common.data.id.TenantId; - byte[] get(String key); +public interface TransportOtaPackageService { - byte[] get(String key, int chunkSize, int chunk); + ListenableFuture get(TenantId tenantId, OtaPackageId packageId, int chunkSize, int chunk); - void put(String key, byte[] value); - - void evict(String key); - - default boolean has(String otaPackageId) { - byte[] data = get(otaPackageId, 1, 0); - return data != null && data.length > 0; - } } diff --git a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportService.java b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportService.java index a63e4a01a2..6ca9ce4501 100644 --- a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportService.java +++ b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportService.java @@ -69,6 +69,7 @@ import org.thingsboard.server.common.transport.auth.GetOrCreateDeviceFromGateway import org.thingsboard.server.common.transport.auth.TransportDeviceInfo; import org.thingsboard.server.common.transport.auth.ValidateDeviceCredentialsResponse; import org.thingsboard.server.common.transport.limits.TransportRateLimitService; +import org.thingsboard.server.common.transport.ota.TransportOtaPackageService; import org.thingsboard.server.queue.util.DataDecodingEncodingService; import org.thingsboard.server.common.transport.util.JsonUtils; import org.thingsboard.server.gen.transport.TransportProtos; @@ -157,6 +158,10 @@ public class DefaultTransportService implements TransportService { @Autowired @Lazy private TbApiUsageClient apiUsageClient; + @Autowired + @Lazy + private TransportOtaPackageService otaPackageService; + private final Map statsMap = new LinkedHashMap<>(); private final Gson gson = new Gson(); @@ -727,6 +732,16 @@ public class DefaultTransportService implements TransportService { } } + @Override + public void process(TransportProtos.SendOtaPackageBodyRequestMsg msg, TransportServiceCallback callback) { + TbProtoQueueMsg protoMsg = + new TbProtoQueueMsg<>(UUID.randomUUID(), TransportProtos.TransportApiRequestMsg.newBuilder().setSendOtaPackageBodyRequestMsg(msg).build()); + + AsyncCallbackTemplate.withCallback(transportApiRequestTemplate.send(protoMsg), response -> { + callback.onSuccess(response.getValue().getSendOtaPackageBodyResponseMsg()); + }, callback::onError, transportCallbackExecutor); + } + @Override public void reportActivity(TransportProtos.SessionInfoProto sessionInfo) { reportActivityInternal(sessionInfo);