简柏林 / thingskit · Commits

Commit 6ea39e835e3203bfb4bcc6d6e6f7cf7e4552d16d

Authored by Andrii Shvaika
1 parent 356d4ff2

Transport Rate Limits are now configurable via Tenant Profile

Inline Side-by-side
Showing 30 changed files with 785 additions and 211 deletions
... ... @@ -94,7 +94,8 @@
94 94 "name": "Device Profile Node",
95 95 "debugMode": false,
96 96 "configuration": {
97   - "persistAlarmRulesState": false
  97 + "persistAlarmRulesState": false,
  98 + "fetchAlarmRulesStateOnStart": false
98 99 }
99 100 }
100 101 ],
... ...
... ... @@ -92,6 +92,7 @@ public class TenantController extends BaseController {
92 92 installScripts.createDefaultRuleChains(tenant.getId());
93 93 }
94 94 tenantProfileCache.evict(tenant.getId());
  95 + tbClusterService.onTenantChange(tenant, null);
95 96 return tenant;
96 97 } catch (Exception e) {
97 98 throw handleException(e);
... ... @@ -105,9 +106,10 @@ public class TenantController extends BaseController {
105 106 checkParameter("tenantId", strTenantId);
106 107 try {
107 108 TenantId tenantId = new TenantId(toUUID(strTenantId));
108   - checkTenantId(tenantId, Operation.DELETE);
  109 + Tenant tenant = checkTenantId(tenantId, Operation.DELETE);
109 110 tenantService.deleteTenant(tenantId);
110 111 tenantProfileCache.evict(tenantId);
  112 + tbClusterService.onTenantDelete(tenant, null);
111 113 tbClusterService.onEntityStateChange(tenantId, tenantId, ComponentLifecycleEvent.DELETED);
112 114 } catch (Exception e) {
113 115 throw handleException(e);
... ...
... ... @@ -34,6 +34,7 @@ import org.thingsboard.server.common.data.id.TenantProfileId;
34 34 import org.thingsboard.server.common.data.page.PageData;
35 35 import org.thingsboard.server.common.data.page.PageLink;
36 36 import org.thingsboard.server.common.data.plugin.ComponentLifecycleEvent;
  37 +import org.thingsboard.server.dao.exception.DataValidationException;
37 38 import org.thingsboard.server.queue.util.TbCoreComponent;
38 39 import org.thingsboard.server.service.security.permission.Operation;
39 40 import org.thingsboard.server.service.security.permission.Resource;
... ... @@ -96,6 +97,7 @@ public class TenantProfileController extends BaseController {
96 97
97 98 tenantProfile = checkNotNull(tenantProfileService.saveTenantProfile(getTenantId(), tenantProfile));
98 99 tenantProfileCache.put(tenantProfile);
  100 + tbClusterService.onTenantProfileChange(tenantProfile, null);
99 101 tbClusterService.onEntityStateChange(TenantId.SYS_TENANT_ID, tenantProfile.getId(),
100 102 newTenantProfile ? ComponentLifecycleEvent.CREATED : ComponentLifecycleEvent.UPDATED);
101 103 return tenantProfile;
... ... @@ -111,8 +113,9 @@ public class TenantProfileController extends BaseController {
111 113 checkParameter("tenantProfileId", strTenantProfileId);
112 114 try {
113 115 TenantProfileId tenantProfileId = new TenantProfileId(toUUID(strTenantProfileId));
114   - checkTenantProfileId(tenantProfileId, Operation.DELETE);
  116 + TenantProfile profile = checkTenantProfileId(tenantProfileId, Operation.DELETE);
115 117 tenantProfileService.deleteTenantProfile(getTenantId(), tenantProfileId);
  118 + tbClusterService.onTenantProfileDelete(profile, null);
116 119 } catch (Exception e) {
117 120 throw handleException(e);
118 121 }
... ...
... ... @@ -61,7 +61,7 @@ public class DefaultTbDeviceProfileCache implements TbDeviceProfileCache {
61 61 profile = deviceProfileService.findDeviceProfileById(tenantId, deviceProfileId);
62 62 if (profile != null) {
63 63 deviceProfilesMap.put(deviceProfileId, profile);
64   - log.info("[{}] Fetch device profile into cache: {}", profile.getId(), profile);
  64 + log.debug("[{}] Fetch device profile into cache: {}", profile.getId(), profile);
65 65 }
66 66 }
67 67 } finally {
... ... @@ -91,7 +91,7 @@ public class DefaultTbDeviceProfileCache implements TbDeviceProfileCache {
91 91 public void put(DeviceProfile profile) {
92 92 if (profile.getId() != null) {
93 93 deviceProfilesMap.put(profile.getId(), profile);
94   - log.info("[{}] pushed device profile to cache: {}", profile.getId(), profile);
  94 + log.debug("[{}] pushed device profile to cache: {}", profile.getId(), profile);
95 95 notifyListeners(profile);
96 96 }
97 97 }
... ... @@ -99,7 +99,7 @@ public class DefaultTbDeviceProfileCache implements TbDeviceProfileCache {
99 99 @Override
100 100 public void evict(TenantId tenantId, DeviceProfileId profileId) {
101 101 DeviceProfile oldProfile = deviceProfilesMap.remove(profileId);
102   - log.info("[{}] evict device profile from cache: {}", profileId, oldProfile);
  102 + log.debug("[{}] evict device profile from cache: {}", profileId, oldProfile);
103 103 DeviceProfile newProfile = get(tenantId, profileId);
104 104 if (newProfile != null) {
105 105 notifyListeners(newProfile);
... ... @@ -117,6 +117,16 @@ public class DefaultTbDeviceProfileCache implements TbDeviceProfileCache {
117 117 }
118 118
119 119 @Override
  120 + public DeviceProfile find(DeviceProfileId deviceProfileId) {
  121 + return deviceProfileService.findDeviceProfileById(TenantId.SYS_TENANT_ID, deviceProfileId);
  122 + }
  123 +
  124 + @Override
  125 + public DeviceProfile findOrCreateDeviceProfile(TenantId tenantId, String profileName) {
  126 + return deviceProfileService.findOrCreateDeviceProfile(tenantId, profileName);
  127 + }
  128 +
  129 + @Override
120 130 public void removeListener(TenantId tenantId, EntityId listenerId) {
121 131 ConcurrentMap<EntityId, Consumer<DeviceProfile>> tenantListeners = listeners.get(tenantId);
122 132 if (tenantListeners != null) {
... ...
... ... @@ -29,4 +29,7 @@ public interface TbDeviceProfileCache extends RuleEngineDeviceProfileCache {
29 29
30 30 void evict(DeviceId id);
31 31
  32 + DeviceProfile find(DeviceProfileId deviceProfileId);
  33 +
  34 + DeviceProfile findOrCreateDeviceProfile(TenantId tenantId, String deviceType);
32 35 }
... ...
... ... @@ -23,11 +23,15 @@ import org.springframework.stereotype.Service;
23 23 import org.thingsboard.rule.engine.api.msg.ToDeviceActorNotificationMsg;
24 24 import org.thingsboard.server.common.data.DeviceProfile;
25 25 import org.thingsboard.server.common.data.EntityType;
  26 +import org.thingsboard.server.common.data.HasName;
  27 +import org.thingsboard.server.common.data.Tenant;
  28 +import org.thingsboard.server.common.data.TenantProfile;
26 29 import org.thingsboard.server.common.data.id.DeviceId;
27 30 import org.thingsboard.server.common.data.id.DeviceProfileId;
28 31 import org.thingsboard.server.common.data.id.EntityId;
29 32 import org.thingsboard.server.common.data.id.RuleChainId;
30 33 import org.thingsboard.server.common.data.id.TenantId;
  34 +import org.thingsboard.server.common.data.id.TenantProfileId;
31 35 import org.thingsboard.server.common.data.plugin.ComponentLifecycleEvent;
32 36 import org.thingsboard.server.common.msg.TbMsg;
33 37 import org.thingsboard.server.common.msg.plugin.ComponentLifecycleMsg;
... ... @@ -189,21 +193,51 @@ public class DefaultTbClusterService implements TbClusterService {
189 193
190 194 @Override
191 195 public void onDeviceProfileChange(DeviceProfile deviceProfile, TbQueueCallback callback) {
192   - log.trace("[{}][{}] Processing device profile [{}] change event", deviceProfile.getTenantId(), deviceProfile.getId(), deviceProfile.getName());
193   - TransportProtos.DeviceProfileUpdateMsg profileUpdateMsg = TransportProtos.DeviceProfileUpdateMsg.newBuilder()
194   - .setData(ByteString.copyFrom(encodingService.encode(deviceProfile))).build();
195   - ToTransportMsg transportMsg = ToTransportMsg.newBuilder().setDeviceProfileUpdateMsg(profileUpdateMsg).build();
196   - broadcast(transportMsg);
  196 + onEntityChange(deviceProfile.getTenantId(), deviceProfile.getId(), deviceProfile, callback);
  197 + }
  198 +
  199 + @Override
  200 + public void onTenantProfileChange(TenantProfile tenantProfile, TbQueueCallback callback) {
  201 + onEntityChange(TenantId.SYS_TENANT_ID, tenantProfile.getId(), tenantProfile, callback);
  202 + }
  203 +
  204 + @Override
  205 + public void onTenantChange(Tenant tenant, TbQueueCallback callback) {
  206 + onEntityChange(TenantId.SYS_TENANT_ID, tenant.getId(), tenant, callback);
  207 + }
  208 +
  209 + @Override
  210 + public void onDeviceProfileDelete(DeviceProfile entity, TbQueueCallback callback) {
  211 + onEntityDelete(entity.getTenantId(), entity.getId(), entity.getName(), callback);
197 212 }
198 213
199 214 @Override
200   - public void onDeviceProfileDelete(DeviceProfile deviceProfile, TbQueueCallback callback) {
201   - log.trace("[{}][{}] Processing device profile [{}] delete event", deviceProfile.getTenantId(), deviceProfile.getId(), deviceProfile.getName());
202   - TransportProtos.DeviceProfileDeleteMsg profileDeleteMsg = TransportProtos.DeviceProfileDeleteMsg.newBuilder()
203   - .setProfileIdMSB(deviceProfile.getId().getId().getMostSignificantBits())
204   - .setProfileIdLSB(deviceProfile.getId().getId().getLeastSignificantBits())
  215 + public void onTenantProfileDelete(TenantProfile entity, TbQueueCallback callback) {
  216 + onEntityDelete(TenantId.SYS_TENANT_ID, entity.getId(), entity.getName(), callback);
  217 + }
  218 +
  219 + @Override
  220 + public void onTenantDelete(Tenant entity, TbQueueCallback callback) {
  221 + onEntityDelete(TenantId.SYS_TENANT_ID, entity.getId(), entity.getName(), callback);
  222 + }
  223 +
  224 + public <T extends HasName> void onEntityChange(TenantId tenantId, EntityId entityid, T entity, TbQueueCallback callback) {
  225 + log.trace("[{}][{}][{}] Processing [{}] change event", tenantId, entityid.getEntityType(), entityid.getId(), entity.getName());
  226 + TransportProtos.EntityUpdateMsg entityUpdateMsg = TransportProtos.EntityUpdateMsg.newBuilder()
  227 + .setEntityType(entityid.getEntityType().name())
  228 + .setData(ByteString.copyFrom(encodingService.encode(entity))).build();
  229 + ToTransportMsg transportMsg = ToTransportMsg.newBuilder().setEntityUpdateMsg(entityUpdateMsg).build();
  230 + broadcast(transportMsg);
  231 + }
  232 +
  233 + private void onEntityDelete(TenantId tenantId, EntityId entityId, String name, TbQueueCallback callback) {
  234 + log.trace("[{}][{}][{}] Processing [{}] delete event", tenantId, entityId.getEntityType(), entityId.getId(), name);
  235 + TransportProtos.EntityDeleteMsg entityDeleteMsg = TransportProtos.EntityDeleteMsg.newBuilder()
  236 + .setEntityType(entityId.getEntityType().name())
  237 + .setEntityIdMSB(entityId.getId().getMostSignificantBits())
  238 + .setEntityIdLSB(entityId.getId().getLeastSignificantBits())
205 239 .build();
206   - ToTransportMsg transportMsg = ToTransportMsg.newBuilder().setDeviceProfileDeleteMsg(profileDeleteMsg).build();
  240 + ToTransportMsg transportMsg = ToTransportMsg.newBuilder().setEntityDeleteMsg(entityDeleteMsg).build();
207 241 broadcast(transportMsg);
208 242 }
209 243
... ...
... ... @@ -17,7 +17,8 @@ package org.thingsboard.server.service.queue;
17 17
18 18 import org.thingsboard.rule.engine.api.msg.ToDeviceActorNotificationMsg;
19 19 import org.thingsboard.server.common.data.DeviceProfile;
20   -import org.thingsboard.server.common.data.id.DeviceProfileId;
  20 +import org.thingsboard.server.common.data.Tenant;
  21 +import org.thingsboard.server.common.data.TenantProfile;
21 22 import org.thingsboard.server.common.data.id.EntityId;
22 23 import org.thingsboard.server.common.data.id.TenantId;
23 24 import org.thingsboard.server.common.data.plugin.ComponentLifecycleEvent;
... ... @@ -53,5 +54,13 @@ public interface TbClusterService {
53 54
54 55 void onDeviceProfileChange(DeviceProfile deviceProfile, TbQueueCallback callback);
55 56
56   - void onDeviceProfileDelete(DeviceProfile deviceProfileId, TbQueueCallback callback);
  57 + void onDeviceProfileDelete(DeviceProfile deviceProfile, TbQueueCallback callback);
  58 +
  59 + void onTenantProfileChange(TenantProfile tenantProfile, TbQueueCallback callback);
  60 +
  61 + void onTenantProfileDelete(TenantProfile tenantProfile, TbQueueCallback callback);
  62 +
  63 + void onTenantChange(Tenant tenant, TbQueueCallback callback);
  64 +
  65 + void onTenantDelete(Tenant tenant, TbQueueCallback callback);
57 66 }
... ...
... ... @@ -28,6 +28,7 @@ import org.springframework.util.StringUtils;
28 28 import org.thingsboard.server.common.data.DataConstants;
29 29 import org.thingsboard.server.common.data.Device;
30 30 import org.thingsboard.server.common.data.DeviceProfile;
  31 +import org.thingsboard.server.common.data.EntityType;
31 32 import org.thingsboard.server.common.data.TenantProfile;
32 33 import org.thingsboard.server.common.data.device.credentials.BasicMqttCredentials;
33 34 import org.thingsboard.server.common.data.device.credentials.ProvisionDeviceCredentialsData;
... ... @@ -45,21 +46,19 @@ import org.thingsboard.server.common.msg.TbMsgDataType;
45 46 import org.thingsboard.server.common.msg.TbMsgMetaData;
46 47 import org.thingsboard.server.common.transport.util.DataDecodingEncodingService;
47 48 import org.thingsboard.server.dao.device.DeviceCredentialsService;
48   -import org.thingsboard.server.dao.device.DeviceProfileService;
49 49 import org.thingsboard.server.dao.device.DeviceProvisionService;
50 50 import org.thingsboard.server.dao.device.DeviceService;
51 51 import org.thingsboard.server.dao.device.provision.ProvisionRequest;
52 52 import org.thingsboard.server.dao.device.provision.ProvisionResponse;
53 53 import org.thingsboard.server.dao.relation.RelationService;
54   -import org.thingsboard.server.dao.tenant.TenantProfileService;
55 54 import org.thingsboard.server.dao.tenant.TenantService;
56 55 import org.thingsboard.server.dao.util.mapping.JacksonUtil;
57 56 import org.thingsboard.server.gen.transport.TransportProtos;
58 57 import org.thingsboard.server.gen.transport.TransportProtos.DeviceInfoProto;
59 58 import org.thingsboard.server.gen.transport.TransportProtos.GetOrCreateDeviceFromGatewayRequestMsg;
60 59 import org.thingsboard.server.gen.transport.TransportProtos.GetOrCreateDeviceFromGatewayResponseMsg;
61   -import org.thingsboard.server.gen.transport.TransportProtos.GetTenantRoutingInfoRequestMsg;
62   -import org.thingsboard.server.gen.transport.TransportProtos.GetTenantRoutingInfoResponseMsg;
  60 +import org.thingsboard.server.gen.transport.TransportProtos.GetEntityProfileRequestMsg;
  61 +import org.thingsboard.server.gen.transport.TransportProtos.GetEntityProfileResponseMsg;
63 62 import org.thingsboard.server.gen.transport.TransportProtos.ProvisionDeviceRequestMsg;
64 63 import org.thingsboard.server.gen.transport.TransportProtos.TransportApiRequestMsg;
65 64 import org.thingsboard.server.gen.transport.TransportProtos.TransportApiResponseMsg;
... ... @@ -70,6 +69,7 @@ import org.thingsboard.server.queue.common.TbProtoQueueMsg;
70 69 import org.thingsboard.server.queue.util.TbCoreComponent;
71 70 import org.thingsboard.server.dao.device.provision.ProvisionFailedException;
72 71 import org.thingsboard.server.service.executors.DbCallbackExecutorService;
  72 +import org.thingsboard.server.service.profile.TbDeviceProfileCache;
73 73 import org.thingsboard.server.service.profile.TbTenantProfileCache;
74 74 import org.thingsboard.server.service.queue.TbClusterService;
75 75 import org.thingsboard.server.service.state.DeviceStateService;
... ... @@ -90,9 +90,7 @@ public class DefaultTransportApiService implements TransportApiService {
90 90
91 91 private static final ObjectMapper mapper = new ObjectMapper();
92 92
93   - //TODO: Constructor dependencies;
94   - private final DeviceProfileService deviceProfileService;
95   - private final TenantService tenantService;
  93 + private final TbDeviceProfileCache deviceProfileCache;
96 94 private final TbTenantProfileCache tenantProfileCache;
97 95 private final DeviceService deviceService;
98 96 private final RelationService relationService;
... ... @@ -103,17 +101,15 @@ public class DefaultTransportApiService implements TransportApiService {
103 101 private final DataDecodingEncodingService dataDecodingEncodingService;
104 102 private final DeviceProvisionService deviceProvisionService;
105 103
106   -
107 104 private final ConcurrentMap<String, ReentrantLock> deviceCreationLocks = new ConcurrentHashMap<>();
108 105
109   - public DefaultTransportApiService(DeviceProfileService deviceProfileService, TenantService tenantService,
  106 + public DefaultTransportApiService(TbDeviceProfileCache deviceProfileCache,
110 107 TbTenantProfileCache tenantProfileCache, DeviceService deviceService,
111 108 RelationService relationService, DeviceCredentialsService deviceCredentialsService,
112 109 DeviceStateService deviceStateService, DbCallbackExecutorService dbCallbackExecutorService,
113 110 TbClusterService tbClusterService, DataDecodingEncodingService dataDecodingEncodingService,
114 111 DeviceProvisionService deviceProvisionService) {
115   - this.deviceProfileService = deviceProfileService;
116   - this.tenantService = tenantService;
  112 + this.deviceProfileCache = deviceProfileCache;
117 113 this.tenantProfileCache = tenantProfileCache;
118 114 this.deviceService = deviceService;
119 115 this.relationService = relationService;
... ... @@ -143,11 +139,8 @@ public class DefaultTransportApiService implements TransportApiService {
143 139 } else if (transportApiRequestMsg.hasGetOrCreateDeviceRequestMsg()) {
144 140 return Futures.transform(handle(transportApiRequestMsg.getGetOrCreateDeviceRequestMsg()),
145 141 value -> new TbProtoQueueMsg<>(tbProtoQueueMsg.getKey(), value, tbProtoQueueMsg.getHeaders()), MoreExecutors.directExecutor());
146   - } else if (transportApiRequestMsg.hasGetTenantRoutingInfoRequestMsg()) {
147   - return Futures.transform(handle(transportApiRequestMsg.getGetTenantRoutingInfoRequestMsg()),
148   - value -> new TbProtoQueueMsg<>(tbProtoQueueMsg.getKey(), value, tbProtoQueueMsg.getHeaders()), MoreExecutors.directExecutor());
149   - } else if (transportApiRequestMsg.hasGetDeviceProfileRequestMsg()) {
150   - return Futures.transform(handle(transportApiRequestMsg.getGetDeviceProfileRequestMsg()),
  142 + } else if (transportApiRequestMsg.hasEntityProfileRequestMsg()) {
  143 + return Futures.transform(handle(transportApiRequestMsg.getEntityProfileRequestMsg()),
151 144 value -> new TbProtoQueueMsg<>(tbProtoQueueMsg.getKey(), value, tbProtoQueueMsg.getHeaders()), MoreExecutors.directExecutor());
152 145 } else if (transportApiRequestMsg.hasProvisionDeviceRequestMsg()) {
153 146 return Futures.transform(handle(transportApiRequestMsg.getProvisionDeviceRequestMsg()),
... ... @@ -238,7 +231,7 @@ public class DefaultTransportApiService implements TransportApiService {
238 231 device.setName(requestMsg.getDeviceName());
239 232 device.setType(requestMsg.getDeviceType());
240 233 device.setCustomerId(gateway.getCustomerId());
241   - DeviceProfile deviceProfile = deviceProfileService.findOrCreateDeviceProfile(gateway.getTenantId(), requestMsg.getDeviceType());
  234 + DeviceProfile deviceProfile = deviceProfileCache.findOrCreateDeviceProfile(gateway.getTenantId(), requestMsg.getDeviceType());
242 235 device.setDeviceProfileId(deviceProfile.getId());
243 236 device = deviceService.saveDevice(device);
244 237 relationService.saveRelationAsync(TenantId.SYS_TENANT_ID, new EntityRelation(gateway.getId(), device.getId(), "Created"));
... ... @@ -258,7 +251,7 @@ public class DefaultTransportApiService implements TransportApiService {
258 251 }
259 252 GetOrCreateDeviceFromGatewayResponseMsg.Builder builder = GetOrCreateDeviceFromGatewayResponseMsg.newBuilder()
260 253 .setDeviceInfo(getDeviceInfoProto(device));
261   - DeviceProfile deviceProfile = deviceProfileService.findDeviceProfileById(device.getTenantId(), device.getDeviceProfileId());
  254 + DeviceProfile deviceProfile = deviceProfileCache.get(device.getTenantId(), device.getDeviceProfileId());
262 255 if (deviceProfile != null) {
263 256 builder.setProfileBody(ByteString.copyFrom(dataDecodingEncodingService.encode(deviceProfile)));
264 257 } else {
... ... @@ -320,23 +313,21 @@ public class DefaultTransportApiService implements TransportApiService {
320 313 .build();
321 314 }
322 315
323   - private ListenableFuture<TransportApiResponseMsg> handle(GetTenantRoutingInfoRequestMsg requestMsg) {
324   - TenantId tenantId = new TenantId(new UUID(requestMsg.getTenantIdMSB(), requestMsg.getTenantIdLSB()));
325   -
326   - ListenableFuture<TenantProfile> tenantProfileFuture = Futures.immediateFuture(tenantProfileCache.get(tenantId));
327   - return Futures.transform(tenantProfileFuture, tenantProfile -> TransportApiResponseMsg.newBuilder()
328   - .setGetTenantRoutingInfoResponseMsg(GetTenantRoutingInfoResponseMsg.newBuilder().setIsolatedTbCore(tenantProfile.isIsolatedTbCore())
329   - .setIsolatedTbRuleEngine(tenantProfile.isIsolatedTbRuleEngine()).build()).build(), dbCallbackExecutorService);
330   - }
331   -
332   - private ListenableFuture<TransportApiResponseMsg> handle(TransportProtos.GetDeviceProfileRequestMsg requestMsg) {
333   - DeviceProfileId profileId = new DeviceProfileId(new UUID(requestMsg.getProfileIdMSB(), requestMsg.getProfileIdLSB()));
334   - DeviceProfile deviceProfile = deviceProfileService.findDeviceProfileById(TenantId.SYS_TENANT_ID, profileId);
335   - return Futures.immediateFuture(TransportApiResponseMsg.newBuilder()
336   - .setGetDeviceProfileResponseMsg(
337   - TransportProtos.GetDeviceProfileResponseMsg.newBuilder()
338   - .setData(ByteString.copyFrom(dataDecodingEncodingService.encode(deviceProfile)))
339   - .build()).build());
  316 + private ListenableFuture<TransportApiResponseMsg> handle(GetEntityProfileRequestMsg requestMsg) {
  317 + EntityType entityType = EntityType.valueOf(requestMsg.getEntityType());
  318 + UUID entityUuid = new UUID(requestMsg.getEntityIdMSB(), requestMsg.getEntityIdLSB());
  319 + ByteString data;
  320 + if (entityType.equals(EntityType.DEVICE_PROFILE)) {
  321 + DeviceProfileId deviceProfileId = new DeviceProfileId(entityUuid);
  322 + DeviceProfile deviceProfile = deviceProfileCache.find(deviceProfileId);
  323 + data = ByteString.copyFrom(dataDecodingEncodingService.encode(deviceProfile));
  324 + } else if (entityType.equals(EntityType.TENANT)) {
  325 + TenantProfile tenantProfile = tenantProfileCache.get(new TenantId(entityUuid));
  326 + data = ByteString.copyFrom(dataDecodingEncodingService.encode(tenantProfile));
  327 + } else {
  328 + throw new RuntimeException("Invalid entity profile request: " + entityType);
  329 + }
  330 + return Futures.immediateFuture(TransportApiResponseMsg.newBuilder().setEntityProfileResponseMsg(GetEntityProfileResponseMsg.newBuilder().setData(data).build()).build());
340 331 }
341 332
342 333 private ListenableFuture<TransportApiResponseMsg> getDeviceInfo(DeviceId deviceId, DeviceCredentials credentials) {
... ... @@ -348,7 +339,7 @@ public class DefaultTransportApiService implements TransportApiService {
348 339 try {
349 340 ValidateDeviceCredentialsResponseMsg.Builder builder = ValidateDeviceCredentialsResponseMsg.newBuilder();
350 341 builder.setDeviceInfo(getDeviceInfoProto(device));
351   - DeviceProfile deviceProfile = deviceProfileService.findDeviceProfileById(device.getTenantId(), device.getDeviceProfileId());
  342 + DeviceProfile deviceProfile = deviceProfileCache.get(device.getTenantId(), device.getDeviceProfileId());
352 343 if (deviceProfile != null) {
353 344 builder.setProfileBody(ByteString.copyFrom(dataDecodingEncodingService.encode(deviceProfile)));
354 345 } else {
... ...
... ... @@ -502,7 +502,7 @@ js:
502 502 remote:
503 503 # Maximum allowed JavaScript execution errors before JavaScript will be blacklisted
504 504 max_errors: "${REMOTE_JS_SANDBOX_MAX_ERRORS:3}"
505   - # Maximum time in seconds for black listed function to stay in the list.
  505 + # Maximum time in seconds for black listed function to stay in 1:the list.
506 506 max_black_list_duration_sec: "${REMOTE_JS_SANDBOX_MAX_BLACKLIST_DURATION_SEC:60}"
507 507 stats:
508 508 enabled: "${TB_JS_REMOTE_STATS_ENABLED:false}"
... ... @@ -512,10 +512,6 @@ transport:
512 512 sessions:
513 513 inactivity_timeout: "${TB_TRANSPORT_SESSIONS_INACTIVITY_TIMEOUT:300000}"
514 514 report_timeout: "${TB_TRANSPORT_SESSIONS_REPORT_TIMEOUT:30000}"
515   - rate_limits:
516   - enabled: "${TB_TRANSPORT_RATE_LIMITS_ENABLED:false}"
517   - tenant: "${TB_TRANSPORT_RATE_LIMITS_TENANT:1000:1,20000:60}"
518   - device: "${TB_TRANSPORT_RATE_LIMITS_DEVICE:10:1,300:60}"
519 515 json:
520 516 # Cast String data types to Numeric if possible when processing Telemetry/Attributes JSON
521 517 type_cast_enabled: "${JSON_TYPE_CAST_ENABLED:true}"
... ...
... ... @@ -177,32 +177,26 @@ message GetOrCreateDeviceFromGatewayResponseMsg {
177 177 bytes profileBody = 2;
178 178 }
179 179
180   -message GetTenantRoutingInfoRequestMsg {
181   - int64 tenantIdMSB = 1;
182   - int64 tenantIdLSB = 2;
183   -}
184   -
185   -message GetTenantRoutingInfoResponseMsg {
186   - bool isolatedTbCore = 1;
187   - bool isolatedTbRuleEngine = 2;
188   -}
189   -
190   -message GetDeviceProfileRequestMsg {
191   - int64 profileIdMSB = 1;
192   - int64 profileIdLSB = 2;
  180 +message GetEntityProfileRequestMsg {
  181 + string entityType = 1;
  182 + int64 entityIdMSB = 2;
  183 + int64 entityIdLSB = 3;
193 184 }
194 185
195   -message GetDeviceProfileResponseMsg {
196   - bytes data = 1;
  186 +message GetEntityProfileResponseMsg {
  187 + string entityType = 1;
  188 + bytes data = 2;
197 189 }
198 190
199   -message DeviceProfileUpdateMsg {
200   - bytes data = 1;
  191 +message EntityUpdateMsg {
  192 + string entityType = 1;
  193 + bytes data = 2;
201 194 }
202 195
203   -message DeviceProfileDeleteMsg {
204   - int64 profileIdMSB = 1;
205   - int64 profileIdLSB = 2;
  196 +message EntityDeleteMsg {
  197 + string entityType = 1;
  198 + int64 entityIdMSB = 2;
  199 + int64 entityIdLSB = 3;
206 200 }
207 201
208 202 message SessionCloseNotificationProto {
... ... @@ -482,8 +476,7 @@ message TransportApiRequestMsg {
482 476 ValidateDeviceTokenRequestMsg validateTokenRequestMsg = 1;
483 477 ValidateDeviceX509CertRequestMsg validateX509CertRequestMsg = 2;
484 478 GetOrCreateDeviceFromGatewayRequestMsg getOrCreateDeviceRequestMsg = 3;
485   - GetTenantRoutingInfoRequestMsg getTenantRoutingInfoRequestMsg = 4;
486   - GetDeviceProfileRequestMsg getDeviceProfileRequestMsg = 5;
  479 + GetEntityProfileRequestMsg entityProfileRequestMsg = 4;
487 480 ValidateBasicMqttCredRequestMsg validateBasicMqttCredRequestMsg = 6;
488 481 ProvisionDeviceRequestMsg provisionDeviceRequestMsg = 7;
489 482 }
... ... @@ -492,9 +485,8 @@ message TransportApiRequestMsg {
492 485 message TransportApiResponseMsg {
493 486 ValidateDeviceCredentialsResponseMsg validateCredResponseMsg = 1;
494 487 GetOrCreateDeviceFromGatewayResponseMsg getOrCreateDeviceResponseMsg = 2;
495   - GetTenantRoutingInfoResponseMsg getTenantRoutingInfoResponseMsg = 4;
496   - GetDeviceProfileResponseMsg getDeviceProfileResponseMsg = 5;
497   - ProvisionDeviceResponseMsg provisionDeviceResponseMsg = 6;
  488 + GetEntityProfileResponseMsg entityProfileResponseMsg = 3;
  489 + ProvisionDeviceResponseMsg provisionDeviceResponseMsg = 4;
498 490 }
499 491
500 492 /* Messages that are handled by ThingsBoard Core Service */
... ... @@ -535,7 +527,8 @@ message ToTransportMsg {
535 527 AttributeUpdateNotificationMsg attributeUpdateNotification = 5;
536 528 ToDeviceRpcRequestMsg toDeviceRequest = 6;
537 529 ToServerRpcResponseMsg toServerResponse = 7;
538   - DeviceProfileUpdateMsg deviceProfileUpdateMsg = 8;
539   - DeviceProfileDeleteMsg deviceProfileDeleteMsg = 9;
  530 + /* For Tenant, TenantProfile and DeviceProfile */
  531 + EntityUpdateMsg entityUpdateMsg = 8;
  532 + EntityDeleteMsg entityDeleteMsg = 9;
540 533 ProvisionDeviceResponseMsg provisionResponse = 10;
541 534 }
... ...
common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/TransportDeviceProfileCache.java renamed from common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/TransportProfileCache.java
... ... @@ -21,7 +21,7 @@ import org.thingsboard.server.common.data.id.DeviceProfileId;
21 21
22 22 import java.util.Optional;
23 23
24   -public interface TransportProfileCache {
  24 +public interface TransportDeviceProfileCache {
25 25
26 26 DeviceProfile getOrCreate(DeviceProfileId id, ByteString profileBody);
27 27
... ...
... ... @@ -20,11 +20,12 @@ import org.thingsboard.server.common.data.DeviceTransportType;
20 20 import org.thingsboard.server.common.data.id.DeviceProfileId;
21 21 import org.thingsboard.server.common.transport.auth.GetOrCreateDeviceFromGatewayResponse;
22 22 import org.thingsboard.server.common.transport.auth.ValidateDeviceCredentialsResponse;
  23 +import org.thingsboard.server.gen.transport.TransportProtos;
23 24 import org.thingsboard.server.gen.transport.TransportProtos.ClaimDeviceMsg;
24 25 import org.thingsboard.server.gen.transport.TransportProtos.GetAttributeRequestMsg;
25 26 import org.thingsboard.server.gen.transport.TransportProtos.GetOrCreateDeviceFromGatewayRequestMsg;
26   -import org.thingsboard.server.gen.transport.TransportProtos.GetTenantRoutingInfoRequestMsg;
27   -import org.thingsboard.server.gen.transport.TransportProtos.GetTenantRoutingInfoResponseMsg;
  27 +import org.thingsboard.server.gen.transport.TransportProtos.GetEntityProfileRequestMsg;
  28 +import org.thingsboard.server.gen.transport.TransportProtos.GetEntityProfileResponseMsg;
28 29 import org.thingsboard.server.gen.transport.TransportProtos.PostAttributeMsg;
29 30 import org.thingsboard.server.gen.transport.TransportProtos.PostTelemetryMsg;
30 31 import org.thingsboard.server.gen.transport.TransportProtos.ProvisionDeviceRequestMsg;
... ... @@ -47,7 +48,7 @@ import java.util.concurrent.ScheduledExecutorService;
47 48 */
48 49 public interface TransportService {
49 50
50   - GetTenantRoutingInfoResponseMsg getRoutingInfo(GetTenantRoutingInfoRequestMsg msg);
  51 + GetEntityProfileResponseMsg getRoutingInfo(GetEntityProfileRequestMsg msg);
51 52
52 53 void process(DeviceTransportType transportType, ValidateDeviceTokenRequestMsg msg,
53 54 TransportServiceCallback<ValidateDeviceCredentialsResponse> callback);
... ... @@ -64,8 +65,6 @@ public interface TransportService {
64 65 void process(ProvisionDeviceRequestMsg msg,
65 66 TransportServiceCallback<ProvisionDeviceResponseMsg> callback);
66 67
67   - void getDeviceProfile(DeviceProfileId deviceProfileId, TransportServiceCallback<DeviceProfile> callback);
68   -
69 68 void onProfileUpdate(DeviceProfile deviceProfile);
70 69
71 70 boolean checkLimits(SessionInfoProto sessionInfo, Object msg, TransportServiceCallback<Void> callback);
... ...
  1 +/**
  2 + * Copyright © 2016-2020 The Thingsboard Authors
  3 + *
  4 + * Licensed under the Apache License, Version 2.0 (the "License");
  5 + * you may not use this file except in compliance with the License.
  6 + * You may obtain a copy of the License at
  7 + *
  8 + * http://www.apache.org/licenses/LICENSE-2.0
  9 + *
  10 + * Unless required by applicable law or agreed to in writing, software
  11 + * distributed under the License is distributed on an "AS IS" BASIS,
  12 + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13 + * See the License for the specific language governing permissions and
  14 + * limitations under the License.
  15 + */
  16 +package org.thingsboard.server.common.transport;
  17 +
  18 +import com.google.protobuf.ByteString;
  19 +import org.thingsboard.server.common.data.TenantProfile;
  20 +import org.thingsboard.server.common.data.id.TenantId;
  21 +import org.thingsboard.server.common.data.id.TenantProfileId;
  22 +import org.thingsboard.server.common.transport.profile.TenantProfileUpdateResult;
  23 +import org.thingsboard.server.queue.discovery.TenantRoutingInfo;
  24 +import org.thingsboard.server.queue.discovery.TenantRoutingInfoService;
  25 +
  26 +import java.util.Set;
  27 +
  28 +public interface TransportTenantProfileCache {
  29 +
  30 + TenantProfile get(TenantId tenantId);
  31 +
  32 + TenantProfileUpdateResult put(ByteString profileBody);
  33 +
  34 + boolean put(TenantId tenantId, TenantProfileId profileId);
  35 +
  36 + Set<TenantId> remove(TenantProfileId profileId);
  37 +
  38 +}
... ...
  1 +/**
  2 + * Copyright © 2016-2020 The Thingsboard Authors
  3 + *
  4 + * Licensed under the Apache License, Version 2.0 (the "License");
  5 + * you may not use this file except in compliance with the License.
  6 + * You may obtain a copy of the License at
  7 + *
  8 + * http://www.apache.org/licenses/LICENSE-2.0
  9 + *
  10 + * Unless required by applicable law or agreed to in writing, software
  11 + * distributed under the License is distributed on an "AS IS" BASIS,
  12 + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13 + * See the License for the specific language governing permissions and
  14 + * limitations under the License.
  15 + */
  16 +package org.thingsboard.server.common.transport.limits;
  17 +
  18 +import lombok.extern.slf4j.Slf4j;
  19 +import org.springframework.stereotype.Component;
  20 +import org.springframework.util.StringUtils;
  21 +import org.thingsboard.server.common.msg.tools.TbRateLimits;
  22 +
  23 +@Slf4j
  24 +@Component
  25 +public class DefaultTransportRateLimitFactory implements TransportRateLimitFactory {
  26 +
  27 + private static final DummyTransportRateLimit ALWAYS_TRUE = new DummyTransportRateLimit();
  28 +
  29 + @Override
  30 + public TransportRateLimit create(TransportRateLimitType type, Object configuration) {
  31 + if (!StringUtils.isEmpty(configuration)) {
  32 + try {
  33 + return new SimpleTransportRateLimit(new TbRateLimits(configuration.toString()), configuration.toString());
  34 + } catch (Exception e) {
  35 + log.warn("[{}] Failed to init rate limit with configuration: {}", type, configuration, e);
  36 + return ALWAYS_TRUE;
  37 + }
  38 + } else {
  39 + return ALWAYS_TRUE;
  40 + }
  41 + }
  42 +
  43 + @Override
  44 + public TransportRateLimit createDefault(TransportRateLimitType type) {
  45 + return ALWAYS_TRUE;
  46 + }
  47 +}
... ...
  1 +/**
  2 + * Copyright © 2016-2020 The Thingsboard Authors
  3 + *
  4 + * Licensed under the Apache License, Version 2.0 (the "License");
  5 + * you may not use this file except in compliance with the License.
  6 + * You may obtain a copy of the License at
  7 + *
  8 + * http://www.apache.org/licenses/LICENSE-2.0
  9 + *
  10 + * Unless required by applicable law or agreed to in writing, software
  11 + * distributed under the License is distributed on an "AS IS" BASIS,
  12 + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13 + * See the License for the specific language governing permissions and
  14 + * limitations under the License.
  15 + */
  16 +package org.thingsboard.server.common.transport.limits;
  17 +
  18 +import lombok.extern.slf4j.Slf4j;
  19 +import org.springframework.stereotype.Service;
  20 +import org.thingsboard.server.common.data.TenantProfile;
  21 +import org.thingsboard.server.common.data.TenantProfileData;
  22 +import org.thingsboard.server.common.data.id.DeviceId;
  23 +import org.thingsboard.server.common.data.id.TenantId;
  24 +import org.thingsboard.server.common.transport.TransportTenantProfileCache;
  25 +import org.thingsboard.server.common.transport.profile.TenantProfileUpdateResult;
  26 +
  27 +import java.util.concurrent.ConcurrentHashMap;
  28 +import java.util.concurrent.ConcurrentMap;
  29 +
  30 +@Service
  31 +@Slf4j
  32 +public class DefaultTransportRateLimitService implements TransportRateLimitService {
  33 +
  34 + private final ConcurrentMap<TenantId, TransportRateLimit[]> perTenantLimits = new ConcurrentHashMap<>();
  35 + private final ConcurrentMap<DeviceId, TransportRateLimit[]> perDeviceLimits = new ConcurrentHashMap<>();
  36 +
  37 + private final TransportRateLimitFactory rateLimitFactory;
  38 + private final TransportTenantProfileCache tenantProfileCache;
  39 +
  40 + public DefaultTransportRateLimitService(TransportRateLimitFactory rateLimitFactory, TransportTenantProfileCache tenantProfileCache) {
  41 + this.rateLimitFactory = rateLimitFactory;
  42 + this.tenantProfileCache = tenantProfileCache;
  43 + }
  44 +
  45 + @Override
  46 + public TransportRateLimit getRateLimit(TenantId tenantId, TransportRateLimitType limitType) {
  47 + TransportRateLimit[] limits = perTenantLimits.get(tenantId);
  48 + if (limits == null) {
  49 + limits = fetchProfileAndInit(tenantId);
  50 + perTenantLimits.put(tenantId, limits);
  51 + }
  52 + return limits[limitType.ordinal()];
  53 + }
  54 +
  55 + @Override
  56 + public TransportRateLimit getRateLimit(TenantId tenantId, DeviceId deviceId, TransportRateLimitType limitType) {
  57 + TransportRateLimit[] limits = perDeviceLimits.get(deviceId);
  58 + if (limits == null) {
  59 + limits = fetchProfileAndInit(tenantId);
  60 + perDeviceLimits.put(deviceId, limits);
  61 + }
  62 + return limits[limitType.ordinal()];
  63 + }
  64 +
  65 + @Override
  66 + public void update(TenantProfileUpdateResult update) {
  67 + TransportRateLimit[] newLimits = createTransportRateLimits(update.getProfile());
  68 + for (TenantId tenantId : update.getAffectedTenants()) {
  69 + mergeLimits(tenantId, newLimits);
  70 + }
  71 + }
  72 +
  73 + @Override
  74 + public void update(TenantId tenantId) {
  75 + mergeLimits(tenantId, fetchProfileAndInit(tenantId));
  76 + }
  77 +
  78 + public void mergeLimits(TenantId tenantId, TransportRateLimit[] newRateLimits) {
  79 + TransportRateLimit[] oldRateLimits = perTenantLimits.get(tenantId);
  80 + if (oldRateLimits == null) {
  81 + perTenantLimits.put(tenantId, newRateLimits);
  82 + } else {
  83 + for (int i = 0; i < TransportRateLimitType.values().length; i++) {
  84 + TransportRateLimit newLimit = newRateLimits[i];
  85 + TransportRateLimit oldLimit = oldRateLimits[i];
  86 + if (newLimit != null && (oldLimit == null || !oldLimit.getConfiguration().equals(newLimit.getConfiguration()))) {
  87 + oldRateLimits[i] = newLimit;
  88 + }
  89 + }
  90 + }
  91 + }
  92 +
  93 + @Override
  94 + public void remove(TenantId tenantId) {
  95 + perTenantLimits.remove(tenantId);
  96 + }
  97 +
  98 + @Override
  99 + public void remove(DeviceId deviceId) {
  100 + perDeviceLimits.remove(deviceId);
  101 + }
  102 +
  103 + private TransportRateLimit[] fetchProfileAndInit(TenantId tenantId) {
  104 + return perTenantLimits.computeIfAbsent(tenantId, tmp -> createTransportRateLimits(tenantProfileCache.get(tenantId)));
  105 + }
  106 +
  107 + private TransportRateLimit[] createTransportRateLimits(TenantProfile tenantProfile) {
  108 + TenantProfileData profileData = tenantProfile.getProfileData();
  109 + TransportRateLimit[] rateLimits = new TransportRateLimit[TransportRateLimitType.values().length];
  110 + for (TransportRateLimitType type : TransportRateLimitType.values()) {
  111 + rateLimits[type.ordinal()] = rateLimitFactory.create(type, profileData.getProperties().get(type.getConfigurationKey()));
  112 + }
  113 + return rateLimits;
  114 + }
  115 +}
... ...
  1 +/**
  2 + * Copyright © 2016-2020 The Thingsboard Authors
  3 + *
  4 + * Licensed under the Apache License, Version 2.0 (the "License");
  5 + * you may not use this file except in compliance with the License.
  6 + * You may obtain a copy of the License at
  7 + *
  8 + * http://www.apache.org/licenses/LICENSE-2.0
  9 + *
  10 + * Unless required by applicable law or agreed to in writing, software
  11 + * distributed under the License is distributed on an "AS IS" BASIS,
  12 + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13 + * See the License for the specific language governing permissions and
  14 + * limitations under the License.
  15 + */
  16 +package org.thingsboard.server.common.transport.limits;
  17 +
  18 +public class DummyTransportRateLimit implements TransportRateLimit {
  19 +
  20 + @Override
  21 + public String getConfiguration() {
  22 + return "";
  23 + }
  24 +
  25 + @Override
  26 + public boolean tryConsume() {
  27 + return true;
  28 + }
  29 +
  30 +}
... ...
  1 +/**
  2 + * Copyright © 2016-2020 The Thingsboard Authors
  3 + *
  4 + * Licensed under the Apache License, Version 2.0 (the "License");
  5 + * you may not use this file except in compliance with the License.
  6 + * You may obtain a copy of the License at
  7 + *
  8 + * http://www.apache.org/licenses/LICENSE-2.0
  9 + *
  10 + * Unless required by applicable law or agreed to in writing, software
  11 + * distributed under the License is distributed on an "AS IS" BASIS,
  12 + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13 + * See the License for the specific language governing permissions and
  14 + * limitations under the License.
  15 + */
  16 +package org.thingsboard.server.common.transport.limits;
  17 +
  18 +import lombok.Getter;
  19 +import lombok.RequiredArgsConstructor;
  20 +import org.thingsboard.server.common.msg.tools.TbRateLimits;
  21 +
  22 +@RequiredArgsConstructor
  23 +public class SimpleTransportRateLimit implements TransportRateLimit {
  24 +
  25 + private final TbRateLimits rateLimit;
  26 + @Getter
  27 + private final String configuration;
  28 +
  29 + @Override
  30 + public boolean tryConsume() {
  31 + return rateLimit.tryConsume();
  32 + }
  33 +
  34 +}
... ...
  1 +/**
  2 + * Copyright © 2016-2020 The Thingsboard Authors
  3 + *
  4 + * Licensed under the Apache License, Version 2.0 (the "License");
  5 + * you may not use this file except in compliance with the License.
  6 + * You may obtain a copy of the License at
  7 + *
  8 + * http://www.apache.org/licenses/LICENSE-2.0
  9 + *
  10 + * Unless required by applicable law or agreed to in writing, software
  11 + * distributed under the License is distributed on an "AS IS" BASIS,
  12 + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13 + * See the License for the specific language governing permissions and
  14 + * limitations under the License.
  15 + */
  16 +package org.thingsboard.server.common.transport.limits;
  17 +
  18 +public interface TransportRateLimit {
  19 +
  20 + String getConfiguration();
  21 +
  22 + boolean tryConsume();
  23 +
  24 +}
... ...
  1 +/**
  2 + * Copyright © 2016-2020 The Thingsboard Authors
  3 + *
  4 + * Licensed under the Apache License, Version 2.0 (the "License");
  5 + * you may not use this file except in compliance with the License.
  6 + * You may obtain a copy of the License at
  7 + *
  8 + * http://www.apache.org/licenses/LICENSE-2.0
  9 + *
  10 + * Unless required by applicable law or agreed to in writing, software
  11 + * distributed under the License is distributed on an "AS IS" BASIS,
  12 + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13 + * See the License for the specific language governing permissions and
  14 + * limitations under the License.
  15 + */
  16 +package org.thingsboard.server.common.transport.limits;
  17 +
  18 +public interface TransportRateLimitFactory {
  19 +
  20 + TransportRateLimit create(TransportRateLimitType type, Object config);
  21 +
  22 + TransportRateLimit createDefault(TransportRateLimitType type);
  23 +
  24 +}
... ...
  1 +/**
  2 + * Copyright © 2016-2020 The Thingsboard Authors
  3 + *
  4 + * Licensed under the Apache License, Version 2.0 (the "License");
  5 + * you may not use this file except in compliance with the License.
  6 + * You may obtain a copy of the License at
  7 + *
  8 + * http://www.apache.org/licenses/LICENSE-2.0
  9 + *
  10 + * Unless required by applicable law or agreed to in writing, software
  11 + * distributed under the License is distributed on an "AS IS" BASIS,
  12 + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13 + * See the License for the specific language governing permissions and
  14 + * limitations under the License.
  15 + */
  16 +package org.thingsboard.server.common.transport.limits;
  17 +
  18 +import org.thingsboard.server.common.data.id.DeviceId;
  19 +import org.thingsboard.server.common.data.id.TenantId;
  20 +import org.thingsboard.server.common.transport.profile.TenantProfileUpdateResult;
  21 +
  22 +public interface TransportRateLimitService {
  23 +
  24 + TransportRateLimit getRateLimit(TenantId tenantId, TransportRateLimitType limit);
  25 +
  26 + TransportRateLimit getRateLimit(TenantId tenantId, DeviceId deviceId, TransportRateLimitType limit);
  27 +
  28 + void update(TenantProfileUpdateResult update);
  29 +
  30 + void update(TenantId tenantId);
  31 +
  32 + void remove(TenantId tenantId);
  33 +
  34 + void remove(DeviceId deviceId);
  35 +
  36 +}
... ...
  1 +/**
  2 + * Copyright © 2016-2020 The Thingsboard Authors
  3 + *
  4 + * Licensed under the Apache License, Version 2.0 (the "License");
  5 + * you may not use this file except in compliance with the License.
  6 + * You may obtain a copy of the License at
  7 + *
  8 + * http://www.apache.org/licenses/LICENSE-2.0
  9 + *
  10 + * Unless required by applicable law or agreed to in writing, software
  11 + * distributed under the License is distributed on an "AS IS" BASIS,
  12 + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13 + * See the License for the specific language governing permissions and
  14 + * limitations under the License.
  15 + */
  16 +package org.thingsboard.server.common.transport.limits;
  17 +
  18 +import lombok.Getter;
  19 +
  20 +public enum TransportRateLimitType {
  21 +
  22 + TENANT_MAX_MSGS("transport.tenant.max.msg"),
  23 + TENANT_MAX_DATA_POINTS("transport.tenant.max.dataPoints"),
  24 + DEVICE_MAX_MSGS("transport.device.max.msg"),
  25 + DEVICE_MAX_DATA_POINTS("transport.device.max.dataPoints");
  26 +
  27 + @Getter
  28 + private final String configurationKey;
  29 +
  30 + TransportRateLimitType(String configurationKey) {
  31 + this.configurationKey = configurationKey;
  32 + }
  33 +}
... ...
  1 +/**
  2 + * Copyright © 2016-2020 The Thingsboard Authors
  3 + *
  4 + * Licensed under the Apache License, Version 2.0 (the "License");
  5 + * you may not use this file except in compliance with the License.
  6 + * You may obtain a copy of the License at
  7 + *
  8 + * http://www.apache.org/licenses/LICENSE-2.0
  9 + *
  10 + * Unless required by applicable law or agreed to in writing, software
  11 + * distributed under the License is distributed on an "AS IS" BASIS,
  12 + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13 + * See the License for the specific language governing permissions and
  14 + * limitations under the License.
  15 + */
  16 +package org.thingsboard.server.common.transport.profile;
  17 +
  18 +import lombok.Data;
  19 +import org.thingsboard.server.common.data.TenantProfile;
  20 +import org.thingsboard.server.common.data.id.TenantId;
  21 +
  22 +import java.util.Set;
  23 +
  24 +@Data
  25 +public class TenantProfileUpdateResult {
  26 +
  27 + private final TenantProfile profile;
  28 + private final Set<TenantId> affectedTenants;
  29 +
  30 +}
... ...
common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportDeviceProfileCache.java renamed from common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportProfileCache.java
... ... @@ -21,7 +21,7 @@ import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
21 21 import org.springframework.stereotype.Component;
22 22 import org.thingsboard.server.common.data.DeviceProfile;
23 23 import org.thingsboard.server.common.data.id.DeviceProfileId;
24   -import org.thingsboard.server.common.transport.TransportProfileCache;
  24 +import org.thingsboard.server.common.transport.TransportDeviceProfileCache;
25 25 import org.thingsboard.server.common.transport.util.DataDecodingEncodingService;
26 26
27 27 import java.util.Optional;
... ... @@ -31,13 +31,13 @@ import java.util.concurrent.ConcurrentMap;
31 31 @Slf4j
32 32 @Component
33 33 @ConditionalOnExpression("('${service.type:null}'=='monolith' && '${transport.api_enabled:true}'=='true') || '${service.type:null}'=='tb-transport'")
34   -public class DefaultTransportProfileCache implements TransportProfileCache {
  34 +public class DefaultTransportDeviceProfileCache implements TransportDeviceProfileCache {
35 35
36 36 private final ConcurrentMap<DeviceProfileId, DeviceProfile> deviceProfiles = new ConcurrentHashMap<>();
37 37
38 38 private final DataDecodingEncodingService dataDecodingEncodingService;
39 39
40   - public DefaultTransportProfileCache(DataDecodingEncodingService dataDecodingEncodingService) {
  40 + public DefaultTransportDeviceProfileCache(DataDecodingEncodingService dataDecodingEncodingService) {
41 41 this.dataDecodingEncodingService = dataDecodingEncodingService;
42 42 }
43 43
... ...
... ... @@ -29,25 +29,35 @@ import org.thingsboard.common.util.ThingsBoardThreadFactory;
29 29 import org.thingsboard.server.common.data.DeviceProfile;
30 30 import org.thingsboard.server.common.data.DeviceTransportType;
31 31 import org.thingsboard.server.common.data.EntityType;
  32 +import org.thingsboard.server.common.data.Tenant;
32 33 import org.thingsboard.server.common.data.id.DeviceId;
33 34 import org.thingsboard.server.common.data.id.DeviceProfileId;
34 35 import org.thingsboard.server.common.data.id.RuleChainId;
35 36 import org.thingsboard.server.common.data.id.TenantId;
  37 +import org.thingsboard.server.common.data.id.TenantProfileId;
36 38 import org.thingsboard.server.common.msg.TbMsg;
37 39 import org.thingsboard.server.common.msg.TbMsgMetaData;
38 40 import org.thingsboard.server.common.msg.queue.ServiceQueue;
39 41 import org.thingsboard.server.common.msg.queue.ServiceType;
40 42 import org.thingsboard.server.common.msg.queue.TopicPartitionInfo;
41 43 import org.thingsboard.server.common.msg.session.SessionMsgType;
42   -import org.thingsboard.server.common.msg.tools.TbRateLimits;
43 44 import org.thingsboard.server.common.msg.tools.TbRateLimitsException;
  45 +import org.thingsboard.server.common.stats.MessagesStats;
  46 +import org.thingsboard.server.common.stats.StatsFactory;
  47 +import org.thingsboard.server.common.stats.StatsType;
44 48 import org.thingsboard.server.common.transport.SessionMsgListener;
45   -import org.thingsboard.server.common.transport.TransportProfileCache;
  49 +import org.thingsboard.server.common.transport.TransportDeviceProfileCache;
46 50 import org.thingsboard.server.common.transport.TransportService;
47 51 import org.thingsboard.server.common.transport.TransportServiceCallback;
  52 +import org.thingsboard.server.common.transport.TransportTenantProfileCache;
48 53 import org.thingsboard.server.common.transport.auth.GetOrCreateDeviceFromGatewayResponse;
49 54 import org.thingsboard.server.common.transport.auth.TransportDeviceInfo;
50 55 import org.thingsboard.server.common.transport.auth.ValidateDeviceCredentialsResponse;
  56 +import org.thingsboard.server.common.transport.limits.TransportRateLimit;
  57 +import org.thingsboard.server.common.transport.limits.TransportRateLimitService;
  58 +import org.thingsboard.server.common.transport.limits.TransportRateLimitType;
  59 +import org.thingsboard.server.common.transport.profile.TenantProfileUpdateResult;
  60 +import org.thingsboard.server.common.transport.util.DataDecodingEncodingService;
51 61 import org.thingsboard.server.common.transport.util.JsonUtils;
52 62 import org.thingsboard.server.gen.transport.TransportProtos;
53 63 import org.thingsboard.server.gen.transport.TransportProtos.ProvisionDeviceRequestMsg;
... ... @@ -69,15 +79,13 @@ import org.thingsboard.server.queue.discovery.PartitionService;
69 79 import org.thingsboard.server.queue.discovery.TbServiceInfoProvider;
70 80 import org.thingsboard.server.queue.provider.TbQueueProducerProvider;
71 81 import org.thingsboard.server.queue.provider.TbTransportQueueFactory;
72   -import org.thingsboard.server.common.stats.MessagesStats;
73   -import org.thingsboard.server.common.stats.StatsFactory;
74   -import org.thingsboard.server.common.stats.StatsType;
75 82
76 83 import javax.annotation.PostConstruct;
77 84 import javax.annotation.PreDestroy;
78 85 import java.util.Collections;
79 86 import java.util.List;
80 87 import java.util.Map;
  88 +import java.util.Optional;
81 89 import java.util.Random;
82 90 import java.util.UUID;
83 91 import java.util.concurrent.ConcurrentHashMap;
... ... @@ -98,12 +106,6 @@ import java.util.concurrent.atomic.AtomicInteger;
98 106 @ConditionalOnExpression("('${service.type:null}'=='monolith' && '${transport.api_enabled:true}'=='true') || '${service.type:null}'=='tb-transport'")
99 107 public class DefaultTransportService implements TransportService {
100 108
101   - @Value("${transport.rate_limits.enabled}")
102   - private boolean rateLimitEnabled;
103   - @Value("${transport.rate_limits.tenant}")
104   - private String perTenantLimitsConf;
105   - @Value("${transport.rate_limits.device}")
106   - private String perDevicesLimitsConf;
107 109 @Value("${transport.sessions.inactivity_timeout}")
108 110 private long sessionInactivityTimeout;
109 111 @Value("${transport.sessions.report_timeout}")
... ... @@ -119,7 +121,10 @@ public class DefaultTransportService implements TransportService {
119 121 private final PartitionService partitionService;
120 122 private final TbServiceInfoProvider serviceInfoProvider;
121 123 private final StatsFactory statsFactory;
122   - private final TransportProfileCache transportProfileCache;
  124 + private final TransportDeviceProfileCache deviceProfileCache;
  125 + private final TransportTenantProfileCache tenantProfileCache;
  126 + private final TransportRateLimitService rateLimitService;
  127 + private final DataDecodingEncodingService dataDecodingEncodingService;
123 128
124 129 protected TbQueueRequestTemplate<TbProtoQueueMsg<TransportApiRequestMsg>, TbProtoQueueMsg<TransportApiResponseMsg>> transportApiRequestTemplate;
125 130 protected TbQueueProducer<TbProtoQueueMsg<ToRuleEngineMsg>> ruleEngineMsgProducer;
... ... @@ -132,14 +137,11 @@ public class DefaultTransportService implements TransportService {
132 137
133 138 protected ScheduledExecutorService schedulerExecutor;
134 139 protected ExecutorService transportCallbackExecutor;
  140 + private ExecutorService mainConsumerExecutor;
135 141
136 142 private final ConcurrentMap<UUID, SessionMetaData> sessions = new ConcurrentHashMap<>();
137 143 private final Map<String, RpcRequestMetadata> toServerRpcPendingMap = new ConcurrentHashMap<>();
138   - //TODO 3.2: @ybondarenko Implement cleanup of this maps.
139   - private final ConcurrentMap<TenantId, TbRateLimits> perTenantLimits = new ConcurrentHashMap<>();
140   - private final ConcurrentMap<DeviceId, TbRateLimits> perDeviceLimits = new ConcurrentHashMap<>();
141 144
142   - private ExecutorService mainConsumerExecutor = Executors.newSingleThreadExecutor(ThingsBoardThreadFactory.forName("transport-consumer"));
143 145 private volatile boolean stopped = false;
144 146
145 147 public DefaultTransportService(TbServiceInfoProvider serviceInfoProvider,
... ... @@ -147,22 +149,22 @@ public class DefaultTransportService implements TransportService {
147 149 TbQueueProducerProvider producerProvider,
148 150 PartitionService partitionService,
149 151 StatsFactory statsFactory,
150   - TransportProfileCache transportProfileCache) {
  152 + TransportDeviceProfileCache deviceProfileCache,
  153 + TransportTenantProfileCache tenantProfileCache,
  154 + TransportRateLimitService rateLimitService, DataDecodingEncodingService dataDecodingEncodingService) {
151 155 this.serviceInfoProvider = serviceInfoProvider;
152 156 this.queueProvider = queueProvider;
153 157 this.producerProvider = producerProvider;
154 158 this.partitionService = partitionService;
155 159 this.statsFactory = statsFactory;
156   - this.transportProfileCache = transportProfileCache;
  160 + this.deviceProfileCache = deviceProfileCache;
  161 + this.tenantProfileCache = tenantProfileCache;
  162 + this.rateLimitService = rateLimitService;
  163 + this.dataDecodingEncodingService = dataDecodingEncodingService;
157 164 }
158 165
159 166 @PostConstruct
160 167 public void init() {
161   - if (rateLimitEnabled) {
162   - //Just checking the configuration parameters
163   - new TbRateLimits(perTenantLimitsConf);
164   - new TbRateLimits(perDevicesLimitsConf);
165   - }
166 168 this.ruleEngineProducerStats = statsFactory.createMessagesStats(StatsType.RULE_ENGINE.getName() + ".producer");
167 169 this.tbCoreProducerStats = statsFactory.createMessagesStats(StatsType.CORE.getName() + ".producer");
168 170 this.transportApiStats = statsFactory.createMessagesStats(StatsType.TRANSPORT.getName() + ".producer");
... ... @@ -177,6 +179,7 @@ public class DefaultTransportService implements TransportService {
177 179 TopicPartitionInfo tpi = partitionService.getNotificationsTopic(ServiceType.TB_TRANSPORT, serviceInfoProvider.getServiceId());
178 180 transportNotificationsConsumer.subscribe(Collections.singleton(tpi));
179 181 transportApiRequestTemplate.init();
  182 + mainConsumerExecutor = Executors.newSingleThreadExecutor(ThingsBoardThreadFactory.forName("transport-consumer"));
180 183 mainConsumerExecutor.execute(() -> {
181 184 while (!stopped) {
182 185 try {
... ... @@ -208,10 +211,6 @@ public class DefaultTransportService implements TransportService {
208 211
209 212 @PreDestroy
210 213 public void destroy() {
211   - if (rateLimitEnabled) {
212   - perTenantLimits.clear();
213   - perDeviceLimits.clear();
214   - }
215 214 stopped = true;
216 215
217 216 if (transportNotificationsConsumer != null) {
... ... @@ -232,7 +231,7 @@ public class DefaultTransportService implements TransportService {
232 231 }
233 232
234 233 @Override
235   - public ScheduledExecutorService getSchedulerExecutor(){
  234 + public ScheduledExecutorService getSchedulerExecutor() {
236 235 return this.schedulerExecutor;
237 236 }
238 237
... ... @@ -242,12 +241,12 @@ public class DefaultTransportService implements TransportService {
242 241 }
243 242
244 243 @Override
245   - public TransportProtos.GetTenantRoutingInfoResponseMsg getRoutingInfo(TransportProtos.GetTenantRoutingInfoRequestMsg msg) {
  244 + public TransportProtos.GetEntityProfileResponseMsg getRoutingInfo(TransportProtos.GetEntityProfileRequestMsg msg) {
246 245 TbProtoQueueMsg<TransportProtos.TransportApiRequestMsg> protoMsg =
247   - new TbProtoQueueMsg<>(UUID.randomUUID(), TransportProtos.TransportApiRequestMsg.newBuilder().setGetTenantRoutingInfoRequestMsg(msg).build());
  246 + new TbProtoQueueMsg<>(UUID.randomUUID(), TransportProtos.TransportApiRequestMsg.newBuilder().setEntityProfileRequestMsg(msg).build());
248 247 try {
249 248 TbProtoQueueMsg<TransportApiResponseMsg> response = transportApiRequestTemplate.send(protoMsg).get();
250   - return response.getValue().getGetTenantRoutingInfoResponseMsg();
  249 + return response.getValue().getEntityProfileResponseMsg();
251 250 } catch (InterruptedException | ExecutionException e) {
252 251 throw new RuntimeException(e);
253 252 }
... ... @@ -289,7 +288,7 @@ public class DefaultTransportService implements TransportService {
289 288 result.deviceInfo(tdi);
290 289 ByteString profileBody = msg.getProfileBody();
291 290 if (profileBody != null && !profileBody.isEmpty()) {
292   - DeviceProfile profile = transportProfileCache.getOrCreate(tdi.getDeviceProfileId(), profileBody);
  291 + DeviceProfile profile = deviceProfileCache.getOrCreate(tdi.getDeviceProfileId(), profileBody);
293 292 if (transportType != DeviceTransportType.DEFAULT
294 293 && profile != null && profile.getTransportType() != DeviceTransportType.DEFAULT && profile.getTransportType() != transportType) {
295 294 log.debug("[{}] Device profile [{}] has different transport type: {}, expected: {}", tdi.getDeviceId(), tdi.getDeviceProfileId(), profile.getTransportType(), transportType);
... ... @@ -315,7 +314,7 @@ public class DefaultTransportService implements TransportService {
315 314 result.deviceInfo(tdi);
316 315 ByteString profileBody = msg.getProfileBody();
317 316 if (profileBody != null && !profileBody.isEmpty()) {
318   - result.deviceProfile(transportProfileCache.getOrCreate(tdi.getDeviceProfileId(), profileBody));
  317 + result.deviceProfile(deviceProfileCache.getOrCreate(tdi.getDeviceProfileId(), profileBody));
319 318 }
320 319 }
321 320 return result.build();
... ... @@ -339,8 +338,8 @@ public class DefaultTransportService implements TransportService {
339 338 log.trace("Processing msg: {}", requestMsg);
340 339 TbProtoQueueMsg<TransportApiRequestMsg> protoMsg = new TbProtoQueueMsg<>(UUID.randomUUID(), TransportApiRequestMsg.newBuilder().setProvisionDeviceRequestMsg(requestMsg).build());
341 340 ListenableFuture<ProvisionDeviceResponseMsg> response = Futures.transform(transportApiRequestTemplate.send(protoMsg), tmp ->
342   - tmp.getValue().getProvisionDeviceResponseMsg()
343   - , MoreExecutors.directExecutor());
  341 + tmp.getValue().getProvisionDeviceResponseMsg()
  342 + , MoreExecutors.directExecutor());
344 343 AsyncCallbackTemplate.withCallback(response, callback::onSuccess, callback::onError, transportCallbackExecutor);
345 344 }
346 345
... ... @@ -580,12 +579,11 @@ public class DefaultTransportService implements TransportService {
580 579 if (log.isTraceEnabled()) {
581 580 log.trace("[{}] Processing msg: {}", toSessionId(sessionInfo), msg);
582 581 }
583   - if (!rateLimitEnabled) {
584   - return true;
585   - }
586 582 TenantId tenantId = new TenantId(new UUID(sessionInfo.getTenantIdMSB(), sessionInfo.getTenantIdLSB()));
587   - TbRateLimits rateLimits = perTenantLimits.computeIfAbsent(tenantId, id -> new TbRateLimits(perTenantLimitsConf));
588   - if (!rateLimits.tryConsume()) {
  583 +
  584 + TransportRateLimit tenantRateLimit = rateLimitService.getRateLimit(tenantId, TransportRateLimitType.TENANT_MAX_MSGS);
  585 +
  586 + if (!tenantRateLimit.tryConsume()) {
589 587 if (callback != null) {
590 588 callback.onError(new TbRateLimitsException(EntityType.TENANT));
591 589 }
... ... @@ -595,8 +593,8 @@ public class DefaultTransportService implements TransportService {
595 593 return false;
596 594 }
597 595 DeviceId deviceId = new DeviceId(new UUID(sessionInfo.getDeviceIdMSB(), sessionInfo.getDeviceIdLSB()));
598   - rateLimits = perDeviceLimits.computeIfAbsent(deviceId, id -> new TbRateLimits(perDevicesLimitsConf));
599   - if (!rateLimits.tryConsume()) {
  596 + TransportRateLimit deviceRateLimit = rateLimitService.getRateLimit(tenantId, deviceId, TransportRateLimitType.DEVICE_MAX_MSGS);
  597 + if (!deviceRateLimit.tryConsume()) {
600 598 if (callback != null) {
601 599 callback.onError(new TbRateLimitsException(EntityType.DEVICE));
602 600 }
... ... @@ -637,16 +635,40 @@ public class DefaultTransportService implements TransportService {
637 635 deregisterSession(md.getSessionInfo());
638 636 }
639 637 } else {
640   - if (toSessionMsg.hasDeviceProfileUpdateMsg()) {
641   - DeviceProfile deviceProfile = transportProfileCache.put(toSessionMsg.getDeviceProfileUpdateMsg().getData());
642   - if (deviceProfile != null) {
643   - onProfileUpdate(deviceProfile);
  638 + if (toSessionMsg.hasEntityUpdateMsg()) {
  639 + TransportProtos.EntityUpdateMsg msg = toSessionMsg.getEntityUpdateMsg();
  640 + EntityType entityType = EntityType.valueOf(msg.getEntityType());
  641 + if (EntityType.DEVICE_PROFILE.equals(entityType)) {
  642 + DeviceProfile deviceProfile = deviceProfileCache.put(msg.getData());
  643 + if (deviceProfile != null) {
  644 + onProfileUpdate(deviceProfile);
  645 + }
  646 + } else if (EntityType.TENANT_PROFILE.equals(entityType)) {
  647 + TenantProfileUpdateResult update = tenantProfileCache.put(msg.getData());
  648 + rateLimitService.update(update);
  649 + } else if (EntityType.TENANT.equals(entityType)) {
  650 + Optional<Tenant> profileOpt = dataDecodingEncodingService.decode(msg.getData().toByteArray());
  651 + if (profileOpt.isPresent()) {
  652 + Tenant tenant = profileOpt.get();
  653 + boolean updated = tenantProfileCache.put(tenant.getId(), tenant.getTenantProfileId());
  654 + if (updated) {
  655 + rateLimitService.update(tenant.getId());
  656 + }
  657 + }
  658 + }
  659 + } else if (toSessionMsg.hasEntityDeleteMsg()) {
  660 + TransportProtos.EntityDeleteMsg msg = toSessionMsg.getEntityDeleteMsg();
  661 + EntityType entityType = EntityType.valueOf(msg.getEntityType());
  662 + UUID entityUuid = new UUID(msg.getEntityIdMSB(), msg.getEntityIdLSB());
  663 + if (EntityType.DEVICE_PROFILE.equals(entityType)) {
  664 + deviceProfileCache.evict(new DeviceProfileId(new UUID(msg.getEntityIdMSB(), msg.getEntityIdLSB())));
  665 + } else if (EntityType.TENANT_PROFILE.equals(entityType)) {
  666 + tenantProfileCache.remove(new TenantProfileId(entityUuid));
  667 + } else if (EntityType.TENANT.equals(entityType)) {
  668 + rateLimitService.remove(new TenantId(entityUuid));
  669 + } else if (EntityType.DEVICE.equals(entityType)) {
  670 + rateLimitService.remove(new DeviceId(entityUuid));
644 671 }
645   - } else if (toSessionMsg.hasDeviceProfileDeleteMsg()) {
646   - transportProfileCache.evict(new DeviceProfileId(new UUID(
647   - toSessionMsg.getDeviceProfileDeleteMsg().getProfileIdMSB(),
648   - toSessionMsg.getDeviceProfileDeleteMsg().getProfileIdLSB()
649   - )));
650 672 } else {
651 673 //TODO: should we notify the device actor about missed session?
652 674 log.debug("[{}] Missing session.", sessionId);
... ... @@ -655,38 +677,6 @@ public class DefaultTransportService implements TransportService {
655 677 }
656 678
657 679 @Override
658   - public void getDeviceProfile(DeviceProfileId deviceProfileId, TransportServiceCallback<DeviceProfile> callback) {
659   - DeviceProfile deviceProfile = transportProfileCache.get(deviceProfileId);
660   - if (deviceProfile != null) {
661   - callback.onSuccess(deviceProfile);
662   - } else {
663   - log.trace("Processing device profile request: [{}]", deviceProfileId);
664   - TransportProtos.GetDeviceProfileRequestMsg msg = TransportProtos.GetDeviceProfileRequestMsg.newBuilder()
665   - .setProfileIdMSB(deviceProfileId.getId().getMostSignificantBits())
666   - .setProfileIdLSB(deviceProfileId.getId().getLeastSignificantBits())
667   - .build();
668   - TbProtoQueueMsg<TransportApiRequestMsg> protoMsg = new TbProtoQueueMsg<>(UUID.randomUUID(),
669   - TransportApiRequestMsg.newBuilder().setGetDeviceProfileRequestMsg(msg).build());
670   - AsyncCallbackTemplate.withCallback(transportApiRequestTemplate.send(protoMsg),
671   - response -> {
672   - ByteString devProfileBody = response.getValue().getGetDeviceProfileResponseMsg().getData();
673   - if (devProfileBody != null && !devProfileBody.isEmpty()) {
674   - DeviceProfile profile = transportProfileCache.put(devProfileBody);
675   - if (profile != null) {
676   - callback.onSuccess(profile);
677   - } else {
678   - log.warn("Failed to decode device profile: {}", devProfileBody);
679   - callback.onError(new IllegalArgumentException("Failed to decode device profile!"));
680   - }
681   - } else {
682   - log.warn("Failed to find device profile: [{}]", deviceProfileId);
683   - callback.onError(new IllegalArgumentException("Failed to find device profile!"));
684   - }
685   - }, callback::onError, transportCallbackExecutor);
686   - }
687   - }
688   -
689   - @Override
690 680 public void onProfileUpdate(DeviceProfile deviceProfile) {
691 681 long deviceProfileIdMSB = deviceProfile.getId().getId().getMostSignificantBits();
692 682 long deviceProfileIdLSB = deviceProfile.getId().getId().getLeastSignificantBits();
... ... @@ -750,7 +740,7 @@ public class DefaultTransportService implements TransportService {
750 740
751 741 private RuleChainId resolveRuleChainId(TransportProtos.SessionInfoProto sessionInfo) {
752 742 DeviceProfileId deviceProfileId = new DeviceProfileId(new UUID(sessionInfo.getDeviceProfileIdMSB(), sessionInfo.getDeviceProfileIdLSB()));
753   - DeviceProfile deviceProfile = transportProfileCache.get(deviceProfileId);
  743 + DeviceProfile deviceProfile = deviceProfileCache.get(deviceProfileId);
754 744 RuleChainId ruleChainId;
755 745 if (deviceProfile == null) {
756 746 log.warn("[{}] Device profile is null!", deviceProfileId);
... ... @@ -779,7 +769,7 @@ public class DefaultTransportService implements TransportService {
779 769 }
780 770 }
781 771
782   - private class StatsCallback implements TbQueueCallback {
  772 + private static class StatsCallback implements TbQueueCallback {
783 773 private final TbQueueCallback callback;
784 774 private final MessagesStats stats;
785 775
... ...
  1 +/**
  2 + * Copyright © 2016-2020 The Thingsboard Authors
  3 + *
  4 + * Licensed under the Apache License, Version 2.0 (the "License");
  5 + * you may not use this file except in compliance with the License.
  6 + * You may obtain a copy of the License at
  7 + *
  8 + * http://www.apache.org/licenses/LICENSE-2.0
  9 + *
  10 + * Unless required by applicable law or agreed to in writing, software
  11 + * distributed under the License is distributed on an "AS IS" BASIS,
  12 + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13 + * See the License for the specific language governing permissions and
  14 + * limitations under the License.
  15 + */
  16 +package org.thingsboard.server.common.transport.service;
  17 +
  18 +import com.google.protobuf.ByteString;
  19 +import lombok.extern.slf4j.Slf4j;
  20 +import org.springframework.beans.factory.annotation.Autowired;
  21 +import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
  22 +import org.springframework.context.annotation.Lazy;
  23 +import org.springframework.stereotype.Component;
  24 +import org.thingsboard.server.common.data.DeviceProfile;
  25 +import org.thingsboard.server.common.data.EntityType;
  26 +import org.thingsboard.server.common.data.TenantProfile;
  27 +import org.thingsboard.server.common.data.id.TenantId;
  28 +import org.thingsboard.server.common.data.id.TenantProfileId;
  29 +import org.thingsboard.server.common.transport.TransportService;
  30 +import org.thingsboard.server.common.transport.TransportTenantProfileCache;
  31 +import org.thingsboard.server.common.transport.profile.TenantProfileUpdateResult;
  32 +import org.thingsboard.server.common.transport.util.DataDecodingEncodingService;
  33 +import org.thingsboard.server.gen.transport.TransportProtos;
  34 +import org.thingsboard.server.queue.discovery.TenantRoutingInfo;
  35 +import org.thingsboard.server.queue.discovery.TenantRoutingInfoService;
  36 +
  37 +import java.util.Collections;
  38 +import java.util.Optional;
  39 +import java.util.Set;
  40 +import java.util.concurrent.ConcurrentHashMap;
  41 +import java.util.concurrent.ConcurrentMap;
  42 +import java.util.concurrent.locks.Lock;
  43 +import java.util.concurrent.locks.ReentrantLock;
  44 +
  45 +@Component
  46 +@ConditionalOnExpression("('${service.type:null}'=='monolith' && '${transport.api_enabled:true}'=='true') || '${service.type:null}'=='tb-transport'")
  47 +@Slf4j
  48 +public class DefaultTransportTenantProfileCache implements TransportTenantProfileCache {
  49 +
  50 + private final Lock tenantProfileFetchLock = new ReentrantLock();
  51 + private final ConcurrentMap<TenantProfileId, TenantProfile> profiles = new ConcurrentHashMap<>();
  52 + private final ConcurrentMap<TenantId, TenantProfileId> tenantIds = new ConcurrentHashMap<>();
  53 + private final ConcurrentMap<TenantProfileId, Set<TenantId>> tenantProfileIds = new ConcurrentHashMap<>();
  54 + private final DataDecodingEncodingService dataDecodingEncodingService;
  55 +
  56 + private TransportService transportService;
  57 +
  58 + @Lazy
  59 + @Autowired
  60 + public void setTransportService(TransportService transportService) {
  61 + this.transportService = transportService;
  62 + }
  63 +
  64 + public DefaultTransportTenantProfileCache(DataDecodingEncodingService dataDecodingEncodingService) {
  65 + this.dataDecodingEncodingService = dataDecodingEncodingService;
  66 + }
  67 +
  68 + @Override
  69 + public TenantProfile get(TenantId tenantId) {
  70 + return getTenantProfile(tenantId);
  71 + }
  72 +
  73 + @Override
  74 + public TenantProfileUpdateResult put(ByteString profileBody) {
  75 + Optional<TenantProfile> profileOpt = dataDecodingEncodingService.decode(profileBody.toByteArray());
  76 + if (profileOpt.isPresent()) {
  77 + TenantProfile newProfile = profileOpt.get();
  78 + log.trace("[{}] put: {}", newProfile.getId(), newProfile);
  79 + return new TenantProfileUpdateResult(newProfile, tenantProfileIds.get(newProfile.getId()));
  80 + } else {
  81 + log.warn("Failed to decode profile: {}", profileBody.toString());
  82 + return new TenantProfileUpdateResult(null, Collections.emptySet());
  83 + }
  84 + }
  85 +
  86 + @Override
  87 + public boolean put(TenantId tenantId, TenantProfileId profileId) {
  88 + log.trace("[{}] put: {}", tenantId, profileId);
  89 + TenantProfileId oldProfileId = tenantIds.get(tenantId);
  90 + if (oldProfileId != null && !oldProfileId.equals(profileId)) {
  91 + tenantProfileIds.computeIfAbsent(oldProfileId, id -> ConcurrentHashMap.newKeySet()).remove(tenantId);
  92 + tenantIds.put(tenantId, profileId);
  93 + tenantProfileIds.computeIfAbsent(profileId, id -> ConcurrentHashMap.newKeySet()).add(tenantId);
  94 + return true;
  95 + } else {
  96 + return false;
  97 + }
  98 + }
  99 +
  100 + @Override
  101 + public Set<TenantId> remove(TenantProfileId profileId) {
  102 + Set<TenantId> tenants = tenantProfileIds.remove(profileId);
  103 + if (tenants != null) {
  104 + tenants.forEach(tenantIds::remove);
  105 + }
  106 + profiles.remove(profileId);
  107 + return tenants;
  108 + }
  109 +
  110 + private TenantProfile getTenantProfile(TenantId tenantId) {
  111 + TenantProfile profile = null;
  112 + TenantProfileId tenantProfileId = tenantIds.get(tenantId);
  113 + if (tenantProfileId != null) {
  114 + profile = profiles.get(tenantProfileId);
  115 + }
  116 + if (profile == null) {
  117 + tenantProfileFetchLock.lock();
  118 + try {
  119 + tenantProfileId = tenantIds.get(tenantId);
  120 + if (tenantProfileId != null) {
  121 + profile = profiles.get(tenantProfileId);
  122 + }
  123 + if (profile == null) {
  124 + TransportProtos.GetEntityProfileRequestMsg msg = TransportProtos.GetEntityProfileRequestMsg.newBuilder()
  125 + .setEntityType(EntityType.TENANT.name())
  126 + .setEntityIdMSB(tenantId.getId().getMostSignificantBits())
  127 + .setEntityIdLSB(tenantId.getId().getLeastSignificantBits())
  128 + .build();
  129 + TransportProtos.GetEntityProfileResponseMsg routingInfo = transportService.getRoutingInfo(msg);
  130 + Optional<TenantProfile> profileOpt = dataDecodingEncodingService.decode(routingInfo.getData().toByteArray());
  131 + if (profileOpt.isPresent()) {
  132 + profile = profileOpt.get();
  133 + TenantProfile existingProfile = profiles.get(profile.getId());
  134 + if (existingProfile != null) {
  135 + profile = existingProfile;
  136 + } else {
  137 + profiles.put(profile.getId(), profile);
  138 + }
  139 + tenantProfileIds.computeIfAbsent(profile.getId(), id -> ConcurrentHashMap.newKeySet()).add(tenantId);
  140 + tenantIds.put(tenantId, profile.getId());
  141 + } else {
  142 + log.warn("[{}] Can't decode tenant profile: {}", tenantId, routingInfo.getData());
  143 + throw new RuntimeException("Can't decode tenant profile!");
  144 + }
  145 + }
  146 + } finally {
  147 + tenantProfileFetchLock.unlock();
  148 + }
  149 + }
  150 + return profile;
  151 + }
  152 +
  153 +
  154 +}
... ...
... ... @@ -16,14 +16,11 @@
16 16 package org.thingsboard.server.common.transport.service;
17 17
18 18 import lombok.extern.slf4j.Slf4j;
19   -import org.springframework.beans.factory.annotation.Autowired;
20 19 import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
21   -import org.springframework.context.annotation.Lazy;
22 20 import org.springframework.stereotype.Service;
  21 +import org.thingsboard.server.common.data.TenantProfile;
23 22 import org.thingsboard.server.common.data.id.TenantId;
24   -import org.thingsboard.server.common.transport.TransportService;
25   -import org.thingsboard.server.gen.transport.TransportProtos.GetTenantRoutingInfoRequestMsg;
26   -import org.thingsboard.server.gen.transport.TransportProtos.GetTenantRoutingInfoResponseMsg;
  23 +import org.thingsboard.server.common.transport.TransportTenantProfileCache;
27 24 import org.thingsboard.server.queue.discovery.TenantRoutingInfo;
28 25 import org.thingsboard.server.queue.discovery.TenantRoutingInfoService;
29 26
... ... @@ -32,21 +29,16 @@ import org.thingsboard.server.queue.discovery.TenantRoutingInfoService;
32 29 @ConditionalOnExpression("'${service.type:null}'=='tb-transport'")
33 30 public class TransportTenantRoutingInfoService implements TenantRoutingInfoService {
34 31
35   - private TransportService transportService;
  32 + private TransportTenantProfileCache tenantProfileCache;
36 33
37   - @Lazy
38   - @Autowired
39   - public void setTransportService(TransportService transportService) {
40   - this.transportService = transportService;
  34 + public TransportTenantRoutingInfoService(TransportTenantProfileCache tenantProfileCache) {
  35 + this.tenantProfileCache = tenantProfileCache;
41 36 }
42 37
43 38 @Override
44 39 public TenantRoutingInfo getRoutingInfo(TenantId tenantId) {
45   - GetTenantRoutingInfoRequestMsg msg = GetTenantRoutingInfoRequestMsg.newBuilder()
46   - .setTenantIdMSB(tenantId.getId().getMostSignificantBits())
47   - .setTenantIdLSB(tenantId.getId().getLeastSignificantBits())
48   - .build();
49   - GetTenantRoutingInfoResponseMsg routingInfo = transportService.getRoutingInfo(msg);
50   - return new TenantRoutingInfo(tenantId, routingInfo.getIsolatedTbCore(), routingInfo.getIsolatedTbRuleEngine());
  40 + TenantProfile profile = tenantProfileCache.get(tenantId);
  41 + return new TenantRoutingInfo(tenantId, profile.isIsolatedTbCore(), profile.isIsolatedTbRuleEngine());
51 42 }
  43 +
52 44 }
... ...
... ... @@ -15,8 +15,6 @@
15 15 */
16 16 package org.thingsboard.server.common.transport.util;
17 17
18   -import org.thingsboard.server.common.msg.TbActorMsg;
19   -
20 18 import java.util.Optional;
21 19
22 20 public interface DataDecodingEncodingService {
... ...
... ... @@ -49,10 +49,6 @@ transport:
49 49 sessions:
50 50 inactivity_timeout: "${TB_TRANSPORT_SESSIONS_INACTIVITY_TIMEOUT:300000}"
51 51 report_timeout: "${TB_TRANSPORT_SESSIONS_REPORT_TIMEOUT:30000}"
52   - rate_limits:
53   - enabled: "${TB_TRANSPORT_RATE_LIMITS_ENABLED:false}"
54   - tenant: "${TB_TRANSPORT_RATE_LIMITS_TENANT:1000:1,20000:60}"
55   - device: "${TB_TRANSPORT_RATE_LIMITS_DEVICE:10:1,300:60}"
56 52 json:
57 53 # Cast String data types to Numeric if possible when processing Telemetry/Attributes JSON
58 54 type_cast_enabled: "${JSON_TYPE_CAST_ENABLED:true}"
... ...
... ... @@ -42,10 +42,6 @@ transport:
42 42 sessions:
43 43 inactivity_timeout: "${TB_TRANSPORT_SESSIONS_INACTIVITY_TIMEOUT:300000}"
44 44 report_timeout: "${TB_TRANSPORT_SESSIONS_REPORT_TIMEOUT:30000}"
45   - rate_limits:
46   - enabled: "${TB_TRANSPORT_RATE_LIMITS_ENABLED:false}"
47   - tenant: "${TB_TRANSPORT_RATE_LIMITS_TENANT:1000:1,20000:60}"
48   - device: "${TB_TRANSPORT_RATE_LIMITS_DEVICE:10:1,300:60}"
49 45 json:
50 46 # Cast String data types to Numeric if possible when processing Telemetry/Attributes JSON
51 47 type_cast_enabled: "${JSON_TYPE_CAST_ENABLED:true}"
... ...
... ... @@ -71,10 +71,6 @@ transport:
71 71 sessions:
72 72 inactivity_timeout: "${TB_TRANSPORT_SESSIONS_INACTIVITY_TIMEOUT:300000}"
73 73 report_timeout: "${TB_TRANSPORT_SESSIONS_REPORT_TIMEOUT:30000}"
74   - rate_limits:
75   - enabled: "${TB_TRANSPORT_RATE_LIMITS_ENABLED:false}"
76   - tenant: "${TB_TRANSPORT_RATE_LIMITS_TENANT:1000:1,20000:60}"
77   - device: "${TB_TRANSPORT_RATE_LIMITS_DEVICE:10:1,300:60}"
78 74 json:
79 75 # Cast String data types to Numeric if possible when processing Telemetry/Attributes JSON
80 76 type_cast_enabled: "${JSON_TYPE_CAST_ENABLED:true}"
... ...