Andrii Shvaika
6 years ago
63 changed files with 1055 additions and 2236 deletions
@ -1,83 +0,0 @@ |
|||
/** |
|||
* Copyright © 2016-2020 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.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; |
|||
import org.thingsboard.server.service.cluster.rpc.GrpcSessionListener; |
|||
import org.thingsboard.server.service.executors.ClusterRpcCallbackExecutorService; |
|||
|
|||
/** |
|||
* @author Andrew Shvayka |
|||
*/ |
|||
@Slf4j |
|||
public class BasicRpcSessionListener implements GrpcSessionListener { |
|||
|
|||
private final ClusterRpcCallbackExecutorService callbackExecutorService; |
|||
private final ActorService service; |
|||
private final ActorRef manager; |
|||
private final ActorRef self; |
|||
|
|||
BasicRpcSessionListener(ActorSystemContext context, ActorRef manager, ActorRef self) { |
|||
this.service = context.getActorService(); |
|||
this.callbackExecutorService = context.getClusterRpcCallbackExecutor(); |
|||
this.manager = manager; |
|||
this.self = self; |
|||
} |
|||
|
|||
@Override |
|||
public void onConnected(GrpcSession session) { |
|||
log.info("[{}][{}] session started", session.getRemoteServer(), getType(session)); |
|||
if (!session.isClient()) { |
|||
manager.tell(new RpcSessionConnectedMsg(session.getRemoteServer(), session.getSessionId()), self); |
|||
} |
|||
} |
|||
|
|||
@Override |
|||
public void onDisconnected(GrpcSession session) { |
|||
log.info("[{}][{}] session closed", session.getRemoteServer(), getType(session)); |
|||
manager.tell(new RpcSessionDisconnectedMsg(session.isClient(), session.getRemoteServer()), self); |
|||
} |
|||
|
|||
@Override |
|||
public void onReceiveClusterGrpcMsg(GrpcSession session, ClusterAPIProtos.ClusterMessage clusterMessage) { |
|||
log.trace("Received session actor msg from [{}][{}]: {}", session.getRemoteServer(), getType(session), clusterMessage); |
|||
callbackExecutorService.execute(() -> { |
|||
try { |
|||
service.onReceivedMsg(session.getRemoteServer(), clusterMessage); |
|||
} catch (Exception e) { |
|||
log.debug("[{}][{}] Failed to process cluster message: {}", session.getRemoteServer(), getType(session), clusterMessage, e); |
|||
} |
|||
}); |
|||
} |
|||
|
|||
@Override |
|||
public void onError(GrpcSession session, Throwable t) { |
|||
log.warn("[{}][{}] session got error -> {}", session.getRemoteServer(), getType(session), t); |
|||
manager.tell(new RpcSessionClosedMsg(session.isClient(), session.getRemoteServer()), self); |
|||
session.close(); |
|||
} |
|||
|
|||
private static String getType(GrpcSession session) { |
|||
return session.isClient() ? "Client" : "Server"; |
|||
} |
|||
|
|||
|
|||
} |
|||
@ -1,27 +0,0 @@ |
|||
/** |
|||
* Copyright © 2016-2020 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.actors.rpc; |
|||
|
|||
import lombok.Data; |
|||
import org.thingsboard.server.gen.cluster.ClusterAPIProtos; |
|||
|
|||
/** |
|||
* @author Andrew Shvayka |
|||
*/ |
|||
@Data |
|||
public final class RpcBroadcastMsg { |
|||
private final ClusterAPIProtos.ClusterMessage msg; |
|||
} |
|||
@ -1,230 +0,0 @@ |
|||
/** |
|||
* Copyright © 2016-2020 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.actors.rpc; |
|||
|
|||
import akka.actor.ActorRef; |
|||
import akka.actor.OneForOneStrategy; |
|||
import akka.actor.Props; |
|||
import akka.actor.SupervisorStrategy; |
|||
import akka.event.Logging; |
|||
import akka.event.LoggingAdapter; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.thingsboard.server.actors.ActorSystemContext; |
|||
import org.thingsboard.server.actors.service.ContextAwareActor; |
|||
import org.thingsboard.server.actors.service.ContextBasedCreator; |
|||
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; |
|||
import scala.concurrent.duration.Duration; |
|||
|
|||
import java.util.*; |
|||
|
|||
/** |
|||
* @author Andrew Shvayka |
|||
*/ |
|||
public class RpcManagerActor extends ContextAwareActor { |
|||
|
|||
private final Map<ServerAddress, SessionActorInfo> sessionActors; |
|||
private final Map<ServerAddress, Queue<ClusterAPIProtos.ClusterMessage>> pendingMsgs; |
|||
private final ServerAddress instance; |
|||
|
|||
private RpcManagerActor(ActorSystemContext systemContext) { |
|||
super(systemContext); |
|||
this.sessionActors = new HashMap<>(); |
|||
this.pendingMsgs = new HashMap<>(); |
|||
this.instance = systemContext.getDiscoveryService().getCurrentServer().getServerAddress(); |
|||
|
|||
systemContext.getDiscoveryService().getOtherServers().stream() |
|||
.filter(otherServer -> otherServer.getServerAddress().compareTo(instance) > 0) |
|||
.forEach(otherServer -> onCreateSessionRequest( |
|||
new RpcSessionCreateRequestMsg(UUID.randomUUID(), otherServer.getServerAddress(), null))); |
|||
} |
|||
|
|||
@Override |
|||
protected boolean process(TbActorMsg msg) { |
|||
//TODO Move everything here, to work with TbActorMsg
|
|||
return false; |
|||
} |
|||
|
|||
@Override |
|||
public void onReceive(Object msg) { |
|||
if (msg instanceof ClusterAPIProtos.ClusterMessage) { |
|||
onMsg((ClusterAPIProtos.ClusterMessage) msg); |
|||
} else if (msg instanceof RpcBroadcastMsg) { |
|||
onMsg((RpcBroadcastMsg) msg); |
|||
} else if (msg instanceof RpcSessionCreateRequestMsg) { |
|||
onCreateSessionRequest((RpcSessionCreateRequestMsg) msg); |
|||
} else if (msg instanceof RpcSessionConnectedMsg) { |
|||
onSessionConnected((RpcSessionConnectedMsg) msg); |
|||
} else if (msg instanceof RpcSessionDisconnectedMsg) { |
|||
onSessionDisconnected((RpcSessionDisconnectedMsg) msg); |
|||
} else if (msg instanceof RpcSessionClosedMsg) { |
|||
onSessionClosed((RpcSessionClosedMsg) msg); |
|||
} else if (msg instanceof ClusterEventMsg) { |
|||
onClusterEvent((ClusterEventMsg) msg); |
|||
} |
|||
} |
|||
|
|||
private void onMsg(RpcBroadcastMsg msg) { |
|||
log.debug("Forwarding msg to session actors {}", msg); |
|||
sessionActors.keySet().forEach(address -> { |
|||
ClusterAPIProtos.ClusterMessage msgWithServerAddress = msg.getMsg() |
|||
.toBuilder() |
|||
.setServerAddress(ClusterAPIProtos.ServerAddress |
|||
.newBuilder() |
|||
.setHost(address.getHost()) |
|||
.setPort(address.getPort()) |
|||
.build()) |
|||
.build(); |
|||
onMsg(msgWithServerAddress); |
|||
}); |
|||
pendingMsgs.values().forEach(queue -> queue.add(msg.getMsg())); |
|||
} |
|||
|
|||
private void onMsg(ClusterAPIProtos.ClusterMessage msg) { |
|||
if (msg.hasServerAddress()) { |
|||
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, msg); |
|||
session.getActor().tell(msg, ActorRef.noSender()); |
|||
} else { |
|||
log.debug("{} Storing msg to pending queue: {}", address, msg); |
|||
Queue<ClusterAPIProtos.ClusterMessage> queue = pendingMsgs.get(address); |
|||
if (queue == null) { |
|||
queue = new LinkedList<>(); |
|||
pendingMsgs.put(new ServerAddress( |
|||
msg.getServerAddress().getHost(), msg.getServerAddress().getPort(), ServerType.CORE), queue); |
|||
} |
|||
queue.add(msg); |
|||
} |
|||
} else { |
|||
log.warn("Cluster msg doesn't have server address [{}]", msg); |
|||
} |
|||
} |
|||
|
|||
@Override |
|||
public void postStop() { |
|||
sessionActors.clear(); |
|||
pendingMsgs.clear(); |
|||
} |
|||
|
|||
private void onClusterEvent(ClusterEventMsg msg) { |
|||
ServerAddress server = msg.getServerAddress(); |
|||
if (server.compareTo(instance) > 0) { |
|||
if (msg.isAdded()) { |
|||
onCreateSessionRequest(new RpcSessionCreateRequestMsg(UUID.randomUUID(), server, null)); |
|||
} else { |
|||
onSessionClose(false, server); |
|||
} |
|||
} |
|||
} |
|||
|
|||
private void onSessionConnected(RpcSessionConnectedMsg msg) { |
|||
register(msg.getRemoteAddress(), msg.getId(), context().sender()); |
|||
} |
|||
|
|||
private void onSessionDisconnected(RpcSessionDisconnectedMsg msg) { |
|||
boolean reconnect = msg.isClient() && isRegistered(msg.getRemoteAddress()); |
|||
onSessionClose(reconnect, msg.getRemoteAddress()); |
|||
} |
|||
|
|||
private void onSessionClosed(RpcSessionClosedMsg msg) { |
|||
boolean reconnect = msg.isClient() && isRegistered(msg.getRemoteAddress()); |
|||
onSessionClose(reconnect, msg.getRemoteAddress()); |
|||
} |
|||
|
|||
private boolean isRegistered(ServerAddress address) { |
|||
for (ServerInstance server : systemContext.getDiscoveryService().getOtherServers()) { |
|||
if (server.getServerAddress().equals(address)) { |
|||
return true; |
|||
} |
|||
} |
|||
return false; |
|||
} |
|||
|
|||
private void onSessionClose(boolean reconnect, ServerAddress remoteAddress) { |
|||
log.info("[{}] session closed. Should reconnect: {}", remoteAddress, reconnect); |
|||
SessionActorInfo sessionRef = sessionActors.get(remoteAddress); |
|||
if (sessionRef != null && context().sender() != null && context().sender().equals(sessionRef.actor)) { |
|||
context().stop(sessionRef.actor); |
|||
sessionActors.remove(remoteAddress); |
|||
pendingMsgs.remove(remoteAddress); |
|||
if (reconnect) { |
|||
onCreateSessionRequest(new RpcSessionCreateRequestMsg(sessionRef.sessionId, remoteAddress, null)); |
|||
} |
|||
} |
|||
} |
|||
|
|||
private void onCreateSessionRequest(RpcSessionCreateRequestMsg msg) { |
|||
if (msg.getRemoteAddress() != null) { |
|||
if (!sessionActors.containsKey(msg.getRemoteAddress())) { |
|||
ActorRef actorRef = createSessionActor(msg); |
|||
register(msg.getRemoteAddress(), msg.getMsgUid(), actorRef); |
|||
} |
|||
} else { |
|||
createSessionActor(msg); |
|||
} |
|||
} |
|||
|
|||
private void register(ServerAddress remoteAddress, UUID uuid, ActorRef sender) { |
|||
sessionActors.put(remoteAddress, new SessionActorInfo(uuid, sender)); |
|||
log.info("[{}][{}] Registering session actor.", remoteAddress, uuid); |
|||
Queue<ClusterAPIProtos.ClusterMessage> data = pendingMsgs.remove(remoteAddress); |
|||
if (data != null) { |
|||
log.info("[{}][{}] Forwarding {} pending messages.", remoteAddress, uuid, data.size()); |
|||
data.forEach(msg -> sender.tell(new RpcSessionTellMsg(msg), ActorRef.noSender())); |
|||
} else { |
|||
log.info("[{}][{}] No pending messages to forward.", remoteAddress, uuid); |
|||
} |
|||
} |
|||
|
|||
private ActorRef createSessionActor(RpcSessionCreateRequestMsg msg) { |
|||
log.info("[{}] Creating session actor.", msg.getMsgUid()); |
|||
ActorRef actor = context().actorOf( |
|||
Props.create(new RpcSessionActor.ActorCreator(systemContext, msg.getMsgUid())) |
|||
.withDispatcher(DefaultActorService.RPC_DISPATCHER_NAME)); |
|||
actor.tell(msg, context().self()); |
|||
return actor; |
|||
} |
|||
|
|||
public static class ActorCreator extends ContextBasedCreator<RpcManagerActor> { |
|||
private static final long serialVersionUID = 1L; |
|||
|
|||
public ActorCreator(ActorSystemContext context) { |
|||
super(context); |
|||
} |
|||
|
|||
@Override |
|||
public RpcManagerActor create() { |
|||
return new RpcManagerActor(context); |
|||
} |
|||
} |
|||
|
|||
@Override |
|||
public SupervisorStrategy supervisorStrategy() { |
|||
return strategy; |
|||
} |
|||
|
|||
private final SupervisorStrategy strategy = new OneForOneStrategy(3, Duration.create("1 minute"), t -> { |
|||
log.warn("Unknown failure", t); |
|||
return SupervisorStrategy.resume(); |
|||
}); |
|||
} |
|||
@ -1,135 +0,0 @@ |
|||
/** |
|||
* Copyright © 2016-2020 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.actors.rpc; |
|||
|
|||
import io.grpc.ManagedChannel; |
|||
import io.grpc.ManagedChannelBuilder; |
|||
import io.grpc.stub.StreamObserver; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.thingsboard.server.actors.ActorSystemContext; |
|||
import org.thingsboard.server.actors.service.ContextAwareActor; |
|||
import org.thingsboard.server.actors.service.ContextBasedCreator; |
|||
import org.thingsboard.server.common.msg.TbActorMsg; |
|||
import org.thingsboard.server.common.msg.cluster.ServerAddress; |
|||
import org.thingsboard.server.gen.cluster.ClusterAPIProtos; |
|||
import org.thingsboard.server.gen.cluster.ClusterRpcServiceGrpc; |
|||
import org.thingsboard.server.service.cluster.rpc.GrpcSession; |
|||
import org.thingsboard.server.service.cluster.rpc.GrpcSessionListener; |
|||
|
|||
import java.util.UUID; |
|||
|
|||
import static org.thingsboard.server.gen.cluster.ClusterAPIProtos.MessageType.CONNECT_RPC_MESSAGE; |
|||
|
|||
/** |
|||
* @author Andrew Shvayka |
|||
*/ |
|||
@Slf4j |
|||
public class RpcSessionActor extends ContextAwareActor { |
|||
|
|||
|
|||
private final UUID sessionId; |
|||
private GrpcSession session; |
|||
private GrpcSessionListener listener; |
|||
|
|||
private RpcSessionActor(ActorSystemContext systemContext, UUID sessionId) { |
|||
super(systemContext); |
|||
this.sessionId = sessionId; |
|||
} |
|||
|
|||
@Override |
|||
protected boolean process(TbActorMsg msg) { |
|||
//TODO Move everything here, to work with TbActorMsg
|
|||
return false; |
|||
} |
|||
|
|||
@Override |
|||
public void onReceive(Object msg) { |
|||
if (msg instanceof ClusterAPIProtos.ClusterMessage) { |
|||
tell((ClusterAPIProtos.ClusterMessage) msg); |
|||
} else if (msg instanceof RpcSessionCreateRequestMsg) { |
|||
initSession((RpcSessionCreateRequestMsg) msg); |
|||
} |
|||
} |
|||
|
|||
private void tell(ClusterAPIProtos.ClusterMessage msg) { |
|||
if (session != null) { |
|||
session.sendMsg(msg); |
|||
} else { |
|||
log.trace("Failed to send message due to missing session!"); |
|||
} |
|||
} |
|||
|
|||
@Override |
|||
public void postStop() { |
|||
if (session != null) { |
|||
log.info("Closing session -> {}", session.getRemoteServer()); |
|||
try { |
|||
session.close(); |
|||
} catch (RuntimeException e) { |
|||
log.trace("Failed to close session!", e); |
|||
} |
|||
} |
|||
} |
|||
|
|||
private void initSession(RpcSessionCreateRequestMsg msg) { |
|||
log.info("[{}] Initializing session", context().self()); |
|||
ServerAddress remoteServer = msg.getRemoteAddress(); |
|||
listener = new BasicRpcSessionListener(systemContext, context().parent(), context().self()); |
|||
if (msg.getRemoteAddress() == null) { |
|||
// Server session
|
|||
session = new GrpcSession(listener); |
|||
session.setOutputStream(msg.getResponseObserver()); |
|||
session.initInputStream(); |
|||
session.initOutputStream(); |
|||
systemContext.getRpcService().onSessionCreated(msg.getMsgUid(), session.getInputStream()); |
|||
} else { |
|||
// Client session
|
|||
ManagedChannel channel = ManagedChannelBuilder.forAddress(remoteServer.getHost(), remoteServer.getPort()).usePlaintext().build(); |
|||
session = new GrpcSession(remoteServer, listener, channel); |
|||
session.initInputStream(); |
|||
|
|||
ClusterRpcServiceGrpc.ClusterRpcServiceStub stub = ClusterRpcServiceGrpc.newStub(channel); |
|||
StreamObserver<ClusterAPIProtos.ClusterMessage> outputStream = stub.handleMsgs(session.getInputStream()); |
|||
|
|||
session.setOutputStream(outputStream); |
|||
session.initOutputStream(); |
|||
outputStream.onNext(toConnectMsg()); |
|||
} |
|||
} |
|||
|
|||
public static class ActorCreator extends ContextBasedCreator<RpcSessionActor> { |
|||
private static final long serialVersionUID = 1L; |
|||
|
|||
private final UUID sessionId; |
|||
|
|||
public ActorCreator(ActorSystemContext context, UUID sessionId) { |
|||
super(context); |
|||
this.sessionId = sessionId; |
|||
} |
|||
|
|||
@Override |
|||
public RpcSessionActor create() { |
|||
return new RpcSessionActor(context, sessionId); |
|||
} |
|||
} |
|||
|
|||
private ClusterAPIProtos.ClusterMessage toConnectMsg() { |
|||
ServerAddress instance = systemContext.getDiscoveryService().getCurrentServer().getServerAddress(); |
|||
return ClusterAPIProtos.ClusterMessage.newBuilder().setMessageType(CONNECT_RPC_MESSAGE).setServerAddress( |
|||
ClusterAPIProtos.ServerAddress.newBuilder().setHost(instance.getHost()) |
|||
.setPort(instance.getPort()).build()).build(); |
|||
} |
|||
} |
|||
@ -1,29 +0,0 @@ |
|||
/** |
|||
* Copyright © 2016-2020 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.actors.rpc; |
|||
|
|||
import lombok.Data; |
|||
import org.thingsboard.server.common.msg.cluster.ServerAddress; |
|||
|
|||
/** |
|||
* @author Andrew Shvayka |
|||
*/ |
|||
@Data |
|||
public final class RpcSessionClosedMsg { |
|||
|
|||
private final boolean client; |
|||
private final ServerAddress remoteAddress; |
|||
} |
|||
@ -1,31 +0,0 @@ |
|||
/** |
|||
* Copyright © 2016-2020 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.actors.rpc; |
|||
|
|||
import lombok.Data; |
|||
import org.thingsboard.server.common.msg.cluster.ServerAddress; |
|||
|
|||
import java.util.UUID; |
|||
|
|||
/** |
|||
* @author Andrew Shvayka |
|||
*/ |
|||
@Data |
|||
public final class RpcSessionConnectedMsg { |
|||
|
|||
private final ServerAddress remoteAddress; |
|||
private final UUID id; |
|||
} |
|||
@ -1,35 +0,0 @@ |
|||
/** |
|||
* Copyright © 2016-2020 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.actors.rpc; |
|||
|
|||
import io.grpc.stub.StreamObserver; |
|||
import lombok.Data; |
|||
import org.thingsboard.server.common.msg.cluster.ServerAddress; |
|||
import org.thingsboard.server.gen.cluster.ClusterAPIProtos; |
|||
|
|||
import java.util.UUID; |
|||
|
|||
/** |
|||
* @author Andrew Shvayka |
|||
*/ |
|||
@Data |
|||
public final class RpcSessionCreateRequestMsg { |
|||
|
|||
private final UUID msgUid; |
|||
private final ServerAddress remoteAddress; |
|||
private final StreamObserver<ClusterAPIProtos.ClusterMessage> responseObserver; |
|||
|
|||
} |
|||
@ -1,29 +0,0 @@ |
|||
/** |
|||
* Copyright © 2016-2020 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.actors.rpc; |
|||
|
|||
import lombok.Data; |
|||
import org.thingsboard.server.common.msg.cluster.ServerAddress; |
|||
|
|||
/** |
|||
* @author Andrew Shvayka |
|||
*/ |
|||
@Data |
|||
public final class RpcSessionDisconnectedMsg { |
|||
|
|||
private final boolean client; |
|||
private final ServerAddress remoteAddress; |
|||
} |
|||
@ -1,27 +0,0 @@ |
|||
/** |
|||
* Copyright © 2016-2020 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.actors.rpc; |
|||
|
|||
import lombok.Data; |
|||
import org.thingsboard.server.gen.cluster.ClusterAPIProtos; |
|||
|
|||
/** |
|||
* @author Andrew Shvayka |
|||
*/ |
|||
@Data |
|||
public final class RpcSessionTellMsg { |
|||
private final ClusterAPIProtos.ClusterMessage msg; |
|||
} |
|||
@ -1,30 +0,0 @@ |
|||
/** |
|||
* Copyright © 2016-2020 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.actors.rpc; |
|||
|
|||
import akka.actor.ActorRef; |
|||
import lombok.Data; |
|||
|
|||
import java.util.UUID; |
|||
|
|||
/** |
|||
* @author Andrew Shvayka |
|||
*/ |
|||
@Data |
|||
public final class SessionActorInfo { |
|||
protected final UUID sessionId; |
|||
protected final ActorRef actor; |
|||
} |
|||
@ -1,55 +0,0 @@ |
|||
/** |
|||
* Copyright © 2016-2020 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.discovery; |
|||
|
|||
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.common.msg.cluster.ServerAddress; |
|||
import org.thingsboard.server.common.msg.cluster.ServerType; |
|||
|
|||
import javax.annotation.PostConstruct; |
|||
|
|||
import static org.thingsboard.server.utils.MiscUtils.missingProperty; |
|||
|
|||
/** |
|||
* @author Andrew Shvayka |
|||
*/ |
|||
@Service |
|||
@Slf4j |
|||
public class CurrentServerInstanceService implements ServerInstanceService { |
|||
|
|||
@Value("${rpc.bind_host}") |
|||
private String rpcHost; |
|||
@Value("${rpc.bind_port}") |
|||
private Integer rpcPort; |
|||
|
|||
private ServerInstance self; |
|||
|
|||
@PostConstruct |
|||
public void init() { |
|||
Assert.hasLength(rpcHost, missingProperty("rpc.bind_host")); |
|||
Assert.notNull(rpcPort, missingProperty("rpc.bind_port")); |
|||
self = new ServerInstance(new ServerAddress(rpcHost, rpcPort, ServerType.CORE)); |
|||
log.info("Current server instance: [{};{}]", self.getHost(), self.getPort()); |
|||
} |
|||
|
|||
@Override |
|||
public ServerInstance getSelf() { |
|||
return self; |
|||
} |
|||
} |
|||
@ -1,33 +0,0 @@ |
|||
/** |
|||
* Copyright © 2016-2020 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.discovery; |
|||
|
|||
import java.util.List; |
|||
|
|||
/** |
|||
* @author Andrew Shvayka |
|||
*/ |
|||
public interface DiscoveryService { |
|||
|
|||
void publishCurrentServer(); |
|||
|
|||
void unpublishCurrentServer(); |
|||
|
|||
ServerInstance getCurrentServer(); |
|||
|
|||
List<ServerInstance> getOtherServers(); |
|||
|
|||
} |
|||
@ -1,28 +0,0 @@ |
|||
/** |
|||
* Copyright © 2016-2020 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.discovery; |
|||
|
|||
/** |
|||
* @author Andrew Shvayka |
|||
*/ |
|||
public interface DiscoveryServiceListener { |
|||
|
|||
void onServerAdded(ServerInstance server); |
|||
|
|||
void onServerUpdated(ServerInstance server); |
|||
|
|||
void onServerRemoved(ServerInstance server); |
|||
} |
|||
@ -1,67 +0,0 @@ |
|||
/** |
|||
* Copyright © 2016-2020 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.discovery; |
|||
|
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.apache.commons.lang3.RandomStringUtils; |
|||
import org.springframework.beans.factory.annotation.Autowired; |
|||
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; |
|||
import org.springframework.context.annotation.DependsOn; |
|||
import org.springframework.stereotype.Service; |
|||
|
|||
import javax.annotation.PostConstruct; |
|||
import java.util.Collections; |
|||
import java.util.List; |
|||
|
|||
/** |
|||
* @author Andrew Shvayka |
|||
*/ |
|||
@Service |
|||
@ConditionalOnProperty(prefix = "zk", value = "enabled", havingValue = "false", matchIfMissing = true) |
|||
@Slf4j |
|||
@DependsOn("environmentLogService") |
|||
public class DummyDiscoveryService implements DiscoveryService { |
|||
|
|||
@Autowired |
|||
private ServerInstanceService serverInstance; |
|||
|
|||
@PostConstruct |
|||
public void init() { |
|||
log.info("Initializing..."); |
|||
} |
|||
|
|||
@Override |
|||
public void publishCurrentServer() { |
|||
//Do nothing
|
|||
} |
|||
|
|||
@Override |
|||
public void unpublishCurrentServer() { |
|||
//Do nothing
|
|||
} |
|||
|
|||
@Override |
|||
public ServerInstance getCurrentServer() { |
|||
return serverInstance.getSelf(); |
|||
} |
|||
|
|||
@Override |
|||
public List<ServerInstance> getOtherServers() { |
|||
return Collections.emptyList(); |
|||
} |
|||
|
|||
|
|||
} |
|||
@ -1,47 +0,0 @@ |
|||
/** |
|||
* Copyright © 2016-2020 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.discovery; |
|||
|
|||
import lombok.EqualsAndHashCode; |
|||
import lombok.Getter; |
|||
import lombok.ToString; |
|||
import org.thingsboard.server.common.msg.cluster.ServerAddress; |
|||
|
|||
/** |
|||
* @author Andrew Shvayka |
|||
*/ |
|||
@ToString |
|||
@EqualsAndHashCode(exclude = {"serverInfo", "serverAddress"}) |
|||
public final class ServerInstance implements Comparable<ServerInstance> { |
|||
|
|||
@Getter |
|||
private final String host; |
|||
@Getter |
|||
private final int port; |
|||
@Getter |
|||
private final ServerAddress serverAddress; |
|||
|
|||
public ServerInstance(ServerAddress serverAddress) { |
|||
this.serverAddress = serverAddress; |
|||
this.host = serverAddress.getHost(); |
|||
this.port = serverAddress.getPort(); |
|||
} |
|||
|
|||
@Override |
|||
public int compareTo(ServerInstance o) { |
|||
return this.serverAddress.compareTo(o.serverAddress); |
|||
} |
|||
} |
|||
@ -1,24 +0,0 @@ |
|||
/** |
|||
* Copyright © 2016-2020 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.discovery; |
|||
|
|||
/** |
|||
* @author Andrew Shvayka |
|||
*/ |
|||
public interface ServerInstanceService { |
|||
|
|||
ServerInstance getSelf(); |
|||
} |
|||
@ -1,330 +0,0 @@ |
|||
/** |
|||
* Copyright © 2016-2020 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.discovery; |
|||
|
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.apache.commons.lang3.RandomStringUtils; |
|||
import org.apache.commons.lang3.SerializationException; |
|||
import org.apache.commons.lang3.SerializationUtils; |
|||
import org.apache.curator.framework.CuratorFramework; |
|||
import org.apache.curator.framework.CuratorFrameworkFactory; |
|||
import org.apache.curator.framework.imps.CuratorFrameworkState; |
|||
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; |
|||
import org.apache.zookeeper.KeeperException; |
|||
import org.springframework.beans.factory.annotation.Autowired; |
|||
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.context.event.EventListener; |
|||
import org.springframework.stereotype.Service; |
|||
import org.springframework.util.Assert; |
|||
import org.thingsboard.common.util.ThingsBoardThreadFactory; |
|||
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; |
|||
|
|||
import javax.annotation.PostConstruct; |
|||
import javax.annotation.PreDestroy; |
|||
import java.util.List; |
|||
import java.util.NoSuchElementException; |
|||
import java.util.concurrent.ExecutorService; |
|||
import java.util.concurrent.Executors; |
|||
import java.util.stream.Collectors; |
|||
|
|||
import static org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent.Type.CHILD_REMOVED; |
|||
|
|||
/** |
|||
* @author Andrew Shvayka |
|||
*/ |
|||
@Service |
|||
@ConditionalOnProperty(prefix = "zk", value = "enabled", havingValue = "true", matchIfMissing = false) |
|||
@Slf4j |
|||
public class ZkDiscoveryService implements DiscoveryService, PathChildrenCacheListener { |
|||
|
|||
@Value("${zk.url}") |
|||
private String zkUrl; |
|||
@Value("${zk.retry_interval_ms}") |
|||
private Integer zkRetryInterval; |
|||
@Value("${zk.connection_timeout_ms}") |
|||
private Integer zkConnectionTimeout; |
|||
@Value("${zk.session_timeout_ms}") |
|||
private Integer zkSessionTimeout; |
|||
@Value("${zk.zk_dir}") |
|||
private String zkDir; |
|||
|
|||
private String zkNodesDir; |
|||
|
|||
@Autowired |
|||
private ServerInstanceService serverInstance; |
|||
|
|||
@Autowired |
|||
@Lazy |
|||
private TelemetrySubscriptionService tsSubService; |
|||
|
|||
@Autowired |
|||
@Lazy |
|||
private DeviceStateService deviceStateService; |
|||
|
|||
@Autowired |
|||
@Lazy |
|||
private ActorService actorService; |
|||
|
|||
@Autowired |
|||
@Lazy |
|||
private ClusterRoutingService routingService; |
|||
|
|||
private ExecutorService reconnectExecutorService; |
|||
|
|||
private CuratorFramework client; |
|||
private PathChildrenCache cache; |
|||
private String nodePath; |
|||
|
|||
private volatile boolean stopped = true; |
|||
|
|||
@PostConstruct |
|||
public void init() { |
|||
log.info("Initializing..."); |
|||
Assert.hasLength(zkUrl, MiscUtils.missingProperty("zk.url")); |
|||
Assert.notNull(zkRetryInterval, MiscUtils.missingProperty("zk.retry_interval_ms")); |
|||
Assert.notNull(zkConnectionTimeout, MiscUtils.missingProperty("zk.connection_timeout_ms")); |
|||
Assert.notNull(zkSessionTimeout, MiscUtils.missingProperty("zk.session_timeout_ms")); |
|||
|
|||
reconnectExecutorService = Executors.newSingleThreadExecutor(ThingsBoardThreadFactory.forName("zk-discovery")); |
|||
|
|||
log.info("Initializing discovery service using ZK connect string: {}", zkUrl); |
|||
|
|||
zkNodesDir = zkDir + "/nodes"; |
|||
initZkClient(); |
|||
} |
|||
|
|||
private void initZkClient() { |
|||
try { |
|||
client = CuratorFrameworkFactory.newClient(zkUrl, zkSessionTimeout, zkConnectionTimeout, new RetryForever(zkRetryInterval)); |
|||
client.start(); |
|||
client.blockUntilConnected(); |
|||
cache = new PathChildrenCache(client, zkNodesDir, true); |
|||
cache.getListenable().addListener(this); |
|||
cache.start(); |
|||
stopped = false; |
|||
log.info("ZK client connected"); |
|||
} catch (Exception e) { |
|||
log.error("Failed to connect to ZK: {}", e.getMessage(), e); |
|||
CloseableUtils.closeQuietly(cache); |
|||
CloseableUtils.closeQuietly(client); |
|||
throw new RuntimeException(e); |
|||
} |
|||
} |
|||
|
|||
private void destroyZkClient() { |
|||
stopped = true; |
|||
try { |
|||
unpublishCurrentServer(); |
|||
} catch (Exception e) {} |
|||
CloseableUtils.closeQuietly(cache); |
|||
CloseableUtils.closeQuietly(client); |
|||
log.info("ZK client disconnected"); |
|||
} |
|||
|
|||
@PreDestroy |
|||
public void destroy() { |
|||
destroyZkClient(); |
|||
reconnectExecutorService.shutdownNow(); |
|||
log.info("Stopped discovery service"); |
|||
} |
|||
|
|||
@Override |
|||
public synchronized void publishCurrentServer() { |
|||
ServerInstance self = this.serverInstance.getSelf(); |
|||
if (currentServerExists()) { |
|||
log.info("[{}:{}] ZK node for current instance already exists, NOT created new one: {}", self.getHost(), self.getPort(), nodePath); |
|||
} else { |
|||
try { |
|||
log.info("[{}:{}] Creating ZK node for current instance", self.getHost(), self.getPort()); |
|||
nodePath = client.create() |
|||
.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 currentServerExists() { |
|||
if (nodePath == null) { |
|||
return false; |
|||
} |
|||
try { |
|||
ServerInstance self = this.serverInstance.getSelf(); |
|||
ServerAddress registeredServerAdress = null; |
|||
registeredServerAdress = SerializationUtils.deserialize(client.getData().forPath(nodePath)); |
|||
if (self.getServerAddress() != null && self.getServerAddress().equals(registeredServerAdress)) { |
|||
return true; |
|||
} |
|||
} catch (KeeperException.NoNodeException e) { |
|||
log.info("ZK node does not exist: {}", nodePath); |
|||
} catch (Exception e) { |
|||
log.error("Couldn't check if ZK node exists", e); |
|||
} |
|||
return false; |
|||
} |
|||
|
|||
private ConnectionStateListener checkReconnect(ServerInstance self) { |
|||
return (client, newState) -> { |
|||
log.info("[{}:{}] ZK state changed: {}", self.getHost(), self.getPort(), newState); |
|||
if (newState == ConnectionState.LOST) { |
|||
reconnectExecutorService.submit(this::reconnect); |
|||
} |
|||
}; |
|||
} |
|||
|
|||
private volatile boolean reconnectInProgress = false; |
|||
|
|||
private synchronized void reconnect() { |
|||
if (!reconnectInProgress) { |
|||
reconnectInProgress = true; |
|||
try { |
|||
destroyZkClient(); |
|||
initZkClient(); |
|||
publishCurrentServer(); |
|||
} catch (Exception e) { |
|||
log.error("Failed to reconnect to ZK: {}", e.getMessage(), e); |
|||
} finally { |
|||
reconnectInProgress = false; |
|||
} |
|||
} |
|||
} |
|||
|
|||
@Override |
|||
public void unpublishCurrentServer() { |
|||
try { |
|||
if (nodePath != null) { |
|||
client.delete().forPath(nodePath); |
|||
} |
|||
} catch (Exception e) { |
|||
log.error("Failed to delete ZK node {}", nodePath, e); |
|||
throw new RuntimeException(e); |
|||
} |
|||
} |
|||
|
|||
@Override |
|||
public ServerInstance getCurrentServer() { |
|||
return serverInstance.getSelf(); |
|||
} |
|||
|
|||
@Override |
|||
public List<ServerInstance> getOtherServers() { |
|||
return cache.getCurrentData().stream() |
|||
.filter(cd -> !cd.getPath().equals(nodePath)) |
|||
.map(cd -> { |
|||
try { |
|||
return new ServerInstance((ServerAddress) SerializationUtils.deserialize(cd.getData())); |
|||
} catch (NoSuchElementException e) { |
|||
log.error("Failed to decode ZK node", e); |
|||
throw new RuntimeException(e); |
|||
} |
|||
}) |
|||
.collect(Collectors.toList()); |
|||
} |
|||
|
|||
@EventListener(ApplicationReadyEvent.class) |
|||
public void onApplicationEvent(ApplicationReadyEvent applicationReadyEvent) { |
|||
log.info("Received application ready event. Starting current ZK node."); |
|||
if (stopped) { |
|||
log.debug("Ignoring application ready event. Service is stopped."); |
|||
return; |
|||
} |
|||
if (client.getState() != CuratorFrameworkState.STARTED) { |
|||
log.debug("Ignoring application ready event, ZK client is not started, ZK client state [{}]", client.getState()); |
|||
return; |
|||
} |
|||
publishCurrentServer(); |
|||
getOtherServers().forEach( |
|||
server -> log.info("Found active server: [{}:{}]", server.getHost(), server.getPort()) |
|||
); |
|||
} |
|||
|
|||
@Override |
|||
public void childEvent(CuratorFramework curatorFramework, PathChildrenCacheEvent pathChildrenCacheEvent) throws Exception { |
|||
if (stopped) { |
|||
log.debug("Ignoring {}. Service is stopped.", pathChildrenCacheEvent); |
|||
return; |
|||
} |
|||
if (client.getState() != CuratorFrameworkState.STARTED) { |
|||
log.debug("Ignoring {}, ZK client is not started, ZK client state [{}]", pathChildrenCacheEvent, client.getState()); |
|||
return; |
|||
} |
|||
ChildData data = pathChildrenCacheEvent.getData(); |
|||
if (data == null) { |
|||
log.debug("Ignoring {} due to empty child data", pathChildrenCacheEvent); |
|||
return; |
|||
} else if (data.getData() == null) { |
|||
log.debug("Ignoring {} due to empty child's data", pathChildrenCacheEvent); |
|||
return; |
|||
} else if (nodePath != null && nodePath.equals(data.getPath())) { |
|||
if (pathChildrenCacheEvent.getType() == CHILD_REMOVED) { |
|||
log.info("ZK node for current instance is somehow deleted."); |
|||
publishCurrentServer(); |
|||
} |
|||
log.debug("Ignoring event about current server {}", pathChildrenCacheEvent); |
|||
return; |
|||
} |
|||
ServerInstance instance; |
|||
try { |
|||
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); |
|||
throw e; |
|||
} |
|||
log.info("Processing [{}] event for [{}:{}]", pathChildrenCacheEvent.getType(), instance.getHost(), instance.getPort()); |
|||
switch (pathChildrenCacheEvent.getType()) { |
|||
case CHILD_ADDED: |
|||
routingService.onServerAdded(instance); |
|||
tsSubService.onClusterUpdate(); |
|||
deviceStateService.onClusterUpdate(); |
|||
actorService.onServerAdded(instance); |
|||
break; |
|||
case CHILD_UPDATED: |
|||
routingService.onServerUpdated(instance); |
|||
actorService.onServerUpdated(instance); |
|||
break; |
|||
case CHILD_REMOVED: |
|||
routingService.onServerRemoved(instance); |
|||
tsSubService.onClusterUpdate(); |
|||
deviceStateService.onClusterUpdate(); |
|||
actorService.onServerRemoved(instance); |
|||
break; |
|||
default: |
|||
break; |
|||
} |
|||
} |
|||
} |
|||
@ -1,35 +0,0 @@ |
|||
/** |
|||
* Copyright © 2016-2020 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 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; |
|||
|
|||
/** |
|||
* @author Andrew Shvayka |
|||
*/ |
|||
public interface ClusterRoutingService extends DiscoveryServiceListener { |
|||
|
|||
ServerAddress getCurrentServer(); |
|||
|
|||
Optional<ServerAddress> resolveById(EntityId entityId); |
|||
|
|||
} |
|||
@ -1,153 +0,0 @@ |
|||
/** |
|||
* Copyright © 2016-2020 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 com.google.common.hash.HashCode; |
|||
import com.google.common.hash.HashFunction; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.springframework.beans.factory.annotation.Autowired; |
|||
import org.springframework.beans.factory.annotation.Value; |
|||
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; |
|||
import java.util.concurrent.ConcurrentSkipListMap; |
|||
|
|||
/** |
|||
* Cluster service implementation based on consistent hash ring |
|||
*/ |
|||
|
|||
@Service |
|||
@Slf4j |
|||
public class ConsistentClusterRoutingService implements ClusterRoutingService { |
|||
|
|||
@Autowired |
|||
private DiscoveryService discoveryService; |
|||
|
|||
@Value("${cluster.hash_function_name}") |
|||
private String hashFunctionName; |
|||
@Value("${cluster.vitrual_nodes_size}") |
|||
private Integer virtualNodesSize; |
|||
|
|||
private ServerInstance currentServer; |
|||
|
|||
private HashFunction hashFunction; |
|||
|
|||
private ConsistentHashCircle[] circles; |
|||
private ConsistentHashCircle rootCircle; |
|||
|
|||
@PostConstruct |
|||
public void init() { |
|||
log.info("Initializing Cluster routing service!"); |
|||
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); |
|||
} |
|||
logCircle(); |
|||
log.info("Cluster routing service initialized!"); |
|||
} |
|||
|
|||
@Override |
|||
public ServerAddress getCurrentServer() { |
|||
return discoveryService.getCurrentServer().getServerAddress(); |
|||
} |
|||
|
|||
@Override |
|||
public Optional<ServerAddress> resolveById(EntityId entityId) { |
|||
return resolveByUuid(rootCircle, entityId.getId()); |
|||
} |
|||
|
|||
private Optional<ServerAddress> resolveByUuid(ConsistentHashCircle circle, UUID uuid) { |
|||
Assert.notNull(uuid); |
|||
if (circle.isEmpty()) { |
|||
return Optional.empty(); |
|||
} |
|||
Long hash = hashFunction.newHasher().putLong(uuid.getMostSignificantBits()) |
|||
.putLong(uuid.getLeastSignificantBits()).hash().asLong(); |
|||
if (!circle.containsKey(hash)) { |
|||
ConcurrentNavigableMap<Long, ServerInstance> tailMap = |
|||
circle.tailMap(hash); |
|||
hash = tailMap.isEmpty() ? |
|||
circle.firstKey() : tailMap.firstKey(); |
|||
} |
|||
ServerInstance result = circle.get(hash); |
|||
if (!currentServer.equals(result)) { |
|||
return Optional.of(result.getServerAddress()); |
|||
} else { |
|||
return Optional.empty(); |
|||
} |
|||
} |
|||
|
|||
@Override |
|||
public void onServerAdded(ServerInstance server) { |
|||
log.info("On server added event: {}", server); |
|||
addNode(server); |
|||
logCircle(); |
|||
} |
|||
|
|||
@Override |
|||
public void onServerUpdated(ServerInstance server) { |
|||
log.debug("Ignoring server onUpdate event: {}", server); |
|||
} |
|||
|
|||
@Override |
|||
public void onServerRemoved(ServerInstance server) { |
|||
log.info("On server removed event: {}", server); |
|||
removeNode(server); |
|||
logCircle(); |
|||
} |
|||
|
|||
private void addNode(ServerInstance instance) { |
|||
for (int i = 0; i < virtualNodesSize; i++) { |
|||
circles[instance.getServerAddress().getServerType().ordinal()].put(hash(instance, i).asLong(), instance); |
|||
} |
|||
} |
|||
|
|||
private void removeNode(ServerInstance instance) { |
|||
for (int i = 0; i < virtualNodesSize; i++) { |
|||
circles[instance.getServerAddress().getServerType().ordinal()].remove(hash(instance, i).asLong()); |
|||
} |
|||
} |
|||
|
|||
private HashCode hash(ServerInstance instance, int i) { |
|||
return hashFunction.newHasher().putString(instance.getHost(), MiscUtils.UTF8).putInt(instance.getPort()).putInt(i).hash(); |
|||
} |
|||
|
|||
private void logCircle() { |
|||
log.trace("Consistent Hash Circle Start"); |
|||
Arrays.asList(circles).forEach(ConsistentHashCircle::log); |
|||
log.trace("Consistent Hash Circle End"); |
|||
} |
|||
|
|||
} |
|||
@ -1,237 +0,0 @@ |
|||
/** |
|||
* Copyright © 2016-2020 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.transport; |
|||
|
|||
import akka.actor.ActorRef; |
|||
import io.github.bucket4j.Bandwidth; |
|||
import io.github.bucket4j.BlockingBucket; |
|||
import io.github.bucket4j.Bucket4j; |
|||
import io.github.bucket4j.local.LocalBucket; |
|||
import io.github.bucket4j.local.LocalBucketBuilder; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.springframework.beans.factory.annotation.Autowired; |
|||
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.event.EventListener; |
|||
import org.springframework.scheduling.annotation.Scheduled; |
|||
import org.springframework.stereotype.Service; |
|||
import org.thingsboard.common.util.ThingsBoardThreadFactory; |
|||
import org.thingsboard.server.TbQueueCallback; |
|||
import org.thingsboard.server.TbQueueConsumer; |
|||
import org.thingsboard.server.TbQueueMsgMetadata; |
|||
import org.thingsboard.server.TbQueueProducer; |
|||
import org.thingsboard.server.actors.ActorSystemContext; |
|||
import org.thingsboard.server.common.TbProtoQueueMsg; |
|||
import org.thingsboard.server.common.msg.cluster.ServerAddress; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.DeviceActorToTransportMsg; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.ToTransportMsg; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.TransportToDeviceActorMsg; |
|||
import org.thingsboard.server.provider.TbCoreQueueProvider; |
|||
import org.thingsboard.server.service.cluster.routing.ClusterRoutingService; |
|||
import org.thingsboard.server.service.encoding.DataDecodingEncodingService; |
|||
import org.thingsboard.server.service.queue.TbCoreConsumerStats; |
|||
import org.thingsboard.server.service.transport.msg.TransportToDeviceActorMsgWrapper; |
|||
|
|||
import javax.annotation.PostConstruct; |
|||
import javax.annotation.PreDestroy; |
|||
import java.time.Duration; |
|||
import java.util.List; |
|||
import java.util.Optional; |
|||
import java.util.UUID; |
|||
import java.util.concurrent.ExecutorService; |
|||
import java.util.concurrent.Executors; |
|||
import java.util.concurrent.TimeUnit; |
|||
import java.util.function.Consumer; |
|||
|
|||
/** |
|||
* Created by ashvayka on 09.10.18. |
|||
*/ |
|||
@Slf4j |
|||
@Service |
|||
@ConditionalOnProperty(prefix = "transport", value = "type", havingValue = "remote") |
|||
public class RemoteRuleEngineTransportService { |
|||
|
|||
@Value("${transport.remote.rule_engine.topic}") |
|||
private String ruleEngineTopic; |
|||
@Value("${transport.remote.notifications.topic}") |
|||
private String notificationsTopic; |
|||
@Value("${transport.remote.rule_engine.poll_interval}") |
|||
private int pollDuration; |
|||
@Value("${transport.remote.rule_engine.auto_commit_interval}") |
|||
private int autoCommitInterval; |
|||
|
|||
@Value("${transport.remote.rule_engine.poll_records_pack_size}") |
|||
private int pollRecordsPackSize; |
|||
@Value("${transport.remote.rule_engine.max_poll_records_per_second}") |
|||
private long pollRecordsPerSecond; |
|||
@Value("${transport.remote.rule_engine.max_poll_records_per_minute}") |
|||
private long pollRecordsPerMinute; |
|||
@Value("${transport.remote.rule_engine.stats.enabled:false}") |
|||
private boolean statsEnabled; |
|||
|
|||
@Autowired |
|||
private ActorSystemContext actorContext; |
|||
|
|||
//TODO: completely replace this routing with the Kafka routing by partition ids.
|
|||
@Autowired |
|||
private ClusterRoutingService routingService; |
|||
@Autowired |
|||
private ClusterRpcService rpcService; |
|||
@Autowired |
|||
private DataDecodingEncodingService encodingService; |
|||
|
|||
@Autowired |
|||
private TbCoreQueueProvider coreQueueProvider; |
|||
|
|||
private TbQueueConsumer<TbProtoQueueMsg<ToRuleEngineMsg>> ruleEngineConsumer; |
|||
|
|||
private TbQueueProducer<TbProtoQueueMsg<ToTransportMsg>> notificationsProducer; |
|||
|
|||
private ExecutorService mainConsumerExecutor = Executors.newSingleThreadExecutor(ThingsBoardThreadFactory.forName("tb-main-consumer")); |
|||
|
|||
private volatile boolean stopped = false; |
|||
|
|||
private final TbCoreConsumerStats stats = new TbCoreConsumerStats(); |
|||
|
|||
@PostConstruct |
|||
public void init() { |
|||
notificationsProducer = coreQueueProvider.getTransportMsgProducer(); |
|||
notificationsProducer.init(); |
|||
|
|||
//TODO: 2.5
|
|||
// ruleEngineConsumer =
|
|||
// ruleEngineConsumer.subscribe();
|
|||
} |
|||
|
|||
@EventListener(ApplicationReadyEvent.class) |
|||
public void onApplicationEvent(ApplicationReadyEvent applicationReadyEvent) { |
|||
log.info("Received application ready event. Starting polling for events."); |
|||
LocalBucketBuilder builder = Bucket4j.builder(); |
|||
builder.addLimit(Bandwidth.simple(pollRecordsPerSecond, Duration.ofSeconds(1))); |
|||
builder.addLimit(Bandwidth.simple(pollRecordsPerMinute, Duration.ofMinutes(1))); |
|||
LocalBucket pollRateBucket = builder.build(); |
|||
BlockingBucket blockingPollRateBucket = pollRateBucket.asScheduler(); |
|||
|
|||
mainConsumerExecutor.execute(() -> { |
|||
while (!stopped) { |
|||
try { |
|||
List<TbProtoQueueMsg<ToRuleEngineMsg>> msgs = ruleEngineConsumer.poll(pollDuration); |
|||
int recordsCount = msgs.size(); |
|||
if (recordsCount > 0) { |
|||
while (!blockingPollRateBucket.tryConsume(recordsCount, TimeUnit.SECONDS.toNanos(5))) { |
|||
log.info("Rule Engine consumer is busy. Required tokens: [{}]. Available tokens: [{}].", recordsCount, pollRateBucket.getAvailableTokens()); |
|||
Thread.sleep(TimeUnit.SECONDS.toMillis(1)); |
|||
} |
|||
log.trace("Processing {} records", recordsCount); |
|||
} |
|||
msgs.forEach(msg -> { |
|||
try { |
|||
ToRuleEngineMsg toRuleEngineMsg = msg.getValue(); |
|||
log.trace("Forwarding message to rule engine {}", toRuleEngineMsg); |
|||
if (toRuleEngineMsg.hasToDeviceActorMsg()) { |
|||
forwardToDeviceActor(toRuleEngineMsg.getToDeviceActorMsg()); |
|||
} |
|||
} catch (Throwable e) { |
|||
log.warn("Failed to process the notification.", e); |
|||
} |
|||
}); |
|||
} catch (Exception e) { |
|||
log.warn("Failed to obtain messages from queue.", e); |
|||
try { |
|||
Thread.sleep(pollDuration); |
|||
} catch (InterruptedException e2) { |
|||
log.trace("Failed to wait until the server has capacity to handle new requests", e2); |
|||
} |
|||
} |
|||
} |
|||
}); |
|||
} |
|||
|
|||
@Scheduled(fixedDelayString = "${transport.remote.rule_engine.stats.print_interval_ms}") |
|||
public void printStats() { |
|||
if (statsEnabled) { |
|||
stats.printStats(); |
|||
} |
|||
} |
|||
|
|||
@Override |
|||
public void process(String nodeId, DeviceActorToTransportMsg msg) { |
|||
process(nodeId, msg, null, null); |
|||
} |
|||
|
|||
@Override |
|||
public void process(String nodeId, DeviceActorToTransportMsg msg, Runnable onSuccess, Consumer<Throwable> onFailure) { |
|||
String topic = notificationsTopic + "." + nodeId; |
|||
UUID sessionId = new UUID(msg.getSessionIdMSB(), msg.getSessionIdLSB()); |
|||
ToTransportMsg transportMsg = ToTransportMsg.newBuilder().setToDeviceSessionMsg(msg).build(); |
|||
log.trace("[{}][{}] Pushing session data to topic: {}", topic, sessionId, transportMsg); |
|||
//TODO: 2.5 id
|
|||
TbProtoQueueMsg<ToTransportMsg> queueMsg = new TbProtoQueueMsg<>(sessionId, transportMsg); |
|||
notificationsProducer.send(topic, queueMsg, new QueueCallbackAdaptor(onSuccess, onFailure)); |
|||
} |
|||
|
|||
private void forwardToDeviceActor(TransportToDeviceActorMsg toDeviceActorMsg) { |
|||
if (statsEnabled) { |
|||
stats.log(toDeviceActorMsg); |
|||
} |
|||
TransportToDeviceActorMsgWrapper wrapper = new TransportToDeviceActorMsgWrapper(toDeviceActorMsg); |
|||
Optional<ServerAddress> address = routingService.resolveById(wrapper.getDeviceId()); |
|||
if (address.isPresent()) { |
|||
log.trace("[{}] Pushing message to remote server: {}", address.get(), toDeviceActorMsg); |
|||
rpcService.tell(encodingService.convertToProtoDataMessage(address.get(), wrapper)); |
|||
} else { |
|||
log.trace("Pushing message to local server: {}", toDeviceActorMsg); |
|||
actorContext.getAppActor().tell(wrapper, ActorRef.noSender()); |
|||
} |
|||
} |
|||
|
|||
@PreDestroy |
|||
public void destroy() { |
|||
stopped = true; |
|||
if (ruleEngineConsumer != null) { |
|||
ruleEngineConsumer.unsubscribe(); |
|||
} |
|||
if (mainConsumerExecutor != null) { |
|||
mainConsumerExecutor.shutdownNow(); |
|||
} |
|||
} |
|||
|
|||
private static class QueueCallbackAdaptor implements TbQueueCallback { |
|||
private final Runnable onSuccess; |
|||
private final Consumer<Throwable> onFailure; |
|||
|
|||
QueueCallbackAdaptor(Runnable onSuccess, Consumer<Throwable> onFailure) { |
|||
this.onSuccess = onSuccess; |
|||
this.onFailure = onFailure; |
|||
} |
|||
|
|||
@Override |
|||
public void onSuccess(TbQueueMsgMetadata metadata) { |
|||
if (onSuccess != null) { |
|||
onSuccess.run(); |
|||
} |
|||
} |
|||
|
|||
@Override |
|||
public void onFailure(Throwable t) { |
|||
if (onFailure != null) { |
|||
onFailure.accept(t); |
|||
} |
|||
} |
|||
} |
|||
} |
|||
@ -1,31 +0,0 @@ |
|||
/** |
|||
* Copyright © 2016-2020 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.transport; |
|||
|
|||
import org.thingsboard.server.gen.transport.TransportProtos.TransportApiRequestMsg; |
|||
import org.thingsboard.server.kafka.TbKafkaDecoder; |
|||
|
|||
import java.io.IOException; |
|||
|
|||
/** |
|||
* Created by ashvayka on 05.10.18. |
|||
*/ |
|||
public class TransportApiRequestDecoder implements TbKafkaDecoder<TransportApiRequestMsg> { |
|||
@Override |
|||
public TransportApiRequestMsg decode(byte[] data) throws IOException { |
|||
return TransportApiRequestMsg.parseFrom(data); |
|||
} |
|||
} |
|||
@ -1,30 +0,0 @@ |
|||
/** |
|||
* Copyright © 2016-2020 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.transport; |
|||
|
|||
import org.thingsboard.server.kafka.TbKafkaEncoder; |
|||
|
|||
import org.thingsboard.server.gen.transport.TransportProtos.TransportApiResponseMsg; |
|||
|
|||
/** |
|||
* Created by ashvayka on 05.10.18. |
|||
*/ |
|||
public class TransportApiResponseEncoder implements TbKafkaEncoder<TransportApiResponseMsg> { |
|||
@Override |
|||
public byte[] encode(TransportApiResponseMsg value) { |
|||
return value.toByteArray(); |
|||
} |
|||
} |
|||
@ -0,0 +1,221 @@ |
|||
package org.thingsboard.server.discovery; |
|||
|
|||
import com.google.common.hash.HashCode; |
|||
import com.google.common.hash.HashFunction; |
|||
import com.google.common.hash.Hashing; |
|||
import lombok.Getter; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.springframework.beans.factory.annotation.Autowired; |
|||
import org.springframework.beans.factory.annotation.Value; |
|||
import org.springframework.stereotype.Service; |
|||
import org.thingsboard.server.common.data.id.EntityId; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
import org.thingsboard.server.gen.transport.TransportProtos; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.ServiceInfo; |
|||
|
|||
import javax.annotation.PostConstruct; |
|||
import java.nio.charset.StandardCharsets; |
|||
import java.util.ArrayList; |
|||
import java.util.HashMap; |
|||
import java.util.HashSet; |
|||
import java.util.List; |
|||
import java.util.Map; |
|||
import java.util.Objects; |
|||
import java.util.Set; |
|||
import java.util.UUID; |
|||
import java.util.concurrent.ConcurrentHashMap; |
|||
import java.util.concurrent.ConcurrentMap; |
|||
import java.util.concurrent.ConcurrentNavigableMap; |
|||
|
|||
@Service |
|||
@Slf4j |
|||
public class ConsistentHashPartitionService implements PartitionService { |
|||
|
|||
@Value("${queue.core.topic}") |
|||
private String coreTopic; |
|||
@Value("${queue.core.partitions:100}") |
|||
private Integer corePartitions; |
|||
@Value("${queue.rule_engine.topic}") |
|||
private String ruleEngineTopic; |
|||
@Value("${queue.rule_engine.partitions:100}") |
|||
private Integer ruleEnginePartitions; |
|||
@Value("${queue.partitions.hash_function_name:murmur3_32}") |
|||
private String hashFunctionName; |
|||
@Value("${queue.partitions.virtual_nodes_size:16}") |
|||
private Integer virtualNodesSize; |
|||
|
|||
private final TbServiceInfoProvider serviceInfoProvider; |
|||
private final ConcurrentMap<ServiceType, String> partitionTopics = new ConcurrentHashMap<>(); |
|||
private final ConcurrentMap<ServiceType, Integer> partitionSizes = new ConcurrentHashMap<>(); |
|||
private ConcurrentMap<ServiceKey, List<Integer>> myPartitions = new ConcurrentHashMap<>(); |
|||
//TODO: Fetch this from the database, together with size of partitions for each service for each tenant.
|
|||
private ConcurrentMap<TenantId, Set<ServiceType>> isolatedTenants = new ConcurrentHashMap<>(); |
|||
|
|||
private HashFunction hashFunction; |
|||
|
|||
public ConsistentHashPartitionService(TbServiceInfoProvider serviceInfoProvider) { |
|||
this.serviceInfoProvider = serviceInfoProvider; |
|||
} |
|||
|
|||
@PostConstruct |
|||
public void init() { |
|||
this.hashFunction = forName(hashFunctionName); |
|||
partitionSizes.put(ServiceType.TB_CORE, corePartitions); |
|||
partitionSizes.put(ServiceType.TB_RULE_ENGINE, ruleEnginePartitions); |
|||
partitionTopics.put(ServiceType.TB_CORE, coreTopic); |
|||
partitionTopics.put(ServiceType.TB_RULE_ENGINE, ruleEngineTopic); |
|||
} |
|||
|
|||
@Override |
|||
public List<TopicPartitionInfo> getCurrentPartitions(ServiceType serviceType) { |
|||
ServiceInfo currentService = serviceInfoProvider.getServiceInfo(); |
|||
TenantId tenantId = getTenantId(currentService); |
|||
ServiceKey serviceKey = new ServiceKey(serviceType, tenantId); |
|||
List<Integer> partitions = myPartitions.get(serviceKey); |
|||
List<TopicPartitionInfo> topicPartitions = new ArrayList<>(); |
|||
for (Integer partition : partitions) { |
|||
TopicPartitionInfo.TopicPartitionInfoBuilder tpi = TopicPartitionInfo.builder(); |
|||
tpi.topic(partitionTopics.get(serviceType)); |
|||
tpi.partition(partition); |
|||
if (!tenantId.isNullUid()) { |
|||
tpi.tenantId(tenantId); |
|||
} |
|||
topicPartitions.add(tpi.build()); |
|||
} |
|||
return topicPartitions; |
|||
} |
|||
|
|||
@Override |
|||
public TopicPartitionInfo resolve(ServiceType serviceType, TenantId tenantId, EntityId entityId) { |
|||
boolean isolated = isolatedTenants.get(tenantId) != null && isolatedTenants.get(tenantId).contains(serviceType); |
|||
int hash = hashFunction.newHasher() |
|||
.putLong(entityId.getId().getMostSignificantBits()) |
|||
.putLong(entityId.getId().getLeastSignificantBits()).hash().asInt(); |
|||
int partition = Math.abs(hash % partitionSizes.get(serviceType)); |
|||
TopicPartitionInfo.TopicPartitionInfoBuilder tpi = TopicPartitionInfo.builder(); |
|||
tpi.topic(partitionTopics.get(serviceType)); |
|||
tpi.partition(partition); |
|||
if (isolated) { |
|||
tpi.tenantId(tenantId); |
|||
} |
|||
return tpi.build(); |
|||
} |
|||
|
|||
@Override |
|||
public void recalculatePartitions(ServiceInfo currentService, List<ServiceInfo> otherServices) { |
|||
logServiceInfo(currentService); |
|||
otherServices.forEach(this::logServiceInfo); |
|||
|
|||
Map<ServiceType, ConsistentHashCircle<ServiceInfo>> newCircles = new HashMap<>(ServiceType.values().length); |
|||
for (ServiceType serverType : ServiceType.values()) { |
|||
newCircles.put(serverType, new ConsistentHashCircle<>()); |
|||
} |
|||
addNode(newCircles, currentService); |
|||
for (ServiceInfo other : otherServices) { |
|||
addNode(newCircles, other); |
|||
TenantId tenantId = getTenantId(other); |
|||
if (!tenantId.isNullUid()) { |
|||
isolatedTenants.putIfAbsent(tenantId, new HashSet<>()); |
|||
for (String serviceType : other.getServiceTypesList()) { |
|||
isolatedTenants.get(tenantId).add(ServiceType.valueOf(serviceType.toUpperCase())); |
|||
} |
|||
|
|||
} |
|||
} |
|||
ConcurrentMap<ServiceKey, List<Integer>> oldPartitions = myPartitions; |
|||
myPartitions = new ConcurrentHashMap<>(); |
|||
partitionSizes.forEach((type, size) -> { |
|||
for (int i = 0; i < size; i++) { |
|||
ServiceInfo serviceInfo = resolveByPartitionIdx(newCircles.get(type), i); |
|||
if (currentService.equals(serviceInfo)) { |
|||
myPartitions.putIfAbsent(new ServiceKey(type, getTenantId(serviceInfo)), new ArrayList<>()); |
|||
} |
|||
} |
|||
}); |
|||
myPartitions.forEach((serviceKey, partitions) -> { |
|||
if (!partitions.equals(oldPartitions.get(serviceKey))) { |
|||
log.info("[{}] NEW PARTITIONS: {}", serviceKey, partitions); |
|||
} |
|||
}); |
|||
} |
|||
|
|||
private void logServiceInfo(TransportProtos.ServiceInfo server) { |
|||
TenantId tenantId = getTenantId(server); |
|||
if (tenantId.isNullUid()) { |
|||
log.info("[{}] Found common server: [{}]", server.getServiceId(), server.getServiceTypesList()); |
|||
} else { |
|||
log.info("[{}][{}] Found specific server: [{}]", server.getServiceId(), tenantId, server.getServiceTypesList()); |
|||
} |
|||
} |
|||
|
|||
private TenantId getTenantId(TransportProtos.ServiceInfo serviceInfo) { |
|||
return new TenantId(new UUID(serviceInfo.getTenantIdMSB(), serviceInfo.getTenantIdLSB())); |
|||
} |
|||
|
|||
private void addNode(Map<ServiceType, ConsistentHashCircle<ServiceInfo>> circles, ServiceInfo instance) { |
|||
for (String serviceTypeStr : instance.getServiceTypesList()) { |
|||
ServiceType serviceType = ServiceType.valueOf(serviceTypeStr.toUpperCase()); |
|||
for (int i = 0; i < virtualNodesSize; i++) { |
|||
circles.get(serviceType).put(hash(instance, i).asLong(), instance); |
|||
} |
|||
} |
|||
} |
|||
|
|||
private ServiceInfo resolveByPartitionIdx(ConsistentHashCircle<ServiceInfo> circle, Integer partitionIdx) { |
|||
if (circle.isEmpty()) { |
|||
return null; |
|||
} |
|||
Long hash = hashFunction.newHasher().putInt(partitionIdx).hash().asLong(); |
|||
if (!circle.containsKey(hash)) { |
|||
ConcurrentNavigableMap<Long, ServiceInfo> tailMap = circle.tailMap(hash); |
|||
hash = tailMap.isEmpty() ? |
|||
circle.firstKey() : tailMap.firstKey(); |
|||
} |
|||
return circle.get(hash); |
|||
} |
|||
|
|||
private HashCode hash(ServiceInfo instance, int i) { |
|||
return hashFunction.newHasher().putString(instance.getServiceId(), StandardCharsets.UTF_8).putInt(i).hash(); |
|||
} |
|||
|
|||
private static class ServiceKey { |
|||
@Getter |
|||
private final ServiceType serviceType; |
|||
@Getter |
|||
private final TenantId tenantId; |
|||
|
|||
public ServiceKey(ServiceType serviceType, TenantId tenantId) { |
|||
this.serviceType = serviceType; |
|||
this.tenantId = tenantId; |
|||
} |
|||
|
|||
@Override |
|||
public boolean equals(Object o) { |
|||
if (this == o) return true; |
|||
if (o == null || getClass() != o.getClass()) return false; |
|||
ServiceKey that = (ServiceKey) o; |
|||
return serviceType == that.serviceType && |
|||
Objects.equals(tenantId, that.tenantId); |
|||
} |
|||
|
|||
@Override |
|||
public int hashCode() { |
|||
return Objects.hash(serviceType, tenantId); |
|||
} |
|||
} |
|||
|
|||
public static HashFunction forName(String name) { |
|||
switch (name) { |
|||
case "murmur3_32": |
|||
return Hashing.murmur3_32(); |
|||
case "murmur3_128": |
|||
return Hashing.murmur3_128(); |
|||
case "crc32": |
|||
return Hashing.crc32(); |
|||
case "md5": |
|||
return Hashing.md5(); |
|||
default: |
|||
throw new IllegalArgumentException("Can't find hash function with name " + name); |
|||
} |
|||
} |
|||
} |
|||
@ -0,0 +1,32 @@ |
|||
package org.thingsboard.server.discovery; |
|||
|
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; |
|||
import org.springframework.boot.context.event.ApplicationReadyEvent; |
|||
import org.springframework.context.annotation.DependsOn; |
|||
import org.springframework.context.event.EventListener; |
|||
import org.springframework.stereotype.Service; |
|||
|
|||
import java.util.Collections; |
|||
|
|||
@Service |
|||
@ConditionalOnProperty(prefix = "zk", value = "enabled", havingValue = "false", matchIfMissing = true) |
|||
@Slf4j |
|||
@DependsOn("environmentLogService") |
|||
public class DummyDiscoveryService implements PartitionDiscoveryService { |
|||
|
|||
private final TbServiceInfoProvider serviceInfoProvider; |
|||
private final PartitionService partitionService; |
|||
|
|||
|
|||
public DummyDiscoveryService(TbServiceInfoProvider serviceInfoProvider, PartitionService partitionService) { |
|||
this.serviceInfoProvider = serviceInfoProvider; |
|||
this.partitionService = partitionService; |
|||
} |
|||
|
|||
@EventListener(ApplicationReadyEvent.class) |
|||
public void onApplicationEvent(ApplicationReadyEvent event) { |
|||
partitionService.recalculatePartitions(serviceInfoProvider.getServiceInfo(), Collections.emptyList()); |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,16 @@ |
|||
package org.thingsboard.server.discovery; |
|||
|
|||
import org.thingsboard.server.common.data.id.EntityId; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
import org.thingsboard.server.gen.transport.TransportProtos; |
|||
|
|||
import java.util.List; |
|||
|
|||
public interface PartitionService { |
|||
|
|||
List<TopicPartitionInfo> getCurrentPartitions(ServiceType serviceType); |
|||
|
|||
TopicPartitionInfo resolve(ServiceType serviceType, TenantId tenantId, EntityId entityId); |
|||
|
|||
void recalculatePartitions(TransportProtos.ServiceInfo currentService, List<TransportProtos.ServiceInfo> otherServices); |
|||
} |
|||
@ -0,0 +1,119 @@ |
|||
/** |
|||
* Copyright © 2016-2020 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.provider; |
|||
|
|||
import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression; |
|||
import org.springframework.stereotype.Component; |
|||
import org.thingsboard.server.TbQueueConsumer; |
|||
import org.thingsboard.server.TbQueueCoreSettings; |
|||
import org.thingsboard.server.TbQueueProducer; |
|||
import org.thingsboard.server.common.TbProtoQueueMsg; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.ToTransportMsg; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.TransportApiRequestMsg; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.TransportApiResponseMsg; |
|||
import org.thingsboard.server.kafka.TBKafkaConsumerTemplate; |
|||
import org.thingsboard.server.kafka.TBKafkaProducerTemplate; |
|||
import org.thingsboard.server.kafka.TbKafkaSettings; |
|||
import org.thingsboard.server.kafka.TbNodeIdProvider; |
|||
|
|||
@Component |
|||
@ConditionalOnExpression("'${queue.type:null}'=='kafka' && '${service.type:null}'=='monolith'") |
|||
public class KafkaMonolithQueueProvider implements TbCoreQueueProvider, TbRuleEngineQueueProvider { |
|||
|
|||
private final TbKafkaSettings kafkaSettings; |
|||
private final TbNodeIdProvider nodeIdProvider; |
|||
private final TbQueueCoreSettings coreSettings; |
|||
|
|||
public KafkaMonolithQueueProvider(TbKafkaSettings kafkaSettings, TbNodeIdProvider nodeIdProvider, TbQueueCoreSettings coreSettings) { |
|||
this.kafkaSettings = kafkaSettings; |
|||
this.nodeIdProvider = nodeIdProvider; |
|||
this.coreSettings = coreSettings; |
|||
} |
|||
|
|||
@Override |
|||
public TbQueueProducer<TbProtoQueueMsg<ToTransportMsg>> getTransportMsgProducer() { |
|||
TBKafkaProducerTemplate.TBKafkaProducerTemplateBuilder<TbProtoQueueMsg<ToTransportMsg>> requestBuilder = TBKafkaProducerTemplate.builder(); |
|||
requestBuilder.settings(kafkaSettings); |
|||
requestBuilder.clientId("producer-transport-" + nodeIdProvider.getNodeId()); |
|||
requestBuilder.defaultTopic(coreSettings.getTopic()); |
|||
return requestBuilder.build(); |
|||
} |
|||
|
|||
@Override |
|||
public TbQueueProducer<TbProtoQueueMsg<ToRuleEngineMsg>> getRuleEngineMsgProducer() { |
|||
TBKafkaProducerTemplate.TBKafkaProducerTemplateBuilder<TbProtoQueueMsg<ToRuleEngineMsg>> requestBuilder = TBKafkaProducerTemplate.builder(); |
|||
requestBuilder.settings(kafkaSettings); |
|||
requestBuilder.clientId("producer-rule-engine-" + nodeIdProvider.getNodeId()); |
|||
requestBuilder.defaultTopic(coreSettings.getTopic()); |
|||
return requestBuilder.build(); |
|||
} |
|||
|
|||
@Override |
|||
public TbQueueProducer<TbProtoQueueMsg<ToCoreMsg>> getTbCoreMsgProducer() { |
|||
TBKafkaProducerTemplate.TBKafkaProducerTemplateBuilder<TbProtoQueueMsg<ToCoreMsg>> requestBuilder = TBKafkaProducerTemplate.builder(); |
|||
requestBuilder.settings(kafkaSettings); |
|||
requestBuilder.clientId("producer-core-" + nodeIdProvider.getNodeId()); |
|||
requestBuilder.defaultTopic(coreSettings.getTopic()); |
|||
return requestBuilder.build(); |
|||
} |
|||
|
|||
@Override |
|||
public TbQueueConsumer<TbProtoQueueMsg<ToRuleEngineMsg>> getToRuleEngineMsgConsumer() { |
|||
TBKafkaConsumerTemplate.TBKafkaConsumerTemplateBuilder<TbProtoQueueMsg<ToRuleEngineMsg>> responseBuilder = TBKafkaConsumerTemplate.builder(); |
|||
responseBuilder.settings(kafkaSettings); |
|||
responseBuilder.topic(coreSettings.getTopic()); |
|||
responseBuilder.clientId("tb-rule-engine-consumer-" + nodeIdProvider.getNodeId()); |
|||
responseBuilder.groupId("tb-rule-engine-" + nodeIdProvider.getNodeId()); |
|||
responseBuilder.autoCommit(true); |
|||
responseBuilder.decoder(msg -> new TbProtoQueueMsg<>(msg.getKey(), ToRuleEngineMsg.parseFrom(msg.getData()), msg.getHeaders())); |
|||
return responseBuilder.build(); |
|||
} |
|||
|
|||
@Override |
|||
public TbQueueConsumer<TbProtoQueueMsg<ToCoreMsg>> getToCoreMsgConsumer() { |
|||
TBKafkaConsumerTemplate.TBKafkaConsumerTemplateBuilder<TbProtoQueueMsg<ToCoreMsg>> consumerBuilder = TBKafkaConsumerTemplate.builder(); |
|||
consumerBuilder.settings(kafkaSettings); |
|||
consumerBuilder.topic(coreSettings.getTopic()); |
|||
consumerBuilder.clientId("tb-core-consumer" + nodeIdProvider.getNodeId()); |
|||
consumerBuilder.groupId("tb-core-node-" + nodeIdProvider.getNodeId()); |
|||
consumerBuilder.decoder(msg -> new TbProtoQueueMsg<>(msg.getKey(), ToCoreMsg.parseFrom(msg.getData()), msg.getHeaders())); |
|||
return consumerBuilder.build(); |
|||
} |
|||
|
|||
@Override |
|||
public TbQueueConsumer<TbProtoQueueMsg<TransportApiRequestMsg>> getTransportApiRequestConsumer() { |
|||
TBKafkaConsumerTemplate.TBKafkaConsumerTemplateBuilder<TbProtoQueueMsg<TransportApiRequestMsg>> responseBuilder = TBKafkaConsumerTemplate.builder(); |
|||
responseBuilder.settings(kafkaSettings); |
|||
responseBuilder.topic(coreSettings.getTopic()); |
|||
responseBuilder.clientId("consumer-transport-" + nodeIdProvider.getNodeId()); |
|||
responseBuilder.groupId("rule-engine-node-" + nodeIdProvider.getNodeId()); |
|||
responseBuilder.autoCommit(true); |
|||
//TODO 2.5
|
|||
responseBuilder.decoder(msg -> new TbProtoQueueMsg<>(msg.getKey(), TransportApiRequestMsg.parseFrom(msg.getData()), msg.getHeaders())); |
|||
return responseBuilder.build(); |
|||
} |
|||
|
|||
@Override |
|||
public TbQueueProducer<TbProtoQueueMsg<TransportApiResponseMsg>> getTransportApiResponseProducer() { |
|||
TBKafkaProducerTemplate.TBKafkaProducerTemplateBuilder<TbProtoQueueMsg<TransportApiResponseMsg>> requestBuilder = TBKafkaProducerTemplate.builder(); |
|||
requestBuilder.settings(kafkaSettings); |
|||
requestBuilder.clientId("transport-api-" + nodeIdProvider.getNodeId()); |
|||
requestBuilder.defaultTopic(coreSettings.getTopic()); |
|||
return requestBuilder.build(); |
|||
} |
|||
} |
|||
@ -0,0 +1,86 @@ |
|||
/** |
|||
* Copyright © 2016-2020 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.provider; |
|||
|
|||
import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression; |
|||
import org.springframework.stereotype.Component; |
|||
import org.thingsboard.server.TbQueueConsumer; |
|||
import org.thingsboard.server.TbQueueCoreSettings; |
|||
import org.thingsboard.server.TbQueueProducer; |
|||
import org.thingsboard.server.common.TbProtoQueueMsg; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.ToTransportMsg; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.TransportApiRequestMsg; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.TransportApiResponseMsg; |
|||
import org.thingsboard.server.kafka.TBKafkaConsumerTemplate; |
|||
import org.thingsboard.server.kafka.TBKafkaProducerTemplate; |
|||
import org.thingsboard.server.kafka.TbKafkaSettings; |
|||
import org.thingsboard.server.kafka.TbNodeIdProvider; |
|||
|
|||
@Component |
|||
@ConditionalOnExpression("'${queue.type:null}'=='kafka' && '${service.type:null}'=='tb-rule-engine'") |
|||
public class KafkaTbRuleEngineQueueProvider implements TbRuleEngineQueueProvider { |
|||
|
|||
private final TbKafkaSettings kafkaSettings; |
|||
private final TbNodeIdProvider nodeIdProvider; |
|||
private final TbQueueCoreSettings coreSettings; |
|||
|
|||
public KafkaTbRuleEngineQueueProvider(TbKafkaSettings kafkaSettings, TbNodeIdProvider nodeIdProvider, TbQueueCoreSettings coreSettings) { |
|||
this.kafkaSettings = kafkaSettings; |
|||
this.nodeIdProvider = nodeIdProvider; |
|||
this.coreSettings = coreSettings; |
|||
} |
|||
|
|||
@Override |
|||
public TbQueueProducer<TbProtoQueueMsg<ToTransportMsg>> getTransportMsgProducer() { |
|||
TBKafkaProducerTemplate.TBKafkaProducerTemplateBuilder<TbProtoQueueMsg<ToTransportMsg>> requestBuilder = TBKafkaProducerTemplate.builder(); |
|||
requestBuilder.settings(kafkaSettings); |
|||
requestBuilder.clientId("producer-transport-" + nodeIdProvider.getNodeId()); |
|||
requestBuilder.defaultTopic(coreSettings.getTopic()); |
|||
return requestBuilder.build(); |
|||
} |
|||
|
|||
@Override |
|||
public TbQueueProducer<TbProtoQueueMsg<ToRuleEngineMsg>> getRuleEngineMsgProducer() { |
|||
TBKafkaProducerTemplate.TBKafkaProducerTemplateBuilder<TbProtoQueueMsg<ToRuleEngineMsg>> requestBuilder = TBKafkaProducerTemplate.builder(); |
|||
requestBuilder.settings(kafkaSettings); |
|||
requestBuilder.clientId("producer-rule-engine-" + nodeIdProvider.getNodeId()); |
|||
requestBuilder.defaultTopic(coreSettings.getTopic()); |
|||
return requestBuilder.build(); |
|||
} |
|||
|
|||
@Override |
|||
public TbQueueProducer<TbProtoQueueMsg<ToCoreMsg>> getTbCoreMsgProducer() { |
|||
TBKafkaProducerTemplate.TBKafkaProducerTemplateBuilder<TbProtoQueueMsg<ToCoreMsg>> requestBuilder = TBKafkaProducerTemplate.builder(); |
|||
requestBuilder.settings(kafkaSettings); |
|||
requestBuilder.clientId("producer-core-" + nodeIdProvider.getNodeId()); |
|||
requestBuilder.defaultTopic(coreSettings.getTopic()); |
|||
return requestBuilder.build(); |
|||
} |
|||
|
|||
@Override |
|||
public TbQueueConsumer<TbProtoQueueMsg<ToRuleEngineMsg>> getToRuleEngineMsgConsumer() { |
|||
TBKafkaConsumerTemplate.TBKafkaConsumerTemplateBuilder<TbProtoQueueMsg<ToRuleEngineMsg>> responseBuilder = TBKafkaConsumerTemplate.builder(); |
|||
responseBuilder.settings(kafkaSettings); |
|||
responseBuilder.topic(coreSettings.getTopic()); |
|||
responseBuilder.clientId("tb-rule-engine-consumer-" + nodeIdProvider.getNodeId()); |
|||
responseBuilder.groupId("tb-rule-engine-" + nodeIdProvider.getNodeId()); |
|||
responseBuilder.autoCommit(true); |
|||
responseBuilder.decoder(msg -> new TbProtoQueueMsg<>(msg.getKey(), ToRuleEngineMsg.parseFrom(msg.getData()), msg.getHeaders())); |
|||
return responseBuilder.build(); |
|||
} |
|||
} |
|||
@ -0,0 +1,61 @@ |
|||
/** |
|||
* Copyright © 2016-2020 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.provider; |
|||
|
|||
import org.thingsboard.server.TbQueueConsumer; |
|||
import org.thingsboard.server.TbQueueProducer; |
|||
import org.thingsboard.server.common.TbProtoQueueMsg; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.ToTransportMsg; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.TransportApiRequestMsg; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.TransportApiResponseMsg; |
|||
|
|||
/** |
|||
* Responsible for initialization of various Producers and Consumers used by TB Core Node. |
|||
* Implementation Depends on the queue queue.type from yml or TB_QUEUE_TYPE environment variable |
|||
*/ |
|||
public interface TbRuleEngineQueueProvider { |
|||
|
|||
/** |
|||
* Used to push messages to instances of TB Transport Service |
|||
* |
|||
* @return |
|||
*/ |
|||
TbQueueProducer<TbProtoQueueMsg<ToTransportMsg>> getTransportMsgProducer(); |
|||
|
|||
/** |
|||
* Used to push messages to instances of TB RuleEngine Service |
|||
* |
|||
* @return |
|||
*/ |
|||
TbQueueProducer<TbProtoQueueMsg<ToRuleEngineMsg>> getRuleEngineMsgProducer(); |
|||
|
|||
/** |
|||
* Used to push messages to other instances of TB Core Service |
|||
* |
|||
* @return |
|||
*/ |
|||
TbQueueProducer<TbProtoQueueMsg<ToCoreMsg>> getTbCoreMsgProducer(); |
|||
|
|||
/** |
|||
* Used to consume messages by TB Core Service |
|||
* |
|||
* @return |
|||
*/ |
|||
TbQueueConsumer<TbProtoQueueMsg<ToRuleEngineMsg>> getToRuleEngineMsgConsumer(); |
|||
|
|||
} |
|||
Loading…
Reference in new issue