Browse Source

Implementation of the Git Queue service client

pull/6591/head
Andrii Shvaika 4 years ago
parent
commit
7071c7a6bc
  1. 21
      application/src/main/java/org/thingsboard/server/controller/BaseController.java
  2. 85
      application/src/main/java/org/thingsboard/server/controller/EntitiesVersionControlController.java
  3. 12
      application/src/main/java/org/thingsboard/server/service/install/SqlDatabaseUpgradeService.java
  4. 8
      application/src/main/java/org/thingsboard/server/service/queue/DefaultTbClusterService.java
  5. 33
      application/src/main/java/org/thingsboard/server/service/sync/vc/CommitGitRequest.java
  6. 215
      application/src/main/java/org/thingsboard/server/service/sync/vc/DefaultEntitiesVersionControlService.java
  7. 287
      application/src/main/java/org/thingsboard/server/service/sync/vc/DefaultGitVersionControlQueueService.java
  8. 37
      application/src/main/java/org/thingsboard/server/service/sync/vc/EntitiesContentGitRequest.java
  9. 19
      application/src/main/java/org/thingsboard/server/service/sync/vc/EntitiesVersionControlService.java
  10. 34
      application/src/main/java/org/thingsboard/server/service/sync/vc/EntityContentGitRequest.java
  11. 57
      application/src/main/java/org/thingsboard/server/service/sync/vc/GitVersionControlQueueService.java
  12. 29
      application/src/main/java/org/thingsboard/server/service/sync/vc/ListBranchesGitRequest.java
  13. 30
      application/src/main/java/org/thingsboard/server/service/sync/vc/ListEntitiesGitRequest.java
  14. 31
      application/src/main/java/org/thingsboard/server/service/sync/vc/ListVersionsGitRequest.java
  15. 452
      application/src/main/java/org/thingsboard/server/service/sync/vc/LocalGitVersionControlService.java
  16. 37
      application/src/main/java/org/thingsboard/server/service/sync/vc/PendingGitRequest.java
  17. 5
      application/src/main/resources/thingsboard.yml
  18. 7
      common/cluster-api/src/main/java/org/thingsboard/server/cluster/TbClusterService.java
  19. 96
      common/cluster-api/src/main/proto/queue.proto
  20. 2
      common/message/src/main/java/org/thingsboard/server/common/msg/queue/ServiceType.java
  21. 10
      common/queue/src/main/java/org/thingsboard/server/queue/discovery/HashPartitionService.java
  22. 6
      common/queue/src/main/java/org/thingsboard/server/queue/kafka/TbKafkaTopicConfigs.java
  23. 7
      common/queue/src/main/java/org/thingsboard/server/queue/provider/AwsSqsMonolithQueueFactory.java
  24. 7
      common/queue/src/main/java/org/thingsboard/server/queue/provider/AwsSqsTbCoreQueueFactory.java
  25. 6
      common/queue/src/main/java/org/thingsboard/server/queue/provider/InMemoryMonolithQueueFactory.java
  26. 36
      common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaMonolithQueueFactory.java
  27. 20
      common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaTbCoreQueueFactory.java
  28. 116
      common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaTbVersionControlQueueFactory.java
  29. 7
      common/queue/src/main/java/org/thingsboard/server/queue/provider/PubSubMonolithQueueFactory.java
  30. 7
      common/queue/src/main/java/org/thingsboard/server/queue/provider/PubSubTbCoreQueueFactory.java
  31. 7
      common/queue/src/main/java/org/thingsboard/server/queue/provider/RabbitMqMonolithQueueFactory.java
  32. 7
      common/queue/src/main/java/org/thingsboard/server/queue/provider/RabbitMqTbCoreQueueFactory.java
  33. 7
      common/queue/src/main/java/org/thingsboard/server/queue/provider/ServiceBusMonolithQueueFactory.java
  34. 7
      common/queue/src/main/java/org/thingsboard/server/queue/provider/ServiceBusTbCoreQueueFactory.java
  35. 8
      common/queue/src/main/java/org/thingsboard/server/queue/provider/TbCoreQueueFactory.java
  36. 8
      common/queue/src/main/java/org/thingsboard/server/queue/provider/TbCoreQueueProducerProvider.java
  37. 8
      common/queue/src/main/java/org/thingsboard/server/queue/provider/TbQueueProducerProvider.java
  38. 6
      common/queue/src/main/java/org/thingsboard/server/queue/provider/TbRuleEngineProducerProvider.java
  39. 6
      common/queue/src/main/java/org/thingsboard/server/queue/provider/TbTransportQueueProducerProvider.java
  40. 84
      common/queue/src/main/java/org/thingsboard/server/queue/provider/TbVersionControlProducerProvider.java
  41. 44
      common/queue/src/main/java/org/thingsboard/server/queue/provider/TbVersionControlQueueFactory.java
  42. 34
      common/queue/src/main/java/org/thingsboard/server/queue/settings/TbQueueVersionControlSettings.java
  43. 26
      common/queue/src/main/java/org/thingsboard/server/queue/util/TbVersionControlComponent.java
  44. 24
      common/version-control/pom.xml
  45. 11
      common/version-control/src/main/java/org/thingsboard/server/service/sync/vc/GitVersionControlService.java
  46. 71
      common/version-control/src/main/proto/vc.proto

21
application/src/main/java/org/thingsboard/server/controller/BaseController.java

@ -18,6 +18,10 @@ package org.thingsboard.server.controller;
import com.fasterxml.jackson.core.JsonProcessingException; import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper; import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ObjectNode; import com.fasterxml.jackson.databind.node.ObjectNode;
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 lombok.Getter; import lombok.Getter;
import lombok.extern.slf4j.Slf4j; import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils; import org.apache.commons.lang3.StringUtils;
@ -27,6 +31,7 @@ import org.springframework.http.MediaType;
import org.springframework.security.core.Authentication; import org.springframework.security.core.Authentication;
import org.springframework.security.core.context.SecurityContextHolder; import org.springframework.security.core.context.SecurityContextHolder;
import org.springframework.web.bind.annotation.ExceptionHandler; import org.springframework.web.bind.annotation.ExceptionHandler;
import org.springframework.web.context.request.async.DeferredResult;
import org.thingsboard.server.cluster.TbClusterService; import org.thingsboard.server.cluster.TbClusterService;
import org.thingsboard.server.common.data.Customer; import org.thingsboard.server.common.data.Customer;
import org.thingsboard.server.common.data.Dashboard; import org.thingsboard.server.common.data.Dashboard;
@ -948,4 +953,20 @@ public abstract class BaseController {
return MediaType.APPLICATION_OCTET_STREAM; return MediaType.APPLICATION_OCTET_STREAM;
} }
} }
protected <T> DeferredResult<T> wrapFuture(ListenableFuture<T> future) {
final DeferredResult<T> deferredResult = new DeferredResult<>();
Futures.addCallback(future, new FutureCallback<>() {
@Override
public void onSuccess(T result) {
deferredResult.setResult(result);
}
@Override
public void onFailure(Throwable t) {
deferredResult.setErrorResult(t);
}
}, MoreExecutors.directExecutor());
return deferredResult;
}
} }

85
application/src/main/java/org/thingsboard/server/controller/EntitiesVersionControlController.java

