dendi-wang/thingsboard
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Docker improvements and Telemetry Service in cluster mode

Browse Source
This commit is contained in:
Andrew Shvayka 2018-05-16 19:19:38 +03:00
parent e4f2c8dffa
commit 2a58bd318a
38 changed files with 143 additions and 167 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
Dockerfile.cassandra
View File

@ -1,13 +0,0 @@
FROM cassandra
WORKDIR /opt/cassandra
COPY dao/src/main/resources/cassandra/schema.cql /opt/cassandra
COPY entrypoint-with-db-init.sh /opt/cassandra/entrypoint-with-db-init.sh
RUN chmod +x /opt/cassandra/entrypoint-with-db-init.sh
ENTRYPOINT ["/opt/cassandra/entrypoint-with-db-init.sh"]
CMD ["cassandra", "-f"]

2
application/src/main/java/org/thingsboard/server/actors/plugin/ValidationCallback.java
View File

@ -15,7 +15,7 @@
*/
package org.thingsboard.server.actors.plugin;
import com.hazelcast.util.function.Consumer;
import java.util.function.Consumer;
import org.thingsboard.server.extensions.api.exception.AccessDeniedException;
import org.thingsboard.server.extensions.api.exception.EntityNotFoundException;
import org.thingsboard.server.extensions.api.exception.InternalErrorException;

3
application/src/main/java/org/thingsboard/server/actors/rpc/BasicRpcSessionListener.java
View File

@ -17,7 +17,6 @@ package org.thingsboard.server.actors.rpc;
import akka.actor.ActorRef;
import lombok.extern.slf4j.Slf4j;
import org.thingsboard.server.actors.ActorSystemContext;
import org.thingsboard.server.actors.service.ActorService;
import org.thingsboard.server.gen.cluster.ClusterAPIProtos;
import org.thingsboard.server.service.cluster.rpc.GrpcSession;
@ -58,7 +57,7 @@ public class BasicRpcSessionListener implements GrpcSessionListener {
log.trace("{} Service [{}] received session actor msg {}", getType(session),
session.getRemoteServer(),
clusterMessage);
service.onRecievedMsg(clusterMessage);
service.onReceivedMsg(session.getRemoteServer(), clusterMessage);
}
@Override

8
application/src/main/java/org/thingsboard/server/actors/rpc/RpcManagerActor.java
View File

