diff --git a/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java b/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java index d33734e14c..2f78672800 100644 --- a/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java +++ b/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java @@ -65,7 +65,7 @@ import org.thingsboard.server.service.mail.MailExecutorService; import org.thingsboard.server.service.queue.MsgQueueService; import org.thingsboard.server.service.rpc.DeviceRpcService; import org.thingsboard.server.service.script.JsExecutorService; -import org.thingsboard.server.service.script.JsSandboxService; +import org.thingsboard.server.service.script.JsInvokeService; import org.thingsboard.server.service.state.DeviceStateService; import org.thingsboard.server.service.telemetry.TelemetrySubscriptionService; @@ -169,7 +169,7 @@ public class ActorSystemContext { @Autowired @Getter - private JsSandboxService jsSandbox; + private JsInvokeService jsSandbox; @Autowired @Getter diff --git a/application/src/main/java/org/thingsboard/server/actors/rpc/RpcManagerActor.java b/application/src/main/java/org/thingsboard/server/actors/rpc/RpcManagerActor.java index 9e38c17aff..608a77fd7a 100644 --- a/application/src/main/java/org/thingsboard/server/actors/rpc/RpcManagerActor.java +++ b/application/src/main/java/org/thingsboard/server/actors/rpc/RpcManagerActor.java @@ -26,6 +26,7 @@ import org.thingsboard.server.actors.service.DefaultActorService; import org.thingsboard.server.common.msg.TbActorMsg; import org.thingsboard.server.common.msg.cluster.ClusterEventMsg; import org.thingsboard.server.common.msg.cluster.ServerAddress; +import org.thingsboard.server.common.msg.cluster.ServerType; import org.thingsboard.server.gen.cluster.ClusterAPIProtos; import org.thingsboard.server.service.cluster.discovery.ServerInstance; @@ -100,8 +101,7 @@ public class RpcManagerActor extends ContextAwareActor { private void onMsg(ClusterAPIProtos.ClusterMessage msg) { if (msg.hasServerAddress()) { - ServerAddress address = new ServerAddress(msg.getServerAddress().getHost(), - msg.getServerAddress().getPort()); + ServerAddress address = new ServerAddress(msg.getServerAddress().getHost(), msg.getServerAddress().getPort(), ServerType.CORE); SessionActorInfo session = sessionActors.get(address); if (session != null) { log.debug("{} Forwarding msg to session actor", address); @@ -112,7 +112,7 @@ public class RpcManagerActor extends ContextAwareActor { if (queue == null) { queue = new LinkedList<>(); pendingMsgs.put(new ServerAddress( - msg.getServerAddress().getHost(), msg.getServerAddress().getPort()), queue); + msg.getServerAddress().getHost(), msg.getServerAddress().getPort(), ServerType.CORE), queue); } queue.add(msg); } diff --git a/application/src/main/java/org/thingsboard/server/actors/service/DefaultActorService.java b/application/src/main/java/org/thingsboard/server/actors/service/DefaultActorService.java index fac5d97175..1e9e23bae3 100644 --- a/application/src/main/java/org/thingsboard/server/actors/service/DefaultActorService.java +++ b/application/src/main/java/org/thingsboard/server/actors/service/DefaultActorService.java @@ -115,8 +115,6 @@ public class DefaultActorService implements ActorService { actorContext.setStatsActor(statsActor); rpcService.init(this); - - discoveryService.addListener(this); log.info("Actor system initialized."); } @@ -202,7 +200,7 @@ public class DefaultActorService implements ActorService { @Override public void onReceivedMsg(ServerAddress source, ClusterAPIProtos.ClusterMessage msg) { - ServerAddress serverAddress = new ServerAddress(source.getHost(), source.getPort()); + ServerAddress serverAddress = new ServerAddress(source.getHost(), source.getPort(), source.getServerType()); log.info("Received msg [{}] from [{}]", msg.getMessageType().name(), serverAddress); if (log.isDebugEnabled()) { log.info("MSG: ", msg); diff --git a/application/src/main/java/org/thingsboard/server/controller/RuleChainController.java b/application/src/main/java/org/thingsboard/server/controller/RuleChainController.java index 82ab1702fa..fbd5faf75b 100644 --- a/application/src/main/java/org/thingsboard/server/controller/RuleChainController.java +++ b/application/src/main/java/org/thingsboard/server/controller/RuleChainController.java @@ -50,7 +50,7 @@ import org.thingsboard.server.common.data.rule.RuleChainMetaData; import org.thingsboard.server.common.msg.TbMsg; import org.thingsboard.server.common.msg.TbMsgMetaData; import org.thingsboard.server.dao.event.EventService; -import org.thingsboard.server.service.script.JsSandboxService; +import org.thingsboard.server.service.script.JsInvokeService; import org.thingsboard.server.service.script.RuleNodeJsScriptEngine; import java.util.List; @@ -71,7 +71,7 @@ public class RuleChainController extends BaseController { private EventService eventService; @Autowired - private JsSandboxService jsSandboxService; + private JsInvokeService jsInvokeService; @PreAuthorize("hasAnyAuthority('TENANT_ADMIN')") @RequestMapping(value = "/ruleChain/{ruleChainId}", method = RequestMethod.GET) @@ -276,7 +276,7 @@ public class RuleChainController extends BaseController { String errorText = ""; ScriptEngine engine = null; try { - engine = new RuleNodeJsScriptEngine(jsSandboxService, getCurrentUser().getId(), script, argNames); + engine = new RuleNodeJsScriptEngine(jsInvokeService, getCurrentUser().getId(), script, argNames); TbMsg inMsg = new TbMsg(UUIDs.timeBased(), msgType, null, new TbMsgMetaData(metadata), data, null, null, 0L); switch (scriptType) { case "update": diff --git a/application/src/main/java/org/thingsboard/server/service/cluster/discovery/CurrentServerInstanceService.java b/application/src/main/java/org/thingsboard/server/service/cluster/discovery/CurrentServerInstanceService.java index 2232ef898c..3ae7b16af1 100644 --- a/application/src/main/java/org/thingsboard/server/service/cluster/discovery/CurrentServerInstanceService.java +++ b/application/src/main/java/org/thingsboard/server/service/cluster/discovery/CurrentServerInstanceService.java @@ -19,7 +19,8 @@ import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Value; import org.springframework.stereotype.Service; import org.springframework.util.Assert; -import org.thingsboard.server.gen.discovery.ServerInstanceProtos; +import org.thingsboard.server.common.msg.cluster.ServerAddress; +import org.thingsboard.server.common.msg.cluster.ServerType; import javax.annotation.PostConstruct; @@ -43,8 +44,7 @@ public class CurrentServerInstanceService implements ServerInstanceService { public void init() { Assert.hasLength(rpcHost, missingProperty("rpc.bind_host")); Assert.notNull(rpcPort, missingProperty("rpc.bind_port")); - - self = new ServerInstance(ServerInstanceProtos.ServerInfo.newBuilder().setHost(rpcHost).setPort(rpcPort).setTs(System.currentTimeMillis()).build()); + self = new ServerInstance(new ServerAddress(rpcHost, rpcPort, ServerType.CORE)); log.info("Current server instance: [{};{}]", self.getHost(), self.getPort()); } diff --git a/application/src/main/java/org/thingsboard/server/service/cluster/discovery/DiscoveryService.java b/application/src/main/java/org/thingsboard/server/service/cluster/discovery/DiscoveryService.java index 516fca9484..f9caafa412 100644 --- a/application/src/main/java/org/thingsboard/server/service/cluster/discovery/DiscoveryService.java +++ b/application/src/main/java/org/thingsboard/server/service/cluster/discovery/DiscoveryService.java @@ -30,8 +30,4 @@ public interface DiscoveryService { List getOtherServers(); - boolean addListener(DiscoveryServiceListener listener); - - boolean removeListener(DiscoveryServiceListener listener); - } diff --git a/application/src/main/java/org/thingsboard/server/service/cluster/discovery/DummyDiscoveryService.java b/application/src/main/java/org/thingsboard/server/service/cluster/discovery/DummyDiscoveryService.java index c21f1aa84c..fd91643cfb 100644 --- a/application/src/main/java/org/thingsboard/server/service/cluster/discovery/DummyDiscoveryService.java +++ b/application/src/main/java/org/thingsboard/server/service/cluster/discovery/DummyDiscoveryService.java @@ -62,13 +62,5 @@ public class DummyDiscoveryService implements DiscoveryService { return Collections.emptyList(); } - @Override - public boolean addListener(DiscoveryServiceListener listener) { - return false; - } - @Override - public boolean removeListener(DiscoveryServiceListener listener) { - return false; - } } diff --git a/application/src/main/java/org/thingsboard/server/service/cluster/discovery/ServerInstance.java b/application/src/main/java/org/thingsboard/server/service/cluster/discovery/ServerInstance.java index 6eee5f3061..8f4525c20d 100644 --- a/application/src/main/java/org/thingsboard/server/service/cluster/discovery/ServerInstance.java +++ b/application/src/main/java/org/thingsboard/server/service/cluster/discovery/ServerInstance.java @@ -19,7 +19,6 @@ import lombok.EqualsAndHashCode; import lombok.Getter; import lombok.ToString; import org.thingsboard.server.common.msg.cluster.ServerAddress; -import org.thingsboard.server.gen.discovery.ServerInstanceProtos; /** * @author Andrew Shvayka @@ -41,12 +40,6 @@ public final class ServerInstance implements Comparable { this.port = serverAddress.getPort(); } - public ServerInstance(ServerInstanceProtos.ServerInfo serverInfo) { - this.host = serverInfo.getHost(); - this.port = serverInfo.getPort(); - this.serverAddress = new ServerAddress(host, port); - } - @Override public int compareTo(ServerInstance o) { return this.serverAddress.compareTo(o.serverAddress); diff --git a/application/src/main/java/org/thingsboard/server/service/cluster/discovery/ZkDiscoveryService.java b/application/src/main/java/org/thingsboard/server/service/cluster/discovery/ZkDiscoveryService.java index c3ffbabbea..19a04d83ab 100644 --- a/application/src/main/java/org/thingsboard/server/service/cluster/discovery/ZkDiscoveryService.java +++ b/application/src/main/java/org/thingsboard/server/service/cluster/discovery/ZkDiscoveryService.java @@ -24,6 +24,8 @@ import org.apache.curator.framework.recipes.cache.ChildData; import org.apache.curator.framework.recipes.cache.PathChildrenCache; import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent; import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener; +import org.apache.curator.framework.state.ConnectionState; +import org.apache.curator.framework.state.ConnectionStateListener; import org.apache.curator.retry.RetryForever; import org.apache.curator.utils.CloseableUtils; import org.apache.zookeeper.CreateMode; @@ -35,7 +37,9 @@ 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.actors.service.ActorService; import org.thingsboard.server.common.msg.cluster.ServerAddress; +import org.thingsboard.server.service.cluster.routing.ClusterRoutingService; import org.thingsboard.server.service.state.DeviceStateService; import org.thingsboard.server.service.telemetry.TelemetrySubscriptionService; import org.thingsboard.server.utils.MiscUtils; @@ -44,7 +48,6 @@ import javax.annotation.PostConstruct; import javax.annotation.PreDestroy; import java.util.List; import java.util.NoSuchElementException; -import java.util.concurrent.CopyOnWriteArrayList; import java.util.stream.Collectors; /** @@ -79,7 +82,13 @@ public class ZkDiscoveryService implements DiscoveryService, PathChildrenCacheLi @Lazy private DeviceStateService deviceStateService; - private final List listeners = new CopyOnWriteArrayList<>(); + @Autowired + @Lazy + private ActorService actorService; + + @Autowired + @Lazy + private ClusterRoutingService routingService; private CuratorFramework client; private PathChildrenCache cache; @@ -127,12 +136,37 @@ public class ZkDiscoveryService implements DiscoveryService, PathChildrenCacheLi .creatingParentsIfNeeded() .withMode(CreateMode.EPHEMERAL_SEQUENTIAL).forPath(zkNodesDir + "/", SerializationUtils.serialize(self.getServerAddress())); log.info("[{}:{}] Created ZK node for current instance: {}", self.getHost(), self.getPort(), nodePath); + client.getConnectionStateListenable().addListener(checkReconnect(self)); } catch (Exception e) { log.error("Failed to create ZK node", e); throw new RuntimeException(e); } } + private boolean reconnectInProgress = false; + + private synchronized ConnectionStateListener checkReconnect(ServerInstance self) { + return (client, newState) -> { + log.info("[{}:{}] ZK state changed: {}", self.getHost(), self.getPort(), newState); + if (newState == ConnectionState.LOST) { + if (!reconnectInProgress) { + reconnectInProgress = true; + reconnect(); + } + } + }; + } + + private void reconnect() { + try { + client.blockUntilConnected(); + } catch (InterruptedException e) { + log.error("Failed to reconnect to ZK: {}", e.getMessage(), e); + } + publishCurrentServer(); + reconnectInProgress = false; + } + @Override public void unpublishCurrentServer() { try { @@ -156,7 +190,7 @@ public class ZkDiscoveryService implements DiscoveryService, PathChildrenCacheLi .filter(cd -> !cd.getPath().equals(nodePath)) .map(cd -> { try { - return new ServerInstance( (ServerAddress) SerializationUtils.deserialize(cd.getData())); + return new ServerInstance((ServerAddress) SerializationUtils.deserialize(cd.getData())); } catch (NoSuchElementException e) { log.error("Failed to decode ZK node", e); throw new RuntimeException(e); @@ -165,16 +199,6 @@ public class ZkDiscoveryService implements DiscoveryService, PathChildrenCacheLi .collect(Collectors.toList()); } - @Override - public boolean addListener(DiscoveryServiceListener listener) { - return listeners.add(listener); - } - - @Override - public boolean removeListener(DiscoveryServiceListener listener) { - return listeners.remove(listener); - } - @Override public void onApplicationEvent(ApplicationReadyEvent applicationReadyEvent) { publishCurrentServer(); @@ -198,7 +222,7 @@ public class ZkDiscoveryService implements DiscoveryService, PathChildrenCacheLi } ServerInstance instance; try { - ServerAddress serverAddress = SerializationUtils.deserialize(data.getData()); + ServerAddress serverAddress = SerializationUtils.deserialize(data.getData()); instance = new ServerInstance(serverAddress); } catch (SerializationException e) { log.error("Failed to decode server instance for node {}", data.getPath(), e); @@ -207,17 +231,20 @@ public class ZkDiscoveryService implements DiscoveryService, PathChildrenCacheLi log.info("Processing [{}] event for [{}:{}]", pathChildrenCacheEvent.getType(), instance.getHost(), instance.getPort()); switch (pathChildrenCacheEvent.getType()) { case CHILD_ADDED: + routingService.onServerAdded(instance); tsSubService.onClusterUpdate(); deviceStateService.onClusterUpdate(); - listeners.forEach(listener -> listener.onServerAdded(instance)); + actorService.onServerAdded(instance); break; case CHILD_UPDATED: - listeners.forEach(listener -> listener.onServerUpdated(instance)); + routingService.onServerUpdated(instance); + actorService.onServerUpdated(instance); break; case CHILD_REMOVED: + routingService.onServerRemoved(instance); tsSubService.onClusterUpdate(); deviceStateService.onClusterUpdate(); - listeners.forEach(listener -> listener.onServerRemoved(instance)); + actorService.onServerRemoved(instance); break; default: break; diff --git a/application/src/main/java/org/thingsboard/server/service/cluster/routing/ClusterRoutingService.java b/application/src/main/java/org/thingsboard/server/service/cluster/routing/ClusterRoutingService.java index 272073dd23..bb76c74e28 100644 --- a/application/src/main/java/org/thingsboard/server/service/cluster/routing/ClusterRoutingService.java +++ b/application/src/main/java/org/thingsboard/server/service/cluster/routing/ClusterRoutingService.java @@ -17,6 +17,8 @@ package org.thingsboard.server.service.cluster.routing; import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.common.msg.cluster.ServerAddress; +import org.thingsboard.server.common.msg.cluster.ServerType; +import org.thingsboard.server.service.cluster.discovery.DiscoveryServiceListener; import java.util.Optional; import java.util.UUID; @@ -24,7 +26,7 @@ import java.util.UUID; /** * @author Andrew Shvayka */ -public interface ClusterRoutingService { +public interface ClusterRoutingService extends DiscoveryServiceListener { ServerAddress getCurrentServer(); @@ -32,4 +34,9 @@ public interface ClusterRoutingService { Optional resolveById(EntityId entityId); + Optional resolveByUuid(ServerType server, UUID uuid); + + Optional resolveById(ServerType server, EntityId entityId); + + } diff --git a/application/src/main/java/org/thingsboard/server/service/cluster/routing/ConsistentClusterRoutingService.java b/application/src/main/java/org/thingsboard/server/service/cluster/routing/ConsistentClusterRoutingService.java index 5087b5ce46..f47f6d2e8e 100644 --- a/application/src/main/java/org/thingsboard/server/service/cluster/routing/ConsistentClusterRoutingService.java +++ b/application/src/main/java/org/thingsboard/server/service/cluster/routing/ConsistentClusterRoutingService.java @@ -24,12 +24,14 @@ import org.springframework.stereotype.Service; import org.springframework.util.Assert; import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.common.msg.cluster.ServerAddress; +import org.thingsboard.server.common.msg.cluster.ServerType; import org.thingsboard.server.service.cluster.discovery.DiscoveryService; import org.thingsboard.server.service.cluster.discovery.DiscoveryServiceListener; import org.thingsboard.server.service.cluster.discovery.ServerInstance; import org.thingsboard.server.utils.MiscUtils; import javax.annotation.PostConstruct; +import java.util.Arrays; import java.util.Optional; import java.util.UUID; import java.util.concurrent.ConcurrentNavigableMap; @@ -41,7 +43,7 @@ import java.util.concurrent.ConcurrentSkipListMap; @Service @Slf4j -public class ConsistentClusterRoutingService implements ClusterRoutingService, DiscoveryServiceListener { +public class ConsistentClusterRoutingService implements ClusterRoutingService { @Autowired private DiscoveryService discoveryService; @@ -55,15 +57,19 @@ public class ConsistentClusterRoutingService implements ClusterRoutingService, D private HashFunction hashFunction; - private final ConcurrentNavigableMap circle = - new ConcurrentSkipListMap<>(); + private ConsistentHashCircle[] circles; + private ConsistentHashCircle rootCircle; @PostConstruct public void init() { log.info("Initializing Cluster routing service!"); - hashFunction = MiscUtils.forName(hashFunctionName); - discoveryService.addListener(this); + this.hashFunction = MiscUtils.forName(hashFunctionName); this.currentServer = discoveryService.getCurrentServer(); + this.circles = new ConsistentHashCircle[ServerType.values().length]; + for (ServerType serverType : ServerType.values()) { + circles[serverType.ordinal()] = new ConsistentHashCircle(); + } + rootCircle = circles[ServerType.CORE.ordinal()]; addNode(discoveryService.getCurrentServer()); for (ServerInstance instance : discoveryService.getOtherServers()) { addNode(instance); @@ -79,11 +85,25 @@ public class ConsistentClusterRoutingService implements ClusterRoutingService, D @Override public Optional resolveById(EntityId entityId) { - return resolveByUuid(entityId.getId()); + return resolveByUuid(rootCircle, entityId.getId()); } @Override public Optional resolveByUuid(UUID uuid) { + return resolveByUuid(rootCircle, uuid); + } + + @Override + public Optional resolveByUuid(ServerType server, UUID uuid) { + return resolveByUuid(circles[server.ordinal()], uuid); + } + + @Override + public Optional resolveById(ServerType server, EntityId entityId) { + return resolveByUuid(circles[server.ordinal()], entityId.getId()); + } + + private Optional resolveByUuid(ConsistentHashCircle circle, UUID uuid) { Assert.notNull(uuid); if (circle.isEmpty()) { return Optional.empty(); @@ -125,13 +145,13 @@ public class ConsistentClusterRoutingService implements ClusterRoutingService, D private void addNode(ServerInstance instance) { for (int i = 0; i < virtualNodesSize; i++) { - circle.put(hash(instance, i).asLong(), instance); + circles[instance.getServerAddress().getServerType().ordinal()].put(hash(instance, i).asLong(), instance); } } private void removeNode(ServerInstance instance) { for (int i = 0; i < virtualNodesSize; i++) { - circle.remove(hash(instance, i).asLong()); + circles[instance.getServerAddress().getServerType().ordinal()].remove(hash(instance, i).asLong()); } } @@ -141,7 +161,7 @@ public class ConsistentClusterRoutingService implements ClusterRoutingService, D private void logCircle() { log.trace("Consistent Hash Circle Start"); - circle.entrySet().forEach((e) -> log.debug("{} -> {}", e.getKey(), e.getValue().getServerAddress())); + Arrays.asList(circles).forEach(ConsistentHashCircle::log); log.trace("Consistent Hash Circle End"); } diff --git a/application/src/main/java/org/thingsboard/server/service/cluster/routing/ConsistentHashCircle.java b/application/src/main/java/org/thingsboard/server/service/cluster/routing/ConsistentHashCircle.java new file mode 100644 index 0000000000..a8c8f37a02 --- /dev/null +++ b/application/src/main/java/org/thingsboard/server/service/cluster/routing/ConsistentHashCircle.java @@ -0,0 +1,63 @@ +/** + * 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.service.cluster.routing; + +import lombok.extern.slf4j.Slf4j; +import org.thingsboard.server.service.cluster.discovery.ServerInstance; + +import java.util.concurrent.ConcurrentNavigableMap; +import java.util.concurrent.ConcurrentSkipListMap; + +/** + * Created by ashvayka on 23.09.18. + */ +@Slf4j +public class ConsistentHashCircle { + private final ConcurrentNavigableMap circle = + new ConcurrentSkipListMap<>(); + + public void put(long hash, ServerInstance instance) { + circle.put(hash, instance); + } + + public void remove(long hash) { + circle.remove(hash); + } + + public boolean isEmpty() { + return circle.isEmpty(); + } + + public boolean containsKey(Long hash) { + return circle.containsKey(hash); + } + + public ConcurrentNavigableMap tailMap(Long hash) { + return circle.tailMap(hash); + } + + public Long firstKey() { + return circle.firstKey(); + } + + public ServerInstance get(Long hash) { + return circle.get(hash); + } + + public void log() { + circle.entrySet().forEach((e) -> log.debug("{} -> {}", e.getKey(), e.getValue().getServerAddress())); + } +} diff --git a/application/src/main/java/org/thingsboard/server/service/cluster/rpc/GrpcSession.java b/application/src/main/java/org/thingsboard/server/service/cluster/rpc/GrpcSession.java index 7216c4319f..55cf52c37b 100644 --- a/application/src/main/java/org/thingsboard/server/service/cluster/rpc/GrpcSession.java +++ b/application/src/main/java/org/thingsboard/server/service/cluster/rpc/GrpcSession.java @@ -19,6 +19,7 @@ import io.grpc.stub.StreamObserver; import lombok.Data; import lombok.extern.slf4j.Slf4j; import org.thingsboard.server.common.msg.cluster.ServerAddress; +import org.thingsboard.server.common.msg.cluster.ServerType; import org.thingsboard.server.gen.cluster.ClusterAPIProtos; import java.io.Closeable; @@ -61,8 +62,8 @@ 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.getServerAddress().getHost(), clusterMessage.getServerAddress().getPort()); - remoteServer = new ServerAddress(rpcAddress.getHost(), rpcAddress.getPort()); + ServerAddress rpcAddress = new ServerAddress(clusterMessage.getServerAddress().getHost(), clusterMessage.getServerAddress().getPort(), ServerType.CORE); + remoteServer = new ServerAddress(rpcAddress.getHost(), rpcAddress.getPort(), ServerType.CORE); listener.onConnected(GrpcSession.this); } if (connected) { diff --git a/application/src/main/java/org/thingsboard/server/service/script/AbstractNashornJsSandboxService.java b/application/src/main/java/org/thingsboard/server/service/script/AbstractNashornJsInvokeService.java similarity index 99% rename from application/src/main/java/org/thingsboard/server/service/script/AbstractNashornJsSandboxService.java rename to application/src/main/java/org/thingsboard/server/service/script/AbstractNashornJsInvokeService.java index 1d89c9d860..4acef05cac 100644 --- a/application/src/main/java/org/thingsboard/server/service/script/AbstractNashornJsSandboxService.java +++ b/application/src/main/java/org/thingsboard/server/service/script/AbstractNashornJsInvokeService.java @@ -40,7 +40,7 @@ import java.util.concurrent.Executors; import java.util.concurrent.atomic.AtomicInteger; @Slf4j -public abstract class AbstractNashornJsSandboxService implements JsSandboxService { +public abstract class AbstractNashornJsInvokeService implements JsInvokeService { private NashornSandbox sandbox; private ScriptEngine engine; diff --git a/application/src/main/java/org/thingsboard/server/service/script/JsSandboxService.java b/application/src/main/java/org/thingsboard/server/service/script/JsInvokeService.java similarity index 96% rename from application/src/main/java/org/thingsboard/server/service/script/JsSandboxService.java rename to application/src/main/java/org/thingsboard/server/service/script/JsInvokeService.java index 5e1c676443..1da9859c23 100644 --- a/application/src/main/java/org/thingsboard/server/service/script/JsSandboxService.java +++ b/application/src/main/java/org/thingsboard/server/service/script/JsInvokeService.java @@ -21,7 +21,7 @@ import org.thingsboard.server.common.data.id.EntityId; import java.util.UUID; -public interface JsSandboxService { +public interface JsInvokeService { ListenableFuture eval(JsScriptType scriptType, String scriptBody, String... argNames); diff --git a/application/src/main/java/org/thingsboard/server/service/script/NashornJsSandboxService.java b/application/src/main/java/org/thingsboard/server/service/script/NashornJsInvokeService.java similarity index 95% rename from application/src/main/java/org/thingsboard/server/service/script/NashornJsSandboxService.java rename to application/src/main/java/org/thingsboard/server/service/script/NashornJsInvokeService.java index 3e8b4e9c55..c17430fac9 100644 --- a/application/src/main/java/org/thingsboard/server/service/script/NashornJsSandboxService.java +++ b/application/src/main/java/org/thingsboard/server/service/script/NashornJsInvokeService.java @@ -22,7 +22,7 @@ import org.springframework.stereotype.Service; @Slf4j @Service -public class NashornJsSandboxService extends AbstractNashornJsSandboxService { +public class NashornJsInvokeService extends AbstractNashornJsInvokeService { @Value("${actors.rule.js_sandbox.use_js_sandbox}") private boolean useJsSandbox; diff --git a/application/src/main/java/org/thingsboard/server/service/script/RuleNodeJsScriptEngine.java b/application/src/main/java/org/thingsboard/server/service/script/RuleNodeJsScriptEngine.java index 767dc05a23..c001a1a725 100644 --- a/application/src/main/java/org/thingsboard/server/service/script/RuleNodeJsScriptEngine.java +++ b/application/src/main/java/org/thingsboard/server/service/script/RuleNodeJsScriptEngine.java @@ -37,12 +37,12 @@ import java.util.concurrent.ExecutionException; public class RuleNodeJsScriptEngine implements org.thingsboard.rule.engine.api.ScriptEngine { private static final ObjectMapper mapper = new ObjectMapper(); - private final JsSandboxService sandboxService; + private final JsInvokeService sandboxService; private final UUID scriptId; private final EntityId entityId; - public RuleNodeJsScriptEngine(JsSandboxService sandboxService, EntityId entityId, String script, String... argNames) { + public RuleNodeJsScriptEngine(JsInvokeService sandboxService, EntityId entityId, String script, String... argNames) { this.sandboxService = sandboxService; this.entityId = entityId; try { diff --git a/application/src/main/java/org/thingsboard/server/service/telemetry/DefaultTelemetrySubscriptionService.java b/application/src/main/java/org/thingsboard/server/service/telemetry/DefaultTelemetrySubscriptionService.java index 548c41739e..5c3b88a253 100644 --- a/application/src/main/java/org/thingsboard/server/service/telemetry/DefaultTelemetrySubscriptionService.java +++ b/application/src/main/java/org/thingsboard/server/service/telemetry/DefaultTelemetrySubscriptionService.java @@ -232,7 +232,7 @@ public class DefaultTelemetrySubscriptionService implements TelemetrySubscriptio new SubscriptionState(proto.getSessionId(), proto.getSubscriptionId(), EntityIdFactory.getByTypeAndId(proto.getEntityType(), proto.getEntityId()), TelemetryFeature.valueOf(proto.getType()), proto.getAllKeys(), statesMap, proto.getScope()), - false, new ServerAddress(serverAddress.getHost(), serverAddress.getPort())); + false, new ServerAddress(serverAddress.getHost(), serverAddress.getPort(), serverAddress.getServerType())); addRemoteWsSubscription(serverAddress, proto.getSessionId(), subscription); } diff --git a/application/src/main/proto/jsinvoke.proto b/application/src/main/proto/jsinvoke.proto new file mode 100644 index 0000000000..c465309689 --- /dev/null +++ b/application/src/main/proto/jsinvoke.proto @@ -0,0 +1,37 @@ +/** + * 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. + */ +syntax = "proto3"; +package cluster; + +option java_package = "org.thingsboard.server.gen.js"; +option java_outer_classname = "JsInvokeProtos"; + +service JsInvokeRpcService { + rpc handleMsgs(stream JsInvokeRequest) returns (stream JsInvokeResponse) {} +} + +message JsInvokeRequest { + string scriptId = 1; + string scriptBody = 2; + repeated string args = 3; +} + +message JsInvokeResponse { + string result = 1; + string errorName = 2; + string errorDetails = 3; +} + diff --git a/application/src/test/java/org/thingsboard/server/service/script/RuleNodeJsScriptEngineTest.java b/application/src/test/java/org/thingsboard/server/service/script/RuleNodeJsScriptEngineTest.java index 88961dc256..730121f4fa 100644 --- a/application/src/test/java/org/thingsboard/server/service/script/RuleNodeJsScriptEngineTest.java +++ b/application/src/test/java/org/thingsboard/server/service/script/RuleNodeJsScriptEngineTest.java @@ -31,7 +31,6 @@ import java.util.Map; import java.util.Set; import java.util.UUID; import java.util.concurrent.*; -import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; import static org.junit.Assert.*; @@ -39,13 +38,13 @@ import static org.junit.Assert.*; public class RuleNodeJsScriptEngineTest { private ScriptEngine scriptEngine; - private TestNashornJsSandboxService jsSandboxService; + private TestNashornJsInvokeService jsSandboxService; private EntityId ruleNodeId = new RuleNodeId(UUIDs.timeBased()); @Before public void beforeTest() throws Exception { - jsSandboxService = new TestNashornJsSandboxService(false, 1, 100, 3); + jsSandboxService = new TestNashornJsInvokeService(false, 1, 100, 3); } @After diff --git a/application/src/test/java/org/thingsboard/server/service/script/TestNashornJsSandboxService.java b/application/src/test/java/org/thingsboard/server/service/script/TestNashornJsInvokeService.java similarity index 73% rename from application/src/test/java/org/thingsboard/server/service/script/TestNashornJsSandboxService.java rename to application/src/test/java/org/thingsboard/server/service/script/TestNashornJsInvokeService.java index 731c4bb147..3a83f40bf3 100644 --- a/application/src/test/java/org/thingsboard/server/service/script/TestNashornJsSandboxService.java +++ b/application/src/test/java/org/thingsboard/server/service/script/TestNashornJsInvokeService.java @@ -16,23 +16,14 @@ package org.thingsboard.server.service.script; -import com.google.common.util.concurrent.ListenableFuture; -import delight.nashornsandbox.NashornSandbox; -import delight.nashornsandbox.NashornSandboxes; - -import javax.script.ScriptException; -import java.util.UUID; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.Executors; - -public class TestNashornJsSandboxService extends AbstractNashornJsSandboxService { +public class TestNashornJsInvokeService extends AbstractNashornJsInvokeService { private boolean useJsSandbox; private final int monitorThreadPoolSize; private final long maxCpuTime; private final int maxErrors; - public TestNashornJsSandboxService(boolean useJsSandbox, int monitorThreadPoolSize, long maxCpuTime, int maxErrors) { + public TestNashornJsInvokeService(boolean useJsSandbox, int monitorThreadPoolSize, long maxCpuTime, int maxErrors) { this.useJsSandbox = useJsSandbox; this.monitorThreadPoolSize = monitorThreadPoolSize; this.maxCpuTime = maxCpuTime; diff --git a/common/message/src/main/java/org/thingsboard/server/common/msg/cluster/ServerAddress.java b/common/message/src/main/java/org/thingsboard/server/common/msg/cluster/ServerAddress.java index 4b65d6fe54..60c9d12b51 100644 --- a/common/message/src/main/java/org/thingsboard/server/common/msg/cluster/ServerAddress.java +++ b/common/message/src/main/java/org/thingsboard/server/common/msg/cluster/ServerAddress.java @@ -29,6 +29,7 @@ public class ServerAddress implements Comparable, Serializable { private final String host; private final int port; + private final ServerType serverType; @Override public int compareTo(ServerAddress o) { diff --git a/application/src/main/proto/discovery.proto b/common/message/src/main/java/org/thingsboard/server/common/msg/cluster/ServerType.java similarity index 72% rename from application/src/main/proto/discovery.proto rename to common/message/src/main/java/org/thingsboard/server/common/msg/cluster/ServerType.java index be4ec92d2f..f9b057c463 100644 --- a/application/src/main/proto/discovery.proto +++ b/common/message/src/main/java/org/thingsboard/server/common/msg/cluster/ServerType.java @@ -13,14 +13,12 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -syntax = "proto3"; -package discovery; +package org.thingsboard.server.common.msg.cluster; -option java_package = "org.thingsboard.server.gen.discovery"; -option java_outer_classname = "ServerInstanceProtos"; - -message ServerInfo { - string host = 1; - int32 port = 2; - int64 ts = 3; +/** + * Created by ashvayka on 23.09.18. + */ +public enum ServerType { + //Should match content of enum in discovery.proto. + CORE, JS_EVALUATOR }