@ -15,16 +15,26 @@
*/ */
package org.thingsboard.server.controller; package org.thingsboard.server.controller;
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 io.swagger.annotations.ApiOperation; import io.swagger.annotations.ApiOperation;
import lombok.Data; import lombok.Data;
import lombok.RequiredArgsConstructor; import lombok.RequiredArgsConstructor;
import org.springframework.http.HttpStatus;
import org.springframework.http.ResponseEntity;
import org.springframework.security.access.prepost.PreAuthorize; import org.springframework.security.access.prepost.PreAuthorize;
import org.springframework.web.bind.annotation.*; import org.springframework.web.bind.annotation.*;
import org.springframework.web.context.request.async.DeferredResult;
import org.thingsboard.server.common.data.EntityType; import org.thingsboard.server.common.data.EntityType;
import org.thingsboard.server.common.data.StringUtils; import org.thingsboard.server.common.data.StringUtils;
import org.thingsboard.server.common.data.exception.ThingsboardException; import org.thingsboard.server.common.data.exception.ThingsboardException;
import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.common.data.id.EntityId;
import org.thingsboard.server.common.data.id.EntityIdFactory; import org.thingsboard.server.common.data.id.EntityIdFactory;
import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.dao.device.claim.ClaimResponse;
import org.thingsboard.server.dao.device.claim.ClaimResult;
import org.thingsboard.server.service.security.model.SecurityUser; import org.thingsboard.server.service.security.model.SecurityUser;
import org.thingsboard.server.service.sync.vc.EntitiesVersionControlService; import org.thingsboard.server.service.sync.vc.EntitiesVersionControlService;
import org.thingsboard.server.common.data.sync.vc.EntityVersion; import org.thingsboard.server.common.data.sync.vc.EntityVersion;
@ -34,6 +44,7 @@ import org.thingsboard.server.common.data.sync.vc.VersionedEntityInfo;
import org.thingsboard.server.common.data.sync.vc.request.create.VersionCreateRequest; import org.thingsboard.server.common.data.sync.vc.request.create.VersionCreateRequest;
import org.thingsboard.server.common.data.sync.vc.request.load.VersionLoadRequest; import org.thingsboard.server.common.data.sync.vc.request.load.VersionLoadRequest;
import javax.annotation.Nullable;
import java.util.ArrayList; import java.util.ArrayList;
import java.util.List; import java.util.List;
import java.util.UUID; import java.util.UUID;
@ -90,16 +101,15 @@ public class EntitiesVersionControlController extends BaseController {
" }\n" + " }\n" +
"}\n```") "}\n```")
@PostMapping("/version") @PostMapping("/version")
public VersionCreationResult saveEntitiesVersion(@RequestBody VersionCreateRequest request) throws ThingsboardException { public DeferredResult<VersionCreationResult> saveEntitiesVersion(@RequestBody VersionCreateRequest request) throws ThingsboardException {
SecurityUser user = getCurrentUser(); SecurityUser user = getCurrentUser();
try { try {
return versionControlService.saveEntitiesVersion(user, request); return wrapFuture(versionControlService.saveEntitiesVersion(user, request));
} catch (Exception e) { } catch (Exception e) {
throw handleException(e); throw handleException(e);
} }
} }
@ApiOperation(value = "", notes = "" + @ApiOperation(value = "", notes = "" +
"```\n[\n" + "```\n[\n" +
" {\n" + " {\n" +
@ -108,12 +118,12 @@ public class EntitiesVersionControlController extends BaseController {
" }\n" + " }\n" +
"]\n```") "]\n```")
@GetMapping("/version/{branch}/{entityType}/{externalEntityUuid}") @GetMapping("/version/{branch}/{entityType}/{externalEntityUuid}")
public List<EntityVersion> listEntityVersions(@PathVariable String branch, public DeferredResult<List<EntityVersion>> listEntityVersions(@PathVariable String branch,
@PathVariable EntityType entityType, @PathVariable EntityType entityType,
@PathVariable UUID externalEntityUuid) throws ThingsboardException { @PathVariable UUID externalEntityUuid) throws ThingsboardException {
try { try {
EntityId externalEntityId = EntityIdFactory.getByTypeAndUuid(entityType, externalEntityUuid); EntityId externalEntityId = EntityIdFactory.getByTypeAndUuid(entityType, externalEntityUuid);
return versionControlService.listEntityVersions(getTenantId(), branch, externalEntityId); return wrapFuture(versionControlService.listEntityVersions(getTenantId(), branch, externalEntityId));
} catch (Exception e) { } catch (Exception e) {
throw handleException(e); throw handleException(e);
} }
@ -127,10 +137,10 @@ public class EntitiesVersionControlController extends BaseController {
" }\n" + " }\n" +
"]\n```") "]\n```")
@GetMapping("/version/{branch}/{entityType}") @GetMapping("/version/{branch}/{entityType}")
public List<EntityVersion> listEntityTypeVersions(@PathVariable String branch, public DeferredResult<List<EntityVersion>> listEntityTypeVersions(@PathVariable String branch,
@PathVariable EntityType entityType) throws ThingsboardException { @PathVariable EntityType entityType) throws ThingsboardException {
try { try {
return versionControlService.listEntityTypeVersions(getTenantId(), branch, entityType); return wrapFuture(versionControlService.listEntityTypeVersions(getTenantId(), branch, entityType));
} catch (Exception e) { } catch (Exception e) {
throw handleException(e); throw handleException(e);
} }
@ -152,9 +162,9 @@ public class EntitiesVersionControlController extends BaseController {
" }\n" + " }\n" +
"]\n```") "]\n```")
@GetMapping("/version/{branch}") @GetMapping("/version/{branch}")
public List<EntityVersion> listVersions(@PathVariable String branch) throws ThingsboardException { public DeferredResult<List<EntityVersion>> listVersions(@PathVariable String branch) throws ThingsboardException {
try { try {
return versionControlService.listVersions(getTenantId(), branch); return wrapFuture(versionControlService.listVersions(getTenantId(), branch));
} catch (Exception e) { } catch (Exception e) {
throw handleException(e); throw handleException(e);
} }
@ -162,21 +172,21 @@ public class EntitiesVersionControlController extends BaseController {
@GetMapping("/entity/{branch}/{entityType}/{versionId}") @GetMapping("/entity/{branch}/{entityType}/{versionId}")
public List<VersionedEntityInfo> listEntitiesAtVersion(@PathVariable String branch, public DeferredResult<List<VersionedEntityInfo>> listEntitiesAtVersion(@PathVariable String branch,
@PathVariable EntityType entityType, @PathVariable EntityType entityType,
@PathVariable String versionId) throws ThingsboardException { @PathVariable String versionId) throws ThingsboardException {
try { try {
return versionControlService.listEntitiesAtVersion(getTenantId(), branch, versionId, entityType); return wrapFuture(versionControlService.listEntitiesAtVersion(getTenantId(), branch, versionId, entityType));
} catch (Exception e) { } catch (Exception e) {
throw handleException(e); throw handleException(e);
} }
} }
@GetMapping("/entity/{branch}/{versionId}") @GetMapping("/entity/{branch}/{versionId}")
public List<VersionedEntityInfo> listAllEntitiesAtVersion(@PathVariable String branch, public DeferredResult<List<VersionedEntityInfo>> listAllEntitiesAtVersion(@PathVariable String branch,
@PathVariable String versionId) throws ThingsboardException { @PathVariable String versionId) throws ThingsboardException {
try { try {
return versionControlService.listAllEntitiesAtVersion(getTenantId(), branch, versionId); return wrapFuture(versionControlService.listAllEntitiesAtVersion(getTenantId(), branch, versionId));
} catch (Exception e) { } catch (Exception e) {
throw handleException(e); throw handleException(e);
} }
@ -216,20 +226,10 @@ public class EntitiesVersionControlController extends BaseController {
" }\n" + " }\n" +
"}\n```") "}\n```")
@PostMapping("/entity") @PostMapping("/entity")
public List<VersionLoadResult> loadEntitiesVersion(@RequestBody VersionLoadRequest request) throws ThingsboardException { public DeferredResult<List<VersionLoadResult>> loadEntitiesVersion(@RequestBody VersionLoadRequest request) throws ThingsboardException {
SecurityUser user = getCurrentUser(); SecurityUser user = getCurrentUser();
try { try {
String versionId = request.getVersionId(); return wrapFuture(versionControlService.loadEntitiesVersion(user, request));
if (versionId == null) {
List<EntityVersion> versions = versionControlService.listVersions(user.getTenantId(), request.getBranch());
if (versions.size() > 0) {
versionId = versions.get(0).getId();
} else {
throw new IllegalArgumentException("No versions available in branch");
}
}
return versionControlService.loadEntitiesVersion(user, request);
} catch (Exception e) { } catch (Exception e) {
throw handleException(e); throw handleException(e);
} }
@ -252,19 +252,22 @@ public class EntitiesVersionControlController extends BaseController {
" }\n" + " }\n" +
"]\n\n```") "]\n\n```")
@GetMapping("/branches") @GetMapping("/branches")
public List<BranchInfo> listBranches() throws ThingsboardException { public DeferredResult<List<BranchInfo>> listBranches() throws ThingsboardException {
try { try {
List<String> remoteBranches = versionControlService.listBranches(getTenantId()); final TenantId tenantId = getTenantId();
List<BranchInfo> infos = new ArrayList<>(); ListenableFuture<List<String>> branches = versionControlService.listBranches(tenantId);
return wrapFuture(Futures.transform(branches, remoteBranches -> {
List<BranchInfo> infos = new ArrayList<>();
String defaultBranch = versionControlService.getVersionControlSettings(getTenantId()).getDefaultBranch(); String defaultBranch = versionControlService.getVersionControlSettings(tenantId).getDefaultBranch();
if (StringUtils.isNotEmpty(defaultBranch)) { if (StringUtils.isNotEmpty(defaultBranch)) {
remoteBranches.remove(defaultBranch); remoteBranches.remove(defaultBranch);
infos.add(new BranchInfo(defaultBranch, true)); infos.add(new BranchInfo(defaultBranch, true));
} }
remoteBranches.forEach(branch -> infos.add(new BranchInfo(branch, false))); remoteBranches.forEach(branch -> infos.add(new BranchInfo(branch, false)));
return infos; return infos;
}, MoreExecutors.directExecutor()));
} catch (Exception e) { } catch (Exception e) {
throw handleException(e); throw handleException(e);
} }

12
application/src/main/java/org/thingsboard/server/service/install/SqlDatabaseUpgradeService.java

@ -647,18 +647,6 @@ public class SqlDatabaseUpgradeService implements DatabaseEntitiesUpgradeService
log.error("Failed updating schema!!!", e); log.error("Failed updating schema!!!", e);
} }
break; break;
case "3.3.4":
try (Connection conn = DriverManager.getConnection(dbUrl, dbUserName, dbPassword)) {
log.info("Updating schema ...");
schemaUpdateFile = Paths.get(installScripts.getDataDir(), "upgrade", "3.3.4", SCHEMA_UPDATE_SQL);
loadSql(schemaUpdateFile, conn);
log.info("Updating schema settings...");
conn.createStatement().execute("UPDATE tb_schema_settings SET schema_version = 3004000;");
log.info("Schema updated.");
} catch (Exception e) {
log.error("Failed updating schema!!!", e);
}
break;
default: default:
throw new RuntimeException("Unable to upgrade SQL database, unsupported fromVersion: " + fromVersion); throw new RuntimeException("Unable to upgrade SQL database, unsupported fromVersion: " + fromVersion);
} }

8
application/src/main/java/org/thingsboard/server/service/queue/DefaultTbClusterService.java

@ -135,6 +135,14 @@ public class DefaultTbClusterService implements TbClusterService {
toCoreMsgs.incrementAndGet(); toCoreMsgs.incrementAndGet();
} }
@Override
public void pushMsgToVersionControl(TenantId tenantId, TransportProtos.ToVersionControlServiceMsg msg, TbQueueCallback callback) {
TopicPartitionInfo tpi = partitionService.resolve(ServiceType.TB_VC_EXECUTOR, tenantId, tenantId);
log.trace("PUSHING msg: {} to:{}", msg, tpi);
producerProvider.getTbVersionControlMsgProducer().send(tpi, new TbProtoQueueMsg<>(tenantId.getId(), msg), callback);
toCoreMsgs.incrementAndGet();
}
@Override @Override
public void pushNotificationToCore(String serviceId, FromDeviceRpcResponse response, TbQueueCallback callback) { public void pushNotificationToCore(String serviceId, FromDeviceRpcResponse response, TbQueueCallback callback) {
TopicPartitionInfo tpi = notificationsTopicService.getNotificationsTopic(ServiceType.TB_CORE, serviceId); TopicPartitionInfo tpi = notificationsTopicService.getNotificationsTopic(ServiceType.TB_CORE, serviceId);

33
application/src/main/java/org/thingsboard/server/service/sync/vc/CommitGitRequest.java

@ -0,0 +1,33 @@
/**
* Copyright © 2016-2022 The Thingsboard Authors
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.thingsboard.server.service.sync.vc;
import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.data.sync.vc.VersionCreationResult;
import org.thingsboard.server.common.data.sync.vc.request.create.VersionCreateRequest;
import java.util.UUID;
public class CommitGitRequest extends PendingGitRequest<VersionCreationResult> {
private final VersionCreateRequest request;
public CommitGitRequest(TenantId tenantId, VersionCreateRequest request) {
super(tenantId);
this.request = request;
}
}

215
application/src/main/java/org/thingsboard/server/service/sync/vc/DefaultEntitiesVersionControlService.java

@ -15,12 +15,18 @@
*/ */
package org.thingsboard.server.service.sync.vc; package org.thingsboard.server.service.sync.vc;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.ListeningExecutorService;
import com.google.common.util.concurrent.MoreExecutors;
import lombok.RequiredArgsConstructor; import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j; import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.ObjectUtils; import org.apache.commons.lang3.ObjectUtils;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service; import org.springframework.stereotype.Service;
import org.springframework.transaction.support.TransactionTemplate; import org.springframework.transaction.support.TransactionTemplate;
import org.thingsboard.common.util.JacksonUtil; import org.thingsboard.common.util.JacksonUtil;
import org.thingsboard.common.util.ThingsBoardExecutors;
import org.thingsboard.server.common.data.AdminSettings; import org.thingsboard.server.common.data.AdminSettings;
import org.thingsboard.server.common.data.EntityType; import org.thingsboard.server.common.data.EntityType;
import org.thingsboard.server.common.data.ExportableEntity; import org.thingsboard.server.common.data.ExportableEntity;
@ -33,7 +39,6 @@ import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.data.sync.vc.*; import org.thingsboard.server.common.data.sync.vc.*;
import org.thingsboard.server.common.data.sync.vc.request.load.EntityTypeVersionLoadConfig; import org.thingsboard.server.common.data.sync.vc.request.load.EntityTypeVersionLoadConfig;
import org.thingsboard.server.dao.DaoUtil; import org.thingsboard.server.dao.DaoUtil;
import org.thingsboard.server.dao.entity.EntityService;
import org.thingsboard.server.dao.settings.AdminSettingsService; import org.thingsboard.server.dao.settings.AdminSettingsService;
import org.thingsboard.server.queue.util.TbCoreComponent; import org.thingsboard.server.queue.util.TbCoreComponent;
import org.thingsboard.server.service.security.model.SecurityUser; import org.thingsboard.server.service.security.model.SecurityUser;
@ -56,6 +61,8 @@ import org.thingsboard.server.common.data.sync.vc.request.load.VersionLoadConfig
import org.thingsboard.server.common.data.sync.vc.request.load.VersionLoadRequest; import org.thingsboard.server.common.data.sync.vc.request.load.VersionLoadRequest;
import org.thingsboard.server.common.data.sync.ThrowingRunnable; import org.thingsboard.server.common.data.sync.ThrowingRunnable;
import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.util.ArrayList; import java.util.ArrayList;
import java.util.HashMap; import java.util.HashMap;
import java.util.HashSet; import java.util.HashSet;
@ -72,118 +79,137 @@ import java.util.stream.Collectors;
@Slf4j @Slf4j
public class DefaultEntitiesVersionControlService implements EntitiesVersionControlService { public class DefaultEntitiesVersionControlService implements EntitiesVersionControlService {
private final GitVersionControlService gitService; private final GitVersionControlQueueService gitServiceQueue;
private final EntitiesExportImportService exportImportService; private final EntitiesExportImportService exportImportService;
private final ExportableEntitiesService exportableEntitiesService; private final ExportableEntitiesService exportableEntitiesService;
private final AdminSettingsService adminSettingsService; private final AdminSettingsService adminSettingsService;
private final EntityService entityService;
private final TransactionTemplate transactionTemplate; private final TransactionTemplate transactionTemplate;
private ListeningExecutorService executor;
@Value("${vc.thread_pool_size:4}")
private int threadPoolSize;
@PostConstruct
public void init() {
executor = MoreExecutors.listeningDecorator(ThingsBoardExecutors.newWorkStealingPool(threadPoolSize, DefaultEntitiesVersionControlService.class));
}
@PreDestroy
public void shutdown() {
if (executor != null) {
executor.shutdownNow();
}
}
@SuppressWarnings("UnstableApiUsage")
@Override @Override
public VersionCreationResult saveEntitiesVersion(SecurityUser user, VersionCreateRequest request) throws Exception { public ListenableFuture<VersionCreationResult> saveEntitiesVersion(SecurityUser user, VersionCreateRequest request) throws Exception {
var commit = gitService.prepareCommit(user.getTenantId(), request); var pendingCommit = gitServiceQueue.prepareCommit(user.getTenantId(), request);
switch (request.getType()) { return Futures.transformAsync(pendingCommit, commit -> {
case SINGLE_ENTITY: { List<ListenableFuture<Void>> gitFutures = new ArrayList<>();
SingleEntityVersionCreateRequest versionCreateRequest = (SingleEntityVersionCreateRequest) request; switch (request.getType()) {
saveEntityData(user, commit, versionCreateRequest.getEntityId(), versionCreateRequest.getConfig()); case SINGLE_ENTITY: {
break; SingleEntityVersionCreateRequest versionCreateRequest = (SingleEntityVersionCreateRequest) request;
} gitFutures.add(saveEntityData(user, commit, versionCreateRequest.getEntityId(), versionCreateRequest.getConfig()));
case COMPLEX: { break;
ComplexVersionCreateRequest versionCreateRequest = (ComplexVersionCreateRequest) request; }
versionCreateRequest.getEntityTypes().forEach((entityType, config) -> { case COMPLEX: {
if (ObjectUtils.defaultIfNull(config.getSyncStrategy(), versionCreateRequest.getSyncStrategy()) == SyncStrategy.OVERWRITE) { ComplexVersionCreateRequest versionCreateRequest = (ComplexVersionCreateRequest) request;
gitService.deleteAll(commit, entityType); versionCreateRequest.getEntityTypes().forEach((entityType, config) -> {
} if (ObjectUtils.defaultIfNull(config.getSyncStrategy(), versionCreateRequest.getSyncStrategy()) == SyncStrategy.OVERWRITE) {
gitFutures.add(gitServiceQueue.deleteAll(commit, entityType));
}
if (config.isAllEntities()) { if (config.isAllEntities()) {
DaoUtil.processInBatches(pageLink -> { DaoUtil.processInBatches(pageLink -> exportableEntitiesService.findEntitiesByTenantId(user.getTenantId(), entityType, pageLink)
return exportableEntitiesService.findEntitiesByTenantId(user.getTenantId(), entityType, pageLink); , 100, entity -> {
}, 100, entity -> { try {
try { gitFutures.add(saveEntityData(user, commit, entity.getId(), config));
saveEntityData(user, commit, entity.getId(), config); } catch (Exception e) {
} catch (Exception e) { throw new RuntimeException(e);
throw new RuntimeException(e); }
} });
}); } else {
} else { for (UUID entityId : config.getEntityIds()) {
for (UUID entityId : config.getEntityIds()) { try {
try { gitFutures.add(saveEntityData(user, commit, EntityIdFactory.getByTypeAndUuid(entityType, entityId), config));
saveEntityData(user, commit, EntityIdFactory.getByTypeAndUuid(entityType, entityId), config); } catch (Exception e) {
} catch (Exception e) { throw new RuntimeException(e);
throw new RuntimeException(e); }
} }
} }
} });
break;
}); }
break;
} }
} return Futures.transformAsync(Futures.allAsList(gitFutures), success -> gitServiceQueue.push(commit), executor);
}, executor);
return gitService.push(commit);
} }
private void saveEntityData(SecurityUser user, PendingCommit commit, EntityId entityId, VersionCreateConfig config) throws Exception { private ListenableFuture<Void> saveEntityData(SecurityUser user, CommitGitRequest commit, EntityId entityId, VersionCreateConfig config) throws Exception {
EntityExportData<ExportableEntity<EntityId>> entityData = exportImportService.exportEntity(user, entityId, EntityExportSettings.builder() EntityExportData<ExportableEntity<EntityId>> entityData = exportImportService.exportEntity(user, entityId, EntityExportSettings.builder()
.exportRelations(config.isSaveRelations()) .exportRelations(config.isSaveRelations())
.build()); .build());
gitService.addToCommit(commit, entityData); return gitServiceQueue.addToCommit(commit, entityData);
} }
@Override @Override
public List<EntityVersion> listEntityVersions(TenantId tenantId, String branch, EntityId externalId) throws Exception { public ListenableFuture<List<EntityVersion>> listEntityVersions(TenantId tenantId, String branch, EntityId externalId) throws Exception {
return gitService.listVersions(tenantId, branch, externalId); return gitServiceQueue.listVersions(tenantId, branch, externalId);
} }
@Override @Override
public List<EntityVersion> listEntityTypeVersions(TenantId tenantId, String branch, EntityType entityType) throws Exception { public ListenableFuture<List<EntityVersion>> listEntityTypeVersions(TenantId tenantId, String branch, EntityType entityType) throws Exception {
return gitService.listVersions(tenantId, branch, entityType); return gitServiceQueue.listVersions(tenantId, branch, entityType);
} }
@Override @Override
public List<EntityVersion> listVersions(TenantId tenantId, String branch) throws Exception { public ListenableFuture<List<EntityVersion>> listVersions(TenantId tenantId, String branch) throws Exception {
return gitService.listVersions(tenantId, branch); return gitServiceQueue.listVersions(tenantId, branch);
} }
@Override @Override
public List<VersionedEntityInfo> listEntitiesAtVersion(TenantId tenantId, String branch, String versionId, EntityType entityType) throws Exception { public ListenableFuture<List<VersionedEntityInfo>> listEntitiesAtVersion(TenantId tenantId, String branch, String versionId, EntityType entityType) throws Exception {
return gitService.listEntitiesAtVersion(tenantId, branch, versionId, entityType); return gitServiceQueue.listEntitiesAtVersion(tenantId, branch, versionId, entityType);
} }
@Override @Override
public List<VersionedEntityInfo> listAllEntitiesAtVersion(TenantId tenantId, String branch, String versionId) throws Exception { public ListenableFuture<List<VersionedEntityInfo>> listAllEntitiesAtVersion(TenantId tenantId, String branch, String versionId) throws Exception {
return gitService.listEntitiesAtVersion(tenantId, branch, versionId); return gitServiceQueue.listEntitiesAtVersion(tenantId, branch, versionId);
} }
@SuppressWarnings({"UnstableApiUsage", "rawtypes"})
@Override @Override
public List<VersionLoadResult> loadEntitiesVersion(SecurityUser user, VersionLoadRequest request) throws Exception { public ListenableFuture<List<VersionLoadResult>> loadEntitiesVersion(SecurityUser user, VersionLoadRequest request) throws Exception {
switch (request.getType()) { switch (request.getType()) {
case SINGLE_ENTITY: { case SINGLE_ENTITY: {
SingleEntityVersionLoadRequest versionLoadRequest = (SingleEntityVersionLoadRequest) request; SingleEntityVersionLoadRequest versionLoadRequest = (SingleEntityVersionLoadRequest) request;
VersionLoadConfig config = versionLoadRequest.getConfig(); VersionLoadConfig config = versionLoadRequest.getConfig();
EntityImportResult<?> importResult = transactionTemplate.execute(status -> { ListenableFuture<EntityExportData> future = gitServiceQueue.getEntity(user.getTenantId(), request.getVersionId(), versionLoadRequest.getExternalEntityId());
try { Futures.transform(future, entityData -> {
EntityExportData entityData = gitService.getEntity(user.getTenantId(), request.getVersionId(), versionLoadRequest.getExternalEntityId()); EntityImportResult<?> importResult = transactionTemplate.execute(status -> {
return exportImportService.importEntity(user, entityData, EntityImportSettings.builder() try {
.updateRelations(config.isLoadRelations()) return exportImportService.importEntity(user, entityData, EntityImportSettings.builder()
.findExistingByName(config.isFindExistingEntityByName()) .updateRelations(config.isLoadRelations())
.build(), true, true); .findExistingByName(config.isFindExistingEntityByName())
} catch (Exception e) { .build(), true, true);
throw new RuntimeException(e); } catch (Exception e) {
} throw new RuntimeException(e);
}); }
return List.of(VersionLoadResult.builder() });
.entityType(importResult.getEntityType()) return List.of(VersionLoadResult.builder()
.created(importResult.getOldEntity() == null ? 1 : 0) .entityType(importResult.getEntityType())
.updated(importResult.getOldEntity() != null ? 1 : 0) .created(importResult.getOldEntity() == null ? 1 : 0)
.deleted(0) .updated(importResult.getOldEntity() != null ? 1 : 0)
.build()); .deleted(0)
.build());
}, executor);
} }
case ENTITY_TYPE: { case ENTITY_TYPE: {
EntityTypeVersionLoadRequest versionLoadRequest = (EntityTypeVersionLoadRequest) request; EntityTypeVersionLoadRequest versionLoadRequest = (EntityTypeVersionLoadRequest) request;
return transactionTemplate.execute(status -> { return executor.submit(() -> transactionTemplate.execute(status -> {
Map<EntityType, VersionLoadResult> results = new HashMap<>(); Map<EntityType, VersionLoadResult> results = new HashMap<>();
Map<EntityType, Set<EntityId>> importedEntities = new HashMap<>(); Map<EntityType, Set<EntityId>> importedEntities = new HashMap<>();
List<ThrowingRunnable> saveReferencesCallbacks = new ArrayList<>(); List<ThrowingRunnable> saveReferencesCallbacks = new ArrayList<>();
@ -199,9 +225,9 @@ public class DefaultEntitiesVersionControlService implements EntitiesVersionCont
try { try {
int limit = 100; int limit = 100;
int offset = 0; int offset = 0;
List<EntityExportData<?>> entityDataList; List<EntityExportData> entityDataList;
do { do {
entityDataList = gitService.getEntities(user.getTenantId(), request.getBranch(), request.getVersionId(), entityType, offset, limit); entityDataList = gitServiceQueue.getEntities(user.getTenantId(), request.getVersionId(), entityType, offset, limit).get();
for (EntityExportData entityData : entityDataList) { for (EntityExportData entityData : entityDataList) {
EntityImportResult<?> importResult = exportImportService.importEntity(user, entityData, EntityImportSettings.builder() EntityImportResult<?> importResult = exportImportService.importEntity(user, entityData, EntityImportSettings.builder()
.updateRelations(config.isLoadRelations()) .updateRelations(config.isLoadRelations())
@ -215,7 +241,7 @@ public class DefaultEntitiesVersionControlService implements EntitiesVersionCont
} }
offset += limit; offset += limit;
importedEntities.computeIfAbsent(entityType, t -> new HashSet<>()) importedEntities.computeIfAbsent(entityType, t -> new HashSet<>())
.addAll(entityDataList.stream().map(entityData -> entityData.getEntity().getId()).collect(Collectors.toSet())); .addAll(entityDataList.stream().map(entityData -> entityData.getEntity().getExternalId()).collect(Collectors.toSet()));
} while (entityDataList.size() == limit); } while (entityDataList.size() == limit);
} catch (Exception e) { } catch (Exception e) {
throw new RuntimeException(e); throw new RuntimeException(e);
@ -263,7 +289,7 @@ public class DefaultEntitiesVersionControlService implements EntitiesVersionCont
} }
} }
return new ArrayList<>(results.values()); return new ArrayList<>(results.values());
}); }));
} }
default: default:
throw new IllegalArgumentException("Unsupported version load request"); throw new IllegalArgumentException("Unsupported version load request");
@ -272,8 +298,8 @@ public class DefaultEntitiesVersionControlService implements EntitiesVersionCont
@Override @Override
public List<String> listBranches(TenantId tenantId) throws Exception { public ListenableFuture<List<String>> listBranches(TenantId tenantId) throws Exception {
return gitService.listBranches(tenantId); return gitServiceQueue.listBranches(tenantId);
} }
@Override @Override
@ -304,32 +330,35 @@ public class DefaultEntitiesVersionControlService implements EntitiesVersionCont
} catch (Exception e) { } catch (Exception e) {
throw new RuntimeException("Failed to load version control settings!", e); throw new RuntimeException("Failed to load version control settings!", e);
} }
try { //TODO: ashvayka
gitService.clearRepository(tenantId); // try {
gitService.initRepository(tenantId, savedVersionControlSettings); // gitService.clearRepository(tenantId);
} catch (Exception e) { // gitService.initRepository(tenantId, savedVersionControlSettings);
throw new RuntimeException("Failed to init repository!", e); // } catch (Exception e) {
} // throw new RuntimeException("Failed to init repository!", e);
// }
return savedVersionControlSettings; return savedVersionControlSettings;
} }
@Override @Override
public void deleteVersionControlSettings(TenantId tenantId) { public void deleteVersionControlSettings(TenantId tenantId) {
if (adminSettingsService.deleteAdminSettings(tenantId, SETTINGS_KEY)) { //TODO: ashvayka
gitService.clearRepository(tenantId); // if (adminSettingsService.deleteAdminSettings(tenantId, SETTINGS_KEY)) {
} // gitService.clearRepository(tenantId);
// }
} }
@Override @Override
public void checkVersionControlAccess(TenantId tenantId, EntitiesVersionControlSettings settings) throws ThingsboardException { public void checkVersionControlAccess(TenantId tenantId, EntitiesVersionControlSettings settings) throws ThingsboardException {
EntitiesVersionControlSettings storedSettings = getVersionControlSettings(tenantId); EntitiesVersionControlSettings storedSettings = getVersionControlSettings(tenantId);
settings = this.restoreCredentials(settings, storedSettings); settings = this.restoreCredentials(settings, storedSettings);
try { //TODO: ashvayka
gitService.testRepository(tenantId, settings); // try {
} catch (Exception e) { // gitService.testRepository(tenantId, settings);
throw new ThingsboardException(String.format("Unable to access repository: %s", e.getMessage()), // } catch (Exception e) {
ThingsboardErrorCode.GENERAL); // throw new ThingsboardException(String.format("Unable to access repository: %s", e.getMessage()),
} // ThingsboardErrorCode.GENERAL);
// }
} }
private EntitiesVersionControlSettings restoreCredentials(EntitiesVersionControlSettings settings, EntitiesVersionControlSettings storedSettings) { private EntitiesVersionControlSettings restoreCredentials(EntitiesVersionControlSettings settings, EntitiesVersionControlSettings storedSettings) {

287
application/src/main/java/org/thingsboard/server/service/sync/vc/DefaultGitVersionControlQueueService.java

@ -0,0 +1,287 @@
/**
* Copyright © 2016-2022 The Thingsboard Authors
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.thingsboard.server.service.sync.vc;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.ObjectWriter;
import com.fasterxml.jackson.databind.SerializationFeature;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.SettableFuture;
import lombok.RequiredArgsConstructor;
import org.springframework.stereotype.Service;
import org.thingsboard.common.util.JacksonUtil;
import org.thingsboard.server.cluster.TbClusterService;
import org.thingsboard.server.common.data.EntityType;
import org.thingsboard.server.common.data.ExportableEntity;
import org.thingsboard.server.common.data.id.EntityId;
import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.data.sync.ie.EntityExportData;
import org.thingsboard.server.common.data.sync.vc.EntityVersion;
import org.thingsboard.server.common.data.sync.vc.VersionCreationResult;
import org.thingsboard.server.common.data.sync.vc.VersionedEntityInfo;
import org.thingsboard.server.common.data.sync.vc.request.create.VersionCreateRequest;
import org.thingsboard.server.gen.transport.TransportProtos;
import org.thingsboard.server.gen.transport.TransportProtos.CommitRequestMsg;
import org.thingsboard.server.gen.transport.TransportProtos.ListEntitiesRequestMsg;
import org.thingsboard.server.gen.transport.TransportProtos.ListVersionsRequestMsg;
import org.thingsboard.server.gen.transport.TransportProtos.EntityContentRequestMsg;
import org.thingsboard.server.gen.transport.TransportProtos.EntitiesContentRequestMsg;
import org.thingsboard.server.gen.transport.TransportProtos.PrepareMsg;
import org.thingsboard.server.gen.transport.TransportProtos.ToVersionControlServiceMsg;
import org.thingsboard.server.queue.TbQueueCallback;
import org.thingsboard.server.queue.TbQueueMsgMetadata;
import org.thingsboard.server.queue.discovery.TbServiceInfoProvider;
import org.thingsboard.server.queue.util.TbCoreComponent;
import java.io.IOException;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.function.Function;
@TbCoreComponent
@Service
@RequiredArgsConstructor
public class DefaultGitVersionControlQueueService implements GitVersionControlQueueService {
private final ObjectWriter jsonWriter = new ObjectMapper().writer(SerializationFeature.INDENT_OUTPUT);
private final TbServiceInfoProvider serviceInfoProvider;
private final TbClusterService clusterService;
private final Map<UUID, PendingGitRequest<?>> pendingRequestMap = new HashMap<>();
@Override
public ListenableFuture<CommitGitRequest> prepareCommit(TenantId tenantId, VersionCreateRequest request) {
SettableFuture<CommitGitRequest> future = SettableFuture.create();
CommitGitRequest commit = new CommitGitRequest(tenantId, request);
registerAndSend(commit, builder -> builder.setCommitRequest(
CommitRequestMsg.newBuilder().setPrepareMsg(getCommitPrepareMsg(request)).build()
).build(), wrap(future, commit));
return future;
}
@Override
public ListenableFuture<Void> addToCommit(CommitGitRequest commit, EntityExportData<ExportableEntity<EntityId>> entityData) {
SettableFuture<Void> future = SettableFuture.create();
String path = getRelativePath(entityData.getEntityType(), entityData.getEntity().getId());
String entityDataJson;
try {
entityDataJson = jsonWriter.writeValueAsString(entityData);
} catch (IOException e) {
//TODO: analyze and return meaningful exceptions that we can show to the client;
throw new RuntimeException(e);
}
registerAndSend(commit, builder -> builder.setCommitRequest(
CommitRequestMsg.newBuilder().setAddMsg(
TransportProtos.AddMsg.newBuilder()
.setRelativePath(path).setEntityDataJson(entityDataJson).build()
).build()
).build(), wrap(commit.getFuture(), null));
return future;
}
@Override
public ListenableFuture<Void> deleteAll(CommitGitRequest commit, EntityType entityType) {
SettableFuture<Void> future = SettableFuture.create();
String path = getRelativePath(entityType, null);
registerAndSend(commit, builder -> builder.setCommitRequest(
CommitRequestMsg.newBuilder().setDeleteMsg(
TransportProtos.DeleteMsg.newBuilder().setRelativePath(path).build()
).build()
).build(), wrap(commit.getFuture(), null));
return future;
}
@Override
public ListenableFuture<VersionCreationResult> push(CommitGitRequest commit) {
registerAndSend(commit, builder -> builder.setCommitRequest(
CommitRequestMsg.newBuilder().setPushMsg(
TransportProtos.PushMsg.newBuilder().build()
).build()
).build(), wrap(commit.getFuture()));
return commit.getFuture();
}
@Override
public ListenableFuture<List<EntityVersion>> listVersions(TenantId tenantId, String branch) {
return listVersions(tenantId, ListVersionsRequestMsg.newBuilder()
.setBranchName(branch).build());
}
@Override
public ListenableFuture<List<EntityVersion>> listVersions(TenantId tenantId, String branch, EntityType entityType) {
return listVersions(tenantId, ListVersionsRequestMsg.newBuilder()
.setBranchName(branch).setEntityType(entityType.name())
.build());
}
@Override
public ListenableFuture<List<EntityVersion>> listVersions(TenantId tenantId, String branch, EntityId entityId) {
return listVersions(tenantId, ListVersionsRequestMsg.newBuilder()
.setBranchName(branch)
.setEntityType(entityId.getEntityType().name())
.setEntityIdMSB(entityId.getId().getMostSignificantBits())
.setEntityIdLSB(entityId.getId().getLeastSignificantBits())
.build());
}
private ListenableFuture<List<EntityVersion>> listVersions(TenantId tenantId, ListVersionsRequestMsg requestMsg) {
ListVersionsGitRequest request = new ListVersionsGitRequest(tenantId);
registerAndSend(request, builder -> builder.setListVersionRequest(requestMsg).build(), wrap(request.getFuture()));
return request.getFuture();
}
@Override
public ListenableFuture<List<VersionedEntityInfo>> listEntitiesAtVersion(TenantId tenantId, String branch, String versionId, EntityType entityType) {
return listEntitiesAtVersion(tenantId, ListEntitiesRequestMsg.newBuilder()
.setBranchName(branch)
.setVersionId(versionId)
.setEntityType(entityType.name())
.build());
}
@Override
public ListenableFuture<List<VersionedEntityInfo>> listEntitiesAtVersion(TenantId tenantId, String branch, String versionId) {
return listEntitiesAtVersion(tenantId, ListEntitiesRequestMsg.newBuilder()
.setBranchName(branch)
.setVersionId(versionId)
.build());
}
private ListenableFuture<List<VersionedEntityInfo>> listEntitiesAtVersion(TenantId tenantId, TransportProtos.ListEntitiesRequestMsg requestMsg) {
ListEntitiesGitRequest request = new ListEntitiesGitRequest(tenantId);
registerAndSend(request, builder -> builder.setListEntitiesRequest(requestMsg).build(), wrap(request.getFuture()));
return request.getFuture();
}
@Override
public ListenableFuture<List<String>> listBranches(TenantId tenantId) {
ListBranchesGitRequest request = new ListBranchesGitRequest(tenantId);
registerAndSend(request, builder -> builder.setListBranchesRequest(TransportProtos.ListBranchesRequestMsg.newBuilder().build()).build(), wrap(request.getFuture()));
return request.getFuture();
}
private <T> void registerAndSend(PendingGitRequest<T> request, Function<ToVersionControlServiceMsg.Builder, ToVersionControlServiceMsg> enrichFunction, TbQueueCallback callback) {
if (!request.getFuture().isDone()) {
pendingRequestMap.putIfAbsent(request.getRequestId(), request);
clusterService.pushMsgToVersionControl(request.getTenantId(), enrichFunction.apply(newRequestProto(request)), callback);
} else {
throw new RuntimeException("Future is already done!");
}
}
@Override
public ListenableFuture<EntityExportData> getEntity(TenantId tenantId, String versionId, EntityId entityId) {
EntityContentGitRequest request = new EntityContentGitRequest(tenantId, versionId, entityId);
registerAndSend(request, builder -> builder.setEntityContentRequest(EntityContentRequestMsg.newBuilder()
.setVersionId(versionId)
.setEntityType(entityId.getEntityType().name())
.setEntityIdMSB(entityId.getId().getMostSignificantBits())
.setEntityIdLSB(entityId.getId().getLeastSignificantBits())).build()
, wrap(request.getFuture()));
return request.getFuture();
// try {
// String entityDataJson = gitRepositoryService.getFileContentAtCommit(tenantId,
// getRelativePath(entityId.getEntityType(), entityId.getId().toString()), versionId);
// return JacksonUtil.fromString(entityDataJson, EntityExportData.class);
// } catch (Exception e) {
// //TODO: analyze and return meaningful exceptions that we can show to the client;
// throw new RuntimeException(e);
// }
}
@Override
public ListenableFuture<List<EntityExportData>> getEntities(TenantId tenantId, String versionId, EntityType entityType, int offset, int limit) {
EntitiesContentGitRequest request = new EntitiesContentGitRequest(tenantId, versionId, entityType);
registerAndSend(request, builder -> builder.setEntitiesContentRequest(EntitiesContentRequestMsg.newBuilder()
.setVersionId(versionId)
.setEntityType(entityType.name())
.setOffset(offset)
.setLimit(limit)
).build()
, wrap(request.getFuture()));
return request.getFuture();
}
private static <T> TbQueueCallback wrap(SettableFuture<T> future) {
return new TbQueueCallback() {
@Override
public void onSuccess(TbQueueMsgMetadata metadata) {
}
@Override
public void onFailure(Throwable t) {
future.setException(t);
}
};
}
private static <T> TbQueueCallback wrap(SettableFuture<T> future, T value) {
return new TbQueueCallback() {
@Override
public void onSuccess(TbQueueMsgMetadata metadata) {
future.set(value);
}
@Override
public void onFailure(Throwable t) {
future.setException(t);
}
};
}
private static String getRelativePath(EntityType entityType, EntityId entityId) {
String path = entityType.name().toLowerCase();
if (entityId != null) {
path += "/" + entityId + ".json";
}
return path;
}
private static PrepareMsg getCommitPrepareMsg(VersionCreateRequest request) {
return PrepareMsg.newBuilder().setCommitMsg(request.getVersionName()).setBranchName(request.getBranch()).build();
}
private ToVersionControlServiceMsg.Builder newRequestProto(PendingGitRequest<?> request) {
var tenantId = request.getTenantId();
var requestId = request.getRequestId();
return ToVersionControlServiceMsg.newBuilder()
.setNodeId(serviceInfoProvider.getServiceId())
.setTenantIdMSB(tenantId.getId().getMostSignificantBits())
.setTenantIdLSB(tenantId.getId().getLeastSignificantBits())
.setRequestIdMSB(requestId.getMostSignificantBits())
.setRequestIdLSB(requestId.getLeastSignificantBits());
}
}

37
application/src/main/java/org/thingsboard/server/service/sync/vc/EntitiesContentGitRequest.java

@ -0,0 +1,37 @@
/**
* Copyright © 2016-2022 The Thingsboard Authors
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.thingsboard.server.service.sync.vc;
import lombok.Getter;
import org.thingsboard.server.common.data.EntityType;
import org.thingsboard.server.common.data.id.EntityId;
import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.data.sync.ie.EntityExportData;
import java.util.List;
@Getter
public class EntitiesContentGitRequest extends PendingGitRequest<List<EntityExportData>> {
private final String versionId;
private final EntityType entityType;
public EntitiesContentGitRequest(TenantId tenantId, String versionId, EntityType entityType) {
super(tenantId);
this.versionId = versionId;
this.entityType = entityType;
}
}

19
application/src/main/java/org/thingsboard/server/service/sync/vc/EntitiesVersionControlService.java

@ -15,6 +15,7 @@
*/ */
package org.thingsboard.server.service.sync.vc; package org.thingsboard.server.service.sync.vc;
import com.google.common.util.concurrent.ListenableFuture;
import org.thingsboard.server.common.data.EntityType; import org.thingsboard.server.common.data.EntityType;
import org.thingsboard.server.common.data.exception.ThingsboardException; import org.thingsboard.server.common.data.exception.ThingsboardException;
import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.common.data.id.EntityId;
@ -34,25 +35,23 @@ public interface EntitiesVersionControlService {
String SETTINGS_KEY = "entitiesVersionControl"; String SETTINGS_KEY = "entitiesVersionControl";
VersionCreationResult saveEntitiesVersion(SecurityUser user, VersionCreateRequest request) throws Exception; ListenableFuture<VersionCreationResult> saveEntitiesVersion(SecurityUser user, VersionCreateRequest request) throws Exception;
ListenableFuture<List<EntityVersion>> listEntityVersions(TenantId tenantId, String branch, EntityId externalId) throws Exception;
List<EntityVersion> listEntityVersions(TenantId tenantId, String branch, EntityId externalId) throws Exception; ListenableFuture<List<EntityVersion>> listEntityTypeVersions(TenantId tenantId, String branch, EntityType entityType) throws Exception;
List<EntityVersion> listEntityTypeVersions(TenantId tenantId, String branch, EntityType entityType) throws Exception; ListenableFuture<List<EntityVersion>> listVersions(TenantId tenantId, String branch) throws Exception;
List<EntityVersion> listVersions(TenantId tenantId, String branch) throws Exception; ListenableFuture<List<VersionedEntityInfo>> listEntitiesAtVersion(TenantId tenantId, String branch, String versionId, EntityType entityType) throws Exception;
ListenableFuture<List<VersionedEntityInfo>> listAllEntitiesAtVersion(TenantId tenantId, String branch, String versionId) throws Exception;
List<VersionedEntityInfo> listEntitiesAtVersion(TenantId tenantId, String branch, String versionId, EntityType entityType) throws Exception;
List<VersionedEntityInfo> listAllEntitiesAtVersion(TenantId tenantId, String branch, String versionId) throws Exception; ListenableFuture<List<VersionLoadResult>> loadEntitiesVersion(SecurityUser user, VersionLoadRequest request) throws Exception;
List<VersionLoadResult> loadEntitiesVersion(SecurityUser user, VersionLoadRequest request) throws Exception; ListenableFuture<List<String>> listBranches(TenantId tenantId) throws Exception;
List<String> listBranches(TenantId tenantId) throws Exception;
EntitiesVersionControlSettings getVersionControlSettings(TenantId tenantId); EntitiesVersionControlSettings getVersionControlSettings(TenantId tenantId);

34
application/src/main/java/org/thingsboard/server/service/sync/vc/EntityContentGitRequest.java

@ -0,0 +1,34 @@
/**
* Copyright © 2016-2022 The Thingsboard Authors
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.thingsboard.server.service.sync.vc;
import lombok.Getter;
import org.thingsboard.server.common.data.id.EntityId;
import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.data.sync.ie.EntityExportData;
@Getter
public class EntityContentGitRequest extends PendingGitRequest<EntityExportData> {
private final String versionId;
private final EntityId entityId;
public EntityContentGitRequest(TenantId tenantId, String versionId, EntityId entityId) {
super(tenantId);
this.versionId = versionId;
this.entityId = entityId;
}
}

57
application/src/main/java/org/thingsboard/server/service/sync/vc/GitVersionControlQueueService.java

@ -0,0 +1,57 @@
/**
* Copyright © 2016-2022 The Thingsboard Authors
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.thingsboard.server.service.sync.vc;
import com.google.common.util.concurrent.ListenableFuture;
import org.thingsboard.server.common.data.EntityType;
import org.thingsboard.server.common.data.ExportableEntity;
import org.thingsboard.server.common.data.id.EntityId;
import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.data.sync.ie.EntityExportData;
import org.thingsboard.server.common.data.sync.vc.EntityVersion;
import org.thingsboard.server.common.data.sync.vc.VersionCreationResult;
import org.thingsboard.server.common.data.sync.vc.VersionedEntityInfo;
import org.thingsboard.server.common.data.sync.vc.request.create.VersionCreateRequest;
import java.util.List;
public interface GitVersionControlQueueService {
ListenableFuture<CommitGitRequest> prepareCommit(TenantId tenantId, VersionCreateRequest request);
ListenableFuture<Void> addToCommit(CommitGitRequest commit, EntityExportData<ExportableEntity<EntityId>> entityData);
ListenableFuture<Void> deleteAll(CommitGitRequest pendingCommit, EntityType entityType);
ListenableFuture<VersionCreationResult> push(CommitGitRequest commit);
ListenableFuture<List<EntityVersion>> listVersions(TenantId tenantId, String branch);
ListenableFuture<List<EntityVersion>> listVersions(TenantId tenantId, String branch, EntityType entityType);
ListenableFuture<List<EntityVersion>> listVersions(TenantId tenantId, String branch, EntityId entityId);
ListenableFuture<List<VersionedEntityInfo>> listEntitiesAtVersion(TenantId tenantId, String branch, String versionId, EntityType entityType);
ListenableFuture<List<VersionedEntityInfo>> listEntitiesAtVersion(TenantId tenantId, String branch, String versionId);
ListenableFuture<List<String>> listBranches(TenantId tenantId);
ListenableFuture<EntityExportData> getEntity(TenantId tenantId, String versionId, EntityId entityId);
ListenableFuture<List<EntityExportData>> getEntities(TenantId tenantId, String versionId, EntityType entityType, int offset, int limit);
}

29
application/src/main/java/org/thingsboard/server/service/sync/vc/ListBranchesGitRequest.java

@ -0,0 +1,29 @@
/**
* Copyright © 2016-2022 The Thingsboard Authors
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.thingsboard.server.service.sync.vc;
import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.data.sync.vc.VersionedEntityInfo;
import java.util.List;
public class ListBranchesGitRequest extends PendingGitRequest<List<String>> {
public ListBranchesGitRequest(TenantId tenantId) {
super(tenantId);
}
}

30
application/src/main/java/org/thingsboard/server/service/sync/vc/ListEntitiesGitRequest.java

@ -0,0 +1,30 @@
/**
* Copyright © 2016-2022 The Thingsboard Authors
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.thingsboard.server.service.sync.vc;
import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.data.sync.vc.EntityVersion;
import org.thingsboard.server.common.data.sync.vc.VersionedEntityInfo;
import java.util.List;
public class ListEntitiesGitRequest extends PendingGitRequest<List<VersionedEntityInfo>> {
public ListEntitiesGitRequest(TenantId tenantId) {
super(tenantId);
}
}

31
application/src/main/java/org/thingsboard/server/service/sync/vc/ListVersionsGitRequest.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.service.sync.vc;
import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.data.sync.vc.EntityVersion;
import org.thingsboard.server.common.data.sync.vc.VersionCreationResult;
import org.thingsboard.server.common.data.sync.vc.request.create.VersionCreateRequest;
import java.util.List;
public class ListVersionsGitRequest extends PendingGitRequest<List<EntityVersion>> {
public ListVersionsGitRequest(TenantId tenantId) {
super(tenantId);
}
}

452
application/src/main/java/org/thingsboard/server/service/sync/vc/LocalGitVersionControlService.java

@ -19,18 +19,15 @@ import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.ObjectWriter; import com.fasterxml.jackson.databind.ObjectWriter;
import com.fasterxml.jackson.databind.SerializationFeature; import com.fasterxml.jackson.databind.SerializationFeature;
import lombok.RequiredArgsConstructor; import lombok.RequiredArgsConstructor;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j; import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.stereotype.Service; import org.springframework.stereotype.Service;
import org.thingsboard.common.util.JacksonUtil; import org.thingsboard.common.util.JacksonUtil;
import org.thingsboard.server.common.data.AdminSettings; import org.thingsboard.server.common.data.AdminSettings;
import org.thingsboard.server.common.data.DataConstants;
import org.thingsboard.server.common.data.EntityType; import org.thingsboard.server.common.data.EntityType;
import org.thingsboard.server.common.data.ExportableEntity; import org.thingsboard.server.common.data.ExportableEntity;
import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.common.data.id.EntityId;
import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.data.kv.KvEntry;
import org.thingsboard.server.common.data.sync.ie.EntityExportData; import org.thingsboard.server.common.data.sync.ie.EntityExportData;
import org.thingsboard.server.common.data.sync.vc.EntitiesVersionControlSettings; import org.thingsboard.server.common.data.sync.vc.EntitiesVersionControlSettings;
import org.thingsboard.server.common.data.sync.vc.EntityVersion; import org.thingsboard.server.common.data.sync.vc.EntityVersion;
@ -38,7 +35,6 @@ import org.thingsboard.server.common.data.sync.vc.VersionCreationResult;
import org.thingsboard.server.common.data.sync.vc.VersionedEntityInfo; import org.thingsboard.server.common.data.sync.vc.VersionedEntityInfo;
import org.thingsboard.server.common.data.sync.vc.request.create.VersionCreateRequest; import org.thingsboard.server.common.data.sync.vc.request.create.VersionCreateRequest;
import org.thingsboard.server.dao.DaoUtil; import org.thingsboard.server.dao.DaoUtil;
import org.thingsboard.server.dao.attributes.AttributesService;
import org.thingsboard.server.dao.settings.AdminSettingsService; import org.thingsboard.server.dao.settings.AdminSettingsService;
import org.thingsboard.server.dao.tenant.TenantDao; import org.thingsboard.server.dao.tenant.TenantDao;
import org.thingsboard.server.queue.util.AfterStartUp; import org.thingsboard.server.queue.util.AfterStartUp;
@ -60,235 +56,235 @@ import java.util.stream.Collectors;
@RequiredArgsConstructor @RequiredArgsConstructor
@Service @Service
@ConditionalOnProperty(prefix = "vc", value = "git.service", havingValue = "local", matchIfMissing = true) @ConditionalOnProperty(prefix = "vc", value = "git.service", havingValue = "local", matchIfMissing = true)
public class LocalGitVersionControlService implements GitVersionControlService { public class LocalGitVersionControlService {
private final ObjectWriter jsonWriter = new ObjectMapper().writer(SerializationFeature.INDENT_OUTPUT); private final ObjectWriter jsonWriter = new ObjectMapper().writer(SerializationFeature.INDENT_OUTPUT);
private final GitRepositoryService gitRepositoryService; private final GitRepositoryService gitRepositoryService;
private final TenantDao tenantDao; private final TenantDao tenantDao;
private final AdminSettingsService adminSettingsService; private final AdminSettingsService adminSettingsService;
private final ConcurrentMap<TenantId, Lock> tenantRepoLocks = new ConcurrentHashMap<>(); private final ConcurrentMap<TenantId, Lock> tenantRepoLocks = new ConcurrentHashMap<>();
private final Map<TenantId, PendingCommit> pendingCommitMap = new HashMap<>(); // private final Map<TenantId, PendingCommit> pendingCommitMap = new HashMap<>();
//
@AfterStartUp // @AfterStartUp
public void init() { // public void init() {
DaoUtil.processInBatches(tenantDao::findTenantsIds, 100, tenantId -> { // DaoUtil.processInBatches(tenantDao::findTenantsIds, 100, tenantId -> {
EntitiesVersionControlSettings settings = getVersionControlSettings(tenantId); // EntitiesVersionControlSettings settings = getVersionControlSettings(tenantId);
if (settings != null) { // if (settings != null) {
try { // try {
gitRepositoryService.initRepository(tenantId, settings); // gitRepositoryService.initRepository(tenantId, settings);
} catch (Exception e) { // } catch (Exception e) {
log.warn("Failed to init repository for tenant {}", tenantId, e); // log.warn("Failed to init repository for tenant {}", tenantId, e);
} // }
} // }
}); // });
} // }
//
@Override // @Override
public void testRepository(TenantId tenantId, EntitiesVersionControlSettings settings) { // public void testRepository(TenantId tenantId, EntitiesVersionControlSettings settings) {
var lock = getRepoLock(tenantId); // var lock = getRepoLock(tenantId);
lock.lock(); // lock.lock();
try { // try {
gitRepositoryService.testRepository(tenantId, settings); // gitRepositoryService.testRepository(tenantId, settings);
} catch (Exception e) { // } catch (Exception e) {
//TODO: analyze and return meaningful exceptions that we can show to the client; // //TODO: analyze and return meaningful exceptions that we can show to the client;
throw new RuntimeException(e); // throw new RuntimeException(e);
} finally { // } finally {
lock.unlock(); // lock.unlock();
} // }
} // }
//
@Override // @Override
public void initRepository(TenantId tenantId, EntitiesVersionControlSettings settings) { // public void initRepository(TenantId tenantId, EntitiesVersionControlSettings settings) {
var lock = getRepoLock(tenantId); // var lock = getRepoLock(tenantId);
lock.lock(); // lock.lock();
try { // try {
gitRepositoryService.initRepository(tenantId, settings); // gitRepositoryService.initRepository(tenantId, settings);
} catch (Exception e) { // } catch (Exception e) {
//TODO: analyze and return meaningful exceptions that we can show to the client; // //TODO: analyze and return meaningful exceptions that we can show to the client;
throw new RuntimeException(e); // throw new RuntimeException(e);
} finally { // } finally {
lock.unlock(); // lock.unlock();
} // }
} // }
//
@Override // @Override
public void clearRepository(TenantId tenantId) { // public void clearRepository(TenantId tenantId) {
var lock = getRepoLock(tenantId); // var lock = getRepoLock(tenantId);
lock.lock(); // lock.lock();
try { // try {
gitRepositoryService.clearRepository(tenantId); // gitRepositoryService.clearRepository(tenantId);
} catch (Exception e) { // } catch (Exception e) {
//TODO: analyze and return meaningful exceptions that we can show to the client; // //TODO: analyze and return meaningful exceptions that we can show to the client;
throw new RuntimeException(e); // throw new RuntimeException(e);
} finally { // } finally {
lock.unlock(); // lock.unlock();
} // }
} // }
//
@Override // @Override
public PendingCommit prepareCommit(TenantId tenantId, VersionCreateRequest request) { // public PendingCommit prepareCommit(TenantId tenantId, VersionCreateRequest request) {
var lock = getRepoLock(tenantId); // var lock = getRepoLock(tenantId);
lock.lock(); // lock.lock();
try { // try {
var pendingCommit = new PendingCommit(tenantId, request); // var pendingCommit = new PendingCommit(tenantId, request);
PendingCommit old = pendingCommitMap.put(tenantId, pendingCommit); // PendingCommit old = pendingCommitMap.put(tenantId, pendingCommit);
if (old != null) { // if (old != null) {
gitRepositoryService.abort(old); // gitRepositoryService.abort(old);
} // }
gitRepositoryService.prepareCommit(pendingCommit); // gitRepositoryService.prepareCommit(pendingCommit);
return pendingCommit; // return pendingCommit;
} finally { // } finally {
lock.unlock(); // lock.unlock();
} // }
} // }
//
@Override // @Override
public void deleteAll(PendingCommit commit, EntityType entityType) { // public void deleteAll(PendingCommit commit, EntityType entityType) {
doInsideLock(commit, c -> { // doInsideLock(commit, c -> {
try { // try {
gitRepositoryService.deleteFolderContent(commit, getRelativePath(entityType, null)); // gitRepositoryService.deleteFolderContent(commit, getRelativePath(entityType, null));
} catch (IOException e) { // } catch (IOException e) {
//TODO: analyze and return meaningful exceptions that we can show to the client; // //TODO: analyze and return meaningful exceptions that we can show to the client;
throw new RuntimeException(e); // throw new RuntimeException(e);
} // }
}); // });
} // }
//
@Override // @Override
public void addToCommit(PendingCommit commit, EntityExportData<ExportableEntity<EntityId>> entityData) { // public void addToCommit(PendingCommit commit, EntityExportData<ExportableEntity<EntityId>> entityData) {
doInsideLock(commit, c -> { // doInsideLock(commit, c -> {
String entityDataJson; // String entityDataJson;
try { // try {
entityDataJson = jsonWriter.writeValueAsString(entityData); // entityDataJson = jsonWriter.writeValueAsString(entityData);
gitRepositoryService.add(c, getRelativePath(entityData.getEntityType(), // gitRepositoryService.add(c, getRelativePath(entityData.getEntityType(),
entityData.getEntity().getId().toString()), entityDataJson); // entityData.getEntity().getId().toString()), entityDataJson);
} catch (IOException e) { // } catch (IOException e) {
//TODO: analyze and return meaningful exceptions that we can show to the client; // //TODO: analyze and return meaningful exceptions that we can show to the client;
throw new RuntimeException(e); // throw new RuntimeException(e);
} // }
}); // });
} // }
//
@Override // @Override
public VersionCreationResult push(PendingCommit commit) { // public VersionCreationResult push(PendingCommit commit) {
return executeInsideLock(commit, gitRepositoryService::push); // return executeInsideLock(commit, gitRepositoryService::push);
} // }
//
@Override // @Override
public List<EntityVersion> listVersions(TenantId tenantId, String branch) { // public List<EntityVersion> listVersions(TenantId tenantId, String branch) {
return listVersions(tenantId, branch, (String) null); // return listVersions(tenantId, branch, (String) null);
} // }
//
@Override // @Override
public List<EntityVersion> listVersions(TenantId tenantId, String branch, EntityType entityType) { // public List<EntityVersion> listVersions(TenantId tenantId, String branch, EntityType entityType) {
return listVersions(tenantId, branch, getRelativePath(entityType, null)); // return listVersions(tenantId, branch, getRelativePath(entityType, null));
} // }
//
@Override // @Override
public List<EntityVersion> listVersions(TenantId tenantId, String branch, EntityId entityId) { // public List<EntityVersion> listVersions(TenantId tenantId, String branch, EntityId entityId) {
return listVersions(tenantId, branch, getRelativePath(entityId.getEntityType(), entityId.getId().toString())); // return listVersions(tenantId, branch, getRelativePath(entityId.getEntityType(), entityId.getId().toString()));
} // }
//
@Override // @Override
public List<VersionedEntityInfo> listEntitiesAtVersion(TenantId tenantId, String branch, String versionId, EntityType entityType) { // public List<VersionedEntityInfo> listEntitiesAtVersion(TenantId tenantId, String branch, String versionId, EntityType entityType) {
try { // try {
return gitRepositoryService.listEntitiesAtVersion(tenantId, branch, versionId, entityType != null ? getRelativePath(entityType, null) : null); // return gitRepositoryService.listEntitiesAtVersion(tenantId, branch, versionId, entityType != null ? getRelativePath(entityType, null) : null);
} catch (Exception e) { // } catch (Exception e) {
//TODO: analyze and return meaningful exceptions that we can show to the client; // //TODO: analyze and return meaningful exceptions that we can show to the client;
throw new RuntimeException(e); // throw new RuntimeException(e);
} // }
} // }
//
@Override // @Override
public List<VersionedEntityInfo> listEntitiesAtVersion(TenantId tenantId, String branch, String versionId) { // public List<VersionedEntityInfo> listEntitiesAtVersion(TenantId tenantId, String branch, String versionId) {
return listEntitiesAtVersion(tenantId, branch, versionId, null); // return listEntitiesAtVersion(tenantId, branch, versionId, null);
} // }
//
@Override // @Override
public List<String> listBranches(TenantId tenantId) { // public List<String> listBranches(TenantId tenantId) {
return gitRepositoryService.listBranches(tenantId); // return gitRepositoryService.listBranches(tenantId);
} // }
//
@Override // @Override
public List<EntityExportData<?>> getEntities(TenantId tenantId, String branch, String versionId, EntityType entityType, int offset, int limit) { // public List<EntityExportData<?>> getEntities(TenantId tenantId, String branch, String versionId, EntityType entityType, int offset, int limit) {
return listEntitiesAtVersion(tenantId, branch, versionId, entityType).stream() // return listEntitiesAtVersion(tenantId, branch, versionId, entityType).stream()
.skip(offset).limit(limit) // .skip(offset).limit(limit)
.map(entityInfo -> getEntity(tenantId, versionId, entityInfo.getExternalId())) // .map(entityInfo -> getEntity(tenantId, versionId, entityInfo.getExternalId()))
.collect(Collectors.toList()); // .collect(Collectors.toList());
} // }
//
@Override // @Override
public EntityExportData<?> getEntity(TenantId tenantId, String versionId, EntityId entityId) { // public EntityExportData<?> getEntity(TenantId tenantId, String versionId, EntityId entityId) {
try { // try {
String entityDataJson = gitRepositoryService.getFileContentAtCommit(tenantId, // String entityDataJson = gitRepositoryService.getFileContentAtCommit(tenantId,
getRelativePath(entityId.getEntityType(), entityId.getId().toString()), versionId); // getRelativePath(entityId.getEntityType(), entityId.getId().toString()), versionId);
return JacksonUtil.fromString(entityDataJson, EntityExportData.class); // return JacksonUtil.fromString(entityDataJson, EntityExportData.class);
} catch (Exception e) { // } catch (Exception e) {
//TODO: analyze and return meaningful exceptions that we can show to the client; // //TODO: analyze and return meaningful exceptions that we can show to the client;
throw new RuntimeException(e); // throw new RuntimeException(e);
} // }
} // }
//
private EntitiesVersionControlSettings getVersionControlSettings(TenantId tenantId) { // private EntitiesVersionControlSettings getVersionControlSettings(TenantId tenantId) {
AdminSettings adminSettings = adminSettingsService.findAdminSettingsByKey(tenantId, EntitiesVersionControlService.SETTINGS_KEY); // AdminSettings adminSettings = adminSettingsService.findAdminSettingsByKey(tenantId, EntitiesVersionControlService.SETTINGS_KEY);
if (adminSettings != null) { // if (adminSettings != null) {
try { // try {
return JacksonUtil.convertValue(adminSettings.getJsonValue(), EntitiesVersionControlSettings.class); // return JacksonUtil.convertValue(adminSettings.getJsonValue(), EntitiesVersionControlSettings.class);
} catch (Exception e) { // } catch (Exception e) {
throw new RuntimeException("Failed to load version control settings!", e); // throw new RuntimeException("Failed to load version control settings!", e);
} // }
} // }
return null; // return null;
} // }
//
private List<EntityVersion> listVersions(TenantId tenantId, String branch, String path) { // private List<EntityVersion> listVersions(TenantId tenantId, String branch, String path) {
try { // try {
return gitRepositoryService.listVersions(tenantId, branch, path); // return gitRepositoryService.listVersions(tenantId, branch, path);
} catch (Exception e) { // } catch (Exception e) {
//TODO: analyze and return meaningful exceptions that we can show to the client; // //TODO: analyze and return meaningful exceptions that we can show to the client;
throw new RuntimeException(e); // throw new RuntimeException(e);
} // }
} // }
//
private void doInsideLock(PendingCommit commit, Consumer<PendingCommit> r) { // private void doInsideLock(PendingCommit commit, Consumer<PendingCommit> r) {
var lock = getRepoLock(commit.getTenantId()); // var lock = getRepoLock(commit.getTenantId());
lock.lock(); // lock.lock();
try { // try {
checkCommit(commit); // checkCommit(commit);
r.accept(commit); // r.accept(commit);
} finally { // } finally {
lock.unlock(); // lock.unlock();
} // }
} // }
//
private <T> T executeInsideLock(PendingCommit commit, Function<PendingCommit, T> c) { // private <T> T executeInsideLock(PendingCommit commit, Function<PendingCommit, T> c) {
var lock = getRepoLock(commit.getTenantId()); // var lock = getRepoLock(commit.getTenantId());
lock.lock(); // lock.lock();
try { // try {
checkCommit(commit); // checkCommit(commit);
return c.apply(commit); // return c.apply(commit);
} finally { // } finally {
lock.unlock(); // lock.unlock();
} // }
} // }
//
private void checkCommit(PendingCommit commit) { // private void checkCommit(PendingCommit commit) {
PendingCommit existing = pendingCommitMap.get(commit.getTenantId()); // PendingCommit existing = pendingCommitMap.get(commit.getTenantId());
if (existing == null || !existing.getTxId().equals(commit.getTxId())) { // if (existing == null || !existing.getRequestId().equals(commit.getRequestId())) {
throw new ConcurrentModificationException(); // throw new ConcurrentModificationException();
} // }
} // }
//
private String getRelativePath(EntityType entityType, String entityId) { // private String getRelativePath(EntityType entityType, String entityId) {
String path = entityType.name().toLowerCase(); // String path = entityType.name().toLowerCase();
if (entityId != null) { // if (entityId != null) {
path += "/" + entityId + ".json"; // path += "/" + entityId + ".json";
} // }
return path; // return path;
} // }
//
private Lock getRepoLock(TenantId tenantId) { // private Lock getRepoLock(TenantId tenantId) {
return tenantRepoLocks.computeIfAbsent(tenantId, t -> new ReentrantLock()); // return tenantRepoLocks.computeIfAbsent(tenantId, t -> new ReentrantLock());
} // }
} }

37
application/src/main/java/org/thingsboard/server/service/sync/vc/PendingGitRequest.java

@ -0,0 +1,37 @@
/**
* Copyright © 2016-2022 The Thingsboard Authors
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.thingsboard.server.service.sync.vc;
import com.google.common.util.concurrent.SettableFuture;
import lombok.Getter;
import lombok.RequiredArgsConstructor;
import org.thingsboard.server.common.data.id.TenantId;
import java.util.UUID;
@Getter
public class PendingGitRequest<T> {
private final UUID requestId;
private final TenantId tenantId;
private final SettableFuture<T> future;
public PendingGitRequest(TenantId tenantId) {
this.requestId = UUID.randomUUID();
this.tenantId = tenantId;
this.future = SettableFuture.create();
}
}

5
application/src/main/resources/thingsboard.yml

@ -1021,6 +1021,9 @@ queue:
stats: stats:
enabled: "${TB_QUEUE_CORE_STATS_ENABLED:true}" enabled: "${TB_QUEUE_CORE_STATS_ENABLED:true}"
print-interval-ms: "${TB_QUEUE_CORE_STATS_PRINT_INTERVAL_MS:60000}" print-interval-ms: "${TB_QUEUE_CORE_STATS_PRINT_INTERVAL_MS:60000}"
vc:
topic: "${TB_QUEUE_VC_TOPIC:tb_version_control}"
partitions: "${TB_QUEUE_VC_PARTITIONS:10}"
js: js:
# JS Eval request topic # JS Eval request topic
request_topic: "${REMOTE_JS_EVAL_REQUEST_TOPIC:js_eval.requests}" request_topic: "${REMOTE_JS_EVAL_REQUEST_TOPIC:js_eval.requests}"
@ -1115,8 +1118,8 @@ metrics:
percentiles: "${METRICS_TIMER_PERCENTILES:0.5}" percentiles: "${METRICS_TIMER_PERCENTILES:0.5}"
vc: vc:
thread_pool_size: "${TB_VC_POOL_SIZE:4}"
git: git:
service: "${JS_VC_GIT_SERVICE:local}" # local/remote
repos-poll-interval: "${TB_VC_GIT_REPOS_POLL_INTERVAL_SEC:60}" repos-poll-interval: "${TB_VC_GIT_REPOS_POLL_INTERVAL_SEC:60}"
management: management:

7
common/cluster-api/src/main/java/org/thingsboard/server/cluster/TbClusterService.java

@ -31,8 +31,9 @@ import org.thingsboard.server.common.msg.TbMsg;
import org.thingsboard.server.common.msg.ToDeviceActorNotificationMsg; import org.thingsboard.server.common.msg.ToDeviceActorNotificationMsg;
import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; import org.thingsboard.server.common.msg.queue.TopicPartitionInfo;
import org.thingsboard.server.common.msg.rpc.FromDeviceRpcResponse; import org.thingsboard.server.common.msg.rpc.FromDeviceRpcResponse;
import org.thingsboard.server.gen.transport.TransportProtos; import org.thingsboard.server.gen.transport.TransportProtos.ToVersionControlServiceMsg;
import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg;
import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg;
import org.thingsboard.server.gen.transport.TransportProtos.ToTransportMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToTransportMsg;
import org.thingsboard.server.queue.TbQueueCallback; import org.thingsboard.server.queue.TbQueueCallback;
import org.thingsboard.server.queue.TbQueueClusterService; import org.thingsboard.server.queue.TbQueueClusterService;
@ -47,9 +48,11 @@ public interface TbClusterService extends TbQueueClusterService {
void pushMsgToCore(ToDeviceActorNotificationMsg msg, TbQueueCallback callback); void pushMsgToCore(ToDeviceActorNotificationMsg msg, TbQueueCallback callback);
void pushMsgToVersionControl(TenantId tenantId, ToVersionControlServiceMsg msg, TbQueueCallback callback);
void pushNotificationToCore(String targetServiceId, FromDeviceRpcResponse response, TbQueueCallback callback); void pushNotificationToCore(String targetServiceId, FromDeviceRpcResponse response, TbQueueCallback callback);
void pushMsgToRuleEngine(TopicPartitionInfo tpi, UUID msgId, TransportProtos.ToRuleEngineMsg msg, TbQueueCallback callback); void pushMsgToRuleEngine(TopicPartitionInfo tpi, UUID msgId, ToRuleEngineMsg msg, TbQueueCallback callback);
void pushMsgToRuleEngine(TenantId tenantId, EntityId entityId, TbMsg msg, TbQueueCallback callback); void pushMsgToRuleEngine(TenantId tenantId, EntityId entityId, TbMsg msg, TbQueueCallback callback);

96
common/cluster-api/src/main/proto/queue.proto

@ -618,8 +618,8 @@ message TbSubscriptionUpdateValueListProto {
} }
message TbSubscriptionUpdateTsValue { message TbSubscriptionUpdateTsValue {
int64 ts = 1; int64 ts = 1;
optional string value = 2; optional string value = 2;
} }
/** /**
@ -676,6 +676,95 @@ message EdgeNotificationMsgProto {
PostAttributeMsg postAttributesMsg = 12; PostAttributeMsg postAttributesMsg = 12;
} }
/**
TB Core to Version Control Service
*/
message CommitRequestMsg {
PrepareMsg prepareMsg = 1;
AddMsg addMsg = 2;
DeleteMsg deleteMsg = 3;
PushMsg pushMsg = 4;
AbortMsg abortMsg = 5;
}
message CommitResponseMsg {
string commitId = 1;
string name = 2;
int32 added = 3;
int32 modified = 4;
int32 removed = 5;
}
message PrepareMsg {
string commitMsg = 1;
string branchName = 2;
}
message AddMsg {
string relativePath = 1;
string entityDataJson = 2;
}
message DeleteMsg {
string relativePath = 1;
}
message PushMsg {
}
message AbortMsg {
}
message ListVersionsRequestMsg {
string branchName = 1;
string entityType = 2;
int64 entityIdMSB = 3;
int64 entityIdLSB = 4;
}
message ListEntitiesRequestMsg {
string branchName = 1;
string versionId = 2;
string entityType = 3;
}
message ListBranchesRequestMsg {
}
message EntityContentRequestMsg {
string versionId = 1;
string entityType = 2;
int64 entityIdMSB = 3;
int64 entityIdLSB = 4;
}
message EntitiesContentRequestMsg {
string versionId = 1;
string entityType = 2;
int32 offset = 3;
int32 limit = 4;
}
message ToVersionControlServiceMsg {
string nodeId = 1;
int64 tenantIdMSB = 2;
int64 tenantIdLSB = 3;
int64 requestIdMSB = 4;
int64 requestIdLSB = 5;
CommitRequestMsg commitRequest = 6;
ListVersionsRequestMsg listVersionRequest = 7;
ListEntitiesRequestMsg listEntitiesRequest = 8;
ListBranchesRequestMsg listBranchesRequest = 9;
EntityContentRequestMsg entityContentRequest = 10;
EntitiesContentRequestMsg entitiesContentRequest = 11;
}
message VersionControlResponseMsg {
int64 requestIdMSB = 1;
int64 requestIdLSB = 2;
CommitResponseMsg commitResponse = 3;
}
/** /**
* Main messages; * Main messages;
*/ */
@ -730,6 +819,7 @@ message ToCoreNotificationMsg {
bytes edgeEventUpdateMsg = 4; bytes edgeEventUpdateMsg = 4;
QueueUpdateMsg queueUpdateMsg = 5; QueueUpdateMsg queueUpdateMsg = 5;
QueueDeleteMsg queueDeleteMsg = 6; QueueDeleteMsg queueDeleteMsg = 6;
VersionControlResponseMsg vcResponseMsg = 7;
} }
/* Messages that are handled by ThingsBoard RuleEngine Service */ /* Messages that are handled by ThingsBoard RuleEngine Service */
@ -793,3 +883,5 @@ message ToOtaPackageStateServiceMsg {
int64 otaPackageIdLSB = 7; int64 otaPackageIdLSB = 7;
string type = 8; string type = 8;
} }

2
common/message/src/main/java/org/thingsboard/server/common/msg/queue/ServiceType.java

@ -17,7 +17,7 @@ package org.thingsboard.server.common.msg.queue;
public enum ServiceType { public enum ServiceType {
TB_CORE, TB_RULE_ENGINE, TB_TRANSPORT, JS_EXECUTOR; TB_CORE, TB_RULE_ENGINE, TB_TRANSPORT, JS_EXECUTOR, TB_VC_EXECUTOR;
public static ServiceType of(String serviceType) { public static ServiceType of(String serviceType) {
return ServiceType.valueOf(serviceType.replace("-", "_").toUpperCase()); return ServiceType.valueOf(serviceType.replace("-", "_").toUpperCase());

10
common/queue/src/main/java/org/thingsboard/server/queue/discovery/HashPartitionService.java

@ -54,6 +54,10 @@ public class HashPartitionService implements PartitionService {
private String coreTopic; private String coreTopic;
@Value("${queue.core.partitions:100}") @Value("${queue.core.partitions:100}")
private Integer corePartitions; private Integer corePartitions;
@Value("${queue.vc.topic}")
private String vcTopic;
@Value("${queue.vc.partitions:10}")
private Integer vcPartitions;
@Value("${queue.partitions.hash_function_name:murmur3_128}") @Value("${queue.partitions.hash_function_name:murmur3_128}")
private String hashFunctionName; private String hashFunctionName;
@ -97,6 +101,10 @@ public class HashPartitionService implements PartitionService {
partitionSizesMap.put(coreKey, corePartitions); partitionSizesMap.put(coreKey, corePartitions);
partitionTopicsMap.put(coreKey, coreTopic); partitionTopicsMap.put(coreKey, coreTopic);
QueueKey vcKey = new QueueKey(ServiceType.TB_VC_EXECUTOR);
partitionSizesMap.put(vcKey, vcPartitions);
partitionTopicsMap.put(vcKey, vcTopic);
List<QueueRoutingInfo> queueRoutingInfoList; List<QueueRoutingInfo> queueRoutingInfoList;
String serviceType = serviceInfoProvider.getServiceType(); String serviceType = serviceInfoProvider.getServiceType();
@ -401,7 +409,7 @@ public class HashPartitionService implements PartitionService {
queueServiceList.computeIfAbsent(key, k -> new ArrayList<>()).add(instance); queueServiceList.computeIfAbsent(key, k -> new ArrayList<>()).add(instance);
} }
}); });
} else if (ServiceType.TB_CORE.equals(serviceType)) { } else if (ServiceType.TB_CORE.equals(serviceType) || ServiceType.TB_VC_EXECUTOR.equals(serviceType)) {
queueServiceList.computeIfAbsent(new QueueKey(serviceType), key -> new ArrayList<>()).add(instance); queueServiceList.computeIfAbsent(new QueueKey(serviceType), key -> new ArrayList<>()).add(instance);
} }
} }

6
common/queue/src/main/java/org/thingsboard/server/queue/kafka/TbKafkaTopicConfigs.java

@ -40,6 +40,9 @@ public class TbKafkaTopicConfigs {
private String jsExecutorProperties; private String jsExecutorProperties;
@Value("${queue.kafka.topic-properties.ota-updates:}") @Value("${queue.kafka.topic-properties.ota-updates:}")
private String fwUpdatesProperties; private String fwUpdatesProperties;
@Value("${queue.kafka.topic-properties.version-control:}")
private String vcProperties;
@Getter @Getter
private Map<String, String> coreConfigs; private Map<String, String> coreConfigs;
@ -53,6 +56,8 @@ public class TbKafkaTopicConfigs {
private Map<String, String> jsExecutorConfigs; private Map<String, String> jsExecutorConfigs;
@Getter @Getter
private Map<String, String> fwUpdatesConfigs; private Map<String, String> fwUpdatesConfigs;
@Getter
private Map<String, String> vcConfigs;
@PostConstruct @PostConstruct
private void init() { private void init() {
@ -62,6 +67,7 @@ public class TbKafkaTopicConfigs {
notificationsConfigs = getConfigs(notificationsProperties); notificationsConfigs = getConfigs(notificationsProperties);
jsExecutorConfigs = getConfigs(jsExecutorProperties); jsExecutorConfigs = getConfigs(jsExecutorProperties);
fwUpdatesConfigs = getConfigs(fwUpdatesProperties); fwUpdatesConfigs = getConfigs(fwUpdatesProperties);
vcConfigs = getConfigs(vcProperties);
} }
private Map<String, String> getConfigs(String properties) { private Map<String, String> getConfigs(String properties) {

7
common/queue/src/main/java/org/thingsboard/server/queue/provider/AwsSqsMonolithQueueFactory.java

@ -23,6 +23,7 @@ import org.thingsboard.server.common.data.queue.Queue;
import org.thingsboard.server.common.msg.queue.ServiceType; import org.thingsboard.server.common.msg.queue.ServiceType;
import org.thingsboard.server.gen.js.JsInvokeProtos.RemoteJsRequest; import org.thingsboard.server.gen.js.JsInvokeProtos.RemoteJsRequest;
import org.thingsboard.server.gen.js.JsInvokeProtos.RemoteJsResponse; import org.thingsboard.server.gen.js.JsInvokeProtos.RemoteJsResponse;
import org.thingsboard.server.gen.transport.TransportProtos;
import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg;
import org.thingsboard.server.gen.transport.TransportProtos.ToCoreNotificationMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreNotificationMsg;
import org.thingsboard.server.gen.transport.TransportProtos.ToOtaPackageStateServiceMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToOtaPackageStateServiceMsg;
@ -205,6 +206,12 @@ public class AwsSqsMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEng
return new TbAwsSqsProducerTemplate<>(coreAdmin, sqsSettings, coreSettings.getOtaPackageTopic()); return new TbAwsSqsProducerTemplate<>(coreAdmin, sqsSettings, coreSettings.getOtaPackageTopic());
} }
@Override
public TbQueueProducer<TbProtoQueueMsg<TransportProtos.ToVersionControlServiceMsg>> createVersionControlMsgProducer() {
//TODO: version-control
return null;
}
@PreDestroy @PreDestroy
private void destroy() { private void destroy() {
if (coreAdmin != null) { if (coreAdmin != null) {

7
common/queue/src/main/java/org/thingsboard/server/queue/provider/AwsSqsTbCoreQueueFactory.java

@ -21,6 +21,7 @@ import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component; import org.springframework.stereotype.Component;
import org.thingsboard.server.common.msg.queue.ServiceType; import org.thingsboard.server.common.msg.queue.ServiceType;
import org.thingsboard.server.gen.js.JsInvokeProtos; import org.thingsboard.server.gen.js.JsInvokeProtos;
import org.thingsboard.server.gen.transport.TransportProtos;
import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg;
import org.thingsboard.server.gen.transport.TransportProtos.ToCoreNotificationMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreNotificationMsg;
import org.thingsboard.server.gen.transport.TransportProtos.ToOtaPackageStateServiceMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToOtaPackageStateServiceMsg;
@ -191,6 +192,12 @@ public class AwsSqsTbCoreQueueFactory implements TbCoreQueueFactory {
return new TbAwsSqsProducerTemplate<>(coreAdmin, sqsSettings, coreSettings.getOtaPackageTopic()); return new TbAwsSqsProducerTemplate<>(coreAdmin, sqsSettings, coreSettings.getOtaPackageTopic());
} }
@Override
public TbQueueProducer<TbProtoQueueMsg<TransportProtos.ToVersionControlServiceMsg>> createVersionControlMsgProducer() {
//TODO: version-control
return null;
}
@PreDestroy @PreDestroy
private void destroy() { private void destroy() {
if (coreAdmin != null) { if (coreAdmin != null) {

6
common/queue/src/main/java/org/thingsboard/server/queue/provider/InMemoryMonolithQueueFactory.java

@ -146,6 +146,12 @@ public class InMemoryMonolithQueueFactory implements TbCoreQueueFactory, TbRuleE
return new InMemoryTbQueueProducer<>(storage, coreSettings.getUsageStatsTopic()); return new InMemoryTbQueueProducer<>(storage, coreSettings.getUsageStatsTopic());
} }
@Override
public TbQueueProducer<TbProtoQueueMsg<TransportProtos.ToVersionControlServiceMsg>> createVersionControlMsgProducer() {
//TODO: version-control
return null;
}
@Scheduled(fixedRateString = "${queue.in_memory.stats.print-interval-ms:60000}") @Scheduled(fixedRateString = "${queue.in_memory.stats.print-interval-ms:60000}")
private void printInMemoryStats() { private void printInMemoryStats() {
storage.printStats(); storage.printStats();

36
common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaMonolithQueueFactory.java

@ -22,6 +22,7 @@ import org.springframework.stereotype.Component;
import org.thingsboard.server.common.data.queue.Queue; import org.thingsboard.server.common.data.queue.Queue;
import org.thingsboard.server.common.msg.queue.ServiceType; import org.thingsboard.server.common.msg.queue.ServiceType;
import org.thingsboard.server.gen.js.JsInvokeProtos; import org.thingsboard.server.gen.js.JsInvokeProtos;
import org.thingsboard.server.gen.transport.TransportProtos;
import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg;
import org.thingsboard.server.gen.transport.TransportProtos.ToCoreNotificationMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreNotificationMsg;
import org.thingsboard.server.gen.transport.TransportProtos.ToOtaPackageStateServiceMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToOtaPackageStateServiceMsg;
@ -51,6 +52,7 @@ import org.thingsboard.server.queue.settings.TbQueueRemoteJsInvokeSettings;
import org.thingsboard.server.queue.settings.TbQueueRuleEngineSettings; import org.thingsboard.server.queue.settings.TbQueueRuleEngineSettings;
import org.thingsboard.server.queue.settings.TbQueueTransportApiSettings; import org.thingsboard.server.queue.settings.TbQueueTransportApiSettings;
import org.thingsboard.server.queue.settings.TbQueueTransportNotificationSettings; import org.thingsboard.server.queue.settings.TbQueueTransportNotificationSettings;
import org.thingsboard.server.queue.settings.TbQueueVersionControlSettings;
import javax.annotation.PreDestroy; import javax.annotation.PreDestroy;
import java.nio.charset.StandardCharsets; import java.nio.charset.StandardCharsets;
@ -58,7 +60,7 @@ import java.util.concurrent.atomic.AtomicLong;
@Component @Component
@ConditionalOnExpression("'${queue.type:null}'=='kafka' && '${service.type:null}'=='monolith'") @ConditionalOnExpression("'${queue.type:null}'=='kafka' && '${service.type:null}'=='monolith'")
public class KafkaMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEngineQueueFactory { public class KafkaMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEngineQueueFactory, TbVersionControlQueueFactory {
private final NotificationsTopicService notificationsTopicService; private final NotificationsTopicService notificationsTopicService;
private final TbKafkaSettings kafkaSettings; private final TbKafkaSettings kafkaSettings;
@ -68,6 +70,7 @@ public class KafkaMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEngi
private final TbQueueTransportApiSettings transportApiSettings; private final TbQueueTransportApiSettings transportApiSettings;
private final TbQueueTransportNotificationSettings transportNotificationSettings; private final TbQueueTransportNotificationSettings transportNotificationSettings;
private final TbQueueRemoteJsInvokeSettings jsInvokeSettings; private final TbQueueRemoteJsInvokeSettings jsInvokeSettings;
private final TbQueueVersionControlSettings vcSettings;
private final TbKafkaConsumerStatsService consumerStatsService; private final TbKafkaConsumerStatsService consumerStatsService;
private final TbQueueAdmin coreAdmin; private final TbQueueAdmin coreAdmin;
@ -76,6 +79,8 @@ public class KafkaMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEngi
private final TbQueueAdmin transportApiAdmin; private final TbQueueAdmin transportApiAdmin;
private final TbQueueAdmin notificationAdmin; private final TbQueueAdmin notificationAdmin;
private final TbQueueAdmin fwUpdatesAdmin; private final TbQueueAdmin fwUpdatesAdmin;
private final TbQueueAdmin vcAdmin;
private final AtomicLong consumerCount = new AtomicLong(); private final AtomicLong consumerCount = new AtomicLong();
public KafkaMonolithQueueFactory(NotificationsTopicService notificationsTopicService, TbKafkaSettings kafkaSettings, public KafkaMonolithQueueFactory(NotificationsTopicService notificationsTopicService, TbKafkaSettings kafkaSettings,
@ -85,6 +90,7 @@ public class KafkaMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEngi
TbQueueTransportApiSettings transportApiSettings, TbQueueTransportApiSettings transportApiSettings,
TbQueueTransportNotificationSettings transportNotificationSettings, TbQueueTransportNotificationSettings transportNotificationSettings,
TbQueueRemoteJsInvokeSettings jsInvokeSettings, TbQueueRemoteJsInvokeSettings jsInvokeSettings,
TbQueueVersionControlSettings vcSettings,
TbKafkaConsumerStatsService consumerStatsService, TbKafkaConsumerStatsService consumerStatsService,
TbKafkaTopicConfigs kafkaTopicConfigs) { TbKafkaTopicConfigs kafkaTopicConfigs) {
this.notificationsTopicService = notificationsTopicService; this.notificationsTopicService = notificationsTopicService;
@ -95,6 +101,7 @@ public class KafkaMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEngi
this.transportApiSettings = transportApiSettings; this.transportApiSettings = transportApiSettings;
this.transportNotificationSettings = transportNotificationSettings; this.transportNotificationSettings = transportNotificationSettings;
this.jsInvokeSettings = jsInvokeSettings; this.jsInvokeSettings = jsInvokeSettings;
this.vcSettings = vcSettings;
this.consumerStatsService = consumerStatsService; this.consumerStatsService = consumerStatsService;
this.coreAdmin = new TbKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getCoreConfigs()); this.coreAdmin = new TbKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getCoreConfigs());
@ -103,6 +110,7 @@ public class KafkaMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEngi
this.transportApiAdmin = new TbKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getTransportApiConfigs()); this.transportApiAdmin = new TbKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getTransportApiConfigs());
this.notificationAdmin = new TbKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getNotificationsConfigs()); this.notificationAdmin = new TbKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getNotificationsConfigs());
this.fwUpdatesAdmin = new TbKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getFwUpdatesConfigs()); this.fwUpdatesAdmin = new TbKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getFwUpdatesConfigs());
this.vcAdmin = new TbKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getVcConfigs());
} }
@Override @Override
@ -155,6 +163,19 @@ public class KafkaMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEngi
return requestBuilder.build(); return requestBuilder.build();
} }
@Override
public TbQueueConsumer<TbProtoQueueMsg<TransportProtos.ToVersionControlServiceMsg>> createToVersionControlMsgConsumer() {
TbKafkaConsumerTemplate.TbKafkaConsumerTemplateBuilder<TbProtoQueueMsg<TransportProtos.ToVersionControlServiceMsg>> consumerBuilder = TbKafkaConsumerTemplate.builder();
consumerBuilder.settings(kafkaSettings);
consumerBuilder.topic(vcSettings.getTopic());
consumerBuilder.clientId("monolith-vc-consumer-" + serviceInfoProvider.getServiceId());
consumerBuilder.groupId("monolith-vc-node");
consumerBuilder.decoder(msg -> new TbProtoQueueMsg<>(msg.getKey(), TransportProtos.ToVersionControlServiceMsg.parseFrom(msg.getData()), msg.getHeaders()));
consumerBuilder.admin(vcAdmin);
consumerBuilder.statsService(consumerStatsService);
return consumerBuilder.build();
}
@Override @Override
public TbQueueConsumer<TbProtoQueueMsg<ToRuleEngineMsg>> createToRuleEngineMsgConsumer(Queue configuration) { public TbQueueConsumer<TbProtoQueueMsg<ToRuleEngineMsg>> createToRuleEngineMsgConsumer(Queue configuration) {
String queueName = configuration.getName(); String queueName = configuration.getName();
@ -311,6 +332,16 @@ public class KafkaMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEngi
return requestBuilder.build(); return requestBuilder.build();
} }
@Override
public TbQueueProducer<TbProtoQueueMsg<TransportProtos.ToVersionControlServiceMsg>> createVersionControlMsgProducer() {
TbKafkaProducerTemplate.TbKafkaProducerTemplateBuilder<TbProtoQueueMsg<TransportProtos.ToVersionControlServiceMsg>> requestBuilder = TbKafkaProducerTemplate.builder();
requestBuilder.settings(kafkaSettings);
requestBuilder.clientId("monolith-vc-producer-" + serviceInfoProvider.getServiceId());
requestBuilder.defaultTopic(vcSettings.getTopic());
requestBuilder.admin(vcAdmin);
return requestBuilder.build();
}
@PreDestroy @PreDestroy
private void destroy() { private void destroy() {
if (coreAdmin != null) { if (coreAdmin != null) {
@ -331,5 +362,8 @@ public class KafkaMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEngi
if (fwUpdatesAdmin != null) { if (fwUpdatesAdmin != null) {
fwUpdatesAdmin.destroy(); fwUpdatesAdmin.destroy();
} }
if (vcAdmin != null) {
vcAdmin.destroy();
}
} }
} }

20
common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaTbCoreQueueFactory.java

@ -28,6 +28,7 @@ import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg;
import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineNotificationMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineNotificationMsg;
import org.thingsboard.server.gen.transport.TransportProtos.ToTransportMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToTransportMsg;
import org.thingsboard.server.gen.transport.TransportProtos.ToUsageStatsServiceMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToUsageStatsServiceMsg;
import org.thingsboard.server.gen.transport.TransportProtos.ToVersionControlServiceMsg;
import org.thingsboard.server.gen.transport.TransportProtos.TransportApiRequestMsg; import org.thingsboard.server.gen.transport.TransportProtos.TransportApiRequestMsg;
import org.thingsboard.server.gen.transport.TransportProtos.TransportApiResponseMsg; import org.thingsboard.server.gen.transport.TransportProtos.TransportApiResponseMsg;
import org.thingsboard.server.queue.TbQueueAdmin; import org.thingsboard.server.queue.TbQueueAdmin;
@ -50,6 +51,7 @@ import org.thingsboard.server.queue.settings.TbQueueRemoteJsInvokeSettings;
import org.thingsboard.server.queue.settings.TbQueueRuleEngineSettings; import org.thingsboard.server.queue.settings.TbQueueRuleEngineSettings;
import org.thingsboard.server.queue.settings.TbQueueTransportApiSettings; import org.thingsboard.server.queue.settings.TbQueueTransportApiSettings;
import org.thingsboard.server.queue.settings.TbQueueTransportNotificationSettings; import org.thingsboard.server.queue.settings.TbQueueTransportNotificationSettings;
import org.thingsboard.server.queue.settings.TbQueueVersionControlSettings;
import javax.annotation.PreDestroy; import javax.annotation.PreDestroy;
import java.nio.charset.StandardCharsets; import java.nio.charset.StandardCharsets;
@ -65,6 +67,7 @@ public class KafkaTbCoreQueueFactory implements TbCoreQueueFactory {
private final TbQueueRuleEngineSettings ruleEngineSettings; private final TbQueueRuleEngineSettings ruleEngineSettings;
private final TbQueueTransportApiSettings transportApiSettings; private final TbQueueTransportApiSettings transportApiSettings;
private final TbQueueRemoteJsInvokeSettings jsInvokeSettings; private final TbQueueRemoteJsInvokeSettings jsInvokeSettings;
private final TbQueueVersionControlSettings vcSettings;
private final TbKafkaConsumerStatsService consumerStatsService; private final TbKafkaConsumerStatsService consumerStatsService;
private final TbQueueTransportNotificationSettings transportNotificationSettings; private final TbQueueTransportNotificationSettings transportNotificationSettings;
@ -74,6 +77,7 @@ public class KafkaTbCoreQueueFactory implements TbCoreQueueFactory {
private final TbQueueAdmin transportApiAdmin; private final TbQueueAdmin transportApiAdmin;
private final TbQueueAdmin notificationAdmin; private final TbQueueAdmin notificationAdmin;
private final TbQueueAdmin fwUpdatesAdmin; private final TbQueueAdmin fwUpdatesAdmin;
private final TbQueueAdmin vcAdmin;
public KafkaTbCoreQueueFactory(NotificationsTopicService notificationsTopicService, public KafkaTbCoreQueueFactory(NotificationsTopicService notificationsTopicService,
TbKafkaSettings kafkaSettings, TbKafkaSettings kafkaSettings,
@ -82,6 +86,7 @@ public class KafkaTbCoreQueueFactory implements TbCoreQueueFactory {
TbQueueRuleEngineSettings ruleEngineSettings, TbQueueRuleEngineSettings ruleEngineSettings,
TbQueueTransportApiSettings transportApiSettings, TbQueueTransportApiSettings transportApiSettings,
TbQueueRemoteJsInvokeSettings jsInvokeSettings, TbQueueRemoteJsInvokeSettings jsInvokeSettings,
TbQueueVersionControlSettings vcSettings,
TbKafkaConsumerStatsService consumerStatsService, TbKafkaConsumerStatsService consumerStatsService,
TbQueueTransportNotificationSettings transportNotificationSettings, TbQueueTransportNotificationSettings transportNotificationSettings,
TbKafkaTopicConfigs kafkaTopicConfigs) { TbKafkaTopicConfigs kafkaTopicConfigs) {
@ -92,6 +97,7 @@ public class KafkaTbCoreQueueFactory implements TbCoreQueueFactory {
this.ruleEngineSettings = ruleEngineSettings; this.ruleEngineSettings = ruleEngineSettings;
this.transportApiSettings = transportApiSettings; this.transportApiSettings = transportApiSettings;
this.jsInvokeSettings = jsInvokeSettings; this.jsInvokeSettings = jsInvokeSettings;
this.vcSettings = vcSettings;
this.consumerStatsService = consumerStatsService; this.consumerStatsService = consumerStatsService;
this.transportNotificationSettings = transportNotificationSettings; this.transportNotificationSettings = transportNotificationSettings;
@ -101,6 +107,7 @@ public class KafkaTbCoreQueueFactory implements TbCoreQueueFactory {
this.transportApiAdmin = new TbKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getTransportApiConfigs()); this.transportApiAdmin = new TbKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getTransportApiConfigs());
this.notificationAdmin = new TbKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getNotificationsConfigs()); this.notificationAdmin = new TbKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getNotificationsConfigs());
this.fwUpdatesAdmin = new TbKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getFwUpdatesConfigs()); this.fwUpdatesAdmin = new TbKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getFwUpdatesConfigs());
this.vcAdmin = new TbKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getVcConfigs());
} }
@Override @Override
@ -282,6 +289,16 @@ public class KafkaTbCoreQueueFactory implements TbCoreQueueFactory {
return requestBuilder.build(); return requestBuilder.build();
} }
@Override
public TbQueueProducer<TbProtoQueueMsg<ToVersionControlServiceMsg>> createVersionControlMsgProducer() {
TbKafkaProducerTemplate.TbKafkaProducerTemplateBuilder<TbProtoQueueMsg<ToVersionControlServiceMsg>> requestBuilder = TbKafkaProducerTemplate.builder();
requestBuilder.settings(kafkaSettings);
requestBuilder.clientId("tb-core-vc-producer-" + serviceInfoProvider.getServiceId());
requestBuilder.defaultTopic(vcSettings.getTopic());
requestBuilder.admin(vcAdmin);
return requestBuilder.build();
}
@PreDestroy @PreDestroy
private void destroy() { private void destroy() {
if (coreAdmin != null) { if (coreAdmin != null) {
@ -302,5 +319,8 @@ public class KafkaTbCoreQueueFactory implements TbCoreQueueFactory {
if (fwUpdatesAdmin != null) { if (fwUpdatesAdmin != null) {
fwUpdatesAdmin.destroy(); fwUpdatesAdmin.destroy();
} }
if (vcAdmin != null) {
vcAdmin.destroy();
}
} }
} }

116
common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaTbVersionControlQueueFactory.java

@ -0,0 +1,116 @@
/**
* 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.queue.provider;
import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
import org.springframework.stereotype.Component;
import org.thingsboard.server.gen.transport.TransportProtos.ToCoreNotificationMsg;
import org.thingsboard.server.gen.transport.TransportProtos.ToUsageStatsServiceMsg;
import org.thingsboard.server.gen.transport.TransportProtos.ToVersionControlServiceMsg;
import org.thingsboard.server.queue.TbQueueAdmin;
import org.thingsboard.server.queue.TbQueueConsumer;
import org.thingsboard.server.queue.TbQueueProducer;
import org.thingsboard.server.queue.common.TbProtoQueueMsg;
import org.thingsboard.server.queue.discovery.TbServiceInfoProvider;
import org.thingsboard.server.queue.kafka.TbKafkaAdmin;
import org.thingsboard.server.queue.kafka.TbKafkaConsumerStatsService;
import org.thingsboard.server.queue.kafka.TbKafkaConsumerTemplate;
import org.thingsboard.server.queue.kafka.TbKafkaProducerTemplate;
import org.thingsboard.server.queue.kafka.TbKafkaSettings;
import org.thingsboard.server.queue.kafka.TbKafkaTopicConfigs;
import org.thingsboard.server.queue.settings.TbQueueCoreSettings;
import org.thingsboard.server.queue.settings.TbQueueVersionControlSettings;
import javax.annotation.PreDestroy;
@Component
@ConditionalOnExpression("'${queue.type:null}'=='kafka' && '${service.type:null}'=='tb-vc-executor'")
public class KafkaTbVersionControlQueueFactory implements TbVersionControlQueueFactory {
private final TbKafkaSettings kafkaSettings;
private final TbServiceInfoProvider serviceInfoProvider;
private final TbQueueCoreSettings coreSettings;
private final TbQueueVersionControlSettings vcSettings;
private final TbKafkaConsumerStatsService consumerStatsService;
private final TbQueueAdmin coreAdmin;
private final TbQueueAdmin vcAdmin;
private final TbQueueAdmin notificationAdmin;
public KafkaTbVersionControlQueueFactory(TbKafkaSettings kafkaSettings,
TbServiceInfoProvider serviceInfoProvider,
TbQueueCoreSettings coreSettings,
TbQueueVersionControlSettings vcSettings,
TbKafkaConsumerStatsService consumerStatsService,
TbKafkaTopicConfigs kafkaTopicConfigs) {
this.kafkaSettings = kafkaSettings;
this.serviceInfoProvider = serviceInfoProvider;
this.coreSettings = coreSettings;
this.vcSettings = vcSettings;
this.consumerStatsService = consumerStatsService;
this.coreAdmin = new TbKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getCoreConfigs());
this.vcAdmin = new TbKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getVcConfigs());
this.notificationAdmin = new TbKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getNotificationsConfigs());
}
@Override
public TbQueueProducer<TbProtoQueueMsg<ToCoreNotificationMsg>> createTbCoreNotificationsMsgProducer() {
TbKafkaProducerTemplate.TbKafkaProducerTemplateBuilder<TbProtoQueueMsg<ToCoreNotificationMsg>> requestBuilder = TbKafkaProducerTemplate.builder();
requestBuilder.settings(kafkaSettings);
requestBuilder.clientId("tb-vc-to-core-notifications-" + serviceInfoProvider.getServiceId());
requestBuilder.defaultTopic(coreSettings.getTopic());
requestBuilder.admin(notificationAdmin);
return requestBuilder.build();
}
@Override
public TbQueueConsumer<TbProtoQueueMsg<ToVersionControlServiceMsg>> createToVersionControlMsgConsumer() {
TbKafkaConsumerTemplate.TbKafkaConsumerTemplateBuilder<TbProtoQueueMsg<ToVersionControlServiceMsg>> consumerBuilder = TbKafkaConsumerTemplate.builder();
consumerBuilder.settings(kafkaSettings);
consumerBuilder.topic(vcSettings.getTopic());
consumerBuilder.clientId("tb-vc-consumer-" + serviceInfoProvider.getServiceId());
consumerBuilder.groupId("tb-vc-node");
consumerBuilder.decoder(msg -> new TbProtoQueueMsg<>(msg.getKey(), ToVersionControlServiceMsg.parseFrom(msg.getData()), msg.getHeaders()));
consumerBuilder.admin(vcAdmin);
consumerBuilder.statsService(consumerStatsService);
return consumerBuilder.build();
}
@Override
public TbQueueProducer<TbProtoQueueMsg<ToUsageStatsServiceMsg>> createToUsageStatsServiceMsgProducer() {
TbKafkaProducerTemplate.TbKafkaProducerTemplateBuilder<TbProtoQueueMsg<ToUsageStatsServiceMsg>> requestBuilder = TbKafkaProducerTemplate.builder();
requestBuilder.settings(kafkaSettings);
requestBuilder.clientId("tb-vc-us-producer-" + serviceInfoProvider.getServiceId());
requestBuilder.defaultTopic(coreSettings.getUsageStatsTopic());
requestBuilder.admin(coreAdmin);
return requestBuilder.build();
}
@PreDestroy
private void destroy() {
if (coreAdmin != null) {
coreAdmin.destroy();
}
if (vcAdmin != null) {
vcAdmin.destroy();
}
if (notificationAdmin != null) {
notificationAdmin.destroy();
}
}
}

7
common/queue/src/main/java/org/thingsboard/server/queue/provider/PubSubMonolithQueueFactory.java

@ -23,6 +23,7 @@ import org.thingsboard.server.common.data.queue.Queue;
import org.thingsboard.server.common.msg.queue.ServiceType; import org.thingsboard.server.common.msg.queue.ServiceType;
import org.thingsboard.server.gen.js.JsInvokeProtos.RemoteJsRequest; import org.thingsboard.server.gen.js.JsInvokeProtos.RemoteJsRequest;
import org.thingsboard.server.gen.js.JsInvokeProtos.RemoteJsResponse; import org.thingsboard.server.gen.js.JsInvokeProtos.RemoteJsResponse;
import org.thingsboard.server.gen.transport.TransportProtos;
import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg;
import org.thingsboard.server.gen.transport.TransportProtos.ToCoreNotificationMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreNotificationMsg;
import org.thingsboard.server.gen.transport.TransportProtos.ToOtaPackageStateServiceMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToOtaPackageStateServiceMsg;
@ -207,6 +208,12 @@ public class PubSubMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEng
return new TbPubSubProducerTemplate<>(coreAdmin, pubSubSettings, coreSettings.getUsageStatsTopic()); return new TbPubSubProducerTemplate<>(coreAdmin, pubSubSettings, coreSettings.getUsageStatsTopic());
} }
@Override
public TbQueueProducer<TbProtoQueueMsg<TransportProtos.ToVersionControlServiceMsg>> createVersionControlMsgProducer() {
//TODO: version-control
return null;
}
@PreDestroy @PreDestroy
private void destroy() { private void destroy() {
if (coreAdmin != null) { if (coreAdmin != null) {

7
common/queue/src/main/java/org/thingsboard/server/queue/provider/PubSubTbCoreQueueFactory.java

@ -21,6 +21,7 @@ import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component; import org.springframework.stereotype.Component;
import org.thingsboard.server.common.msg.queue.ServiceType; import org.thingsboard.server.common.msg.queue.ServiceType;
import org.thingsboard.server.gen.js.JsInvokeProtos; import org.thingsboard.server.gen.js.JsInvokeProtos;
import org.thingsboard.server.gen.transport.TransportProtos;
import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg;
import org.thingsboard.server.gen.transport.TransportProtos.ToCoreNotificationMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreNotificationMsg;
import org.thingsboard.server.gen.transport.TransportProtos.ToOtaPackageStateServiceMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToOtaPackageStateServiceMsg;
@ -191,6 +192,12 @@ public class PubSubTbCoreQueueFactory implements TbCoreQueueFactory {
return new TbPubSubProducerTemplate<>(coreAdmin, pubSubSettings, coreSettings.getUsageStatsTopic()); return new TbPubSubProducerTemplate<>(coreAdmin, pubSubSettings, coreSettings.getUsageStatsTopic());
} }
@Override
public TbQueueProducer<TbProtoQueueMsg<TransportProtos.ToVersionControlServiceMsg>> createVersionControlMsgProducer() {
//TODO: version-control
return null;
}
@PreDestroy @PreDestroy
private void destroy() { private void destroy() {
if (coreAdmin != null) { if (coreAdmin != null) {

7
common/queue/src/main/java/org/thingsboard/server/queue/provider/RabbitMqMonolithQueueFactory.java

@ -23,6 +23,7 @@ import org.thingsboard.server.common.data.queue.Queue;
import org.thingsboard.server.common.msg.queue.ServiceType; import org.thingsboard.server.common.msg.queue.ServiceType;
import org.thingsboard.server.gen.js.JsInvokeProtos.RemoteJsRequest; import org.thingsboard.server.gen.js.JsInvokeProtos.RemoteJsRequest;
import org.thingsboard.server.gen.js.JsInvokeProtos.RemoteJsResponse; import org.thingsboard.server.gen.js.JsInvokeProtos.RemoteJsResponse;
import org.thingsboard.server.gen.transport.TransportProtos;
import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg;
import org.thingsboard.server.gen.transport.TransportProtos.ToCoreNotificationMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreNotificationMsg;
import org.thingsboard.server.gen.transport.TransportProtos.ToOtaPackageStateServiceMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToOtaPackageStateServiceMsg;
@ -205,6 +206,12 @@ public class RabbitMqMonolithQueueFactory implements TbCoreQueueFactory, TbRuleE
return new TbRabbitMqProducerTemplate<>(coreAdmin, rabbitMqSettings, coreSettings.getUsageStatsTopic()); return new TbRabbitMqProducerTemplate<>(coreAdmin, rabbitMqSettings, coreSettings.getUsageStatsTopic());
} }
@Override
public TbQueueProducer<TbProtoQueueMsg<TransportProtos.ToVersionControlServiceMsg>> createVersionControlMsgProducer() {
//TODO: version-control
return null;
}
@PreDestroy @PreDestroy
private void destroy() { private void destroy() {
if (coreAdmin != null) { if (coreAdmin != null) {

7
common/queue/src/main/java/org/thingsboard/server/queue/provider/RabbitMqTbCoreQueueFactory.java

@ -21,6 +21,7 @@ import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component; import org.springframework.stereotype.Component;
import org.thingsboard.server.common.msg.queue.ServiceType; import org.thingsboard.server.common.msg.queue.ServiceType;
import org.thingsboard.server.gen.js.JsInvokeProtos; import org.thingsboard.server.gen.js.JsInvokeProtos;
import org.thingsboard.server.gen.transport.TransportProtos;
import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg;
import org.thingsboard.server.gen.transport.TransportProtos.ToCoreNotificationMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreNotificationMsg;
import org.thingsboard.server.gen.transport.TransportProtos.ToOtaPackageStateServiceMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToOtaPackageStateServiceMsg;
@ -169,6 +170,12 @@ public class RabbitMqTbCoreQueueFactory implements TbCoreQueueFactory {
return builder.build(); return builder.build();
} }
@Override
public TbQueueProducer<TbProtoQueueMsg<TransportProtos.ToVersionControlServiceMsg>> createVersionControlMsgProducer() {
//TODO: version-control
return null;
}
@Override @Override
public TbQueueConsumer<TbProtoQueueMsg<ToUsageStatsServiceMsg>> createToUsageStatsServiceMsgConsumer() { public TbQueueConsumer<TbProtoQueueMsg<ToUsageStatsServiceMsg>> createToUsageStatsServiceMsgConsumer() {
return new TbRabbitMqConsumerTemplate<>(coreAdmin, rabbitMqSettings, coreSettings.getUsageStatsTopic(), return new TbRabbitMqConsumerTemplate<>(coreAdmin, rabbitMqSettings, coreSettings.getUsageStatsTopic(),

7
common/queue/src/main/java/org/thingsboard/server/queue/provider/ServiceBusMonolithQueueFactory.java

@ -22,6 +22,7 @@ import org.springframework.stereotype.Component;
import org.thingsboard.server.common.data.queue.Queue; import org.thingsboard.server.common.data.queue.Queue;
import org.thingsboard.server.common.msg.queue.ServiceType; import org.thingsboard.server.common.msg.queue.ServiceType;
import org.thingsboard.server.gen.js.JsInvokeProtos; import org.thingsboard.server.gen.js.JsInvokeProtos;
import org.thingsboard.server.gen.transport.TransportProtos;
import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg;
import org.thingsboard.server.gen.transport.TransportProtos.ToCoreNotificationMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreNotificationMsg;
import org.thingsboard.server.gen.transport.TransportProtos.ToOtaPackageStateServiceMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToOtaPackageStateServiceMsg;
@ -204,6 +205,12 @@ public class ServiceBusMonolithQueueFactory implements TbCoreQueueFactory, TbRul
return new TbServiceBusProducerTemplate<>(coreAdmin, serviceBusSettings, coreSettings.getUsageStatsTopic()); return new TbServiceBusProducerTemplate<>(coreAdmin, serviceBusSettings, coreSettings.getUsageStatsTopic());
} }
@Override
public TbQueueProducer<TbProtoQueueMsg<TransportProtos.ToVersionControlServiceMsg>> createVersionControlMsgProducer() {
//TODO: version-control
return null;
}
@PreDestroy @PreDestroy
private void destroy() { private void destroy() {
if (coreAdmin != null) { if (coreAdmin != null) {

7
common/queue/src/main/java/org/thingsboard/server/queue/provider/ServiceBusTbCoreQueueFactory.java

@ -21,6 +21,7 @@ import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component; import org.springframework.stereotype.Component;
import org.thingsboard.server.common.msg.queue.ServiceType; import org.thingsboard.server.common.msg.queue.ServiceType;
import org.thingsboard.server.gen.js.JsInvokeProtos; import org.thingsboard.server.gen.js.JsInvokeProtos;
import org.thingsboard.server.gen.transport.TransportProtos;
import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg;
import org.thingsboard.server.gen.transport.TransportProtos.ToCoreNotificationMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreNotificationMsg;
import org.thingsboard.server.gen.transport.TransportProtos.ToOtaPackageStateServiceMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToOtaPackageStateServiceMsg;
@ -191,6 +192,12 @@ public class ServiceBusTbCoreQueueFactory implements TbCoreQueueFactory {
return new TbServiceBusProducerTemplate<>(coreAdmin, serviceBusSettings, coreSettings.getUsageStatsTopic()); return new TbServiceBusProducerTemplate<>(coreAdmin, serviceBusSettings, coreSettings.getUsageStatsTopic());
} }
@Override
public TbQueueProducer<TbProtoQueueMsg<TransportProtos.ToVersionControlServiceMsg>> createVersionControlMsgProducer() {
//TODO: version-control
return null;
}
@PreDestroy @PreDestroy
private void destroy() { private void destroy() {
if (coreAdmin != null) { if (coreAdmin != null) {

8
common/queue/src/main/java/org/thingsboard/server/queue/provider/TbCoreQueueFactory.java

@ -20,6 +20,7 @@ import org.thingsboard.server.gen.transport.TransportProtos.ToOtaPackageStateSer
import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg;
import org.thingsboard.server.gen.transport.TransportProtos.ToCoreNotificationMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreNotificationMsg;
import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg;
import org.thingsboard.server.gen.transport.TransportProtos.ToVersionControlServiceMsg;
import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineNotificationMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineNotificationMsg;
import org.thingsboard.server.gen.transport.TransportProtos.ToTransportMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToTransportMsg;
import org.thingsboard.server.gen.transport.TransportProtos.ToUsageStatsServiceMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToUsageStatsServiceMsg;
@ -122,4 +123,11 @@ public interface TbCoreQueueFactory extends TbUsageStatsClientQueueFactory {
TbQueueProducer<TbProtoQueueMsg<TransportApiResponseMsg>> createTransportApiResponseProducer(); TbQueueProducer<TbProtoQueueMsg<TransportApiResponseMsg>> createTransportApiResponseProducer();
TbQueueRequestTemplate<TbProtoJsQueueMsg<JsInvokeProtos.RemoteJsRequest>, TbProtoQueueMsg<JsInvokeProtos.RemoteJsResponse>> createRemoteJsRequestTemplate(); TbQueueRequestTemplate<TbProtoJsQueueMsg<JsInvokeProtos.RemoteJsRequest>, TbProtoQueueMsg<JsInvokeProtos.RemoteJsResponse>> createRemoteJsRequestTemplate();
/**
* Used to push messages to instances of TB Version Control Service
*
* @return
*/
TbQueueProducer<TbProtoQueueMsg<ToVersionControlServiceMsg>> createVersionControlMsgProducer();
} }

8
common/queue/src/main/java/org/thingsboard/server/queue/provider/TbCoreQueueProducerProvider.java

@ -22,6 +22,7 @@ import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg;
import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineNotificationMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineNotificationMsg;
import org.thingsboard.server.gen.transport.TransportProtos.ToTransportMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToTransportMsg;
import org.thingsboard.server.gen.transport.TransportProtos.ToUsageStatsServiceMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToUsageStatsServiceMsg;
import org.thingsboard.server.gen.transport.TransportProtos.ToVersionControlServiceMsg;
import org.thingsboard.server.queue.TbQueueProducer; import org.thingsboard.server.queue.TbQueueProducer;
import org.thingsboard.server.queue.common.TbProtoQueueMsg; import org.thingsboard.server.queue.common.TbProtoQueueMsg;
import org.thingsboard.server.queue.util.TbCoreComponent; import org.thingsboard.server.queue.util.TbCoreComponent;
@ -39,6 +40,7 @@ public class TbCoreQueueProducerProvider implements TbQueueProducerProvider {
private TbQueueProducer<TbProtoQueueMsg<ToRuleEngineNotificationMsg>> toRuleEngineNotifications; private TbQueueProducer<TbProtoQueueMsg<ToRuleEngineNotificationMsg>> toRuleEngineNotifications;
private TbQueueProducer<TbProtoQueueMsg<ToCoreNotificationMsg>> toTbCoreNotifications; private TbQueueProducer<TbProtoQueueMsg<ToCoreNotificationMsg>> toTbCoreNotifications;
private TbQueueProducer<TbProtoQueueMsg<ToUsageStatsServiceMsg>> toUsageStats; private TbQueueProducer<TbProtoQueueMsg<ToUsageStatsServiceMsg>> toUsageStats;
private TbQueueProducer<TbProtoQueueMsg<ToVersionControlServiceMsg>> toVersionControl;
public TbCoreQueueProducerProvider(TbCoreQueueFactory tbQueueProvider) { public TbCoreQueueProducerProvider(TbCoreQueueFactory tbQueueProvider) {
this.tbQueueProvider = tbQueueProvider; this.tbQueueProvider = tbQueueProvider;
@ -52,6 +54,7 @@ public class TbCoreQueueProducerProvider implements TbQueueProducerProvider {
this.toRuleEngineNotifications = tbQueueProvider.createRuleEngineNotificationsMsgProducer(); this.toRuleEngineNotifications = tbQueueProvider.createRuleEngineNotificationsMsgProducer();
this.toTbCoreNotifications = tbQueueProvider.createTbCoreNotificationsMsgProducer(); this.toTbCoreNotifications = tbQueueProvider.createTbCoreNotificationsMsgProducer();
this.toUsageStats = tbQueueProvider.createToUsageStatsServiceMsgProducer(); this.toUsageStats = tbQueueProvider.createToUsageStatsServiceMsgProducer();
this.toVersionControl = tbQueueProvider.createVersionControlMsgProducer();
} }
@Override @Override
@ -83,4 +86,9 @@ public class TbCoreQueueProducerProvider implements TbQueueProducerProvider {
public TbQueueProducer<TbProtoQueueMsg<ToUsageStatsServiceMsg>> getTbUsageStatsMsgProducer() { public TbQueueProducer<TbProtoQueueMsg<ToUsageStatsServiceMsg>> getTbUsageStatsMsgProducer() {
return toUsageStats; return toUsageStats;
} }
@Override
public TbQueueProducer<TbProtoQueueMsg<ToVersionControlServiceMsg>> getTbVersionControlMsgProducer() {
return toVersionControl;
}
} }

8
common/queue/src/main/java/org/thingsboard/server/queue/provider/TbQueueProducerProvider.java

@ -21,6 +21,7 @@ import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg;
import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineNotificationMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineNotificationMsg;
import org.thingsboard.server.gen.transport.TransportProtos.ToTransportMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToTransportMsg;
import org.thingsboard.server.gen.transport.TransportProtos.ToUsageStatsServiceMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToUsageStatsServiceMsg;
import org.thingsboard.server.gen.transport.TransportProtos.ToVersionControlServiceMsg;
import org.thingsboard.server.queue.TbQueueProducer; import org.thingsboard.server.queue.TbQueueProducer;
import org.thingsboard.server.queue.common.TbProtoQueueMsg; import org.thingsboard.server.queue.common.TbProtoQueueMsg;
@ -70,4 +71,11 @@ public interface TbQueueProducerProvider {
* @return * @return
*/ */
TbQueueProducer<TbProtoQueueMsg<ToUsageStatsServiceMsg>> getTbUsageStatsMsgProducer(); TbQueueProducer<TbProtoQueueMsg<ToUsageStatsServiceMsg>> getTbUsageStatsMsgProducer();
/**
* Used to push messages to other instances of TB Core Service
*
* @return
*/
TbQueueProducer<TbProtoQueueMsg<ToVersionControlServiceMsg>> getTbVersionControlMsgProducer();
} }

6
common/queue/src/main/java/org/thingsboard/server/queue/provider/TbRuleEngineProducerProvider.java

@ -17,6 +17,7 @@ package org.thingsboard.server.queue.provider;
import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression; import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
import org.springframework.stereotype.Service; import org.springframework.stereotype.Service;
import org.thingsboard.server.gen.transport.TransportProtos;
import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg;
import org.thingsboard.server.gen.transport.TransportProtos.ToCoreNotificationMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreNotificationMsg;
import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg;
@ -83,4 +84,9 @@ public class TbRuleEngineProducerProvider implements TbQueueProducerProvider {
public TbQueueProducer<TbProtoQueueMsg<ToUsageStatsServiceMsg>> getTbUsageStatsMsgProducer() { public TbQueueProducer<TbProtoQueueMsg<ToUsageStatsServiceMsg>> getTbUsageStatsMsgProducer() {
return toUsageStats; return toUsageStats;
} }
@Override
public TbQueueProducer<TbProtoQueueMsg<TransportProtos.ToVersionControlServiceMsg>> getTbVersionControlMsgProducer() {
throw new RuntimeException("Not Implemented! Should not be used by Rule Engine!");
}
} }

6
common/queue/src/main/java/org/thingsboard/server/queue/provider/TbTransportQueueProducerProvider.java

@ -17,6 +17,7 @@ package org.thingsboard.server.queue.provider;
import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression; import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
import org.springframework.stereotype.Service; import org.springframework.stereotype.Service;
import org.thingsboard.server.gen.transport.TransportProtos;
import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg;
import org.thingsboard.server.gen.transport.TransportProtos.ToCoreNotificationMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreNotificationMsg;
import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg;
@ -73,6 +74,11 @@ public class TbTransportQueueProducerProvider implements TbQueueProducerProvider
throw new RuntimeException("Not Implemented! Should not be used by Transport!"); throw new RuntimeException("Not Implemented! Should not be used by Transport!");
} }
@Override
public TbQueueProducer<TbProtoQueueMsg<TransportProtos.ToVersionControlServiceMsg>> getTbVersionControlMsgProducer() {
throw new RuntimeException("Not Implemented! Should not be used by Transport!");
}
@Override @Override
public TbQueueProducer<TbProtoQueueMsg<ToUsageStatsServiceMsg>> getTbUsageStatsMsgProducer() { public TbQueueProducer<TbProtoQueueMsg<ToUsageStatsServiceMsg>> getTbUsageStatsMsgProducer() {
return toUsageStats; return toUsageStats;

84
common/queue/src/main/java/org/thingsboard/server/queue/provider/TbVersionControlProducerProvider.java

@ -0,0 +1,84 @@
/**
* 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.queue.provider;
import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
import org.springframework.stereotype.Service;
import org.thingsboard.server.gen.transport.TransportProtos;
import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg;
import org.thingsboard.server.gen.transport.TransportProtos.ToCoreNotificationMsg;
import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg;
import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineNotificationMsg;
import org.thingsboard.server.gen.transport.TransportProtos.ToTransportMsg;
import org.thingsboard.server.gen.transport.TransportProtos.ToUsageStatsServiceMsg;
import org.thingsboard.server.queue.TbQueueProducer;
import org.thingsboard.server.queue.common.TbProtoQueueMsg;
import javax.annotation.PostConstruct;
@Service
@ConditionalOnExpression("'${service.type:null}'=='tb-vc-executor'")
public class TbVersionControlProducerProvider implements TbQueueProducerProvider {
private final TbVersionControlQueueFactory tbQueueProvider;
private TbQueueProducer<TbProtoQueueMsg<ToCoreNotificationMsg>> toTbCoreNotifications;
private TbQueueProducer<TbProtoQueueMsg<ToUsageStatsServiceMsg>> toUsageStats;
public TbVersionControlProducerProvider(TbVersionControlQueueFactory tbQueueProvider) {
this.tbQueueProvider = tbQueueProvider;
}
@PostConstruct
public void init() {
this.toTbCoreNotifications = tbQueueProvider.createTbCoreNotificationsMsgProducer();
this.toUsageStats = tbQueueProvider.createToUsageStatsServiceMsgProducer();
}
@Override
public TbQueueProducer<TbProtoQueueMsg<ToTransportMsg>> getTransportNotificationsMsgProducer() {
throw new RuntimeException("Not Implemented! Should not be used by Version Control Service!");
}
@Override
public TbQueueProducer<TbProtoQueueMsg<ToRuleEngineMsg>> getRuleEngineMsgProducer() {
throw new RuntimeException("Not Implemented! Should not be used by Version Control Service!");
}
@Override
public TbQueueProducer<TbProtoQueueMsg<ToCoreMsg>> getTbCoreMsgProducer() {
throw new RuntimeException("Not Implemented! Should not be used by Version Control Service!");
}
@Override
public TbQueueProducer<TbProtoQueueMsg<ToRuleEngineNotificationMsg>> getRuleEngineNotificationsMsgProducer() {
throw new RuntimeException("Not Implemented! Should not be used by Version Control Service!");
}
@Override
public TbQueueProducer<TbProtoQueueMsg<ToCoreNotificationMsg>> getTbCoreNotificationsMsgProducer() {
return toTbCoreNotifications;
}
@Override
public TbQueueProducer<TbProtoQueueMsg<TransportProtos.ToVersionControlServiceMsg>> getTbVersionControlMsgProducer() {
throw new RuntimeException("Not Implemented! Should not be used by Version Control Service!");
}
@Override
public TbQueueProducer<TbProtoQueueMsg<ToUsageStatsServiceMsg>> getTbUsageStatsMsgProducer() {
return toUsageStats;
}
}

44
common/queue/src/main/java/org/thingsboard/server/queue/provider/TbVersionControlQueueFactory.java

@ -0,0 +1,44 @@
/**
* 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.queue.provider;
import org.thingsboard.server.gen.transport.TransportProtos.ToCoreNotificationMsg;
import org.thingsboard.server.gen.transport.TransportProtos.ToVersionControlServiceMsg;
import org.thingsboard.server.queue.TbQueueConsumer;
import org.thingsboard.server.queue.TbQueueProducer;
import org.thingsboard.server.queue.common.TbProtoQueueMsg;
/**
* Responsible for initialization of various Producers and Consumers used by TB Version Control Node.
* Implementation Depends on the queue queue.type from yml or TB_QUEUE_TYPE environment variable
*/
public interface TbVersionControlQueueFactory extends TbUsageStatsClientQueueFactory {
/**
* Used to push notifications to other instances of TB Core Service
*
* @return
*/
TbQueueProducer<TbProtoQueueMsg<ToCoreNotificationMsg>> createTbCoreNotificationsMsgProducer();
/**
* Used to consume messages from TB Core Service
*
* @return
*/
TbQueueConsumer<TbProtoQueueMsg<ToVersionControlServiceMsg>> createToVersionControlMsgConsumer();
}

34
common/queue/src/main/java/org/thingsboard/server/queue/settings/TbQueueVersionControlSettings.java

@ -0,0 +1,34 @@
/**
* 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.queue.settings;
import lombok.Data;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
@Data
@Component
public class TbQueueVersionControlSettings {
@Value("${queue.vc.topic:tb_version_control}")
private String topic;
@Value("${queue.vc.usage-stats-topic:tb_usage_stats}")
private String usageStatsTopic;
@Value("${queue.vc.partitions:10}")
private int partitions;
}

26
common/queue/src/main/java/org/thingsboard/server/queue/util/TbVersionControlComponent.java

@ -0,0 +1,26 @@
/**
* 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.queue.util;
import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
@Retention(RetentionPolicy.RUNTIME)
@ConditionalOnExpression("'${service.type:null}'=='monolith' || '${service.type:null}'=='tb-vc-executor'")
public @interface TbVersionControlComponent {
}

24
common/version-control/pom.xml

@ -86,21 +86,6 @@
<groupId>org.eclipse.jgit</groupId> <groupId>org.eclipse.jgit</groupId>
<artifactId>org.eclipse.jgit</artifactId> <artifactId>org.eclipse.jgit</artifactId>
</dependency> </dependency>
<dependency>
<groupId>io.grpc</groupId>
<artifactId>grpc-netty-shaded</artifactId>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>io.grpc</groupId>
<artifactId>grpc-protobuf</artifactId>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>io.grpc</groupId>
<artifactId>grpc-stub</artifactId>
<scope>provided</scope>
</dependency>
<dependency> <dependency>
<groupId>org.eclipse.jgit</groupId> <groupId>org.eclipse.jgit</groupId>
<artifactId>org.eclipse.jgit.ssh.apache</artifactId> <artifactId>org.eclipse.jgit.ssh.apache</artifactId>
@ -126,15 +111,6 @@
</dependency> </dependency>
</dependencies> </dependencies>
<build>
<plugins>
<plugin>
<groupId>org.xolstice.maven.plugins</groupId>
<artifactId>protobuf-maven-plugin</artifactId>
</plugin>
</plugins>
</build>
<distributionManagement> <distributionManagement>
<repository> <repository>
<id>thingsboard-repo-deploy</id> <id>thingsboard-repo-deploy</id>

11
common/version-control/src/main/java/org/thingsboard/server/service/sync/vc/GitVersionControlService.java

@ -16,15 +16,12 @@
package org.thingsboard.server.service.sync.vc; package org.thingsboard.server.service.sync.vc;
import org.thingsboard.server.common.data.EntityType; import org.thingsboard.server.common.data.EntityType;
import org.thingsboard.server.common.data.ExportableEntity;
import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.common.data.id.EntityId;
import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.data.sync.ie.EntityExportData; import org.thingsboard.server.common.data.sync.ie.EntityExportData;
import org.thingsboard.server.common.data.sync.vc.EntitiesVersionControlSettings; import org.thingsboard.server.common.data.sync.vc.EntitiesVersionControlSettings;
import org.thingsboard.server.common.data.sync.vc.EntityVersion; import org.thingsboard.server.common.data.sync.vc.EntityVersion;
import org.thingsboard.server.common.data.sync.vc.VersionCreationResult;
import org.thingsboard.server.common.data.sync.vc.VersionedEntityInfo; import org.thingsboard.server.common.data.sync.vc.VersionedEntityInfo;
import org.thingsboard.server.common.data.sync.vc.request.create.VersionCreateRequest;
import java.util.List; import java.util.List;
@ -36,14 +33,6 @@ public interface GitVersionControlService {
void clearRepository(TenantId tenantId); void clearRepository(TenantId tenantId);
PendingCommit prepareCommit(TenantId tenantId, VersionCreateRequest request);
void addToCommit(PendingCommit commit, EntityExportData<ExportableEntity<EntityId>> entityData);
void deleteAll(PendingCommit pendingCommit, EntityType entityType);
VersionCreationResult push(PendingCommit commit);
List<EntityVersion> listVersions(TenantId tenantId, String branch); List<EntityVersion> listVersions(TenantId tenantId, String branch);
List<EntityVersion> listVersions(TenantId tenantId, String branch, EntityType entityType); List<EntityVersion> listVersions(TenantId tenantId, String branch, EntityType entityType);

71
common/version-control/src/main/proto/vc.proto

@ -1,71 +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.
*/
syntax = "proto3";
option java_package = "org.thingsboard.server.gen.vc.v1";
option java_multiple_files = true;
option java_outer_classname = "EdgeProtos";
package vc;
// Interface exported by the ThingsBoard Core.
service TbGitRpcService {
rpc commit(stream CommitRequestMsg) returns (CommitResponseMsg) {}
}
/**
* Data Structures;
*/
message CommitRequestMsg {
string txId = 1;
PrepareMsg prepareMsg = 2;
AddMsg addMsg = 3;
DeleteMsg deleteMsg = 4;
PushMsg pushMsg = 5;
AbortMsg abortMsg = 6;
}
message CommitResponseMsg {
string id = 1;
string name = 2;
int32 added = 3;
int32 modified = 4;
int32 removed = 5;
}
message PrepareMsg {
string tenantId = 1;
string commitMsg = 2;
string branchName = 3;
}
message AddMsg {
string relativePath = 1;
string entityDataJson = 2;
}
message DeleteMsg {
string relativePath = 1;
}
message PushMsg {
}
message AbortMsg {
}
Loading…
Cancel
Save