@ -89,9 +89,9 @@ public class RpcManagerActor extends ContextAwareActor {
}
private void onMsg(ClusterAPIProtos.ClusterMessage msg) {
if (msg.hasServerAdresss()) {
ServerAddress address = new ServerAddress(msg.getServerAdresss().getHost(),
msg.getServerAdresss().getPort());
if (msg.hasServerAddress()) {
ServerAddress address = new ServerAddress(msg.getServerAddress().getHost(),
msg.getServerAddress().getPort());
SessionActorInfo session = sessionActors.get(address);
if (session != null) {
log.debug("{} Forwarding msg to session actor", address);
@ -102,7 +102,7 @@ public class RpcManagerActor extends ContextAwareActor {
if (queue == null) {
queue = new LinkedList<>();
pendingMsgs.put(new ServerAddress(
msg.getServerAdresss().getHost(), msg.getServerAdresss().getPort()), queue);
msg.getServerAddress().getHost(), msg.getServerAddress().getPort()), queue);
}
queue.add(msg);
}

4
application/src/main/java/org/thingsboard/server/actors/rpc/RpcSessionActor.java
View File

@ -78,7 +78,7 @@ public class RpcSessionActor extends ContextAwareActor {
private void initSession(RpcSessionCreateRequestMsg msg) {
log.info("[{}] Initializing session", context().self());
ServerAddress remoteServer = msg.getRemoteAddress();
listener = new BasicRpcSessionListener(systemContext, context().parent(), context().self());
listener = new BasicRpcSessionListener(systemContext.getActorService(), context().parent(), context().self());
if (msg.getRemoteAddress() == null) {
// Server session
session = new GrpcSession(listener);
@ -119,7 +119,7 @@ public class RpcSessionActor extends ContextAwareActor {
private ClusterAPIProtos.ClusterMessage toConnectMsg() {
ServerAddress instance = systemContext.getDiscoveryService().getCurrentServer().getServerAddress();
return ClusterAPIProtos.ClusterMessage.newBuilder().setMessageType(CONNECT_RPC_MESSAGE).setServerAdresss(
return ClusterAPIProtos.ClusterMessage.newBuilder().setMessageType(CONNECT_RPC_MESSAGE).setServerAddress(
ClusterAPIProtos.ServerAddress.newBuilder().setHost(instance.getHost())
.setPort(instance.getPort()).build()).build();
}

14
application/src/main/java/org/thingsboard/server/actors/service/DefaultActorService.java
View File

@ -1,12 +1,12 @@
/**
* Copyright © 2016-2018 The Thingsboard Authors
* <p>
*
* 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
* <p>
* http://www.apache.org/licenses/LICENSE-2.0
* <p>
*
* 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.
@ -56,8 +56,6 @@ import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import static org.thingsboard.server.gen.cluster.ClusterAPIProtos.MessageType.CLUSTER_ACTOR_MESSAGE;
import static org.thingsboard.server.gen.cluster.ClusterAPIProtos.MessageType.CLUSTER_ACTOR_MESSAGE_VALUE;
import static org.thingsboard.server.gen.cluster.ClusterAPIProtos.MessageType.CLUSTER_NETWORK_SERVER_DATA_MESSAGE;
@Service
@Slf4j
@ -211,8 +209,8 @@ public class DefaultActorService implements ActorService {
}
@Override
public void onRecievedMsg(ClusterAPIProtos.ClusterMessage msg) {
ServerAddress serverAddress = new ServerAddress(msg.getServerAddress().getHost(), msg.getServerAddress().getPort());
public void onReceivedMsg(ServerAddress source, ClusterAPIProtos.ClusterMessage msg) {
ServerAddress serverAddress = new ServerAddress(source.getHost(), source.getPort());
switch (msg.getMessageType()) {
case CLUSTER_ACTOR_MESSAGE:
java.util.Optional<TbActorMsg> decodedMsg = actorContext.getEncodingService()

2
application/src/main/java/org/thingsboard/server/actors/session/AbstractSessionActorMsgProcessor.java
View File

@ -87,7 +87,7 @@ abstract class AbstractSessionActorMsgProcessor extends AbstractContextAwareMsgP
return address;
}
protected Optional<ServerAddress> forwardToAppActorIfAdressChanged(ActorContext ctx, DeviceToDeviceActorMsg toForward, Optional<ServerAddress> oldAddress) {
protected Optional<ServerAddress> forwardToAppActorIfAddressChanged(ActorContext ctx, DeviceToDeviceActorMsg toForward, Optional<ServerAddress> oldAddress) {
Optional<ServerAddress> newAddress = systemContext.getRoutingService().resolveById(toForward.getDeviceId());
if (!newAddress.equals(oldAddress)) {

2
application/src/main/java/org/thingsboard/server/actors/session/SyncMsgProcessor.java
View File

@ -73,7 +73,7 @@ class SyncMsgProcessor extends AbstractSessionActorMsgProcessor {
@Override
public void processClusterEvent(ActorContext context, ClusterEventMsg msg) {
if (pendingResponse) {
Optional<ServerAddress> newTargetServer = forwardToAppActorIfAdressChanged(context, pendingMsg, currentTargetServer);
Optional<ServerAddress> newTargetServer = forwardToAppActorIfAddressChanged(context, pendingMsg, currentTargetServer);
if (logger.isDebugEnabled()) {
if (!newTargetServer.equals(currentTargetServer)) {
if (newTargetServer.isPresent()) {

8
application/src/main/java/org/thingsboard/server/service/cluster/discovery/ZkDiscoveryService.java
View File

@ -32,9 +32,11 @@ import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.event.ApplicationReadyEvent;
import org.springframework.context.ApplicationListener;
import org.springframework.context.annotation.Lazy;
import org.springframework.stereotype.Service;
import org.springframework.util.Assert;
import org.thingsboard.server.common.msg.cluster.ServerAddress;
import org.thingsboard.server.service.telemetry.TelemetrySubscriptionService;
import org.thingsboard.server.utils.MiscUtils;
import javax.annotation.PostConstruct;
@ -68,6 +70,10 @@ public class ZkDiscoveryService implements DiscoveryService, PathChildrenCacheLi
@Autowired
private ServerInstanceService serverInstance;
@Autowired
@Lazy
private TelemetrySubscriptionService tsSubService;
private final List<DiscoveryServiceListener> listeners = new CopyOnWriteArrayList<>();
private CuratorFramework client;
@ -196,12 +202,14 @@ public class ZkDiscoveryService implements DiscoveryService, PathChildrenCacheLi
log.info("Processing [{}] event for [{}:{}]", pathChildrenCacheEvent.getType(), instance.getHost(), instance.getPort());
switch (pathChildrenCacheEvent.getType()) {
case CHILD_ADDED:
tsSubService.onClusterUpdate();
listeners.forEach(listener -> listener.onServerAdded(instance));
break;
case CHILD_UPDATED:
listeners.forEach(listener -> listener.onServerUpdated(instance));
break;
case CHILD_REMOVED:
tsSubService.onClusterUpdate();
listeners.forEach(listener -> listener.onServerRemoved(instance));
break;
default:

8
application/src/main/java/org/thingsboard/server/service/cluster/rpc/ClusterGrpcService.java
View File

@ -1,12 +1,12 @@
/**
* Copyright © 2016-2018 The Thingsboard Authors
* <p>
*
* 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
* <p>
* http://www.apache.org/licenses/LICENSE-2.0
* <p>
*
* 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.

2
application/src/main/java/org/thingsboard/server/service/cluster/rpc/GrpcSession.java
View File

@ -61,7 +61,7 @@ public final class GrpcSession implements Closeable {
public void onNext(ClusterAPIProtos.ClusterMessage clusterMessage) {
if (!connected && clusterMessage.getMessageType() == ClusterAPIProtos.MessageType.CONNECT_RPC_MESSAGE) {
connected = true;
ServerAddress rpcAddress = new ServerAddress(clusterMessage.getServerAdresss().getHost(), clusterMessage.getServerAdresss().getPort());
ServerAddress rpcAddress = new ServerAddress(clusterMessage.getServerAddress().getHost(), clusterMessage.getServerAddress().getPort());
remoteServer = new ServerAddress(rpcAddress.getHost(), rpcAddress.getPort());
listener.onConnected(GrpcSession.this);
}

3
application/src/main/java/org/thingsboard/server/service/cluster/rpc/RpcMsgListener.java
View File

@ -17,6 +17,7 @@ package org.thingsboard.server.service.cluster.rpc;
import org.thingsboard.server.actors.rpc.RpcBroadcastMsg;
import org.thingsboard.server.actors.rpc.RpcSessionCreateRequestMsg;
import org.thingsboard.server.common.msg.cluster.ServerAddress;
import org.thingsboard.server.gen.cluster.ClusterAPIProtos;
/**
@ -24,7 +25,7 @@ import org.thingsboard.server.gen.cluster.ClusterAPIProtos;
*/
public interface RpcMsgListener {
void onRecievedMsg(ClusterAPIProtos.ClusterMessage msg);
void onReceivedMsg(ServerAddress remoteServer, ClusterAPIProtos.ClusterMessage msg);
void onSendMsg(ClusterAPIProtos.ClusterMessage msg);
void onRpcSessionCreateRequestMsg(RpcSessionCreateRequestMsg msg);
void onBroadcastMsg(RpcBroadcastMsg msg);

6
application/src/main/java/org/thingsboard/server/service/encoding/ProtoWithJavaSerializationDecodingEncodingService.java
View File

@ -25,7 +25,7 @@ import org.thingsboard.server.gen.cluster.ClusterAPIProtos;
import java.util.Optional;
import static org.thingsboard.server.gen.cluster.ClusterAPIProtos.MessageType.CLUSTER_NETWORK_SERVER_DATA_MESSAGE;
import static org.thingsboard.server.gen.cluster.ClusterAPIProtos.MessageType.CLUSTER_ACTOR_MESSAGE;
@Slf4j
@ -55,12 +55,12 @@ public class ProtoWithJavaSerializationDecodingEncodingService implements DataDe
TbActorMsg msg) {
return ClusterAPIProtos.ClusterMessage
.newBuilder()
.setServerAdresss(ClusterAPIProtos.ServerAddress
.setServerAddress(ClusterAPIProtos.ServerAddress
.newBuilder()
.setHost(serverAddress.getHost())
.setPort(serverAddress.getPort())
.build())
.setMessageType(CLUSTER_NETWORK_SERVER_DATA_MESSAGE)
.setMessageType(CLUSTER_ACTOR_MESSAGE)
.setPayload(ByteString.copyFrom(encode(msg))).build();
}

8
application/src/main/java/org/thingsboard/server/service/telemetry/DefaultTelemetrySubscriptionService.java
View File

@ -1,12 +1,12 @@
/**
* Copyright © 2016-2018 The Thingsboard Authors
* <p>
*
* 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
* <p>
* http://www.apache.org/licenses/LICENSE-2.0
* <p>
*
* 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.

2
application/src/main/java/org/thingsboard/server/service/telemetry/DefaultTelemetryWebSocketService.java
View File

@ -21,7 +21,7 @@ import com.google.common.base.Function;
import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import com.hazelcast.util.function.Consumer;
import java.util.function.Consumer;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

2
application/src/main/resources/thingsboard.yml
View File

@ -107,7 +107,7 @@ mqtt:
# CoAP server parameters
coap:
# Enable/disable coap transport protocol.
enabled: "${COAP_ENABLED:true}"
enabled: "${COAP_ENABLED:false}"
bind_address: "${COAP_BIND_ADDRESS:0.0.0.0}"
bind_port: "${COAP_BIND_PORT:5683}"
adaptor: "${COAP_ADAPTOR_NAME:JsonCoapAdaptor}"

15
application/src/test/java/org/thingsboard/server/mqtt/DbConfigurationTestRule.java
View File

@ -1,3 +1,18 @@
/**
* Copyright © 2016-2018 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.mqtt;
import org.junit.rules.TestRule;

27
base-docker-compose.yml
View File

@ -1,3 +1,19 @@
#
# Copyright © 2016-2018 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.
#
version: '3.3'
services:
zookeeper:
@ -7,20 +23,15 @@ services:
ports:
- "2181:2181"
cassandra-tb:
build:
context: .
dockerfile: Dockerfile.cassandra
image: cassandra
cassandra:
image: cassandra:3.11.2
networks:
- core
ports:
- "7199:7199"
- "9160:9160"
- "9042:9042"
volumes:
- /cassandra:/var/lib/cassandra
- ./db-schema:/docker-entrypoint-initdb.d/
redis:
image: redis:4.0
networks:

12
dao/pom.xml
View File

@ -152,22 +152,10 @@
<groupId>org.apache.curator</groupId>
<artifactId>curator-x-discovery</artifactId>
</dependency>
<dependency>
<groupId>com.hazelcast</groupId>
<artifactId>hazelcast-zookeeper</artifactId>
</dependency>
<dependency>
<groupId>com.hazelcast</groupId>
<artifactId>hazelcast</artifactId>
</dependency>
<dependency>
<groupId>com.github.ben-manes.caffeine</groupId>
<artifactId>caffeine</artifactId>
</dependency>
<dependency>
<groupId>com.hazelcast</groupId>
<artifactId>hazelcast-spring</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-autoconfigure</artifactId>

32
dao/src/main/java/org/thingsboard/server/dao/alarm/BaseAlarmService.java
View File

@ -222,15 +222,14 @@ public class BaseAlarmService extends AbstractEntityService implements AlarmServ
public ListenableFuture<AlarmInfo> findAlarmInfoByIdAsync(AlarmId alarmId) {
log.trace("Executing findAlarmInfoByIdAsync [{}]", alarmId);
validateId(alarmId, "Incorrect alarmId " + alarmId);
return Futures.transform(alarmDao.findAlarmByIdAsync(alarmId.getId()),
(AsyncFunction<Alarm, AlarmInfo>) alarm1 -> {
AlarmInfo alarmInfo = new AlarmInfo(alarm1);
return Futures.transformAsync(alarmDao.findAlarmByIdAsync(alarmId.getId()),
a -> {
AlarmInfo alarmInfo = new AlarmInfo(a);
return Futures.transform(
entityService.fetchEntityNameAsync(alarmInfo.getOriginator()), (Function<String, AlarmInfo>)
originatorName -> {
alarmInfo.setOriginatorName(originatorName);
return alarmInfo;
}
entityService.fetchEntityNameAsync(alarmInfo.getOriginator()), originatorName -> {
alarmInfo.setOriginatorName(originatorName);
return alarmInfo;
}
);
});
}
@ -239,18 +238,17 @@ public class BaseAlarmService extends AbstractEntityService implements AlarmServ
public ListenableFuture<TimePageData<AlarmInfo>> findAlarms(AlarmQuery query) {
ListenableFuture<List<AlarmInfo>> alarms = alarmDao.findAlarms(query);
if (query.getFetchOriginator() != null && query.getFetchOriginator().booleanValue()) {
alarms = Futures.transform(alarms, (AsyncFunction<List<AlarmInfo>, List<AlarmInfo>>) input -> {
alarms = Futures.transformAsync(alarms, input -> {
List<ListenableFuture<AlarmInfo>> alarmFutures = new ArrayList<>(input.size());
for (AlarmInfo alarmInfo : input) {
alarmFutures.add(Futures.transform(
entityService.fetchEntityNameAsync(alarmInfo.getOriginator()), (Function<String, AlarmInfo>)
originatorName -> {
if (originatorName == null) {
originatorName = "Deleted";
}
alarmInfo.setOriginatorName(originatorName);
return alarmInfo;
}
entityService.fetchEntityNameAsync(alarmInfo.getOriginator()), originatorName -> {
if (originatorName == null) {
originatorName = "Deleted";
}
alarmInfo.setOriginatorName(originatorName);
return alarmInfo;
}
));
}
return Futures.successfulAsList(alarmFutures);

4
dao/src/main/java/org/thingsboard/server/dao/alarm/CassandraAlarmDao.java
View File

@ -102,12 +102,12 @@ public class CassandraAlarmDao extends CassandraAbstractModelDao<AlarmEntity, Al
}
String relationType = BaseAlarmService.ALARM_RELATION_PREFIX + searchStatusName;
ListenableFuture<List<EntityRelation>> relations = relationDao.findRelations(affectedEntity, relationType, RelationTypeGroup.ALARM, EntityType.ALARM, query.getPageLink());
return Futures.transform(relations, (AsyncFunction<List<EntityRelation>, List<AlarmInfo>>) input -> {
return Futures.transformAsync(relations, input -> {
List<ListenableFuture<AlarmInfo>> alarmFutures = new ArrayList<>(input.size());
for (EntityRelation relation : input) {
alarmFutures.add(Futures.transform(
findAlarmByIdAsync(relation.getTo().getId()),
(Function<Alarm, AlarmInfo>) AlarmInfo::new));
AlarmInfo::new));
}
return Futures.successfulAsList(alarmFutures);
});

4
dao/src/main/java/org/thingsboard/server/dao/asset/BaseAssetService.java
View File

@ -194,10 +194,10 @@ public class BaseAssetService extends AbstractEntityService implements AssetServ
@Override
public ListenableFuture<List<Asset>> findAssetsByQuery(AssetSearchQuery query) {
ListenableFuture<List<EntityRelation>> relations = relationService.findByQuery(query.toEntitySearchQuery());
ListenableFuture<List<Asset>> assets = Futures.transform(relations, (AsyncFunction<List<EntityRelation>, List<Asset>>) relations1 -> {
ListenableFuture<List<Asset>> assets = Futures.transformAsync(relations, r -> {
EntitySearchDirection direction = query.toEntitySearchQuery().getParameters().getDirection();
List<ListenableFuture<Asset>> futures = new ArrayList<>();
for (EntityRelation relation : relations1) {
for (EntityRelation relation : r) {
EntityId entityId = direction == EntitySearchDirection.FROM ? relation.getTo() : relation.getFrom();
if (entityId.getEntityType() == EntityType.ASSET) {
futures.add(findAssetByIdAsync(new AssetId(entityId.getId())));

9
dao/src/main/java/org/thingsboard/server/dao/cassandra/AbstractCassandraCluster.java
View File

@ -18,8 +18,12 @@ package org.thingsboard.server.dao.cassandra;
import com.datastax.driver.core.*;
import com.datastax.driver.core.ProtocolOptions.Compression;
import com.datastax.driver.mapping.DefaultPropertyMapper;
import com.datastax.driver.mapping.Mapper;
import com.datastax.driver.mapping.MappingConfiguration;
import com.datastax.driver.mapping.MappingManager;
import com.datastax.driver.mapping.PropertyAccessStrategy;
import com.datastax.driver.mapping.PropertyMapper;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
@ -145,7 +149,10 @@ public abstract class AbstractCassandraCluster {
} else {
session = cluster.connect();
}
mappingManager = new MappingManager(session);
DefaultPropertyMapper propertyMapper = new DefaultPropertyMapper();
propertyMapper.setPropertyAccessStrategy(PropertyAccessStrategy.FIELDS);
MappingConfiguration configuration = MappingConfiguration.builder().withPropertyMapper(propertyMapper).build();
mappingManager = new MappingManager(session, configuration);
break;
} catch (Exception e) {
log.warn("Failed to initialize cassandra cluster due to {}. Will retry in {} ms", e.getMessage(), initRetryInterval);

2
dao/src/main/java/org/thingsboard/server/dao/dashboard/CassandraDashboardInfoDao.java
View File

@ -77,7 +77,7 @@ public class CassandraDashboardInfoDao extends CassandraAbstractSearchTextDao<Da
ListenableFuture<List<EntityRelation>> relations = relationDao.findRelations(new CustomerId(customerId), EntityRelation.CONTAINS_TYPE, RelationTypeGroup.DASHBOARD, EntityType.DASHBOARD, pageLink);
return Futures.transform(relations, (AsyncFunction<List<EntityRelation>, List<DashboardInfo>>) input -> {
return Futures.transformAsync(relations, input -> {
List<ListenableFuture<DashboardInfo>> dashboardFutures = new ArrayList<>(input.size());
for (EntityRelation relation : input) {
dashboardFutures.add(findByIdAsync(relation.getTo().getId()));

4
dao/src/main/java/org/thingsboard/server/dao/device/DeviceServiceImpl.java
View File

@ -227,10 +227,10 @@ public class DeviceServiceImpl extends AbstractEntityService implements DeviceSe
@Override
public ListenableFuture<List<Device>> findDevicesByQuery(DeviceSearchQuery query) {
ListenableFuture<List<EntityRelation>> relations = relationService.findByQuery(query.toEntitySearchQuery());
ListenableFuture<List<Device>> devices = Futures.transform(relations, (AsyncFunction<List<EntityRelation>, List<Device>>) relations1 -> {
ListenableFuture<List<Device>> devices = Futures.transformAsync(relations, r -> {
EntitySearchDirection direction = query.toEntitySearchQuery().getParameters().getDirection();
List<ListenableFuture<Device>> futures = new ArrayList<>();
for (EntityRelation relation : relations1) {
for (EntityRelation relation : r) {
EntityId entityId = direction == EntitySearchDirection.FROM ? relation.getTo() : relation.getFrom();
if (entityId.getEntityType() == EntityType.DEVICE) {
futures.add(findDeviceByIdAsync(new DeviceId(entityId.getId())));

4
dao/src/main/java/org/thingsboard/server/dao/nosql/RateLimitedResultSetFuture.java
View File

@ -36,14 +36,14 @@ public class RateLimitedResultSetFuture implements ResultSetFuture {
private final ListenableFuture<Void> rateLimitFuture;
public RateLimitedResultSetFuture(Session session, AsyncRateLimiter rateLimiter, Statement statement) {
this.rateLimitFuture = Futures.withFallback(rateLimiter.acquireAsync(), t -> {
this.rateLimitFuture = Futures.catchingAsync(rateLimiter.acquireAsync(), Throwable.class, t -> {
if (!(t instanceof BufferLimitException)) {
rateLimiter.release();
}
return Futures.immediateFailedFuture(t);
});
this.originalFuture = Futures.transform(rateLimitFuture,
(Function<Void, ResultSetFuture>) i -> executeAsyncWithRelease(rateLimiter, session, statement));
i -> executeAsyncWithRelease(rateLimiter, session, statement));
}

29
dao/src/main/java/org/thingsboard/server/dao/relation/BaseRelationService.java
View File

@ -227,8 +227,8 @@ public class BaseRelationService implements RelationService {
inboundRelationsListTo.add(relationDao.findAllByTo(entity, typeGroup));
}
ListenableFuture<List<List<EntityRelation>>> inboundRelationsTo = Futures.allAsList(inboundRelationsListTo);
ListenableFuture<List<Boolean>> inboundDeletions = Futures.transform(inboundRelationsTo,
(AsyncFunction<List<List<EntityRelation>>, List<Boolean>>) relations -> {
ListenableFuture<List<Boolean>> inboundDeletions = Futures.transformAsync(inboundRelationsTo,
relations -> {
List<ListenableFuture<Boolean>> results = getListenableFutures(relations, cache, true);
return Futures.allAsList(results);
});
@ -240,7 +240,7 @@ public class BaseRelationService implements RelationService {
inboundRelationsListFrom.add(relationDao.findAllByTo(entity, typeGroup));
}
ListenableFuture<List<List<EntityRelation>>> inboundRelationsFrom = Futures.allAsList(inboundRelationsListFrom);
Futures.transform(inboundRelationsFrom, (AsyncFunction<List<List<EntityRelation>>, List<Boolean>>) relations -> {
Futures.transformAsync(inboundRelationsFrom, relations -> {
List<ListenableFuture<Boolean>> results = getListenableFutures(relations, cache, false);
return Futures.allAsList(results);
});
@ -252,7 +252,7 @@ public class BaseRelationService implements RelationService {
private List<ListenableFuture<Boolean>> getListenableFutures(List<List<EntityRelation>> relations, Cache cache, boolean isRemove) {
List<ListenableFuture<Boolean>> results = new ArrayList<>();
for (List<EntityRelation> relationList : relations) {
relationList.stream().forEach(relation -> {
relationList.forEach(relation -> {
checkFromDeleteAsync(cache, results, relation, isRemove);
});
}
@ -325,17 +325,16 @@ public class BaseRelationService implements RelationService {
validate(from);
validateTypeGroup(typeGroup);
ListenableFuture<List<EntityRelation>> relations = relationDao.findAllByFrom(from, typeGroup);
ListenableFuture<List<EntityRelationInfo>> relationsInfo = Futures.transform(relations,
(AsyncFunction<List<EntityRelation>, List<EntityRelationInfo>>) relations1 -> {
return Futures.transformAsync(relations,
relations1 -> {
List<ListenableFuture<EntityRelationInfo>> futures = new ArrayList<>();
relations1.stream().forEach(relation ->
relations1.forEach(relation ->
futures.add(fetchRelationInfoAsync(relation,
relation2 -> relation2.getTo(),
(EntityRelationInfo relationInfo, String entityName) -> relationInfo.setToName(entityName)))
EntityRelation::getTo,
EntityRelationInfo::setToName))
);
return Futures.successfulAsList(futures);
});
return relationsInfo;
}
@Cacheable(cacheNames = RELATIONS_CACHE, key = "{#from, #relationType, #typeGroup}")
@ -381,8 +380,8 @@ public class BaseRelationService implements RelationService {
validate(to);
validateTypeGroup(typeGroup);
ListenableFuture<List<EntityRelation>> relations = relationDao.findAllByTo(to, typeGroup);
ListenableFuture<List<EntityRelationInfo>> relationsInfo = Futures.transform(relations,
(AsyncFunction<List<EntityRelation>, List<EntityRelationInfo>>) relations1 -> {
return Futures.transformAsync(relations,
relations1 -> {
List<ListenableFuture<EntityRelationInfo>> futures = new ArrayList<>();
relations1.stream().forEach(relation ->
futures.add(fetchRelationInfoAsync(relation,
@ -391,7 +390,6 @@ public class BaseRelationService implements RelationService {
);
return Futures.successfulAsList(futures);
});
return relationsInfo;
}
private ListenableFuture<EntityRelationInfo> fetchRelationInfoAsync(EntityRelation relation,
@ -463,8 +461,8 @@ public class BaseRelationService implements RelationService {
log.trace("Executing findInfoByQuery [{}]", query);
ListenableFuture<List<EntityRelation>> relations = findByQuery(query);
EntitySearchDirection direction = query.getParameters().getDirection();
ListenableFuture<List<EntityRelationInfo>> relationsInfo = Futures.transform(relations,
(AsyncFunction<List<EntityRelation>, List<EntityRelationInfo>>) relations1 -> {
return Futures.transformAsync(relations,
relations1 -> {
List<ListenableFuture<EntityRelationInfo>> futures = new ArrayList<>();
relations1.stream().forEach(relation ->
futures.add(fetchRelationInfoAsync(relation,
@ -479,7 +477,6 @@ public class BaseRelationService implements RelationService {
);
return Futures.successfulAsList(futures);
});
return relationsInfo;
}
protected void validate(EntityRelation relation) {

4
dao/src/main/java/org/thingsboard/server/dao/sql/alarm/JpaAlarmDao.java
View File

@ -102,12 +102,12 @@ public class JpaAlarmDao extends JpaAbstractDao<AlarmEntity, Alarm> implements A
}
String relationType = BaseAlarmService.ALARM_RELATION_PREFIX + searchStatusName;
ListenableFuture<List<EntityRelation>> relations = relationDao.findRelations(affectedEntity, relationType, RelationTypeGroup.ALARM, EntityType.ALARM, query.getPageLink());
return Futures.transform(relations, (AsyncFunction<List<EntityRelation>, List<AlarmInfo>>) input -> {
return Futures.transformAsync(relations, input -> {
List<ListenableFuture<AlarmInfo>> alarmFutures = new ArrayList<>(input.size());
for (EntityRelation relation : input) {
alarmFutures.add(Futures.transform(
findAlarmByIdAsync(relation.getTo().getId()),
(Function<Alarm, AlarmInfo>) AlarmInfo::new));
AlarmInfo::new));
}
return Futures.successfulAsList(alarmFutures);
});

2
dao/src/main/java/org/thingsboard/server/dao/sql/dashboard/JpaDashboardInfoDao.java
View File

@ -86,7 +86,7 @@ public class JpaDashboardInfoDao extends JpaAbstractSearchTextDao<DashboardInfoE
ListenableFuture<List<EntityRelation>> relations = relationDao.findRelations(new CustomerId(customerId), EntityRelation.CONTAINS_TYPE, RelationTypeGroup.DASHBOARD, EntityType.DASHBOARD, pageLink);
return Futures.transform(relations, (AsyncFunction<List<EntityRelation>, List<DashboardInfo>>) input -> {
return Futures.transformAsync(relations, input -> {
List<ListenableFuture<DashboardInfo>> dashboardFutures = new ArrayList<>(input.size());
for (EntityRelation relation : input) {
dashboardFutures.add(findByIdAsync(relation.getTo().getId()));

2
dao/src/main/java/org/thingsboard/server/dao/timeseries/CassandraBaseTimeseriesDao.java
View File

@ -217,7 +217,7 @@ public class CassandraBaseTimeseriesDao extends CassandraAbstractAsyncDao implem
ListenableFuture<List<Long>> partitionsListFuture = Futures.transform(partitionsFuture, getPartitionsArrayFunction(), readResultsProcessingExecutor);
ListenableFuture<List<ResultSet>> aggregationChunks = Futures.transform(partitionsListFuture,
ListenableFuture<List<ResultSet>> aggregationChunks = Futures.transformAsync(partitionsListFuture,
getFetchChunksAsyncFunction(entityId, key, aggregation, startTs, endTs), readResultsProcessingExecutor);
return Futures.transform(aggregationChunks, new AggregatePartitionsFunction(aggregation, key, ts), readResultsProcessingExecutor);

12
entrypoint-with-db-init.sh
View File

@ -1,12 +0,0 @@
#!/bin/bash
if [[ $1 = 'cassandra' ]]; then
until cqlsh -f/opt/cassandra/schema.cql; do
echo "cqlsh: Cassandra is unavailable - retrying"
sleep 2
done &
fi
exec /docker-entrypoint.sh "$@"

23
pom.xml
View File

@ -41,10 +41,10 @@
<logback.version>1.2.3</logback.version>
<mockito.version>1.9.5</mockito.version>
<rat.version>0.10</rat.version>
<cassandra.version>3.0.7</cassandra.version>
<cassandra.version>3.5.0</cassandra.version>
<cassandra-unit.version>3.0.0.1</cassandra-unit.version>
<takari-cpsuite.version>1.2.7</takari-cpsuite.version>
<guava.version>18.0</guava.version>
<guava.version>20.0</guava.version>
<caffeine.version>2.6.1</caffeine.version>
<commons-lang3.version>3.4</commons-lang3.version>
<commons-validator.version>1.5.0</commons-validator.version>
@ -61,15 +61,13 @@
<mail.version>1.4.3</mail.version>
<curator.version>2.11.0</curator.version>
<protobuf.version>3.0.2</protobuf.version>
<grpc.version>1.0.0</grpc.version>
<grpc.version>1.12.0</grpc.version>
<lombok.version>1.16.18</lombok.version>
<paho.client.version>1.1.0</paho.client.version>
<netty.version>4.1.22.Final</netty.version>
<os-maven-plugin.version>1.5.0</os-maven-plugin.version>
<rabbitmq.version>3.6.5</rabbitmq.version>
<kafka.version>0.9.0.0</kafka.version>
<hazelcast.version>3.6.6</hazelcast.version>
<hazelcast-zookeeper.version>3.6.1</hazelcast-zookeeper.version>
<surfire.version>2.19.1</surfire.version>
<jar-plugin.version>3.0.2</jar-plugin.version>
<springfox-swagger.version>2.6.1</springfox-swagger.version>
@ -760,26 +758,11 @@
<artifactId>org.eclipse.paho.client.mqttv3</artifactId>
<version>${paho.client.version}</version>
</dependency>
<dependency>
<groupId>com.hazelcast</groupId>
<artifactId>hazelcast-spring</artifactId>
<version>${hazelcast.version}</version>
</dependency>
<dependency>
<groupId>org.apache.curator</groupId>
<artifactId>curator-x-discovery</artifactId>
<version>${curator.version}</version>
</dependency>
<dependency>
<groupId>com.hazelcast</groupId>
<artifactId>hazelcast-zookeeper</artifactId>
<version>${hazelcast-zookeeper.version}</version>
</dependency>
<dependency>
<groupId>com.hazelcast</groupId>
<artifactId>hazelcast</artifactId>
<version>${hazelcast.version}</version>
</dependency>
<dependency>
<groupId>io.springfox</groupId>
<artifactId>springfox-swagger-ui</artifactId>

6
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbClearAlarmNode.java
View File

@ -56,7 +56,7 @@ public class TbClearAlarmNode extends TbAbstractAlarmNode<TbClearAlarmNodeConfig
@Override
protected ListenableFuture<AlarmResult> processAlarm(TbContext ctx, TbMsg msg) {
ListenableFuture<Alarm> latest = ctx.getAlarmService().findLatestByOriginatorAndType(ctx.getTenantId(), msg.getOriginator(), config.getAlarmType());
return Futures.transform(latest, (AsyncFunction<Alarm, AlarmResult>) a -> {
return Futures.transformAsync(latest, a -> {
if (a != null && !a.getStatus().isCleared()) {
return clearAlarm(ctx, msg, a);
}
@ -66,9 +66,9 @@ public class TbClearAlarmNode extends TbAbstractAlarmNode<TbClearAlarmNodeConfig
private ListenableFuture<AlarmResult> clearAlarm(TbContext ctx, TbMsg msg, Alarm alarm) {
ListenableFuture<JsonNode> asyncDetails = buildAlarmDetails(ctx, msg, alarm.getDetails());
return Futures.transform(asyncDetails, (AsyncFunction<JsonNode, AlarmResult>) details -> {
return Futures.transformAsync(asyncDetails, details -> {
ListenableFuture<Boolean> clearFuture = ctx.getAlarmService().clearAlarm(alarm.getId(), details, System.currentTimeMillis());
return Futures.transform(clearFuture, (AsyncFunction<Boolean, AlarmResult>) cleared -> {
return Futures.transformAsync(clearFuture, cleared -> {
if (cleared && details != null) {
alarm.setDetails(details);
}

10
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbCreateAlarmNode.java
View File

@ -58,7 +58,7 @@ public class TbCreateAlarmNode extends TbAbstractAlarmNode<TbCreateAlarmNodeConf
@Override
protected ListenableFuture<AlarmResult> processAlarm(TbContext ctx, TbMsg msg) {
ListenableFuture<Alarm> latest = ctx.getAlarmService().findLatestByOriginatorAndType(ctx.getTenantId(), msg.getOriginator(), config.getAlarmType());
return Futures.transform(latest, (AsyncFunction<Alarm, AlarmResult>) a -> {
return Futures.transformAsync(latest, a -> {
if (a == null || a.getStatus().isCleared()) {
return createNewAlarm(ctx, msg);
} else {
@ -70,10 +70,10 @@ public class TbCreateAlarmNode extends TbAbstractAlarmNode<TbCreateAlarmNodeConf
private ListenableFuture<AlarmResult> createNewAlarm(TbContext ctx, TbMsg msg) {
ListenableFuture<Alarm> asyncAlarm = Futures.transform(buildAlarmDetails(ctx, msg, null),
(Function<JsonNode, Alarm>) details -> buildAlarm(msg, details, ctx.getTenantId()));
details -> buildAlarm(msg, details, ctx.getTenantId()));
ListenableFuture<Alarm> asyncCreated = Futures.transform(asyncAlarm,
(Function<Alarm, Alarm>) alarm -> ctx.getAlarmService().createOrUpdateAlarm(alarm), ctx.getDbCallbackExecutor());
return Futures.transform(asyncCreated, (Function<Alarm, AlarmResult>) alarm -> new AlarmResult(true, false, false, alarm));
alarm -> ctx.getAlarmService().createOrUpdateAlarm(alarm), ctx.getDbCallbackExecutor());
return Futures.transform(asyncCreated, alarm -> new AlarmResult(true, false, false, alarm));
}
private ListenableFuture<AlarmResult> updateAlarm(TbContext ctx, TbMsg msg, Alarm alarm) {
@ -85,7 +85,7 @@ public class TbCreateAlarmNode extends TbAbstractAlarmNode<TbCreateAlarmNodeConf
return ctx.getAlarmService().createOrUpdateAlarm(alarm);
}, ctx.getDbCallbackExecutor());
return Futures.transform(asyncUpdated, (Function<Alarm, AlarmResult>) a -> new AlarmResult(false, true, false, a));
return Futures.transform(asyncUpdated, a -> new AlarmResult(false, true, false, a));
}
private Alarm buildAlarm(TbMsg msg, JsonNode details, TenantId tenantId) {

5
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/util/EntitiesCustomerIdAsyncLoader.java
View File

@ -43,8 +43,7 @@ public class EntitiesCustomerIdAsyncLoader {
}
private static <T extends HasCustomerId> ListenableFuture<CustomerId> getCustomerAsync(ListenableFuture<T> future) {
return Futures.transform(future, (AsyncFunction<HasCustomerId, CustomerId>) in -> {
return in != null ? Futures.immediateFuture(in.getCustomerId())
: Futures.immediateFuture(null);});
return Futures.transformAsync(future, in -> in != null ? Futures.immediateFuture(in.getCustomerId())
: Futures.immediateFuture(null));
}
}

5
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/util/EntitiesRelatedDeviceIdAsyncLoader.java
View File

@ -39,9 +39,8 @@ public class EntitiesRelatedDeviceIdAsyncLoader {
ListenableFuture<List<Device>> asyncDevices = deviceService.findDevicesByQuery(query);
return Futures.transform(asyncDevices, (AsyncFunction<List<Device>, DeviceId>)
d -> CollectionUtils.isNotEmpty(d) ? Futures.immediateFuture(d.get(0).getId())
: Futures.immediateFuture(null));
return Futures.transformAsync(asyncDevices, d -> CollectionUtils.isNotEmpty(d) ? Futures.immediateFuture(d.get(0).getId())
: Futures.immediateFuture(null));
}
private static DeviceSearchQuery buildQuery(EntityId originator, DeviceRelationsQuery deviceRelationsQuery) {

10
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/util/EntitiesRelatedEntityIdAsyncLoader.java
View File

@ -38,13 +38,11 @@ public class EntitiesRelatedEntityIdAsyncLoader {
EntityRelationsQuery query = buildQuery(originator, relationsQuery);
ListenableFuture<List<EntityRelation>> asyncRelation = relationService.findByQuery(query);
if (relationsQuery.getDirection() == EntitySearchDirection.FROM) {
return Futures.transform(asyncRelation, (AsyncFunction<? super List<EntityRelation>, EntityId>)
r -> CollectionUtils.isNotEmpty(r) ? Futures.immediateFuture(r.get(0).getTo())
: Futures.immediateFuture(null));
return Futures.transformAsync(asyncRelation, r -> CollectionUtils.isNotEmpty(r) ? Futures.immediateFuture(r.get(0).getTo())
: Futures.immediateFuture(null));
} else if (relationsQuery.getDirection() == EntitySearchDirection.TO) {
return Futures.transform(asyncRelation, (AsyncFunction<? super List<EntityRelation>, EntityId>)
r -> CollectionUtils.isNotEmpty(r) ? Futures.immediateFuture(r.get(0).getFrom())
: Futures.immediateFuture(null));
return Futures.transformAsync(asyncRelation, r -> CollectionUtils.isNotEmpty(r) ? Futures.immediateFuture(r.get(0).getFrom())
: Futures.immediateFuture(null));
}
return Futures.immediateFailedFuture(new IllegalStateException("Unknown direction"));
}

2
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/util/EntitiesTenantIdAsyncLoader.java
View File

@ -51,7 +51,7 @@ public class EntitiesTenantIdAsyncLoader {
}
private static <T extends HasTenantId> ListenableFuture<TenantId> getTenantAsync(ListenableFuture<T> future) {
return Futures.transform(future, (AsyncFunction<HasTenantId, TenantId>) in -> {
return Futures.transformAsync(future, in -> {
return in != null ? Futures.immediateFuture(in.getTenantId())
: Futures.immediateFuture(null);});
}