Viacheslav Kukhtyn
8 years ago
314 changed files with 10499 additions and 7367 deletions
@ -1,40 +0,0 @@ |
|||
/** |
|||
* Copyright © 2016-2018 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.actors.device; |
|||
|
|||
import lombok.AllArgsConstructor; |
|||
import lombok.Data; |
|||
import org.thingsboard.server.common.data.id.SessionId; |
|||
import org.thingsboard.server.common.msg.cluster.ServerAddress; |
|||
import org.thingsboard.server.common.msg.session.SessionMsgType; |
|||
|
|||
import java.util.Optional; |
|||
|
|||
/** |
|||
* Created by ashvayka on 17.04.18. |
|||
*/ |
|||
@Data |
|||
@AllArgsConstructor |
|||
public final class PendingSessionMsgData { |
|||
|
|||
private final SessionId sessionId; |
|||
private final Optional<ServerAddress> serverAddress; |
|||
private final SessionMsgType sessionMsgType; |
|||
private final int requestId; |
|||
private final boolean replyOnQueueAck; |
|||
private int ackMsgCount; |
|||
|
|||
} |
|||
@ -1,156 +0,0 @@ |
|||
/** |
|||
* Copyright © 2016-2018 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.actors.session; |
|||
|
|||
import akka.actor.ActorContext; |
|||
import akka.event.LoggingAdapter; |
|||
import org.thingsboard.server.actors.ActorSystemContext; |
|||
import org.thingsboard.server.actors.shared.SessionTimeoutMsg; |
|||
import org.thingsboard.server.common.data.id.SessionId; |
|||
import org.thingsboard.server.common.msg.cluster.ClusterEventMsg; |
|||
import org.thingsboard.server.common.msg.cluster.ServerAddress; |
|||
import org.thingsboard.server.common.msg.core.AttributesSubscribeMsg; |
|||
import org.thingsboard.server.common.msg.core.ResponseMsg; |
|||
import org.thingsboard.server.common.msg.core.RpcSubscribeMsg; |
|||
import org.thingsboard.server.common.msg.core.SessionCloseMsg; |
|||
import org.thingsboard.server.common.msg.core.SessionOpenMsg; |
|||
import org.thingsboard.server.common.msg.device.DeviceToDeviceActorMsg; |
|||
import org.thingsboard.server.common.msg.session.BasicSessionActorToAdaptorMsg; |
|||
import org.thingsboard.server.common.msg.session.FromDeviceMsg; |
|||
import org.thingsboard.server.common.msg.session.FromDeviceRequestMsg; |
|||
import org.thingsboard.server.common.msg.session.SessionMsgType; |
|||
import org.thingsboard.server.common.msg.session.SessionType; |
|||
import org.thingsboard.server.common.msg.session.ToDeviceMsg; |
|||
import org.thingsboard.server.common.msg.session.TransportToDeviceSessionActorMsg; |
|||
import org.thingsboard.server.common.msg.session.ex.SessionException; |
|||
|
|||
import java.util.HashMap; |
|||
import java.util.Map; |
|||
import java.util.Optional; |
|||
|
|||
class ASyncMsgProcessor extends AbstractSessionActorMsgProcessor { |
|||
|
|||
private boolean firstMsg = true; |
|||
private Map<Integer, DeviceToDeviceActorMsg> pendingMap = new HashMap<>(); |
|||
private Optional<ServerAddress> currentTargetServer; |
|||
private boolean subscribedToAttributeUpdates; |
|||
private boolean subscribedToRpcCommands; |
|||
|
|||
public ASyncMsgProcessor(ActorSystemContext ctx, LoggingAdapter logger, SessionId sessionId) { |
|||
super(ctx, logger, sessionId); |
|||
} |
|||
|
|||
@Override |
|||
protected void processToDeviceActorMsg(ActorContext ctx, TransportToDeviceSessionActorMsg msg) { |
|||
updateSessionCtx(msg, SessionType.ASYNC); |
|||
DeviceToDeviceActorMsg pendingMsg = toDeviceMsg(msg); |
|||
FromDeviceMsg fromDeviceMsg = pendingMsg.getPayload(); |
|||
if (firstMsg) { |
|||
if (fromDeviceMsg.getMsgType() != SessionMsgType.SESSION_OPEN) { |
|||
toDeviceMsg(new SessionOpenMsg()).ifPresent(m -> forwardToAppActor(ctx, m)); |
|||
} |
|||
firstMsg = false; |
|||
} |
|||
switch (fromDeviceMsg.getMsgType()) { |
|||
case POST_TELEMETRY_REQUEST: |
|||
case POST_ATTRIBUTES_REQUEST: |
|||
FromDeviceRequestMsg requestMsg = (FromDeviceRequestMsg) fromDeviceMsg; |
|||
if (requestMsg.getRequestId() >= 0) { |
|||
logger.debug("[{}] Pending request {} registered", requestMsg.getRequestId(), requestMsg.getMsgType()); |
|||
//TODO: handle duplicates.
|
|||
pendingMap.put(requestMsg.getRequestId(), pendingMsg); |
|||
} |
|||
break; |
|||
case SUBSCRIBE_ATTRIBUTES_REQUEST: |
|||
subscribedToAttributeUpdates = true; |
|||
break; |
|||
case UNSUBSCRIBE_ATTRIBUTES_REQUEST: |
|||
subscribedToAttributeUpdates = false; |
|||
break; |
|||
case SUBSCRIBE_RPC_COMMANDS_REQUEST: |
|||
subscribedToRpcCommands = true; |
|||
break; |
|||
case UNSUBSCRIBE_RPC_COMMANDS_REQUEST: |
|||
subscribedToRpcCommands = false; |
|||
break; |
|||
default: |
|||
break; |
|||
} |
|||
currentTargetServer = forwardToAppActor(ctx, pendingMsg); |
|||
} |
|||
|
|||
@Override |
|||
public void processToDeviceMsg(ActorContext context, ToDeviceMsg msg) { |
|||
try { |
|||
if (msg.getSessionMsgType() != SessionMsgType.SESSION_CLOSE) { |
|||
switch (msg.getSessionMsgType()) { |
|||
case STATUS_CODE_RESPONSE: |
|||
case GET_ATTRIBUTES_RESPONSE: |
|||
ResponseMsg responseMsg = (ResponseMsg) msg; |
|||
if (responseMsg.getRequestId() >= 0) { |
|||
logger.debug("[{}] Pending request processed: {}", responseMsg.getRequestId(), responseMsg); |
|||
pendingMap.remove(responseMsg.getRequestId()); |
|||
} |
|||
break; |
|||
default: |
|||
break; |
|||
} |
|||
sessionCtx.onMsg(new BasicSessionActorToAdaptorMsg(this.sessionCtx, msg)); |
|||
} else { |
|||
sessionCtx.onMsg(org.thingsboard.server.common.msg.session.ctrl.SessionCloseMsg.onCredentialsRevoked(sessionCtx.getSessionId())); |
|||
} |
|||
} catch (SessionException e) { |
|||
logger.warning("Failed to push session response msg", e); |
|||
} |
|||
} |
|||
|
|||
@Override |
|||
public void processTimeoutMsg(ActorContext context, SessionTimeoutMsg msg) { |
|||
// TODO Auto-generated method stub
|
|||
} |
|||
|
|||
@Override |
|||
protected void cleanupSession(ActorContext ctx) { |
|||
toDeviceMsg(new SessionCloseMsg()).ifPresent(m -> forwardToAppActor(ctx, m)); |
|||
} |
|||
|
|||
@Override |
|||
public void processClusterEvent(ActorContext context, ClusterEventMsg msg) { |
|||
if (pendingMap.size() > 0 || subscribedToAttributeUpdates || subscribedToRpcCommands) { |
|||
Optional<ServerAddress> newTargetServer = systemContext.getRoutingService().resolveById(getDeviceId()); |
|||
if (!newTargetServer.equals(currentTargetServer)) { |
|||
firstMsg = true; |
|||
currentTargetServer = newTargetServer; |
|||
pendingMap.values().forEach(v -> { |
|||
forwardToAppActor(context, v, currentTargetServer); |
|||
if (currentTargetServer.isPresent()) { |
|||
logger.debug("[{}] Forwarded msg to new server: {}", sessionId, currentTargetServer.get()); |
|||
} else { |
|||
logger.debug("[{}] Forwarded msg to local server.", sessionId); |
|||
} |
|||
}); |
|||
if (subscribedToAttributeUpdates) { |
|||
toDeviceMsg(new AttributesSubscribeMsg()).ifPresent(m -> forwardToAppActor(context, m, currentTargetServer)); |
|||
logger.debug("[{}] Forwarded attributes subscription.", sessionId); |
|||
} |
|||
if (subscribedToRpcCommands) { |
|||
toDeviceMsg(new RpcSubscribeMsg()).ifPresent(m -> forwardToAppActor(context, m, currentTargetServer)); |
|||
logger.debug("[{}] Forwarded rpc commands subscription.", sessionId); |
|||
} |
|||
} |
|||
} |
|||
} |
|||
} |
|||
@ -1,122 +0,0 @@ |
|||
/** |
|||
* Copyright © 2016-2018 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.actors.session; |
|||
|
|||
import akka.actor.ActorContext; |
|||
import akka.actor.ActorRef; |
|||
import akka.event.LoggingAdapter; |
|||
import org.thingsboard.server.actors.ActorSystemContext; |
|||
import org.thingsboard.server.actors.shared.AbstractContextAwareMsgProcessor; |
|||
import org.thingsboard.server.actors.shared.SessionTimeoutMsg; |
|||
import org.thingsboard.server.common.data.id.DeviceId; |
|||
import org.thingsboard.server.common.data.id.SessionId; |
|||
import org.thingsboard.server.common.msg.cluster.ClusterEventMsg; |
|||
import org.thingsboard.server.common.msg.cluster.SendToClusterMsg; |
|||
import org.thingsboard.server.common.msg.cluster.ServerAddress; |
|||
import org.thingsboard.server.common.msg.device.BasicDeviceToDeviceActorMsg; |
|||
import org.thingsboard.server.common.msg.device.DeviceToDeviceActorMsg; |
|||
import org.thingsboard.server.common.msg.session.AdaptorToSessionActorMsg; |
|||
import org.thingsboard.server.common.msg.session.FromDeviceMsg; |
|||
import org.thingsboard.server.common.msg.session.SessionContext; |
|||
import org.thingsboard.server.common.msg.session.SessionCtrlMsg; |
|||
import org.thingsboard.server.common.msg.session.SessionType; |
|||
import org.thingsboard.server.common.msg.session.ToDeviceMsg; |
|||
import org.thingsboard.server.common.msg.session.TransportToDeviceSessionActorMsg; |
|||
import org.thingsboard.server.common.msg.session.ctrl.SessionCloseMsg; |
|||
|
|||
import java.util.Optional; |
|||
|
|||
abstract class AbstractSessionActorMsgProcessor extends AbstractContextAwareMsgProcessor { |
|||
|
|||
protected final SessionId sessionId; |
|||
protected SessionContext sessionCtx; |
|||
protected DeviceToDeviceActorMsg deviceToDeviceActorMsgPrototype; |
|||
|
|||
protected AbstractSessionActorMsgProcessor(ActorSystemContext ctx, LoggingAdapter logger, SessionId sessionId) { |
|||
super(ctx, logger); |
|||
this.sessionId = sessionId; |
|||
} |
|||
|
|||
protected abstract void processToDeviceActorMsg(ActorContext ctx, TransportToDeviceSessionActorMsg msg); |
|||
|
|||
protected abstract void processTimeoutMsg(ActorContext context, SessionTimeoutMsg msg); |
|||
|
|||
protected abstract void processToDeviceMsg(ActorContext context, ToDeviceMsg msg); |
|||
|
|||
public abstract void processClusterEvent(ActorContext context, ClusterEventMsg msg); |
|||
|
|||
protected void processSessionCtrlMsg(ActorContext ctx, SessionCtrlMsg msg) { |
|||
if (msg instanceof SessionCloseMsg) { |
|||
cleanupSession(ctx); |
|||
terminateSession(ctx, sessionId); |
|||
} |
|||
} |
|||
|
|||
protected void cleanupSession(ActorContext ctx) { |
|||
} |
|||
|
|||
protected void updateSessionCtx(TransportToDeviceSessionActorMsg msg, SessionType type) { |
|||
sessionCtx = msg.getSessionMsg().getSessionContext(); |
|||
deviceToDeviceActorMsgPrototype = new BasicDeviceToDeviceActorMsg(msg, type); |
|||
} |
|||
|
|||
protected DeviceToDeviceActorMsg toDeviceMsg(TransportToDeviceSessionActorMsg msg) { |
|||
AdaptorToSessionActorMsg adaptorMsg = msg.getSessionMsg(); |
|||
return new BasicDeviceToDeviceActorMsg(deviceToDeviceActorMsgPrototype, adaptorMsg.getMsg()); |
|||
} |
|||
|
|||
protected Optional<DeviceToDeviceActorMsg> toDeviceMsg(FromDeviceMsg msg) { |
|||
if (deviceToDeviceActorMsgPrototype != null) { |
|||
return Optional.of(new BasicDeviceToDeviceActorMsg(deviceToDeviceActorMsgPrototype, msg)); |
|||
} else { |
|||
return Optional.empty(); |
|||
} |
|||
} |
|||
|
|||
protected Optional<ServerAddress> forwardToAppActor(ActorContext ctx, DeviceToDeviceActorMsg toForward) { |
|||
Optional<ServerAddress> address = systemContext.getRoutingService().resolveById(toForward.getDeviceId()); |
|||
forwardToAppActor(ctx, toForward, address); |
|||
return address; |
|||
} |
|||
|
|||
protected Optional<ServerAddress> forwardToAppActorIfAddressChanged(ActorContext ctx, DeviceToDeviceActorMsg toForward, Optional<ServerAddress> oldAddress) { |
|||
|
|||
Optional<ServerAddress> newAddress = systemContext.getRoutingService().resolveById(toForward.getDeviceId()); |
|||
if (!newAddress.equals(oldAddress)) { |
|||
getAppActor().tell(new SendToClusterMsg(toForward.getDeviceId(), toForward |
|||
.toOtherAddress(systemContext.getRoutingService().getCurrentServer())), ctx.self()); |
|||
} |
|||
return newAddress; |
|||
} |
|||
|
|||
protected void forwardToAppActor(ActorContext ctx, DeviceToDeviceActorMsg toForward, Optional<ServerAddress> address) { |
|||
if (address.isPresent()) { |
|||
systemContext.getRpcService().tell(systemContext.getEncodingService().convertToProtoDataMessage(address.get(), |
|||
toForward.toOtherAddress(systemContext.getRoutingService().getCurrentServer()))); |
|||
} else { |
|||
getAppActor().tell(toForward, ctx.self()); |
|||
} |
|||
} |
|||
|
|||
public static void terminateSession(ActorContext ctx, SessionId sessionId) { |
|||
ctx.parent().tell(new SessionTerminationMsg(sessionId), ActorRef.noSender()); |
|||
ctx.stop(ctx.self()); |
|||
} |
|||
|
|||
public DeviceId getDeviceId() { |
|||
return deviceToDeviceActorMsgPrototype.getDeviceId(); |
|||
} |
|||
} |
|||
@ -1,143 +0,0 @@ |
|||
/** |
|||
* Copyright © 2016-2018 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.actors.session; |
|||
|
|||
import akka.actor.OneForOneStrategy; |
|||
import akka.actor.SupervisorStrategy; |
|||
import akka.event.Logging; |
|||
import akka.event.LoggingAdapter; |
|||
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.shared.SessionTimeoutMsg; |
|||
import org.thingsboard.server.common.data.id.SessionId; |
|||
import org.thingsboard.server.common.msg.TbActorMsg; |
|||
import org.thingsboard.server.common.msg.cluster.ClusterEventMsg; |
|||
import org.thingsboard.server.common.msg.core.ActorSystemToDeviceSessionActorMsg; |
|||
import org.thingsboard.server.common.msg.session.SessionCtrlMsg; |
|||
import org.thingsboard.server.common.msg.session.SessionMsg; |
|||
import org.thingsboard.server.common.msg.session.SessionType; |
|||
import org.thingsboard.server.common.msg.session.TransportToDeviceSessionActorMsg; |
|||
import org.thingsboard.server.common.msg.session.ctrl.SessionCloseMsg; |
|||
import scala.concurrent.duration.Duration; |
|||
|
|||
public class SessionActor extends ContextAwareActor { |
|||
|
|||
private final LoggingAdapter logger = Logging.getLogger(getContext().system(), this); |
|||
|
|||
private final SessionId sessionId; |
|||
private AbstractSessionActorMsgProcessor processor; |
|||
|
|||
private SessionActor(ActorSystemContext systemContext, SessionId sessionId) { |
|||
super(systemContext); |
|||
this.sessionId = sessionId; |
|||
} |
|||
|
|||
@Override |
|||
public SupervisorStrategy supervisorStrategy() { |
|||
return new OneForOneStrategy(-1, Duration.Inf(), |
|||
throwable -> { |
|||
logger.error(throwable, "Unknown session error"); |
|||
if (throwable instanceof Error) { |
|||
return OneForOneStrategy.escalate(); |
|||
} else { |
|||
return OneForOneStrategy.resume(); |
|||
} |
|||
}); |
|||
} |
|||
|
|||
@Override |
|||
protected boolean process(TbActorMsg msg) { |
|||
switch (msg.getMsgType()) { |
|||
case TRANSPORT_TO_DEVICE_SESSION_ACTOR_MSG: |
|||
processTransportToSessionMsg((TransportToDeviceSessionActorMsg) msg); |
|||
break; |
|||
case ACTOR_SYSTEM_TO_DEVICE_SESSION_ACTOR_MSG: |
|||
processActorsToSessionMsg((ActorSystemToDeviceSessionActorMsg) msg); |
|||
break; |
|||
case SESSION_TIMEOUT_MSG: |
|||
processTimeoutMsg((SessionTimeoutMsg) msg); |
|||
break; |
|||
case SESSION_CTRL_MSG: |
|||
processSessionCloseMsg((SessionCtrlMsg) msg); |
|||
break; |
|||
case CLUSTER_EVENT_MSG: |
|||
processClusterEvent((ClusterEventMsg) msg); |
|||
break; |
|||
default: return false; |
|||
} |
|||
return true; |
|||
} |
|||
|
|||
private void processClusterEvent(ClusterEventMsg msg) { |
|||
processor.processClusterEvent(context(), msg); |
|||
} |
|||
|
|||
private void processTransportToSessionMsg(TransportToDeviceSessionActorMsg msg) { |
|||
initProcessor(msg); |
|||
processor.processToDeviceActorMsg(context(), msg); |
|||
} |
|||
|
|||
private void processActorsToSessionMsg(ActorSystemToDeviceSessionActorMsg msg) { |
|||
processor.processToDeviceMsg(context(), msg.getMsg()); |
|||
} |
|||
|
|||
private void processTimeoutMsg(SessionTimeoutMsg msg) { |
|||
if (processor != null) { |
|||
processor.processTimeoutMsg(context(), msg); |
|||
} else { |
|||
logger.warning("[{}] Can't process timeout msg: {} without processor", sessionId, msg); |
|||
} |
|||
} |
|||
|
|||
private void processSessionCloseMsg(SessionCtrlMsg msg) { |
|||
if (processor != null) { |
|||
processor.processSessionCtrlMsg(context(), msg); |
|||
} else if (msg instanceof SessionCloseMsg) { |
|||
AbstractSessionActorMsgProcessor.terminateSession(context(), sessionId); |
|||
} else { |
|||
logger.warning("[{}] Can't process session ctrl msg: {} without processor", sessionId, msg); |
|||
} |
|||
} |
|||
|
|||
private void initProcessor(TransportToDeviceSessionActorMsg msg) { |
|||
if (processor == null) { |
|||
SessionMsg sessionMsg = (SessionMsg) msg.getSessionMsg(); |
|||
if (sessionMsg.getSessionContext().getSessionType() == SessionType.SYNC) { |
|||
processor = new SyncMsgProcessor(systemContext, logger, sessionId); |
|||
} else { |
|||
processor = new ASyncMsgProcessor(systemContext, logger, sessionId); |
|||
} |
|||
} |
|||
} |
|||
|
|||
public static class ActorCreator extends ContextBasedCreator<SessionActor> { |
|||
private static final long serialVersionUID = 1L; |
|||
|
|||
private final SessionId sessionId; |
|||
|
|||
public ActorCreator(ActorSystemContext context, SessionId sessionId) { |
|||
super(context); |
|||
this.sessionId = sessionId; |
|||
} |
|||
|
|||
@Override |
|||
public SessionActor create() throws Exception { |
|||
return new SessionActor(context, sessionId); |
|||
} |
|||
} |
|||
|
|||
} |
|||
@ -1,180 +0,0 @@ |
|||
/** |
|||
* Copyright © 2016-2018 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.actors.session; |
|||
|
|||
import akka.actor.ActorInitializationException; |
|||
import akka.actor.ActorRef; |
|||
import akka.actor.InvalidActorNameException; |
|||
import akka.actor.LocalActorRef; |
|||
import akka.actor.OneForOneStrategy; |
|||
import akka.actor.Props; |
|||
import akka.actor.SupervisorStrategy; |
|||
import akka.actor.Terminated; |
|||
import akka.event.Logging; |
|||
import akka.event.LoggingAdapter; |
|||
import akka.japi.Function; |
|||
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.actors.shared.SessionTimeoutMsg; |
|||
import org.thingsboard.server.common.data.id.SessionId; |
|||
import org.thingsboard.server.common.msg.TbActorMsg; |
|||
import org.thingsboard.server.common.msg.aware.SessionAwareMsg; |
|||
import org.thingsboard.server.common.msg.cluster.ClusterEventMsg; |
|||
import org.thingsboard.server.common.msg.core.ActorSystemToDeviceSessionActorMsg; |
|||
import org.thingsboard.server.common.msg.core.SessionCloseMsg; |
|||
import org.thingsboard.server.common.msg.session.SessionCtrlMsg; |
|||
import scala.concurrent.duration.Duration; |
|||
|
|||
import java.util.HashMap; |
|||
import java.util.Map; |
|||
|
|||
public class SessionManagerActor extends ContextAwareActor { |
|||
|
|||
private static final int INITIAL_SESSION_MAP_SIZE = 1024; |
|||
|
|||
private final LoggingAdapter log = Logging.getLogger(getContext().system(), this); |
|||
|
|||
private final Map<String, ActorRef> sessionActors; |
|||
|
|||
SessionManagerActor(ActorSystemContext systemContext) { |
|||
super(systemContext); |
|||
this.sessionActors = new HashMap<>(INITIAL_SESSION_MAP_SIZE); |
|||
} |
|||
|
|||
@Override |
|||
public SupervisorStrategy supervisorStrategy() { |
|||
return strategy; |
|||
} |
|||
|
|||
@Override |
|||
protected boolean process(TbActorMsg msg) { |
|||
//TODO Move everything here, to work with TbActorMsg
|
|||
return false; |
|||
} |
|||
|
|||
@Override |
|||
public void onReceive(Object msg) throws Exception { |
|||
if (msg instanceof SessionCtrlMsg) { |
|||
onSessionCtrlMsg((SessionCtrlMsg) msg); |
|||
} else if (msg instanceof SessionAwareMsg) { |
|||
forwardToSessionActor((SessionAwareMsg) msg); |
|||
} else if (msg instanceof SessionTerminationMsg) { |
|||
onSessionTermination((SessionTerminationMsg) msg); |
|||
} else if (msg instanceof Terminated) { |
|||
onTermination((Terminated) msg); |
|||
} else if (msg instanceof SessionTimeoutMsg) { |
|||
onSessionTimeout((SessionTimeoutMsg) msg); |
|||
} else if (msg instanceof ClusterEventMsg) { |
|||
broadcast(msg); |
|||
} |
|||
} |
|||
|
|||
private void broadcast(Object msg) { |
|||
sessionActors.values().forEach(actorRef -> actorRef.tell(msg, ActorRef.noSender())); |
|||
} |
|||
|
|||
private void onSessionTimeout(SessionTimeoutMsg msg) { |
|||
String sessionIdStr = msg.getSessionId().toUidStr(); |
|||
ActorRef sessionActor = sessionActors.get(sessionIdStr); |
|||
if (sessionActor != null) { |
|||
sessionActor.tell(msg, ActorRef.noSender()); |
|||
} |
|||
} |
|||
|
|||
private void onSessionCtrlMsg(SessionCtrlMsg msg) { |
|||
String sessionIdStr = msg.getSessionId().toUidStr(); |
|||
ActorRef sessionActor = sessionActors.get(sessionIdStr); |
|||
if (sessionActor != null) { |
|||
sessionActor.tell(msg, ActorRef.noSender()); |
|||
} |
|||
} |
|||
|
|||
private void onSessionTermination(SessionTerminationMsg msg) { |
|||
String sessionIdStr = msg.getId().toUidStr(); |
|||
ActorRef sessionActor = sessionActors.remove(sessionIdStr); |
|||
if (sessionActor != null) { |
|||
log.debug("[{}] Removed session actor.", sessionIdStr); |
|||
//TODO: onSubscriptionUpdate device actor about session close;
|
|||
} else { |
|||
log.debug("[{}] Session actor was already removed.", sessionIdStr); |
|||
} |
|||
} |
|||
|
|||
private void forwardToSessionActor(SessionAwareMsg msg) { |
|||
if (msg instanceof ActorSystemToDeviceSessionActorMsg || msg instanceof SessionCloseMsg) { |
|||
String sessionIdStr = msg.getSessionId().toUidStr(); |
|||
ActorRef sessionActor = sessionActors.get(sessionIdStr); |
|||
if (sessionActor != null) { |
|||
sessionActor.tell(msg, ActorRef.noSender()); |
|||
} else { |
|||
log.debug("[{}] Session actor was already removed.", sessionIdStr); |
|||
} |
|||
} else { |
|||
try { |
|||
getOrCreateSessionActor(msg.getSessionId()).tell(msg, self()); |
|||
} catch (InvalidActorNameException e) { |
|||
log.info("Invalid msg : {}", msg); |
|||
} |
|||
} |
|||
} |
|||
|
|||
private ActorRef getOrCreateSessionActor(SessionId sessionId) { |
|||
String sessionIdStr = sessionId.toUidStr(); |
|||
ActorRef sessionActor = sessionActors.get(sessionIdStr); |
|||
if (sessionActor == null) { |
|||
log.debug("[{}] Creating session actor.", sessionIdStr); |
|||
sessionActor = context().actorOf( |
|||
Props.create(new SessionActor.ActorCreator(systemContext, sessionId)).withDispatcher(DefaultActorService.SESSION_DISPATCHER_NAME), |
|||
sessionIdStr); |
|||
sessionActors.put(sessionIdStr, sessionActor); |
|||
log.debug("[{}] Created session actor.", sessionIdStr); |
|||
} |
|||
return sessionActor; |
|||
} |
|||
|
|||
private void onTermination(Terminated message) { |
|||
ActorRef terminated = message.actor(); |
|||
if (terminated instanceof LocalActorRef) { |
|||
log.info("Removed actor: {}.", terminated); |
|||
//TODO: cleanup session actors map
|
|||
} else { |
|||
throw new IllegalStateException("Remote actors are not supported!"); |
|||
} |
|||
} |
|||
|
|||
public static class ActorCreator extends ContextBasedCreator<SessionManagerActor> { |
|||
private static final long serialVersionUID = 1L; |
|||
|
|||
public ActorCreator(ActorSystemContext context) { |
|||
super(context); |
|||
} |
|||
|
|||
@Override |
|||
public SessionManagerActor create() throws Exception { |
|||
return new SessionManagerActor(context); |
|||
} |
|||
} |
|||
|
|||
private final SupervisorStrategy strategy = new OneForOneStrategy(3, Duration.create("1 minute"), new Function<Throwable, SupervisorStrategy.Directive>() { |
|||
@Override |
|||
public SupervisorStrategy.Directive apply(Throwable t) { |
|||
logger.error(t, "Unknown failure"); |
|||
return SupervisorStrategy.stop(); |
|||
} |
|||
}); |
|||
} |
|||
@ -1,26 +0,0 @@ |
|||
/** |
|||
* Copyright © 2016-2018 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.actors.session; |
|||
|
|||
import org.thingsboard.server.actors.shared.ActorTerminationMsg; |
|||
import org.thingsboard.server.common.data.id.SessionId; |
|||
|
|||
public class SessionTerminationMsg extends ActorTerminationMsg<SessionId> { |
|||
|
|||
public SessionTerminationMsg(SessionId id) { |
|||
super(id); |
|||
} |
|||
} |
|||
@ -1,95 +0,0 @@ |
|||
/** |
|||
* Copyright © 2016-2018 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.actors.session; |
|||
|
|||
import akka.actor.ActorContext; |
|||
import akka.event.LoggingAdapter; |
|||
import org.thingsboard.server.actors.ActorSystemContext; |
|||
import org.thingsboard.server.actors.shared.SessionTimeoutMsg; |
|||
import org.thingsboard.server.common.data.id.SessionId; |
|||
import org.thingsboard.server.common.msg.cluster.ClusterEventMsg; |
|||
import org.thingsboard.server.common.msg.cluster.ServerAddress; |
|||
import org.thingsboard.server.common.msg.device.DeviceToDeviceActorMsg; |
|||
import org.thingsboard.server.common.msg.session.BasicSessionActorToAdaptorMsg; |
|||
import org.thingsboard.server.common.msg.session.SessionContext; |
|||
import org.thingsboard.server.common.msg.session.SessionType; |
|||
import org.thingsboard.server.common.msg.session.ToDeviceMsg; |
|||
import org.thingsboard.server.common.msg.session.TransportToDeviceSessionActorMsg; |
|||
import org.thingsboard.server.common.msg.session.ctrl.SessionCloseMsg; |
|||
import org.thingsboard.server.common.msg.session.ex.SessionException; |
|||
|
|||
import java.util.Optional; |
|||
|
|||
class SyncMsgProcessor extends AbstractSessionActorMsgProcessor { |
|||
private DeviceToDeviceActorMsg pendingMsg; |
|||
private Optional<ServerAddress> currentTargetServer; |
|||
private boolean pendingResponse; |
|||
|
|||
public SyncMsgProcessor(ActorSystemContext ctx, LoggingAdapter logger, SessionId sessionId) { |
|||
super(ctx, logger, sessionId); |
|||
} |
|||
|
|||
@Override |
|||
protected void processToDeviceActorMsg(ActorContext ctx, TransportToDeviceSessionActorMsg msg) { |
|||
updateSessionCtx(msg, SessionType.SYNC); |
|||
pendingMsg = toDeviceMsg(msg); |
|||
pendingResponse = true; |
|||
currentTargetServer = forwardToAppActor(ctx, pendingMsg); |
|||
scheduleMsgWithDelay(ctx, new SessionTimeoutMsg(sessionId), getTimeout(systemContext, msg.getSessionMsg().getSessionContext()), ctx.parent()); |
|||
} |
|||
|
|||
public void processTimeoutMsg(ActorContext context, SessionTimeoutMsg msg) { |
|||
if (pendingResponse) { |
|||
try { |
|||
sessionCtx.onMsg(SessionCloseMsg.onTimeout(sessionId)); |
|||
} catch (SessionException e) { |
|||
logger.warning("Failed to push session close msg", e); |
|||
} |
|||
terminateSession(context, this.sessionId); |
|||
} |
|||
} |
|||
|
|||
public void processToDeviceMsg(ActorContext context, ToDeviceMsg msg) { |
|||
try { |
|||
sessionCtx.onMsg(new BasicSessionActorToAdaptorMsg(this.sessionCtx, msg)); |
|||
pendingResponse = false; |
|||
} catch (SessionException e) { |
|||
logger.warning("Failed to push session response msg", e); |
|||
} |
|||
terminateSession(context, this.sessionId); |
|||
} |
|||
|
|||
@Override |
|||
public void processClusterEvent(ActorContext context, ClusterEventMsg msg) { |
|||
if (pendingResponse) { |
|||
Optional<ServerAddress> newTargetServer = forwardToAppActorIfAddressChanged(context, pendingMsg, currentTargetServer); |
|||
if (logger.isDebugEnabled()) { |
|||
if (!newTargetServer.equals(currentTargetServer)) { |
|||
if (newTargetServer.isPresent()) { |
|||
logger.debug("[{}] Forwarded msg to new server: {}", sessionId, newTargetServer.get()); |
|||
} else { |
|||
logger.debug("[{}] Forwarded msg to local server.", sessionId); |
|||
} |
|||
} |
|||
} |
|||
currentTargetServer = newTargetServer; |
|||
} |
|||
} |
|||
|
|||
private long getTimeout(ActorSystemContext ctx, SessionContext sessionCtx) { |
|||
return sessionCtx.getTimeout() > 0 ? sessionCtx.getTimeout() : ctx.getSyncSessionTimeout(); |
|||
} |
|||
} |
|||
@ -1,36 +0,0 @@ |
|||
/** |
|||
* Copyright © 2016-2018 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.actors.shared; |
|||
|
|||
import lombok.Data; |
|||
import org.thingsboard.server.common.data.id.SessionId; |
|||
import org.thingsboard.server.common.msg.MsgType; |
|||
import org.thingsboard.server.common.msg.TbActorMsg; |
|||
|
|||
import java.io.Serializable; |
|||
|
|||
@Data |
|||
public class SessionTimeoutMsg implements Serializable, TbActorMsg { |
|||
|
|||
private static final long serialVersionUID = 1L; |
|||
|
|||
private final SessionId sessionId; |
|||
|
|||
@Override |
|||
public MsgType getMsgType() { |
|||
return MsgType.SESSION_TIMEOUT_MSG; |
|||
} |
|||
} |
|||
@ -1,111 +0,0 @@ |
|||
/** |
|||
* Copyright © 2016-2018 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.service.queue; |
|||
|
|||
import com.google.common.util.concurrent.Futures; |
|||
import com.google.common.util.concurrent.ListenableFuture; |
|||
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.TenantId; |
|||
import org.thingsboard.server.common.msg.TbMsg; |
|||
import org.thingsboard.server.common.transport.quota.tenant.TenantQuotaService; |
|||
import org.thingsboard.server.dao.queue.MsgQueue; |
|||
|
|||
import javax.annotation.PostConstruct; |
|||
import javax.annotation.PreDestroy; |
|||
import java.util.Map; |
|||
import java.util.UUID; |
|||
import java.util.concurrent.ConcurrentHashMap; |
|||
import java.util.concurrent.Executors; |
|||
import java.util.concurrent.ScheduledExecutorService; |
|||
import java.util.concurrent.TimeUnit; |
|||
import java.util.concurrent.atomic.AtomicLong; |
|||
|
|||
@Service |
|||
@Slf4j |
|||
public class DefaultMsgQueueService implements MsgQueueService { |
|||
|
|||
@Value("${actors.rule.queue.max_size}") |
|||
private long queueMaxSize; |
|||
|
|||
@Value("${actors.rule.queue.cleanup_period}") |
|||
private long queueCleanUpPeriod; |
|||
|
|||
@Autowired |
|||
private MsgQueue msgQueue; |
|||
|
|||
@Autowired |
|||
private TenantQuotaService quotaService; |
|||
|
|||
private ScheduledExecutorService cleanupExecutor; |
|||
|
|||
private Map<TenantId, AtomicLong> pendingCountPerTenant = new ConcurrentHashMap<>(); |
|||
|
|||
@PostConstruct |
|||
public void init() { |
|||
if (queueCleanUpPeriod > 0) { |
|||
cleanupExecutor = Executors.newSingleThreadScheduledExecutor(); |
|||
cleanupExecutor.scheduleAtFixedRate(() -> cleanup(), |
|||
queueCleanUpPeriod, queueCleanUpPeriod, TimeUnit.SECONDS); |
|||
} |
|||
} |
|||
|
|||
@PreDestroy |
|||
public void stop() { |
|||
if (cleanupExecutor != null) { |
|||
cleanupExecutor.shutdownNow(); |
|||
} |
|||
} |
|||
|
|||
@Override |
|||
public ListenableFuture<Void> put(TenantId tenantId, TbMsg msg, UUID nodeId, long clusterPartition) { |
|||
if(quotaService.isQuotaExceeded(tenantId.getId().toString())) { |
|||
log.warn("Tenant TbMsg Quota exceeded for [{}:{}] . Reject", tenantId.getId()); |
|||
return Futures.immediateFailedFuture(new RuntimeException("Tenant TbMsg Quota exceeded")); |
|||
} |
|||
|
|||
AtomicLong pendingMsgCount = pendingCountPerTenant.computeIfAbsent(tenantId, key -> new AtomicLong()); |
|||
if (pendingMsgCount.incrementAndGet() < queueMaxSize) { |
|||
return msgQueue.put(tenantId, msg, nodeId, clusterPartition); |
|||
} else { |
|||
pendingMsgCount.decrementAndGet(); |
|||
return Futures.immediateFailedFuture(new RuntimeException("Message queue is full!")); |
|||
} |
|||
} |
|||
|
|||
@Override |
|||
public ListenableFuture<Void> ack(TenantId tenantId, TbMsg msg, UUID nodeId, long clusterPartition) { |
|||
ListenableFuture<Void> result = msgQueue.ack(tenantId, msg, nodeId, clusterPartition); |
|||
AtomicLong pendingMsgCount = pendingCountPerTenant.computeIfAbsent(tenantId, key -> new AtomicLong()); |
|||
pendingMsgCount.decrementAndGet(); |
|||
return result; |
|||
} |
|||
|
|||
@Override |
|||
public Iterable<TbMsg> findUnprocessed(TenantId tenantId, UUID nodeId, long clusterPartition) { |
|||
return msgQueue.findUnprocessed(tenantId, nodeId, clusterPartition); |
|||
} |
|||
|
|||
private void cleanup() { |
|||
pendingCountPerTenant.forEach((tenantId, pendingMsgCount) -> { |
|||
pendingMsgCount.set(0); |
|||
msgQueue.cleanUp(tenantId); |
|||
}); |
|||
} |
|||
|
|||
} |
|||
@ -1,32 +0,0 @@ |
|||
/** |
|||
* Copyright © 2016-2018 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.service.queue; |
|||
|
|||
import com.google.common.util.concurrent.ListenableFuture; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
import org.thingsboard.server.common.msg.TbMsg; |
|||
|
|||
import java.util.UUID; |
|||
|
|||
public interface MsgQueueService { |
|||
|
|||
ListenableFuture<Void> put(TenantId tenantId, TbMsg msg, UUID nodeId, long clusterPartition); |
|||
|
|||
ListenableFuture<Void> ack(TenantId tenantId, TbMsg msg, UUID nodeId, long clusterPartition); |
|||
|
|||
Iterable<TbMsg> findUnprocessed(TenantId tenantId, UUID nodeId, long clusterPartition); |
|||
|
|||
} |
|||
@ -0,0 +1,173 @@ |
|||
/** |
|||
* Copyright © 2016-2018 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.service.transport; |
|||
|
|||
import com.fasterxml.jackson.core.JsonProcessingException; |
|||
import com.fasterxml.jackson.databind.ObjectMapper; |
|||
import com.google.common.util.concurrent.Futures; |
|||
import com.google.common.util.concurrent.ListenableFuture; |
|||
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.stereotype.Service; |
|||
import org.thingsboard.server.common.data.Device; |
|||
import org.thingsboard.server.common.data.id.DeviceId; |
|||
import org.thingsboard.server.common.data.relation.EntityRelation; |
|||
import org.thingsboard.server.common.data.security.DeviceCredentials; |
|||
import org.thingsboard.server.common.data.security.DeviceCredentialsType; |
|||
import org.thingsboard.server.dao.device.DeviceCredentialsService; |
|||
import org.thingsboard.server.dao.device.DeviceService; |
|||
import org.thingsboard.server.dao.relation.RelationService; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.DeviceInfoProto; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.GetOrCreateDeviceFromGatewayRequestMsg; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.GetOrCreateDeviceFromGatewayResponseMsg; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.TransportApiRequestMsg; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.TransportApiResponseMsg; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.ValidateDeviceCredentialsResponseMsg; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.ValidateDeviceTokenRequestMsg; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.ValidateDeviceX509CertRequestMsg; |
|||
import org.thingsboard.server.kafka.TBKafkaConsumerTemplate; |
|||
import org.thingsboard.server.kafka.TBKafkaProducerTemplate; |
|||
import org.thingsboard.server.kafka.TbKafkaResponseTemplate; |
|||
import org.thingsboard.server.kafka.TbKafkaSettings; |
|||
import org.thingsboard.server.service.cluster.discovery.DiscoveryService; |
|||
import org.thingsboard.server.service.executors.DbCallbackExecutorService; |
|||
import org.thingsboard.server.service.state.DeviceStateService; |
|||
|
|||
import javax.annotation.PostConstruct; |
|||
import javax.annotation.PreDestroy; |
|||
import java.util.UUID; |
|||
import java.util.concurrent.ExecutorService; |
|||
import java.util.concurrent.Executors; |
|||
import java.util.concurrent.locks.ReentrantLock; |
|||
|
|||
/** |
|||
* Created by ashvayka on 05.10.18. |
|||
*/ |
|||
@Slf4j |
|||
@Service |
|||
public class LocalTransportApiService implements TransportApiService { |
|||
|
|||
private static final ObjectMapper mapper = new ObjectMapper(); |
|||
|
|||
@Autowired |
|||
private DeviceService deviceService; |
|||
|
|||
@Autowired |
|||
private RelationService relationService; |
|||
|
|||
@Autowired |
|||
private DeviceCredentialsService deviceCredentialsService; |
|||
|
|||
@Autowired |
|||
private DeviceStateService deviceStateService; |
|||
|
|||
@Autowired |
|||
private DbCallbackExecutorService dbCallbackExecutorService; |
|||
|
|||
private ReentrantLock deviceCreationLock = new ReentrantLock(); |
|||
|
|||
@Override |
|||
public ListenableFuture<TransportApiResponseMsg> handle(TransportApiRequestMsg transportApiRequestMsg) { |
|||
if (transportApiRequestMsg.hasValidateTokenRequestMsg()) { |
|||
ValidateDeviceTokenRequestMsg msg = transportApiRequestMsg.getValidateTokenRequestMsg(); |
|||
return validateCredentials(msg.getToken(), DeviceCredentialsType.ACCESS_TOKEN); |
|||
} else if (transportApiRequestMsg.hasValidateX509CertRequestMsg()) { |
|||
ValidateDeviceX509CertRequestMsg msg = transportApiRequestMsg.getValidateX509CertRequestMsg(); |
|||
return validateCredentials(msg.getHash(), DeviceCredentialsType.X509_CERTIFICATE); |
|||
} else if (transportApiRequestMsg.hasGetOrCreateDeviceRequestMsg()) { |
|||
return handle(transportApiRequestMsg.getGetOrCreateDeviceRequestMsg()); |
|||
} |
|||
return getEmptyTransportApiResponseFuture(); |
|||
} |
|||
|
|||
private ListenableFuture<TransportApiResponseMsg> validateCredentials(String credentialsId, DeviceCredentialsType credentialsType) { |
|||
//TODO: Make async and enable caching
|
|||
DeviceCredentials credentials = deviceCredentialsService.findDeviceCredentialsByCredentialsId(credentialsId); |
|||
if (credentials != null && credentials.getCredentialsType() == credentialsType) { |
|||
return getDeviceInfo(credentials.getDeviceId()); |
|||
} else { |
|||
return getEmptyTransportApiResponseFuture(); |
|||
} |
|||
} |
|||
|
|||
private ListenableFuture<TransportApiResponseMsg> handle(GetOrCreateDeviceFromGatewayRequestMsg requestMsg) { |
|||
DeviceId gatewayId = new DeviceId(new UUID(requestMsg.getGatewayIdMSB(), requestMsg.getGatewayIdLSB())); |
|||
ListenableFuture<Device> gatewayFuture = deviceService.findDeviceByIdAsync(gatewayId); |
|||
return Futures.transform(gatewayFuture, gateway -> { |
|||
deviceCreationLock.lock(); |
|||
try { |
|||
Device device = deviceService.findDeviceByTenantIdAndName(gateway.getTenantId(), requestMsg.getDeviceName()); |
|||
if (device == null) { |
|||
device = new Device(); |
|||
device.setTenantId(gateway.getTenantId()); |
|||
device.setName(requestMsg.getDeviceName()); |
|||
device.setType(requestMsg.getDeviceType()); |
|||
device.setCustomerId(gateway.getCustomerId()); |
|||
device = deviceService.saveDevice(device); |
|||
relationService.saveRelationAsync(new EntityRelation(gateway.getId(), device.getId(), "Created")); |
|||
deviceStateService.onDeviceAdded(device); |
|||
} |
|||
return TransportApiResponseMsg.newBuilder() |
|||
.setGetOrCreateDeviceResponseMsg(GetOrCreateDeviceFromGatewayResponseMsg.newBuilder().setDeviceInfo(getDeviceInfoProto(device)).build()).build(); |
|||
} catch (JsonProcessingException e) { |
|||
log.warn("[{}] Failed to lookup device by gateway id and name", gatewayId, requestMsg.getDeviceName(), e); |
|||
throw new RuntimeException(e); |
|||
} finally { |
|||
deviceCreationLock.unlock(); |
|||
} |
|||
}, dbCallbackExecutorService); |
|||
} |
|||
|
|||
|
|||
private ListenableFuture<TransportApiResponseMsg> getDeviceInfo(DeviceId deviceId) { |
|||
return Futures.transform(deviceService.findDeviceByIdAsync(deviceId), device -> { |
|||
if (device == null) { |
|||
log.trace("[{}] Failed to lookup device by id", deviceId); |
|||
return getEmptyTransportApiResponse(); |
|||
} |
|||
try { |
|||
return TransportApiResponseMsg.newBuilder() |
|||
.setValidateTokenResponseMsg(ValidateDeviceCredentialsResponseMsg.newBuilder().setDeviceInfo(getDeviceInfoProto(device)).build()).build(); |
|||
} catch (JsonProcessingException e) { |
|||
log.warn("[{}] Failed to lookup device by id", deviceId, e); |
|||
return getEmptyTransportApiResponse(); |
|||
} |
|||
}); |
|||
} |
|||
|
|||
private DeviceInfoProto getDeviceInfoProto(Device device) throws JsonProcessingException { |
|||
return DeviceInfoProto.newBuilder() |
|||
.setTenantIdMSB(device.getTenantId().getId().getMostSignificantBits()) |
|||
.setTenantIdLSB(device.getTenantId().getId().getLeastSignificantBits()) |
|||
.setDeviceIdMSB(device.getId().getId().getMostSignificantBits()) |
|||
.setDeviceIdLSB(device.getId().getId().getLeastSignificantBits()) |
|||
.setDeviceName(device.getName()) |
|||
.setDeviceType(device.getType()) |
|||
.setAdditionalInfo(mapper.writeValueAsString(device.getAdditionalInfo())) |
|||
.build(); |
|||
} |
|||
|
|||
private ListenableFuture<TransportApiResponseMsg> getEmptyTransportApiResponseFuture() { |
|||
return Futures.immediateFuture(getEmptyTransportApiResponse()); |
|||
} |
|||
|
|||
private TransportApiResponseMsg getEmptyTransportApiResponse() { |
|||
return TransportApiResponseMsg.newBuilder() |
|||
.setValidateTokenResponseMsg(ValidateDeviceCredentialsResponseMsg.getDefaultInstance()).build(); |
|||
} |
|||
} |
|||
@ -0,0 +1,225 @@ |
|||
/** |
|||
* Copyright © 2016-2018 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.service.transport; |
|||
|
|||
import akka.actor.ActorRef; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.springframework.beans.factory.annotation.Autowired; |
|||
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; |
|||
import org.springframework.stereotype.Service; |
|||
import org.thingsboard.rule.engine.api.util.DonAsynchron; |
|||
import org.thingsboard.server.actors.ActorSystemContext; |
|||
import org.thingsboard.server.common.msg.cluster.ServerAddress; |
|||
import org.thingsboard.server.common.transport.SessionMsgListener; |
|||
import org.thingsboard.server.common.transport.TransportService; |
|||
import org.thingsboard.server.common.transport.TransportServiceCallback; |
|||
import org.thingsboard.server.common.transport.service.AbstractTransportService; |
|||
import org.thingsboard.server.gen.transport.TransportProtos; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.DeviceActorToTransportMsg; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.GetAttributeRequestMsg; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.GetOrCreateDeviceFromGatewayRequestMsg; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.GetOrCreateDeviceFromGatewayResponseMsg; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.PostAttributeMsg; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.PostTelemetryMsg; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.SessionEventMsg; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.SessionInfoProto; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.SubscribeToAttributeUpdatesMsg; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.SubscribeToRPCMsg; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.ToDeviceRpcResponseMsg; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.ToServerRpcRequestMsg; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.TransportApiRequestMsg; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.TransportToDeviceActorMsg; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.ValidateDeviceCredentialsResponseMsg; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.ValidateDeviceTokenRequestMsg; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.ValidateDeviceX509CertRequestMsg; |
|||
import org.thingsboard.server.service.cluster.routing.ClusterRoutingService; |
|||
import org.thingsboard.server.service.cluster.rpc.ClusterRpcService; |
|||
import org.thingsboard.server.service.encoding.DataDecodingEncodingService; |
|||
import org.thingsboard.server.service.transport.msg.TransportToDeviceActorMsgWrapper; |
|||
|
|||
import javax.annotation.PostConstruct; |
|||
import javax.annotation.PreDestroy; |
|||
import java.util.Optional; |
|||
import java.util.UUID; |
|||
import java.util.concurrent.ConcurrentHashMap; |
|||
import java.util.concurrent.ConcurrentMap; |
|||
import java.util.concurrent.ExecutorService; |
|||
import java.util.concurrent.Executors; |
|||
import java.util.concurrent.ScheduledExecutorService; |
|||
import java.util.concurrent.SynchronousQueue; |
|||
import java.util.concurrent.ThreadPoolExecutor; |
|||
import java.util.concurrent.TimeUnit; |
|||
import java.util.function.Consumer; |
|||
|
|||
/** |
|||
* Created by ashvayka on 12.10.18. |
|||
*/ |
|||
@Slf4j |
|||
@Service |
|||
@ConditionalOnProperty(prefix = "transport", value = "type", havingValue = "local") |
|||
public class LocalTransportService extends AbstractTransportService implements RuleEngineTransportService { |
|||
|
|||
@Autowired |
|||
private TransportApiService transportApiService; |
|||
|
|||
@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; |
|||
|
|||
@PostConstruct |
|||
public void init() { |
|||
super.init(); |
|||
} |
|||
|
|||
@PreDestroy |
|||
public void destroy() { |
|||
super.destroy(); |
|||
} |
|||
|
|||
@Override |
|||
public void process(ValidateDeviceTokenRequestMsg msg, TransportServiceCallback<ValidateDeviceCredentialsResponseMsg> callback) { |
|||
DonAsynchron.withCallback( |
|||
transportApiService.handle(TransportApiRequestMsg.newBuilder().setValidateTokenRequestMsg(msg).build()), |
|||
transportApiResponseMsg -> { |
|||
if (callback != null) { |
|||
callback.onSuccess(transportApiResponseMsg.getValidateTokenResponseMsg()); |
|||
} |
|||
}, |
|||
getThrowableConsumer(callback), transportCallbackExecutor); |
|||
} |
|||
|
|||
@Override |
|||
public void process(ValidateDeviceX509CertRequestMsg msg, TransportServiceCallback<ValidateDeviceCredentialsResponseMsg> callback) { |
|||
DonAsynchron.withCallback( |
|||
transportApiService.handle(TransportApiRequestMsg.newBuilder().setValidateX509CertRequestMsg(msg).build()), |
|||
transportApiResponseMsg -> { |
|||
if (callback != null) { |
|||
callback.onSuccess(transportApiResponseMsg.getValidateTokenResponseMsg()); |
|||
} |
|||
}, |
|||
getThrowableConsumer(callback), transportCallbackExecutor); |
|||
} |
|||
|
|||
@Override |
|||
public void process(GetOrCreateDeviceFromGatewayRequestMsg msg, TransportServiceCallback<GetOrCreateDeviceFromGatewayResponseMsg> callback) { |
|||
DonAsynchron.withCallback( |
|||
transportApiService.handle(TransportApiRequestMsg.newBuilder().setGetOrCreateDeviceRequestMsg(msg).build()), |
|||
transportApiResponseMsg -> { |
|||
if (callback != null) { |
|||
callback.onSuccess(transportApiResponseMsg.getGetOrCreateDeviceResponseMsg()); |
|||
} |
|||
}, |
|||
getThrowableConsumer(callback), transportCallbackExecutor); |
|||
} |
|||
|
|||
@Override |
|||
public void process(SessionInfoProto sessionInfo, SessionEventMsg msg, TransportServiceCallback<Void> callback) { |
|||
if (checkLimits(sessionInfo, callback)) { |
|||
forwardToDeviceActor(TransportToDeviceActorMsg.newBuilder().setSessionInfo(sessionInfo).setSessionEvent(msg).build(), callback); |
|||
} |
|||
} |
|||
|
|||
@Override |
|||
public void process(SessionInfoProto sessionInfo, PostTelemetryMsg msg, TransportServiceCallback<Void> callback) { |
|||
if (checkLimits(sessionInfo, callback)) { |
|||
forwardToDeviceActor(TransportToDeviceActorMsg.newBuilder().setSessionInfo(sessionInfo).setPostTelemetry(msg).build(), callback); |
|||
} |
|||
} |
|||
|
|||
@Override |
|||
public void process(SessionInfoProto sessionInfo, PostAttributeMsg msg, TransportServiceCallback<Void> callback) { |
|||
if (checkLimits(sessionInfo, callback)) { |
|||
forwardToDeviceActor(TransportToDeviceActorMsg.newBuilder().setSessionInfo(sessionInfo).setPostAttributes(msg).build(), callback); |
|||
} |
|||
} |
|||
|
|||
@Override |
|||
public void process(SessionInfoProto sessionInfo, GetAttributeRequestMsg msg, TransportServiceCallback<Void> callback) { |
|||
if (checkLimits(sessionInfo, callback)) { |
|||
forwardToDeviceActor(TransportToDeviceActorMsg.newBuilder().setSessionInfo(sessionInfo).setGetAttributes(msg).build(), callback); |
|||
} |
|||
} |
|||
|
|||
@Override |
|||
public void process(SessionInfoProto sessionInfo, SubscribeToAttributeUpdatesMsg msg, TransportServiceCallback<Void> callback) { |
|||
if (checkLimits(sessionInfo, callback)) { |
|||
forwardToDeviceActor(TransportToDeviceActorMsg.newBuilder().setSessionInfo(sessionInfo).setSubscribeToAttributes(msg).build(), callback); |
|||
} |
|||
} |
|||
|
|||
@Override |
|||
public void process(SessionInfoProto sessionInfo, SubscribeToRPCMsg msg, TransportServiceCallback<Void> callback) { |
|||
if (checkLimits(sessionInfo, callback)) { |
|||
forwardToDeviceActor(TransportToDeviceActorMsg.newBuilder().setSessionInfo(sessionInfo).setSubscribeToRPC(msg).build(), callback); |
|||
} |
|||
} |
|||
|
|||
@Override |
|||
public void process(SessionInfoProto sessionInfo, ToDeviceRpcResponseMsg msg, TransportServiceCallback<Void> callback) { |
|||
if (checkLimits(sessionInfo, callback)) { |
|||
forwardToDeviceActor(TransportToDeviceActorMsg.newBuilder().setSessionInfo(sessionInfo).setToDeviceRPCCallResponse(msg).build(), callback); |
|||
} |
|||
} |
|||
|
|||
@Override |
|||
public void process(SessionInfoProto sessionInfo, ToServerRpcRequestMsg msg, TransportServiceCallback<Void> callback) { |
|||
if (checkLimits(sessionInfo, callback)) { |
|||
forwardToDeviceActor(TransportToDeviceActorMsg.newBuilder().setSessionInfo(sessionInfo).setToServerRPCCallRequest(msg).build(), callback); |
|||
} |
|||
} |
|||
|
|||
@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) { |
|||
processToTransportMsg(msg); |
|||
if (onSuccess != null) { |
|||
onSuccess.run(); |
|||
} |
|||
} |
|||
|
|||
private void forwardToDeviceActor(TransportToDeviceActorMsg toDeviceActorMsg, TransportServiceCallback<Void> callback) { |
|||
TransportToDeviceActorMsgWrapper wrapper = new TransportToDeviceActorMsgWrapper(toDeviceActorMsg); |
|||
Optional<ServerAddress> address = routingService.resolveById(wrapper.getDeviceId()); |
|||
if (address.isPresent()) { |
|||
rpcService.tell(encodingService.convertToProtoDataMessage(address.get(), wrapper)); |
|||
} else { |
|||
actorContext.getAppActor().tell(wrapper, ActorRef.noSender()); |
|||
} |
|||
if (callback != null) { |
|||
callback.onSuccess(null); |
|||
} |
|||
} |
|||
|
|||
private <T> Consumer<Throwable> getThrowableConsumer(TransportServiceCallback<T> callback) { |
|||
return e -> { |
|||
if (callback != null) { |
|||
callback.onError(e); |
|||
} |
|||
}; |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,200 @@ |
|||
/** |
|||
* Copyright © 2016-2018 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.service.transport; |
|||
|
|||
import akka.actor.ActorRef; |
|||
import com.fasterxml.jackson.databind.ObjectMapper; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.apache.kafka.clients.consumer.ConsumerRecords; |
|||
import org.apache.kafka.clients.producer.Callback; |
|||
import org.apache.kafka.clients.producer.RecordMetadata; |
|||
import org.springframework.beans.factory.annotation.Autowired; |
|||
import org.springframework.beans.factory.annotation.Value; |
|||
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; |
|||
import org.springframework.stereotype.Service; |
|||
import org.thingsboard.server.actors.ActorSystemContext; |
|||
import org.thingsboard.server.actors.service.ActorService; |
|||
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.kafka.TBKafkaConsumerTemplate; |
|||
import org.thingsboard.server.kafka.TBKafkaProducerTemplate; |
|||
import org.thingsboard.server.kafka.TbKafkaSettings; |
|||
import org.thingsboard.server.kafka.TbNodeIdProvider; |
|||
import org.thingsboard.server.service.cluster.discovery.DiscoveryService; |
|||
import org.thingsboard.server.service.cluster.routing.ClusterRoutingService; |
|||
import org.thingsboard.server.service.cluster.rpc.ClusterRpcService; |
|||
import org.thingsboard.server.service.encoding.DataDecodingEncodingService; |
|||
import org.thingsboard.server.service.transport.msg.TransportToDeviceActorMsgWrapper; |
|||
|
|||
import javax.annotation.PostConstruct; |
|||
import javax.annotation.PreDestroy; |
|||
import java.time.Duration; |
|||
import java.util.Optional; |
|||
import java.util.UUID; |
|||
import java.util.concurrent.ExecutorService; |
|||
import java.util.concurrent.Executors; |
|||
import java.util.function.Consumer; |
|||
|
|||
/** |
|||
* Created by ashvayka on 09.10.18. |
|||
*/ |
|||
@Slf4j |
|||
@Service |
|||
@ConditionalOnProperty(prefix = "transport", value = "type", havingValue = "remote") |
|||
public class RemoteRuleEngineTransportService implements RuleEngineTransportService { |
|||
|
|||
@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; |
|||
|
|||
@Autowired |
|||
private TbKafkaSettings kafkaSettings; |
|||
|
|||
@Autowired |
|||
private TbNodeIdProvider nodeIdProvider; |
|||
|
|||
@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; |
|||
|
|||
private TBKafkaConsumerTemplate<ToRuleEngineMsg> ruleEngineConsumer; |
|||
private TBKafkaProducerTemplate<ToTransportMsg> notificationsProducer; |
|||
|
|||
private ExecutorService mainConsumerExecutor = Executors.newSingleThreadExecutor(); |
|||
|
|||
private volatile boolean stopped = false; |
|||
|
|||
@PostConstruct |
|||
public void init() { |
|||
TBKafkaProducerTemplate.TBKafkaProducerTemplateBuilder<ToTransportMsg> notificationsProducerBuilder = TBKafkaProducerTemplate.builder(); |
|||
notificationsProducerBuilder.settings(kafkaSettings); |
|||
notificationsProducerBuilder.defaultTopic(notificationsTopic); |
|||
notificationsProducerBuilder.encoder(new ToTransportMsgEncoder()); |
|||
|
|||
notificationsProducer = notificationsProducerBuilder.build(); |
|||
notificationsProducer.init(); |
|||
|
|||
TBKafkaConsumerTemplate.TBKafkaConsumerTemplateBuilder<ToRuleEngineMsg> ruleEngineConsumerBuilder = TBKafkaConsumerTemplate.builder(); |
|||
ruleEngineConsumerBuilder.settings(kafkaSettings); |
|||
ruleEngineConsumerBuilder.topic(ruleEngineTopic); |
|||
ruleEngineConsumerBuilder.clientId("transport-" + nodeIdProvider.getNodeId()); |
|||
ruleEngineConsumerBuilder.groupId("tb-node"); |
|||
ruleEngineConsumerBuilder.autoCommit(true); |
|||
ruleEngineConsumerBuilder.autoCommitIntervalMs(autoCommitInterval); |
|||
ruleEngineConsumerBuilder.decoder(new ToRuleEngineMsgDecoder()); |
|||
|
|||
ruleEngineConsumer = ruleEngineConsumerBuilder.build(); |
|||
ruleEngineConsumer.subscribe(); |
|||
|
|||
mainConsumerExecutor.execute(() -> { |
|||
while (!stopped) { |
|||
try { |
|||
ConsumerRecords<String, byte[]> records = ruleEngineConsumer.poll(Duration.ofMillis(pollDuration)); |
|||
records.forEach(record -> { |
|||
try { |
|||
ToRuleEngineMsg toRuleEngineMsg = ruleEngineConsumer.decode(record); |
|||
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); |
|||
} |
|||
} |
|||
} |
|||
}); |
|||
} |
|||
|
|||
@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) { |
|||
notificationsProducer.send(notificationsTopic + "." + nodeId, |
|||
new UUID(msg.getSessionIdMSB(), msg.getSessionIdLSB()).toString(), |
|||
ToTransportMsg.newBuilder().setToDeviceSessionMsg(msg).build() |
|||
, new QueueCallbackAdaptor(onSuccess, onFailure)); |
|||
} |
|||
|
|||
private void forwardToDeviceActor(TransportToDeviceActorMsg toDeviceActorMsg) { |
|||
TransportToDeviceActorMsgWrapper wrapper = new TransportToDeviceActorMsgWrapper(toDeviceActorMsg); |
|||
Optional<ServerAddress> address = routingService.resolveById(wrapper.getDeviceId()); |
|||
if (address.isPresent()) { |
|||
rpcService.tell(encodingService.convertToProtoDataMessage(address.get(), wrapper)); |
|||
} else { |
|||
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 Callback { |
|||
private final Runnable onSuccess; |
|||
private final Consumer<Throwable> onFailure; |
|||
|
|||
QueueCallbackAdaptor(Runnable onSuccess, Consumer<Throwable> onFailure) { |
|||
this.onSuccess = onSuccess; |
|||
this.onFailure = onFailure; |
|||
} |
|||
|
|||
@Override |
|||
public void onCompletion(RecordMetadata metadata, Exception exception) { |
|||
if (exception == null) { |
|||
if (onSuccess != null) { |
|||
onSuccess.run(); |
|||
} |
|||
} else { |
|||
if (onFailure != null) { |
|||
onFailure.accept(exception); |
|||
} |
|||
} |
|||
} |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,109 @@ |
|||
/** |
|||
* Copyright © 2016-2018 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.service.transport; |
|||
|
|||
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.stereotype.Component; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.TransportApiRequestMsg; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.TransportApiResponseMsg; |
|||
import org.thingsboard.server.kafka.*; |
|||
|
|||
import javax.annotation.PostConstruct; |
|||
import javax.annotation.PreDestroy; |
|||
import java.util.concurrent.ExecutorService; |
|||
import java.util.concurrent.SynchronousQueue; |
|||
import java.util.concurrent.ThreadPoolExecutor; |
|||
import java.util.concurrent.TimeUnit; |
|||
|
|||
/** |
|||
* Created by ashvayka on 05.10.18. |
|||
*/ |
|||
@Slf4j |
|||
@Component |
|||
@ConditionalOnProperty(prefix = "transport", value = "type", havingValue = "remote") |
|||
public class RemoteTransportApiService { |
|||
|
|||
@Value("${transport.remote.transport_api.requests_topic}") |
|||
private String transportApiRequestsTopic; |
|||
@Value("${transport.remote.transport_api.responses_topic}") |
|||
private String transportApiResponsesTopic; |
|||
@Value("${transport.remote.transport_api.max_pending_requests}") |
|||
private int maxPendingRequests; |
|||
@Value("${transport.remote.transport_api.request_timeout}") |
|||
private long requestTimeout; |
|||
@Value("${transport.remote.transport_api.request_poll_interval}") |
|||
private int responsePollDuration; |
|||
@Value("${transport.remote.transport_api.request_auto_commit_interval}") |
|||
private int autoCommitInterval; |
|||
|
|||
@Autowired |
|||
private TbKafkaSettings kafkaSettings; |
|||
|
|||
@Autowired |
|||
private TbNodeIdProvider nodeIdProvider; |
|||
|
|||
@Autowired |
|||
private TransportApiService transportApiService; |
|||
|
|||
private ExecutorService transportCallbackExecutor; |
|||
|
|||
private TbKafkaResponseTemplate<TransportApiRequestMsg, TransportApiResponseMsg> transportApiTemplate; |
|||
|
|||
@PostConstruct |
|||
public void init() { |
|||
this.transportCallbackExecutor = new ThreadPoolExecutor(0, 100, 60L, TimeUnit.SECONDS, new SynchronousQueue<>()); |
|||
|
|||
TBKafkaProducerTemplate.TBKafkaProducerTemplateBuilder<TransportApiResponseMsg> responseBuilder = TBKafkaProducerTemplate.builder(); |
|||
responseBuilder.settings(kafkaSettings); |
|||
responseBuilder.defaultTopic(transportApiResponsesTopic); |
|||
responseBuilder.encoder(new TransportApiResponseEncoder()); |
|||
|
|||
TBKafkaConsumerTemplate.TBKafkaConsumerTemplateBuilder<TransportApiRequestMsg> requestBuilder = TBKafkaConsumerTemplate.builder(); |
|||
requestBuilder.settings(kafkaSettings); |
|||
requestBuilder.topic(transportApiRequestsTopic); |
|||
requestBuilder.clientId(nodeIdProvider.getNodeId()); |
|||
requestBuilder.groupId("tb-node"); |
|||
requestBuilder.autoCommit(true); |
|||
requestBuilder.autoCommitIntervalMs(autoCommitInterval); |
|||
requestBuilder.decoder(new TransportApiRequestDecoder()); |
|||
|
|||
TbKafkaResponseTemplate.TbKafkaResponseTemplateBuilder |
|||
<TransportApiRequestMsg, TransportApiResponseMsg> builder = TbKafkaResponseTemplate.builder(); |
|||
builder.requestTemplate(requestBuilder.build()); |
|||
builder.responseTemplate(responseBuilder.build()); |
|||
builder.maxPendingRequests(maxPendingRequests); |
|||
builder.requestTimeout(requestTimeout); |
|||
builder.pollInterval(responsePollDuration); |
|||
builder.executor(transportCallbackExecutor); |
|||
builder.handler(transportApiService); |
|||
transportApiTemplate = builder.build(); |
|||
transportApiTemplate.init(); |
|||
} |
|||
|
|||
@PreDestroy |
|||
public void destroy() { |
|||
if (transportApiTemplate != null) { |
|||
transportApiTemplate.stop(); |
|||
} |
|||
if (transportCallbackExecutor != null) { |
|||
transportCallbackExecutor.shutdownNow(); |
|||
} |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,31 @@ |
|||
/** |
|||
* Copyright © 2016-2018 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.service.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); |
|||
} |
|||
} |
|||
@ -0,0 +1,50 @@ |
|||
/** |
|||
* Copyright © 2016-2018 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.service.transport.msg; |
|||
|
|||
import lombok.Data; |
|||
import org.thingsboard.server.common.data.id.DeviceId; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
import org.thingsboard.server.common.msg.MsgType; |
|||
import org.thingsboard.server.common.msg.TbActorMsg; |
|||
import org.thingsboard.server.common.msg.aware.DeviceAwareMsg; |
|||
import org.thingsboard.server.common.msg.aware.TenantAwareMsg; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.TransportToDeviceActorMsg; |
|||
|
|||
import java.io.Serializable; |
|||
import java.util.UUID; |
|||
|
|||
/** |
|||
* Created by ashvayka on 09.10.18. |
|||
*/ |
|||
@Data |
|||
public class TransportToDeviceActorMsgWrapper implements TbActorMsg, DeviceAwareMsg, TenantAwareMsg, Serializable { |
|||
|
|||
private final TenantId tenantId; |
|||
private final DeviceId deviceId; |
|||
private final TransportToDeviceActorMsg msg; |
|||
|
|||
public TransportToDeviceActorMsgWrapper(TransportToDeviceActorMsg msg) { |
|||
this.msg = msg; |
|||
this.tenantId = new TenantId(new UUID(msg.getSessionInfo().getTenantIdMSB(), msg.getSessionInfo().getTenantIdLSB())); |
|||
this.deviceId = new DeviceId(new UUID(msg.getSessionInfo().getDeviceIdMSB(), msg.getSessionInfo().getDeviceIdLSB())); |
|||
} |
|||
|
|||
@Override |
|||
public MsgType getMsgType() { |
|||
return MsgType.TRANSPORT_TO_DEVICE_ACTOR_MSG; |
|||
} |
|||
} |
|||
@ -1,30 +0,0 @@ |
|||
/** |
|||
* Copyright © 2016-2018 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.common.msg.core; |
|||
|
|||
import org.thingsboard.server.common.msg.TbActorMsg; |
|||
import org.thingsboard.server.common.msg.aware.SessionAwareMsg; |
|||
import org.thingsboard.server.common.msg.session.ToDeviceMsg; |
|||
|
|||
import java.io.Serializable; |
|||
|
|||
/** |
|||
* @author Andrew Shvayka |
|||
*/ |
|||
public interface ActorSystemToDeviceSessionActorMsg extends SessionAwareMsg, Serializable, TbActorMsg { |
|||
|
|||
ToDeviceMsg getMsg(); |
|||
} |
|||
@ -1,47 +0,0 @@ |
|||
/** |
|||
* Copyright © 2016-2018 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.common.msg.core; |
|||
|
|||
import lombok.ToString; |
|||
import org.thingsboard.server.common.msg.kv.AttributesKVMsg; |
|||
import org.thingsboard.server.common.msg.session.SessionMsgType; |
|||
import org.thingsboard.server.common.msg.session.SessionMsgType; |
|||
import org.thingsboard.server.common.msg.session.ToDeviceMsg; |
|||
|
|||
@ToString |
|||
public class AttributesUpdateNotification implements ToDeviceMsg { |
|||
|
|||
private static final long serialVersionUID = 1L; |
|||
|
|||
private AttributesKVMsg data; |
|||
|
|||
public AttributesUpdateNotification(AttributesKVMsg data) { |
|||
this.data = data; |
|||
} |
|||
|
|||
@Override |
|||
public boolean isSuccess() { |
|||
return true; |
|||
} |
|||
|
|||
public SessionMsgType getSessionMsgType() { |
|||
return SessionMsgType.ATTRIBUTES_UPDATE_NOTIFICATION; |
|||
} |
|||
|
|||
public AttributesKVMsg getData() { |
|||
return data; |
|||
} |
|||
} |
|||
@ -1,28 +0,0 @@ |
|||
/** |
|||
* Copyright © 2016-2018 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.common.msg.core; |
|||
|
|||
import java.util.Set; |
|||
|
|||
import org.thingsboard.server.common.data.kv.AttributeKvEntry; |
|||
import org.thingsboard.server.common.msg.session.FromDeviceMsg; |
|||
import org.thingsboard.server.common.msg.session.FromDeviceRequestMsg; |
|||
|
|||
public interface AttributesUpdateRequest extends FromDeviceRequestMsg { |
|||
|
|||
Set<AttributeKvEntry> getAttributes(); |
|||
|
|||
} |
|||
@ -1,52 +0,0 @@ |
|||
/** |
|||
* Copyright © 2016-2018 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.common.msg.core; |
|||
|
|||
import org.thingsboard.server.common.data.id.SessionId; |
|||
import org.thingsboard.server.common.msg.MsgType; |
|||
import org.thingsboard.server.common.msg.session.ToDeviceMsg; |
|||
|
|||
public class BasicActorSystemToDeviceSessionActorMsg implements ActorSystemToDeviceSessionActorMsg { |
|||
|
|||
private final ToDeviceMsg msg; |
|||
private final SessionId sessionId; |
|||
|
|||
public BasicActorSystemToDeviceSessionActorMsg(ToDeviceMsg msg, SessionId sessionId) { |
|||
super(); |
|||
this.msg = msg; |
|||
this.sessionId = sessionId; |
|||
} |
|||
|
|||
@Override |
|||
public SessionId getSessionId() { |
|||
return sessionId; |
|||
} |
|||
|
|||
@Override |
|||
public ToDeviceMsg getMsg() { |
|||
return msg; |
|||
} |
|||
|
|||
@Override |
|||
public String toString() { |
|||
return "BasicActorSystemToDeviceSessionActorMsg [msg=" + msg + ", sessionId=" + sessionId + "]"; |
|||
} |
|||
|
|||
@Override |
|||
public MsgType getMsgType() { |
|||
return MsgType.ACTOR_SYSTEM_TO_DEVICE_SESSION_ACTOR_MSG; |
|||
} |
|||
} |
|||
@ -1,63 +0,0 @@ |
|||
/** |
|||
* Copyright © 2016-2018 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.common.msg.core; |
|||
|
|||
import java.util.Collection; |
|||
import java.util.LinkedHashSet; |
|||
import java.util.Set; |
|||
|
|||
import org.thingsboard.server.common.data.kv.AttributeKvEntry; |
|||
import org.thingsboard.server.common.msg.session.SessionMsgType; |
|||
|
|||
public class BasicAttributesUpdateRequest extends BasicRequest implements AttributesUpdateRequest { |
|||
|
|||
private static final long serialVersionUID = 1L; |
|||
|
|||
private final Set<AttributeKvEntry> data; |
|||
|
|||
public BasicAttributesUpdateRequest() { |
|||
this(DEFAULT_REQUEST_ID); |
|||
} |
|||
|
|||
public BasicAttributesUpdateRequest(Integer requestId) { |
|||
super(requestId); |
|||
this.data = new LinkedHashSet<>(); |
|||
} |
|||
|
|||
public void add(AttributeKvEntry entry) { |
|||
this.data.add(entry); |
|||
} |
|||
|
|||
public void add(Collection<AttributeKvEntry> entries) { |
|||
this.data.addAll(entries); |
|||
} |
|||
|
|||
@Override |
|||
public SessionMsgType getMsgType() { |
|||
return SessionMsgType.POST_ATTRIBUTES_REQUEST; |
|||
} |
|||
|
|||
@Override |
|||
public Set<AttributeKvEntry> getAttributes() { |
|||
return data; |
|||
} |
|||
|
|||
@Override |
|||
public String toString() { |
|||
return "BasicAttributesUpdateRequest [data=" + data + "]"; |
|||
} |
|||
|
|||
} |
|||
@ -1,45 +0,0 @@ |
|||
/** |
|||
* Copyright © 2016-2018 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.common.msg.core; |
|||
|
|||
import org.thingsboard.server.common.msg.session.SessionMsgType; |
|||
import org.thingsboard.server.common.msg.session.SessionMsgType; |
|||
|
|||
public class BasicCommandAckResponse extends BasicResponseMsg<Integer> implements StatusCodeResponse { |
|||
|
|||
private static final long serialVersionUID = 1L; |
|||
|
|||
public static BasicCommandAckResponse onSuccess(SessionMsgType requestMsgType, Integer requestId) { |
|||
return BasicCommandAckResponse.onSuccess(requestMsgType, requestId, 200); |
|||
} |
|||
|
|||
public static BasicCommandAckResponse onSuccess(SessionMsgType requestMsgType, Integer requestId, Integer code) { |
|||
return new BasicCommandAckResponse(requestMsgType, requestId, true, null, code); |
|||
} |
|||
|
|||
public static BasicCommandAckResponse onError(SessionMsgType requestMsgType, Integer requestId, Exception error) { |
|||
return new BasicCommandAckResponse(requestMsgType, requestId, false, error, null); |
|||
} |
|||
|
|||
private BasicCommandAckResponse(SessionMsgType requestMsgType, Integer requestId, boolean success, Exception error, Integer code) { |
|||
super(requestMsgType, requestId, SessionMsgType.TO_DEVICE_RPC_RESPONSE_ACK, success, error, code); |
|||
} |
|||
|
|||
@Override |
|||
public String toString() { |
|||
return "BasicStatusCodeResponse []"; |
|||
} |
|||
} |
|||
@ -1,59 +0,0 @@ |
|||
/** |
|||
* Copyright © 2016-2018 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.common.msg.core; |
|||
|
|||
import lombok.ToString; |
|||
import org.thingsboard.server.common.msg.session.SessionMsgType; |
|||
import org.thingsboard.server.common.msg.session.SessionMsgType; |
|||
|
|||
import java.util.Collections; |
|||
import java.util.Optional; |
|||
import java.util.Set; |
|||
|
|||
@ToString |
|||
public class BasicGetAttributesRequest extends BasicRequest implements GetAttributesRequest { |
|||
|
|||
private static final long serialVersionUID = 1L; |
|||
|
|||
private final Set<String> clientKeys; |
|||
private final Set<String> sharedKeys; |
|||
|
|||
public BasicGetAttributesRequest(Integer requestId) { |
|||
this(requestId, Collections.emptySet(), Collections.emptySet()); |
|||
} |
|||
|
|||
public BasicGetAttributesRequest(Integer requestId, Set<String> clientKeys, Set<String> sharedKeys) { |
|||
super(requestId); |
|||
this.clientKeys = clientKeys; |
|||
this.sharedKeys = sharedKeys; |
|||
} |
|||
|
|||
@Override |
|||
public SessionMsgType getMsgType() { |
|||
return SessionMsgType.GET_ATTRIBUTES_REQUEST; |
|||
} |
|||
|
|||
@Override |
|||
public Optional<Set<String>> getClientAttributeNames() { |
|||
return Optional.ofNullable(clientKeys); |
|||
} |
|||
|
|||
@Override |
|||
public Optional<Set<String>> getSharedAttributeNames() { |
|||
return Optional.ofNullable(sharedKeys); |
|||
} |
|||
|
|||
} |
|||
@ -1,40 +0,0 @@ |
|||
/** |
|||
* Copyright © 2016-2018 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.common.msg.core; |
|||
|
|||
import lombok.ToString; |
|||
import org.thingsboard.server.common.msg.kv.AttributesKVMsg; |
|||
import org.thingsboard.server.common.msg.session.SessionMsgType; |
|||
import org.thingsboard.server.common.msg.session.SessionMsgType; |
|||
|
|||
@ToString |
|||
public class BasicGetAttributesResponse extends BasicResponseMsg<AttributesKVMsg> implements GetAttributesResponse { |
|||
|
|||
private static final long serialVersionUID = 1L; |
|||
|
|||
public static BasicGetAttributesResponse onSuccess(SessionMsgType requestMsgType, int requestId, AttributesKVMsg code) { |
|||
return new BasicGetAttributesResponse(requestMsgType, requestId, true, null, code); |
|||
} |
|||
|
|||
public static BasicGetAttributesResponse onError(SessionMsgType requestMsgType, int requestId, Exception error) { |
|||
return new BasicGetAttributesResponse(requestMsgType, requestId, false, error, null); |
|||
} |
|||
|
|||
private BasicGetAttributesResponse(SessionMsgType requestMsgType, int requestId, boolean success, Exception error, AttributesKVMsg code) { |
|||
super(requestMsgType, requestId, SessionMsgType.GET_ATTRIBUTES_RESPONSE, success, error, code); |
|||
} |
|||
|
|||
} |
|||
@ -1,79 +0,0 @@ |
|||
/** |
|||
* Copyright © 2016-2018 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.common.msg.core; |
|||
|
|||
import java.io.Serializable; |
|||
import java.util.Optional; |
|||
|
|||
import org.thingsboard.server.common.msg.session.SessionMsgType; |
|||
import org.thingsboard.server.common.msg.session.SessionMsgType; |
|||
|
|||
|
|||
public class BasicResponseMsg<T extends Serializable> implements ResponseMsg<T> { |
|||
|
|||
private static final long serialVersionUID = 1L; |
|||
|
|||
private final SessionMsgType requestMsgType; |
|||
private final Integer requestId; |
|||
private final SessionMsgType sessionMsgType; |
|||
private final boolean success; |
|||
private final T data; |
|||
private final Exception error; |
|||
|
|||
protected BasicResponseMsg(SessionMsgType requestMsgType, Integer requestId, SessionMsgType sessionMsgType, boolean success, Exception error, T data) { |
|||
super(); |
|||
this.requestMsgType = requestMsgType; |
|||
this.requestId = requestId; |
|||
this.sessionMsgType = sessionMsgType; |
|||
this.success = success; |
|||
this.error = error; |
|||
this.data = data; |
|||
} |
|||
|
|||
@Override |
|||
public SessionMsgType getRequestMsgType() { |
|||
return requestMsgType; |
|||
} |
|||
|
|||
@Override |
|||
public Integer getRequestId() { |
|||
return requestId; |
|||
} |
|||
|
|||
@Override |
|||
public boolean isSuccess() { |
|||
return success; |
|||
} |
|||
|
|||
@Override |
|||
public Optional<Exception> getError() { |
|||
return Optional.ofNullable(error); |
|||
} |
|||
|
|||
@Override |
|||
public Optional<T> getData() { |
|||
return Optional.ofNullable(data); |
|||
} |
|||
|
|||
@Override |
|||
public String toString() { |
|||
return "BasicResponseMsg [success=" + success + ", data=" + data + ", error=" + error + "]"; |
|||
} |
|||
|
|||
public SessionMsgType getSessionMsgType() { |
|||
return sessionMsgType; |
|||
} |
|||
} |
|||
@ -1,42 +0,0 @@ |
|||
/** |
|||
* Copyright © 2016-2018 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.common.msg.core; |
|||
|
|||
import lombok.ToString; |
|||
import org.thingsboard.server.common.msg.session.SessionMsgType; |
|||
import org.thingsboard.server.common.msg.session.SessionMsgType; |
|||
|
|||
@ToString |
|||
public class BasicStatusCodeResponse extends BasicResponseMsg<Integer> implements StatusCodeResponse { |
|||
|
|||
private static final long serialVersionUID = 1L; |
|||
|
|||
public static BasicStatusCodeResponse onSuccess(SessionMsgType requestMsgType, Integer requestId) { |
|||
return BasicStatusCodeResponse.onSuccess(requestMsgType, requestId, 0); |
|||
} |
|||
|
|||
public static BasicStatusCodeResponse onSuccess(SessionMsgType requestMsgType, Integer requestId, Integer code) { |
|||
return new BasicStatusCodeResponse(requestMsgType, requestId, true, null, code); |
|||
} |
|||
|
|||
public static BasicStatusCodeResponse onError(SessionMsgType requestMsgType, Integer requestId, Exception error) { |
|||
return new BasicStatusCodeResponse(requestMsgType, requestId, false, error, null); |
|||
} |
|||
|
|||
private BasicStatusCodeResponse(SessionMsgType requestMsgType, Integer requestId, boolean success, Exception error, Integer code) { |
|||
super(requestMsgType, requestId, SessionMsgType.STATUS_CODE_RESPONSE, success, error, code); |
|||
} |
|||
} |
|||
@ -1,66 +0,0 @@ |
|||
/** |
|||
* Copyright © 2016-2018 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.common.msg.core; |
|||
|
|||
import java.util.ArrayList; |
|||
import java.util.HashMap; |
|||
import java.util.List; |
|||
import java.util.Map; |
|||
|
|||
import org.thingsboard.server.common.data.kv.KvEntry; |
|||
import org.thingsboard.server.common.msg.session.SessionMsgType; |
|||
import org.thingsboard.server.common.msg.session.SessionMsgType; |
|||
|
|||
public class BasicTelemetryUploadRequest extends BasicRequest implements TelemetryUploadRequest { |
|||
|
|||
private static final long serialVersionUID = 1L; |
|||
|
|||
private final Map<Long, List<KvEntry>> data; |
|||
|
|||
public BasicTelemetryUploadRequest() { |
|||
this(DEFAULT_REQUEST_ID); |
|||
} |
|||
|
|||
public BasicTelemetryUploadRequest(Integer requestId) { |
|||
super(requestId); |
|||
this.data = new HashMap<>(); |
|||
} |
|||
|
|||
public void add(long ts, KvEntry entry) { |
|||
List<KvEntry> tsEntries = data.get(ts); |
|||
if (tsEntries == null) { |
|||
tsEntries = new ArrayList<>(); |
|||
data.put(ts, tsEntries); |
|||
} |
|||
tsEntries.add(entry); |
|||
} |
|||
|
|||
@Override |
|||
public SessionMsgType getMsgType() { |
|||
return SessionMsgType.POST_TELEMETRY_REQUEST; |
|||
} |
|||
|
|||
@Override |
|||
public Map<Long, List<KvEntry>> getData() { |
|||
return data; |
|||
} |
|||
|
|||
@Override |
|||
public String toString() { |
|||
return "BasicTelemetryUploadRequest [data=" + data + "]"; |
|||
} |
|||
|
|||
} |
|||
@ -1,29 +0,0 @@ |
|||
/** |
|||
* Copyright © 2016-2018 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.common.msg.core; |
|||
|
|||
import java.util.Optional; |
|||
import java.util.Set; |
|||
|
|||
import org.thingsboard.server.common.msg.session.FromDeviceMsg; |
|||
import org.thingsboard.server.common.msg.session.FromDeviceRequestMsg; |
|||
|
|||
public interface GetAttributesRequest extends FromDeviceRequestMsg { |
|||
|
|||
Optional<Set<String>> getClientAttributeNames(); |
|||
Optional<Set<String>> getSharedAttributeNames(); |
|||
|
|||
} |
|||
@ -1,22 +0,0 @@ |
|||
/** |
|||
* Copyright © 2016-2018 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.common.msg.core; |
|||
|
|||
import org.thingsboard.server.common.msg.kv.AttributesKVMsg; |
|||
|
|||
public interface GetAttributesResponse extends ResponseMsg<AttributesKVMsg> { |
|||
|
|||
} |
|||
@ -1,33 +0,0 @@ |
|||
/** |
|||
* Copyright © 2016-2018 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.common.msg.core; |
|||
|
|||
import java.io.Serializable; |
|||
import java.util.Optional; |
|||
|
|||
import org.thingsboard.server.common.msg.session.SessionMsgType; |
|||
import org.thingsboard.server.common.msg.session.ToDeviceMsg; |
|||
|
|||
public interface ResponseMsg<T extends Serializable> extends ToDeviceMsg { |
|||
|
|||
SessionMsgType getRequestMsgType(); |
|||
|
|||
Integer getRequestId(); |
|||
|
|||
Optional<Exception> getError(); |
|||
|
|||
Optional<T> getData(); |
|||
} |
|||
@ -1,30 +0,0 @@ |
|||
/** |
|||
* Copyright © 2016-2018 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.common.msg.core; |
|||
|
|||
import org.thingsboard.server.common.msg.session.FromDeviceMsg; |
|||
import org.thingsboard.server.common.msg.session.SessionMsgType; |
|||
import org.thingsboard.server.common.msg.session.SessionMsgType; |
|||
|
|||
/** |
|||
* @author Andrew Shvayka |
|||
*/ |
|||
public class RpcUnsubscribeMsg implements FromDeviceMsg { |
|||
@Override |
|||
public SessionMsgType getMsgType() { |
|||
return SessionMsgType.UNSUBSCRIBE_RPC_COMMANDS_REQUEST; |
|||
} |
|||
} |
|||
@ -1,53 +0,0 @@ |
|||
/** |
|||
* Copyright © 2016-2018 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.common.msg.core; |
|||
|
|||
import lombok.Data; |
|||
import org.thingsboard.server.common.msg.session.SessionMsgType; |
|||
import org.thingsboard.server.common.msg.session.SessionMsgType; |
|||
import org.thingsboard.server.common.msg.session.ToDeviceMsg; |
|||
|
|||
/** |
|||
* @author Andrew Shvayka |
|||
*/ |
|||
@Data |
|||
public class RuleEngineErrorMsg implements ToDeviceMsg { |
|||
|
|||
private final SessionMsgType inSessionMsgType; |
|||
private final RuleEngineError error; |
|||
|
|||
@Override |
|||
public boolean isSuccess() { |
|||
return false; |
|||
} |
|||
|
|||
public SessionMsgType getSessionMsgType() { |
|||
return SessionMsgType.RULE_ENGINE_ERROR; |
|||
} |
|||
|
|||
public String getErrorMsg() { |
|||
switch (error) { |
|||
case QUEUE_PUT_TIMEOUT: |
|||
return "Timeout during persistence of the message to the queue!"; |
|||
case SERVER_ERROR: |
|||
return "Error during processing of message by the server!"; |
|||
case TIMEOUT: |
|||
return "Timeout during processing of message by the server!"; |
|||
default: |
|||
throw new RuntimeException("Error " + error + " is not supported!"); |
|||
} |
|||
} |
|||
} |
|||
@ -1,38 +0,0 @@ |
|||
/** |
|||
* Copyright © 2016-2018 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.common.msg.core; |
|||
|
|||
import lombok.ToString; |
|||
import org.thingsboard.server.common.msg.kv.AttributesKVMsg; |
|||
import org.thingsboard.server.common.msg.session.SessionMsgType; |
|||
import org.thingsboard.server.common.msg.session.SessionMsgType; |
|||
import org.thingsboard.server.common.msg.session.ToDeviceMsg; |
|||
|
|||
@ToString |
|||
public class SessionCloseNotification implements ToDeviceMsg { |
|||
|
|||
private static final long serialVersionUID = 1L; |
|||
|
|||
@Override |
|||
public boolean isSuccess() { |
|||
return true; |
|||
} |
|||
|
|||
public SessionMsgType getSessionMsgType() { |
|||
return SessionMsgType.SESSION_CLOSE; |
|||
} |
|||
|
|||
} |
|||
@ -1,35 +0,0 @@ |
|||
/** |
|||
* Copyright © 2016-2018 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.common.msg.core; |
|||
|
|||
import lombok.Data; |
|||
import org.thingsboard.server.common.data.id.SessionId; |
|||
import org.thingsboard.server.common.msg.aware.SessionAwareMsg; |
|||
import org.thingsboard.server.common.msg.session.FromDeviceMsg; |
|||
import org.thingsboard.server.common.msg.session.SessionMsgType; |
|||
import org.thingsboard.server.common.msg.session.SessionMsgType; |
|||
|
|||
/** |
|||
* @author Andrew Shvayka |
|||
*/ |
|||
@Data |
|||
public class SessionOpenMsg implements FromDeviceMsg { |
|||
|
|||
@Override |
|||
public SessionMsgType getMsgType() { |
|||
return SessionMsgType.SESSION_OPEN; |
|||
} |
|||
} |
|||
@ -1,20 +0,0 @@ |
|||
/** |
|||
* Copyright © 2016-2018 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.common.msg.core; |
|||
|
|||
public interface StatusCodeResponse extends ResponseMsg<Integer>{ |
|||
|
|||
} |
|||
@ -1,29 +0,0 @@ |
|||
/** |
|||
* Copyright © 2016-2018 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.common.msg.core; |
|||
|
|||
import java.util.List; |
|||
import java.util.Map; |
|||
|
|||
import org.thingsboard.server.common.data.kv.KvEntry; |
|||
import org.thingsboard.server.common.msg.session.FromDeviceMsg; |
|||
import org.thingsboard.server.common.msg.session.FromDeviceRequestMsg; |
|||
|
|||
public interface TelemetryUploadRequest extends FromDeviceRequestMsg { |
|||
|
|||
Map<Long, List<KvEntry>> getData(); |
|||
|
|||
} |
|||
@ -1,41 +0,0 @@ |
|||
/** |
|||
* Copyright © 2016-2018 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.common.msg.core; |
|||
|
|||
import lombok.Data; |
|||
import org.thingsboard.server.common.msg.session.SessionMsgType; |
|||
import org.thingsboard.server.common.msg.session.SessionMsgType; |
|||
import org.thingsboard.server.common.msg.session.ToDeviceMsg; |
|||
|
|||
/** |
|||
* @author Andrew Shvayka |
|||
*/ |
|||
@Data |
|||
public class ToDeviceRpcRequestMsg implements ToDeviceMsg { |
|||
|
|||
private final int requestId; |
|||
private final String method; |
|||
private final String params; |
|||
|
|||
public SessionMsgType getSessionMsgType() { |
|||
return SessionMsgType.TO_DEVICE_RPC_REQUEST; |
|||
} |
|||
|
|||
@Override |
|||
public boolean isSuccess() { |
|||
return true; |
|||
} |
|||
} |
|||
@ -1,36 +0,0 @@ |
|||
/** |
|||
* Copyright © 2016-2018 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.common.msg.core; |
|||
|
|||
import lombok.Data; |
|||
import org.thingsboard.server.common.msg.session.FromDeviceMsg; |
|||
import org.thingsboard.server.common.msg.session.SessionMsgType; |
|||
import org.thingsboard.server.common.msg.session.SessionMsgType; |
|||
|
|||
/** |
|||
* @author Andrew Shvayka |
|||
*/ |
|||
@Data |
|||
public class ToDeviceRpcResponseMsg implements FromDeviceMsg { |
|||
|
|||
private final int requestId; |
|||
private final String data; |
|||
|
|||
@Override |
|||
public SessionMsgType getMsgType() { |
|||
return SessionMsgType.TO_DEVICE_RPC_RESPONSE; |
|||
} |
|||
} |
|||
@ -1,36 +0,0 @@ |
|||
/** |
|||
* Copyright © 2016-2018 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.common.msg.core; |
|||
|
|||
import lombok.Data; |
|||
import org.thingsboard.server.common.msg.session.FromDeviceRequestMsg; |
|||
import org.thingsboard.server.common.msg.session.SessionMsgType; |
|||
|
|||
/** |
|||
* @author Andrew Shvayka |
|||
*/ |
|||
@Data |
|||
public class ToServerRpcRequestMsg implements FromDeviceRequestMsg { |
|||
|
|||
private final Integer requestId; |
|||
private final String method; |
|||
private final String params; |
|||
|
|||
@Override |
|||
public SessionMsgType getMsgType() { |
|||
return SessionMsgType.TO_SERVER_RPC_REQUEST; |
|||
} |
|||
} |
|||
@ -1,107 +0,0 @@ |
|||
/** |
|||
* Copyright © 2016-2018 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.common.msg.device; |
|||
|
|||
import lombok.ToString; |
|||
import org.thingsboard.server.common.data.id.CustomerId; |
|||
import org.thingsboard.server.common.data.id.DeviceId; |
|||
import org.thingsboard.server.common.data.id.SessionId; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
import org.thingsboard.server.common.msg.MsgType; |
|||
import org.thingsboard.server.common.msg.cluster.ServerAddress; |
|||
import org.thingsboard.server.common.msg.session.FromDeviceMsg; |
|||
import org.thingsboard.server.common.msg.session.SessionType; |
|||
import org.thingsboard.server.common.msg.session.TransportToDeviceSessionActorMsg; |
|||
|
|||
import java.util.Optional; |
|||
|
|||
@ToString |
|||
public class BasicDeviceToDeviceActorMsg implements DeviceToDeviceActorMsg { |
|||
|
|||
private static final long serialVersionUID = -1866795134993115408L; |
|||
|
|||
private final TenantId tenantId; |
|||
private final CustomerId customerId; |
|||
private final DeviceId deviceId; |
|||
private final SessionId sessionId; |
|||
private final SessionType sessionType; |
|||
private final ServerAddress serverAddress; |
|||
private final FromDeviceMsg msg; |
|||
|
|||
public BasicDeviceToDeviceActorMsg(DeviceToDeviceActorMsg other, FromDeviceMsg msg) { |
|||
this(null, other.getTenantId(), other.getCustomerId(), other.getDeviceId(), other.getSessionId(), other.getSessionType(), msg); |
|||
} |
|||
|
|||
public BasicDeviceToDeviceActorMsg(TransportToDeviceSessionActorMsg msg, SessionType sessionType) { |
|||
this(null, msg.getTenantId(), msg.getCustomerId(), msg.getDeviceId(), msg.getSessionId(), sessionType, msg.getSessionMsg().getMsg()); |
|||
} |
|||
|
|||
private BasicDeviceToDeviceActorMsg(ServerAddress serverAddress, TenantId tenantId, CustomerId customerId, DeviceId deviceId, SessionId sessionId, SessionType sessionType, |
|||
FromDeviceMsg msg) { |
|||
super(); |
|||
this.serverAddress = serverAddress; |
|||
this.tenantId = tenantId; |
|||
this.customerId = customerId; |
|||
this.deviceId = deviceId; |
|||
this.sessionId = sessionId; |
|||
this.sessionType = sessionType; |
|||
this.msg = msg; |
|||
} |
|||
|
|||
@Override |
|||
public DeviceId getDeviceId() { |
|||
return deviceId; |
|||
} |
|||
|
|||
@Override |
|||
public CustomerId getCustomerId() { |
|||
return customerId; |
|||
} |
|||
|
|||
public TenantId getTenantId() { |
|||
return tenantId; |
|||
} |
|||
|
|||
@Override |
|||
public SessionId getSessionId() { |
|||
return sessionId; |
|||
} |
|||
|
|||
@Override |
|||
public SessionType getSessionType() { |
|||
return sessionType; |
|||
} |
|||
|
|||
@Override |
|||
public Optional<ServerAddress> getServerAddress() { |
|||
return Optional.ofNullable(serverAddress); |
|||
} |
|||
|
|||
@Override |
|||
public FromDeviceMsg getPayload() { |
|||
return msg; |
|||
} |
|||
|
|||
@Override |
|||
public DeviceToDeviceActorMsg toOtherAddress(ServerAddress otherAddress) { |
|||
return new BasicDeviceToDeviceActorMsg(otherAddress, tenantId, customerId, deviceId, sessionId, sessionType, msg); |
|||
} |
|||
|
|||
@Override |
|||
public MsgType getMsgType() { |
|||
return MsgType.DEVICE_SESSION_TO_DEVICE_ACTOR_MSG; |
|||
} |
|||
} |
|||
@ -1,41 +0,0 @@ |
|||
/** |
|||
* Copyright © 2016-2018 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.common.msg.device; |
|||
|
|||
import java.io.Serializable; |
|||
import java.util.Optional; |
|||
|
|||
import org.thingsboard.server.common.data.id.SessionId; |
|||
import org.thingsboard.server.common.msg.TbActorMsg; |
|||
import org.thingsboard.server.common.msg.aware.CustomerAwareMsg; |
|||
import org.thingsboard.server.common.msg.aware.DeviceAwareMsg; |
|||
import org.thingsboard.server.common.msg.aware.TenantAwareMsg; |
|||
import org.thingsboard.server.common.msg.cluster.ServerAddress; |
|||
import org.thingsboard.server.common.msg.session.FromDeviceMsg; |
|||
import org.thingsboard.server.common.msg.session.SessionType; |
|||
|
|||
public interface DeviceToDeviceActorMsg extends TbActorMsg, DeviceAwareMsg, CustomerAwareMsg, TenantAwareMsg, Serializable { |
|||
|
|||
SessionId getSessionId(); |
|||
|
|||
SessionType getSessionType(); |
|||
|
|||
Optional<ServerAddress> getServerAddress(); |
|||
|
|||
FromDeviceMsg getPayload(); |
|||
|
|||
DeviceToDeviceActorMsg toOtherAddress(ServerAddress otherAddress); |
|||
} |
|||
@ -1,22 +0,0 @@ |
|||
/** |
|||
* Copyright © 2016-2018 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.common.msg.session; |
|||
|
|||
public interface AdaptorToSessionActorMsg extends SessionMsg { |
|||
|
|||
FromDeviceMsg getMsg(); |
|||
|
|||
} |
|||
@ -1,32 +0,0 @@ |
|||
/** |
|||
* Copyright © 2016-2018 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.common.msg.session; |
|||
|
|||
public class BasicAdaptorToSessionActorMsg extends BasicSessionMsg implements AdaptorToSessionActorMsg { |
|||
|
|||
private final FromDeviceMsg msg; |
|||
|
|||
public BasicAdaptorToSessionActorMsg(SessionContext ctx, FromDeviceMsg msg) { |
|||
super(ctx); |
|||
this.msg = msg; |
|||
} |
|||
|
|||
@Override |
|||
public FromDeviceMsg getMsg() { |
|||
return msg; |
|||
} |
|||
|
|||
} |
|||
@ -1,32 +0,0 @@ |
|||
/** |
|||
* Copyright © 2016-2018 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.common.msg.session; |
|||
|
|||
public class BasicSessionActorToAdaptorMsg extends BasicSessionMsg implements SessionActorToAdaptorMsg { |
|||
|
|||
private final ToDeviceMsg msg; |
|||
|
|||
public BasicSessionActorToAdaptorMsg(SessionContext ctx, ToDeviceMsg msg) { |
|||
super(ctx); |
|||
this.msg = msg; |
|||
} |
|||
|
|||
@Override |
|||
public ToDeviceMsg getMsg() { |
|||
return msg; |
|||
} |
|||
|
|||
} |
|||
@ -1,44 +0,0 @@ |
|||
/** |
|||
* Copyright © 2016-2018 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.common.msg.session; |
|||
|
|||
import org.thingsboard.server.common.data.id.SessionId; |
|||
|
|||
public class BasicSessionMsg implements SessionMsg { |
|||
|
|||
private final SessionContext ctx; |
|||
|
|||
public BasicSessionMsg(SessionContext ctx) { |
|||
super(); |
|||
this.ctx = ctx; |
|||
} |
|||
|
|||
@Override |
|||
public SessionId getSessionId() { |
|||
return ctx.getSessionId(); |
|||
} |
|||
|
|||
@Override |
|||
public SessionContext getSessionContext() { |
|||
return ctx; |
|||
} |
|||
|
|||
@Override |
|||
public String toString() { |
|||
return "BasicSessionMsg [ctx=" + ctx + "]"; |
|||
} |
|||
|
|||
} |
|||
@ -1,74 +0,0 @@ |
|||
/** |
|||
* Copyright © 2016-2018 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.common.msg.session; |
|||
|
|||
import org.thingsboard.server.common.data.Device; |
|||
import org.thingsboard.server.common.data.id.CustomerId; |
|||
import org.thingsboard.server.common.data.id.DeviceId; |
|||
import org.thingsboard.server.common.data.id.SessionId; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
import org.thingsboard.server.common.msg.MsgType; |
|||
|
|||
public class BasicTransportToDeviceSessionActorMsg implements TransportToDeviceSessionActorMsg { |
|||
|
|||
private final TenantId tenantId; |
|||
private final CustomerId customerId; |
|||
private final DeviceId deviceId; |
|||
private final AdaptorToSessionActorMsg msg; |
|||
|
|||
public BasicTransportToDeviceSessionActorMsg(Device device, AdaptorToSessionActorMsg msg) { |
|||
super(); |
|||
this.tenantId = device.getTenantId(); |
|||
this.customerId = device.getCustomerId(); |
|||
this.deviceId = device.getId(); |
|||
this.msg = msg; |
|||
} |
|||
|
|||
@Override |
|||
public DeviceId getDeviceId() { |
|||
return deviceId; |
|||
} |
|||
|
|||
@Override |
|||
public CustomerId getCustomerId() { |
|||
return customerId; |
|||
} |
|||
|
|||
public TenantId getTenantId() { |
|||
return tenantId; |
|||
} |
|||
|
|||
@Override |
|||
public SessionId getSessionId() { |
|||
return msg.getSessionId(); |
|||
} |
|||
|
|||
@Override |
|||
public AdaptorToSessionActorMsg getSessionMsg() { |
|||
return msg; |
|||
} |
|||
|
|||
@Override |
|||
public String toString() { |
|||
return "BasicTransportToDeviceSessionActorMsg [tenantId=" + tenantId + ", customerId=" + customerId + ", deviceId=" + deviceId + ", msg=" + msg |
|||
+ "]"; |
|||
} |
|||
|
|||
@Override |
|||
public MsgType getMsgType() { |
|||
return MsgType.TRANSPORT_TO_DEVICE_SESSION_ACTOR_MSG; |
|||
} |
|||
} |
|||
@ -1,24 +0,0 @@ |
|||
/** |
|||
* Copyright © 2016-2018 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.common.msg.session; |
|||
|
|||
import java.io.Serializable; |
|||
|
|||
public interface FromDeviceMsg extends Serializable { |
|||
|
|||
SessionMsgType getMsgType(); |
|||
|
|||
} |
|||
@ -1,22 +0,0 @@ |
|||
/** |
|||
* Copyright © 2016-2018 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.common.msg.session; |
|||
|
|||
public interface SessionActorToAdaptorMsg extends SessionMsg { |
|||
|
|||
ToDeviceMsg getMsg(); |
|||
|
|||
} |
|||
@ -1,23 +0,0 @@ |
|||
/** |
|||
* Copyright © 2016-2018 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.common.msg.session; |
|||
|
|||
import org.thingsboard.server.common.msg.TbActorMsg; |
|||
import org.thingsboard.server.common.msg.aware.SessionAwareMsg; |
|||
|
|||
public interface SessionCtrlMsg extends SessionAwareMsg, TbActorMsg { |
|||
|
|||
} |
|||
@ -1,24 +0,0 @@ |
|||
/** |
|||
* Copyright © 2016-2018 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.common.msg.session; |
|||
|
|||
import org.thingsboard.server.common.msg.aware.SessionAwareMsg; |
|||
|
|||
public interface SessionMsg extends SessionAwareMsg { |
|||
|
|||
SessionContext getSessionContext(); |
|||
|
|||
} |
|||
@ -1,22 +0,0 @@ |
|||
/** |
|||
* Copyright © 2016-2018 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.common.msg.session; |
|||
|
|||
public enum SessionType { |
|||
|
|||
SYNC, ASYNC; |
|||
|
|||
} |
|||
@ -1,26 +0,0 @@ |
|||
/** |
|||
* Copyright © 2016-2018 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.common.msg.session; |
|||
|
|||
import java.io.Serializable; |
|||
|
|||
public interface ToDeviceMsg extends Serializable { |
|||
|
|||
boolean isSuccess(); |
|||
|
|||
SessionMsgType getSessionMsgType(); |
|||
|
|||
} |
|||
@ -1,28 +0,0 @@ |
|||
/** |
|||
* Copyright © 2016-2018 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.common.msg.session; |
|||
|
|||
import org.thingsboard.server.common.msg.TbActorMsg; |
|||
import org.thingsboard.server.common.msg.aware.CustomerAwareMsg; |
|||
import org.thingsboard.server.common.msg.aware.DeviceAwareMsg; |
|||
import org.thingsboard.server.common.msg.aware.SessionAwareMsg; |
|||
import org.thingsboard.server.common.msg.aware.TenantAwareMsg; |
|||
|
|||
public interface TransportToDeviceSessionActorMsg extends DeviceAwareMsg, CustomerAwareMsg, TenantAwareMsg, SessionAwareMsg, TbActorMsg { |
|||
|
|||
AdaptorToSessionActorMsg getSessionMsg(); |
|||
|
|||
} |
|||
@ -1,68 +0,0 @@ |
|||
/** |
|||
* Copyright © 2016-2018 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.common.msg.session.ctrl; |
|||
|
|||
import org.thingsboard.server.common.data.id.SessionId; |
|||
import org.thingsboard.server.common.msg.MsgType; |
|||
import org.thingsboard.server.common.msg.session.SessionCtrlMsg; |
|||
|
|||
public class SessionCloseMsg implements SessionCtrlMsg { |
|||
|
|||
private final SessionId sessionId; |
|||
private final boolean revoked; |
|||
private final boolean timeout; |
|||
|
|||
public static SessionCloseMsg onDisconnect(SessionId sessionId) { |
|||
return new SessionCloseMsg(sessionId, false, false); |
|||
} |
|||
|
|||
public static SessionCloseMsg onError(SessionId sessionId) { |
|||
return new SessionCloseMsg(sessionId, false, false); |
|||
} |
|||
|
|||
public static SessionCloseMsg onTimeout(SessionId sessionId) { |
|||
return new SessionCloseMsg(sessionId, false, true); |
|||
} |
|||
|
|||
public static SessionCloseMsg onCredentialsRevoked(SessionId sessionId) { |
|||
return new SessionCloseMsg(sessionId, true, false); |
|||
} |
|||
|
|||
private SessionCloseMsg(SessionId sessionId, boolean unauthorized, boolean timeout) { |
|||
super(); |
|||
this.sessionId = sessionId; |
|||
this.revoked = unauthorized; |
|||
this.timeout = timeout; |
|||
} |
|||
|
|||
@Override |
|||
public SessionId getSessionId() { |
|||
return sessionId; |
|||
} |
|||
|
|||
public boolean isCredentialsRevoked() { |
|||
return revoked; |
|||
} |
|||
|
|||
public boolean isTimeout() { |
|||
return timeout; |
|||
} |
|||
|
|||
@Override |
|||
public MsgType getMsgType() { |
|||
return MsgType.SESSION_CTRL_MSG; |
|||
} |
|||
} |
|||
@ -1,36 +0,0 @@ |
|||
/** |
|||
* Copyright © 2016-2018 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.common.msg.timeout; |
|||
|
|||
import org.thingsboard.server.common.msg.MsgType; |
|||
import org.thingsboard.server.common.msg.timeout.TimeoutMsg; |
|||
|
|||
import java.util.UUID; |
|||
|
|||
/** |
|||
* @author Andrew Shvayka |
|||
*/ |
|||
public final class DeviceActorQueueTimeoutMsg extends TimeoutMsg<UUID> { |
|||
|
|||
public DeviceActorQueueTimeoutMsg(UUID id, long timeout) { |
|||
super(id, timeout); |
|||
} |
|||
|
|||
@Override |
|||
public MsgType getMsgType() { |
|||
return MsgType.DEVICE_ACTOR_QUEUE_TIMEOUT_MSG; |
|||
} |
|||
} |
|||
@ -0,0 +1,50 @@ |
|||
/** |
|||
* Copyright © 2016-2018 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.kafka; |
|||
|
|||
import lombok.extern.slf4j.Slf4j; |
|||
|
|||
import java.nio.ByteBuffer; |
|||
import java.nio.charset.StandardCharsets; |
|||
import java.util.UUID; |
|||
|
|||
/** |
|||
* Created by ashvayka on 25.09.18. |
|||
*/ |
|||
@Slf4j |
|||
public abstract class AbstractTbKafkaTemplate { |
|||
protected byte[] uuidToBytes(UUID uuid) { |
|||
ByteBuffer buf = ByteBuffer.allocate(16); |
|||
buf.putLong(uuid.getMostSignificantBits()); |
|||
buf.putLong(uuid.getLeastSignificantBits()); |
|||
return buf.array(); |
|||
} |
|||
|
|||
protected static UUID bytesToUuid(byte[] bytes) { |
|||
ByteBuffer bb = ByteBuffer.wrap(bytes); |
|||
long firstLong = bb.getLong(); |
|||
long secondLong = bb.getLong(); |
|||
return new UUID(firstLong, secondLong); |
|||
} |
|||
|
|||
protected byte[] stringToBytes(String string) { |
|||
return string.getBytes(StandardCharsets.UTF_8); |
|||
} |
|||
|
|||
protected String bytesToString(byte[] data) { |
|||
return new String(data, StandardCharsets.UTF_8); |
|||
} |
|||
} |
|||
@ -0,0 +1,66 @@ |
|||
/** |
|||
* Copyright © 2016-2018 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.kafka; |
|||
|
|||
import com.google.common.util.concurrent.FutureCallback; |
|||
import com.google.common.util.concurrent.Futures; |
|||
import com.google.common.util.concurrent.ListenableFuture; |
|||
|
|||
import java.util.concurrent.Executor; |
|||
import java.util.concurrent.ScheduledExecutorService; |
|||
import java.util.concurrent.TimeUnit; |
|||
import java.util.function.Consumer; |
|||
|
|||
/** |
|||
* Created by ashvayka on 05.10.18. |
|||
*/ |
|||
public class AsyncCallbackTemplate { |
|||
|
|||
public static <T> void withCallbackAndTimeout(ListenableFuture<T> future, |
|||
Consumer<T> onSuccess, |
|||
Consumer<Throwable> onFailure, |
|||
long timeoutInMs, |
|||
ScheduledExecutorService timeoutExecutor, |
|||
Executor callbackExecutor) { |
|||
future = Futures.withTimeout(future, timeoutInMs, TimeUnit.MILLISECONDS, timeoutExecutor); |
|||
withCallback(future, onSuccess, onFailure, callbackExecutor); |
|||
} |
|||
|
|||
public static <T> void withCallback(ListenableFuture<T> future, Consumer<T> onSuccess, |
|||
Consumer<Throwable> onFailure, Executor executor) { |
|||
FutureCallback<T> callback = new FutureCallback<T>() { |
|||
@Override |
|||
public void onSuccess(T result) { |
|||
try { |
|||
onSuccess.accept(result); |
|||
} catch (Throwable th) { |
|||
onFailure(th); |
|||
} |
|||
} |
|||
|
|||
@Override |
|||
public void onFailure(Throwable t) { |
|||
onFailure.accept(t); |
|||
} |
|||
}; |
|||
if (executor != null) { |
|||
Futures.addCallback(future, callback, executor); |
|||
} else { |
|||
Futures.addCallback(future, callback); |
|||
} |
|||
} |
|||
|
|||
} |
|||
Some files were not shown because too many files changed in this diff
Loading…
Reference in new issue