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

BugFix for Gateway API and concurrent device creation

Browse Source
This commit is contained in:
Andrew Shvayka 2018-10-12 13:35:26 +03:00
parent 52ef95ac1b
commit c6800dbd03
10 changed files with 197 additions and 209 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
application/pom.xml
View File

@ -60,14 +60,14 @@
<groupId>org.thingsboard.common</groupId> <groupId>org.thingsboard.common</groupId>
<artifactId>transport</artifactId> <artifactId>transport</artifactId>
</dependency> </dependency>
<dependency> <!--<dependency>-->
<groupId>org.thingsboard.transport</groupId> <!--<groupId>org.thingsboard.transport</groupId>-->
<artifactId>http</artifactId> <!--<artifactId>http</artifactId>-->
</dependency> <!--</dependency>-->
<dependency> <!--<dependency>-->
<groupId>org.thingsboard.transport</groupId> <!--<groupId>org.thingsboard.transport</groupId>-->
<artifactId>coap</artifactId> <!--<artifactId>coap</artifactId>-->
</dependency> <!--</dependency>-->
<dependency> <dependency>
<groupId>org.thingsboard.transport</groupId> <groupId>org.thingsboard.transport</groupId>
<artifactId>mqtt-common</artifactId> <artifactId>mqtt-common</artifactId>

4
application/src/main/java/org/thingsboard/server/actors/device/DeviceActorMessageProcessor.java
View File

@ -455,6 +455,10 @@ public class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcesso
private void processSessionStateMsgs(SessionInfoProto sessionInfo, SessionEventMsg msg) { private void processSessionStateMsgs(SessionInfoProto sessionInfo, SessionEventMsg msg) {
UUID sessionId = getSessionId(sessionInfo); UUID sessionId = getSessionId(sessionInfo);
if (msg.getEvent() == SessionEvent.OPEN) { if (msg.getEvent() == SessionEvent.OPEN) {
if(sessions.containsKey(sessionId)){
logger.debug("[{}] Received duplicate session open event [{}]", deviceId, sessionId);
return;
}
logger.debug("[{}] Processing new session [{}]", deviceId, sessionId); logger.debug("[{}] Processing new session [{}]", deviceId, sessionId);
if (sessions.size() >= systemContext.getMaxConcurrentSessionsPerDevice()) { if (sessions.size() >= systemContext.getMaxConcurrentSessionsPerDevice()) {
UUID sessionIdToRemove = sessions.keySet().stream().findFirst().orElse(null); UUID sessionIdToRemove = sessions.keySet().stream().findFirst().orElse(null);

28
application/src/main/java/org/thingsboard/server/service/transport/RemoteTransportApiService.java
View File

@ -51,6 +51,7 @@ import javax.annotation.PostConstruct;
import java.util.UUID; import java.util.UUID;
import java.util.concurrent.ExecutorService; import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors; import java.util.concurrent.Executors;
import java.util.concurrent.locks.ReentrantLock;
/** /**
* Created by ashvayka on 05.10.18. * Created by ashvayka on 05.10.18.
@ -97,6 +98,8 @@ public class RemoteTransportApiService implements TransportApiService {
private TbKafkaResponseTemplate<TransportApiRequestMsg, TransportApiResponseMsg> transportApiTemplate; private TbKafkaResponseTemplate<TransportApiRequestMsg, TransportApiResponseMsg> transportApiTemplate;
private ReentrantLock deviceCreationLock = new ReentrantLock();
@PostConstruct @PostConstruct
public void init() { public void init() {
this.transportCallbackExecutor = Executors.newCachedThreadPool(); this.transportCallbackExecutor = Executors.newCachedThreadPool();
@ -156,23 +159,26 @@ public class RemoteTransportApiService implements TransportApiService {
DeviceId gatewayId = new DeviceId(new UUID(requestMsg.getGatewayIdMSB(), requestMsg.getGatewayIdLSB())); DeviceId gatewayId = new DeviceId(new UUID(requestMsg.getGatewayIdMSB(), requestMsg.getGatewayIdLSB()));
ListenableFuture<Device> gatewayFuture = deviceService.findDeviceByIdAsync(gatewayId); ListenableFuture<Device> gatewayFuture = deviceService.findDeviceByIdAsync(gatewayId);
return Futures.transform(gatewayFuture, gateway -> { return Futures.transform(gatewayFuture, gateway -> {
Device device = deviceService.findDeviceByTenantIdAndName(gateway.getTenantId(), gateway.getName()); deviceCreationLock.lock();
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);
}
try { 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() return TransportApiResponseMsg.newBuilder()
.setGetOrCreateDeviceResponseMsg(GetOrCreateDeviceFromGatewayResponseMsg.newBuilder().setDeviceInfo(getDeviceInfoProto(device)).build()).build(); .setGetOrCreateDeviceResponseMsg(GetOrCreateDeviceFromGatewayResponseMsg.newBuilder().setDeviceInfo(getDeviceInfoProto(device)).build()).build();
} catch (JsonProcessingException e) { } catch (JsonProcessingException e) {
log.warn("[{}] Failed to lookup device by gateway id and name", gatewayId, requestMsg.getDeviceName(), e); log.warn("[{}] Failed to lookup device by gateway id and name", gatewayId, requestMsg.getDeviceName(), e);
throw new RuntimeException(e); throw new RuntimeException(e);
} finally {
deviceCreationLock.unlock();
} }
}, transportCallbackExecutor); }, transportCallbackExecutor);
} }

32
transport/http/src/main/java/org/thingsboard/server/transport/http/DeviceApiController.java
View File

@ -122,7 +122,8 @@ public class DeviceApiController {
@RequestParam(value = "timeout", required = false, defaultValue = "0") long timeout, @RequestParam(value = "timeout", required = false, defaultValue = "0") long timeout,
HttpServletRequest request) { HttpServletRequest request) {
return subscribe(deviceToken, timeout, new RpcSubscribeMsg(), request); // return subscribe(deviceToken, timeout, new RpcSubscribeMsg(), request);
return null;
} }
@RequestMapping(value = "/{deviceToken}/rpc/{requestId}", method = RequestMethod.POST) @RequestMapping(value = "/{deviceToken}/rpc/{requestId}", method = RequestMethod.POST)
@ -174,15 +175,15 @@ public class DeviceApiController {
public DeferredResult<ResponseEntity> subscribeToAttributes(@PathVariable("deviceToken") String deviceToken, public DeferredResult<ResponseEntity> subscribeToAttributes(@PathVariable("deviceToken") String deviceToken,
@RequestParam(value = "timeout", required = false, defaultValue = "0") long timeout, @RequestParam(value = "timeout", required = false, defaultValue = "0") long timeout,
HttpServletRequest httpRequest) { HttpServletRequest httpRequest) {
return null;
return subscribe(deviceToken, timeout, new AttributesSubscribeMsg(), httpRequest); // return subscribe(deviceToken, timeout, new AttributesSubscribeMsg(), httpRequest);
} }
private DeferredResult<ResponseEntity> subscribe(String deviceToken, long timeout, FromDeviceMsg msg, HttpServletRequest httpRequest) { // private DeferredResult<ResponseEntity> subscribe(String deviceToken, long timeout, FromDeviceMsg msg, HttpServletRequest httpRequest) {
DeferredResult<ResponseEntity> responseWriter = new DeferredResult<ResponseEntity>(); // DeferredResult<ResponseEntity> responseWriter = new DeferredResult<ResponseEntity>();
if (quotaExceeded(httpRequest, responseWriter)) { // if (quotaExceeded(httpRequest, responseWriter)) {
return responseWriter; // return responseWriter;
} // }
// HttpSessionCtx ctx = getHttpSessionCtx(responseWriter, timeout); // HttpSessionCtx ctx = getHttpSessionCtx(responseWriter, timeout);
// if (ctx.login(new DeviceTokenCredentials(deviceToken))) { // if (ctx.login(new DeviceTokenCredentials(deviceToken))) {
// try { // try {
@ -193,21 +194,22 @@ public class DeviceApiController {
// } else { // } else {
// responseWriter.setResult(new ResponseEntity<>(HttpStatus.UNAUTHORIZED)); // responseWriter.setResult(new ResponseEntity<>(HttpStatus.UNAUTHORIZED));
// } // }
return responseWriter; // return responseWriter;
} // }
private HttpSessionCtx getHttpSessionCtx(DeferredResult<ResponseEntity> responseWriter) { private HttpSessionCtx getHttpSessionCtx(DeferredResult<ResponseEntity> responseWriter) {
return getHttpSessionCtx(responseWriter, defaultTimeout); return getHttpSessionCtx(responseWriter, defaultTimeout);
} }
private HttpSessionCtx getHttpSessionCtx(DeferredResult<ResponseEntity> responseWriter, long timeout) { private HttpSessionCtx getHttpSessionCtx(DeferredResult<ResponseEntity> responseWriter, long timeout) {
return new HttpSessionCtx(processor, authService, responseWriter, timeout != 0 ? timeout : defaultTimeout); return null;
// return new HttpSessionCtx(processor, authService, responseWriter, timeout != 0 ? timeout : defaultTimeout);
} }
private void process(HttpSessionCtx ctx, FromDeviceMsg request) { // private void process(HttpSessionCtx ctx, FromDeviceMsg request) {
AdaptorToSessionActorMsg msg = new BasicAdaptorToSessionActorMsg(ctx, request); // AdaptorToSessionActorMsg msg = new BasicAdaptorToSessionActorMsg(ctx, request);
// processor.process(new BasicTransportToDeviceSessionActorMsg(ctx.getDevice(), msg)); //// processor.process(new BasicTransportToDeviceSessionActorMsg(ctx.getDevice(), msg));
} // }
private boolean quotaExceeded(HttpServletRequest request, DeferredResult<ResponseEntity> responseWriter) { private boolean quotaExceeded(HttpServletRequest request, DeferredResult<ResponseEntity> responseWriter) {
if (quotaService.isQuotaExceeded(request.getRemoteAddr())) { if (quotaService.isQuotaExceeded(request.getRemoteAddr())) {

246
transport/http/src/main/java/org/thingsboard/server/transport/http/session/HttpSessionCtx.java
View File

@ -29,6 +29,7 @@ import org.thingsboard.server.common.transport.auth.DeviceAuthService;
import org.thingsboard.server.common.transport.session.DeviceAwareSessionContext; import org.thingsboard.server.common.transport.session.DeviceAwareSessionContext;
import java.util.Optional; import java.util.Optional;
import java.util.UUID;
import java.util.function.Consumer; import java.util.function.Consumer;
/** /**
@ -37,127 +38,136 @@ import java.util.function.Consumer;
@Slf4j @Slf4j
public class HttpSessionCtx extends DeviceAwareSessionContext { public class HttpSessionCtx extends DeviceAwareSessionContext {
private final SessionId sessionId; public HttpSessionCtx(UUID sessionId) {
private final long timeout; super(sessionId);
private final DeferredResult<ResponseEntity> responseWriter;
public HttpSessionCtx(SessionMsgProcessor processor, DeviceAuthService authService, DeferredResult<ResponseEntity> responseWriter, long timeout) {
super();
this.sessionId = new HttpSessionId();
this.responseWriter = responseWriter;
this.timeout = timeout;
} }
@Override @Override
public SessionType getSessionType() { public int nextMsgId() {
return SessionType.SYNC; return 0;
} }
@Override // private final SessionId sessionId;
public void onMsg(SessionActorToAdaptorMsg source) throws SessionException { // private final long timeout;
ToDeviceMsg msg = source.getMsg(); // private final DeferredResult<ResponseEntity> responseWriter;
switch (msg.getSessionMsgType()) { //
case GET_ATTRIBUTES_RESPONSE: // public HttpSessionCtx(SessionMsgProcessor processor, DeviceAuthService authService, DeferredResult<ResponseEntity> responseWriter, long timeout) {
reply((GetAttributesResponse) msg); // super();
return; // this.sessionId = new HttpSessionId();
case STATUS_CODE_RESPONSE: // this.responseWriter = responseWriter;
reply((StatusCodeResponse) msg); // this.timeout = timeout;
return; // }
case ATTRIBUTES_UPDATE_NOTIFICATION: //
reply((AttributesUpdateNotification) msg); // @Override
return; // public SessionType getSessionType() {
case TO_DEVICE_RPC_REQUEST: // return SessionType.SYNC;
reply((ToDeviceRpcRequestMsg) msg); // }
return; //
case TO_SERVER_RPC_RESPONSE: // @Override
reply((ToServerRpcResponseMsg) msg); // public void onMsg(SessionActorToAdaptorMsg source) throws SessionException {
return; // ToDeviceMsg msg = source.getMsg();
case RULE_ENGINE_ERROR: // switch (msg.getSessionMsgType()) {
reply((RuleEngineErrorMsg) msg); // case GET_ATTRIBUTES_RESPONSE:
return; // reply((GetAttributesResponse) msg);
default: // return;
break; // case STATUS_CODE_RESPONSE:
} // reply((StatusCodeResponse) msg);
} // return;
// case ATTRIBUTES_UPDATE_NOTIFICATION:
private void reply(RuleEngineErrorMsg msg) { // reply((AttributesUpdateNotification) msg);
HttpStatus status = HttpStatus.INTERNAL_SERVER_ERROR; // return;
switch (msg.getError()) { // case TO_DEVICE_RPC_REQUEST:
case QUEUE_PUT_TIMEOUT: // reply((ToDeviceRpcRequestMsg) msg);
status = HttpStatus.REQUEST_TIMEOUT; // return;
break; // case TO_SERVER_RPC_RESPONSE:
default: // reply((ToServerRpcResponseMsg) msg);
if (msg.getInSessionMsgType() == SessionMsgType.TO_SERVER_RPC_REQUEST) { // return;
status = HttpStatus.BAD_REQUEST; // case RULE_ENGINE_ERROR:
} // reply((RuleEngineErrorMsg) msg);
break; // return;
} // default:
responseWriter.setResult(new ResponseEntity<>(JsonConverter.toErrorJson(msg.getErrorMsg()).toString(), status)); // break;
} // }
// }
private <T> void reply(ResponseMsg<? extends T> msg, Consumer<T> f) { //
Optional<Exception> msgError = msg.getError(); // private void reply(RuleEngineErrorMsg msg) {
if (!msgError.isPresent()) { // HttpStatus status = HttpStatus.INTERNAL_SERVER_ERROR;
Optional<? extends T> msgData = msg.getData(); // switch (msg.getError()) {
if (msgData.isPresent()) { // case QUEUE_PUT_TIMEOUT:
f.accept(msgData.get()); // status = HttpStatus.REQUEST_TIMEOUT;
} // break;
} else { // default:
Exception e = msgError.get(); // if (msg.getInSessionMsgType() == SessionMsgType.TO_SERVER_RPC_REQUEST) {
responseWriter.setResult(new ResponseEntity<>(e.getMessage(), HttpStatus.INTERNAL_SERVER_ERROR)); // status = HttpStatus.BAD_REQUEST;
} // }
} // break;
// }
private void reply(ToDeviceRpcRequestMsg msg) { // responseWriter.setResult(new ResponseEntity<>(JsonConverter.toErrorJson(msg.getErrorMsg()).toString(), status));
responseWriter.setResult(new ResponseEntity<>(JsonConverter.toJson(msg, true).toString(), HttpStatus.OK)); // }
} //
// private <T> void reply(ResponseMsg<? extends T> msg, Consumer<T> f) {
private void reply(ToServerRpcResponseMsg msg) { // Optional<Exception> msgError = msg.getError();
// responseWriter.setResult(new ResponseEntity<>(JsonConverter.toJson(msg).toString(), HttpStatus.OK)); // if (!msgError.isPresent()) {
} // Optional<? extends T> msgData = msg.getData();
// if (msgData.isPresent()) {
private void reply(AttributesUpdateNotification msg) { // f.accept(msgData.get());
responseWriter.setResult(new ResponseEntity<>(JsonConverter.toJson(msg.getData(), false).toString(), HttpStatus.OK)); // }
} // } else {
// Exception e = msgError.get();
private void reply(GetAttributesResponse msg) { // responseWriter.setResult(new ResponseEntity<>(e.getMessage(), HttpStatus.INTERNAL_SERVER_ERROR));
reply(msg, payload -> { // }
if (payload.getClientAttributes().isEmpty() && payload.getSharedAttributes().isEmpty()) { // }
responseWriter.setResult(new ResponseEntity<>(HttpStatus.NOT_FOUND)); //
} else { // private void reply(ToDeviceRpcRequestMsg msg) {
JsonObject result = JsonConverter.toJson(payload, false); // responseWriter.setResult(new ResponseEntity<>(JsonConverter.toJson(msg, true).toString(), HttpStatus.OK));
responseWriter.setResult(new ResponseEntity<>(result.toString(), HttpStatus.OK)); // }
} //
}); // private void reply(ToServerRpcResponseMsg msg) {
} //// responseWriter.setResult(new ResponseEntity<>(JsonConverter.toJson(msg).toString(), HttpStatus.OK));
// }
private void reply(StatusCodeResponse msg) { //
reply(msg, payload -> { // private void reply(AttributesUpdateNotification msg) {
if (payload == 0) { // responseWriter.setResult(new ResponseEntity<>(JsonConverter.toJson(msg.getData(), false).toString(), HttpStatus.OK));
responseWriter.setResult(new ResponseEntity<>(HttpStatus.OK)); // }
} else { //
responseWriter.setResult(new ResponseEntity<>(HttpStatus.valueOf(payload))); // private void reply(GetAttributesResponse msg) {
} // reply(msg, payload -> {
}); // if (payload.getClientAttributes().isEmpty() && payload.getSharedAttributes().isEmpty()) {
} // responseWriter.setResult(new ResponseEntity<>(HttpStatus.NOT_FOUND));
// } else {
@Override // JsonObject result = JsonConverter.toJson(payload, false);
public void onMsg(SessionCtrlMsg msg) throws SessionException { // responseWriter.setResult(new ResponseEntity<>(result.toString(), HttpStatus.OK));
//Do nothing // }
} // });
// }
@Override //
public boolean isClosed() { // private void reply(StatusCodeResponse msg) {
return false; // reply(msg, payload -> {
} // if (payload == 0) {
// responseWriter.setResult(new ResponseEntity<>(HttpStatus.OK));
@Override // } else {
public long getTimeout() { // responseWriter.setResult(new ResponseEntity<>(HttpStatus.valueOf(payload)));
return timeout; // }
} // });
// }
@Override //
public SessionId getSessionId() { // @Override
return sessionId; // public void onMsg(SessionCtrlMsg msg) throws SessionException {
} // //Do nothing
// }
//
// @Override
// public boolean isClosed() {
// return false;
// }
//
// @Override
// public long getTimeout() {
// return timeout;
// }
//
// @Override
// public SessionId getSessionId() {
// return sessionId;
// }
} }

35
transport/http/src/main/java/org/thingsboard/server/transport/http/session/HttpSessionId.java
View File

@ -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.transport.http.session;
import java.util.UUID;
/**
* @author Andrew Shvayka
*/
public class HttpSessionId implements SessionId {
private final UUID id;
public HttpSessionId() {
this.id = UUID.randomUUID();
}
@Override
public String toUidStr() {
return id.toString();
}
}

24
transport/mqtt-common/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java
View File

@ -54,7 +54,7 @@ import org.thingsboard.server.gen.transport.TransportProtos.ValidateDeviceTokenR
import org.thingsboard.server.gen.transport.TransportProtos.ValidateDeviceX509CertRequestMsg; import org.thingsboard.server.gen.transport.TransportProtos.ValidateDeviceX509CertRequestMsg;
import org.thingsboard.server.transport.mqtt.adaptors.MqttTransportAdaptor; import org.thingsboard.server.transport.mqtt.adaptors.MqttTransportAdaptor;
import org.thingsboard.server.transport.mqtt.session.DeviceSessionCtx; import org.thingsboard.server.transport.mqtt.session.DeviceSessionCtx;
import org.thingsboard.server.transport.mqtt.session.GatewaySessionCtx; import org.thingsboard.server.transport.mqtt.session.GatewaySessionHandler;
import org.thingsboard.server.transport.mqtt.util.SslUtil; import org.thingsboard.server.transport.mqtt.util.SslUtil;
import javax.net.ssl.SSLPeerUnverifiedException; import javax.net.ssl.SSLPeerUnverifiedException;
@ -98,7 +98,7 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement
private volatile SessionInfoProto sessionInfo; private volatile SessionInfoProto sessionInfo;
private volatile InetSocketAddress address; private volatile InetSocketAddress address;
private volatile DeviceSessionCtx deviceSessionCtx; private volatile DeviceSessionCtx deviceSessionCtx;
private volatile GatewaySessionCtx gatewaySessionCtx; private volatile GatewaySessionHandler gatewaySessionHandler;
MqttTransportHandler(MqttTransportContext context) { MqttTransportHandler(MqttTransportContext context) {
this.sessionId = UUID.randomUUID(); this.sessionId = UUID.randomUUID();
@ -175,7 +175,7 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement
log.trace("[{}] Processing publish msg [{}][{}]!", sessionId, topicName, msgId); log.trace("[{}] Processing publish msg [{}][{}]!", sessionId, topicName, msgId);
if (topicName.startsWith(MqttTopics.BASE_GATEWAY_API_TOPIC)) { if (topicName.startsWith(MqttTopics.BASE_GATEWAY_API_TOPIC)) {
if (gatewaySessionCtx != null) { if (gatewaySessionHandler != null) {
handleGatewayPublishMsg(topicName, msgId, mqttMsg); handleGatewayPublishMsg(topicName, msgId, mqttMsg);
} }
} else { } else {
@ -187,22 +187,22 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement
try { try {
switch (topicName) { switch (topicName) {
case MqttTopics.GATEWAY_TELEMETRY_TOPIC: case MqttTopics.GATEWAY_TELEMETRY_TOPIC:
gatewaySessionCtx.onDeviceTelemetry(mqttMsg); gatewaySessionHandler.onDeviceTelemetry(mqttMsg);
break; break;
case MqttTopics.GATEWAY_ATTRIBUTES_TOPIC: case MqttTopics.GATEWAY_ATTRIBUTES_TOPIC:
gatewaySessionCtx.onDeviceAttributes(mqttMsg); gatewaySessionHandler.onDeviceAttributes(mqttMsg);
break; break;
case MqttTopics.GATEWAY_ATTRIBUTES_REQUEST_TOPIC: case MqttTopics.GATEWAY_ATTRIBUTES_REQUEST_TOPIC:
gatewaySessionCtx.onDeviceAttributesRequest(mqttMsg); gatewaySessionHandler.onDeviceAttributesRequest(mqttMsg);
break; break;
case MqttTopics.GATEWAY_RPC_TOPIC: case MqttTopics.GATEWAY_RPC_TOPIC:
gatewaySessionCtx.onDeviceRpcResponse(mqttMsg); gatewaySessionHandler.onDeviceRpcResponse(mqttMsg);
break; break;
case MqttTopics.GATEWAY_CONNECT_TOPIC: case MqttTopics.GATEWAY_CONNECT_TOPIC:
gatewaySessionCtx.onDeviceConnect(mqttMsg); gatewaySessionHandler.onDeviceConnect(mqttMsg);
break; break;
case MqttTopics.GATEWAY_DISCONNECT_TOPIC: case MqttTopics.GATEWAY_DISCONNECT_TOPIC:
gatewaySessionCtx.onDeviceDisconnect(mqttMsg); gatewaySessionHandler.onDeviceDisconnect(mqttMsg);
break; break;
} }
} catch (RuntimeException | AdaptorException e) { } catch (RuntimeException | AdaptorException e) {
@ -405,8 +405,8 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement
if (deviceSessionCtx.isConnected()) { if (deviceSessionCtx.isConnected()) {
transportService.process(sessionInfo, getSessionEventMsg(SessionEvent.CLOSED), null); transportService.process(sessionInfo, getSessionEventMsg(SessionEvent.CLOSED), null);
transportService.deregisterSession(sessionInfo); transportService.deregisterSession(sessionInfo);
if (gatewaySessionCtx != null) { if (gatewaySessionHandler != null) {
gatewaySessionCtx.onGatewayDisconnect(); gatewaySessionHandler.onGatewayDisconnect();
} }
} }
} }
@ -467,7 +467,7 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement
if (infoNode != null) { if (infoNode != null) {
JsonNode gatewayNode = infoNode.get("gateway"); JsonNode gatewayNode = infoNode.get("gateway");
if (gatewayNode != null && gatewayNode.asBoolean()) { if (gatewayNode != null && gatewayNode.asBoolean()) {
gatewaySessionCtx = new GatewaySessionCtx(context, deviceSessionCtx, sessionId); gatewaySessionHandler = new GatewaySessionHandler(context, deviceSessionCtx, sessionId);
} }
} }
} catch (IOException e) { } catch (IOException e) {

4
transport/mqtt-common/src/main/java/org/thingsboard/server/transport/mqtt/session/GatewayDeviceSessionCtx.java
View File

@ -30,10 +30,10 @@ import java.util.concurrent.ConcurrentMap;
@Slf4j @Slf4j
public class GatewayDeviceSessionCtx extends MqttDeviceAwareSessionContext implements SessionMsgListener { public class GatewayDeviceSessionCtx extends MqttDeviceAwareSessionContext implements SessionMsgListener {
private final GatewaySessionCtx parent; private final GatewaySessionHandler parent;
private final SessionInfoProto sessionInfo; private final SessionInfoProto sessionInfo;
public GatewayDeviceSessionCtx(GatewaySessionCtx parent, DeviceInfoProto deviceInfo, ConcurrentMap<String, Integer> mqttQoSMap) { public GatewayDeviceSessionCtx(GatewaySessionHandler parent, DeviceInfoProto deviceInfo, ConcurrentMap<String, Integer> mqttQoSMap) {
super(UUID.randomUUID(), mqttQoSMap); super(UUID.randomUUID(), mqttQoSMap);
this.parent = parent; this.parent = parent;
this.sessionInfo = SessionInfoProto.newBuilder() this.sessionInfo = SessionInfoProto.newBuilder()

15
transport/mqtt-common/src/main/java/org/thingsboard/server/transport/mqtt/session/GatewaySessionCtx.java → transport/mqtt-common/src/main/java/org/thingsboard/server/transport/mqtt/session/GatewaySessionHandler.java
View File

@ -52,13 +52,12 @@ import java.util.Set;
import java.util.UUID; import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap; import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicInteger;
/** /**
* Created by ashvayka on 19.01.17. * Created by ashvayka on 19.01.17.
*/ */
@Slf4j @Slf4j
public class GatewaySessionCtx { public class GatewaySessionHandler {
private static final String DEFAULT_DEVICE_TYPE = "default"; private static final String DEFAULT_DEVICE_TYPE = "default";
private static final String CAN_T_PARSE_VALUE = "Can't parse value: "; private static final String CAN_T_PARSE_VALUE = "Can't parse value: ";
@ -73,7 +72,7 @@ public class GatewaySessionCtx {
private final ChannelHandlerContext channel; private final ChannelHandlerContext channel;
private final DeviceSessionCtx deviceSessionCtx; private final DeviceSessionCtx deviceSessionCtx;
public GatewaySessionCtx(MqttTransportContext context, DeviceSessionCtx deviceSessionCtx, UUID sessionId) { public GatewaySessionHandler(MqttTransportContext context, DeviceSessionCtx deviceSessionCtx, UUID sessionId) {
this.context = context; this.context = context;
this.transportService = context.getTransportService(); this.transportService = context.getTransportService();
this.deviceSessionCtx = deviceSessionCtx; this.deviceSessionCtx = deviceSessionCtx;
@ -114,10 +113,12 @@ public class GatewaySessionCtx {
new TransportServiceCallback<GetOrCreateDeviceFromGatewayResponseMsg>() { new TransportServiceCallback<GetOrCreateDeviceFromGatewayResponseMsg>() {
@Override @Override
public void onSuccess(GetOrCreateDeviceFromGatewayResponseMsg msg) { public void onSuccess(GetOrCreateDeviceFromGatewayResponseMsg msg) {
GatewayDeviceSessionCtx deviceSessionCtx = new GatewayDeviceSessionCtx(GatewaySessionCtx.this, msg.getDeviceInfo(), mqttQoSMap); GatewayDeviceSessionCtx deviceSessionCtx = new GatewayDeviceSessionCtx(GatewaySessionHandler.this, msg.getDeviceInfo(), mqttQoSMap);
if (devices.putIfAbsent(deviceName, deviceSessionCtx) == null) { if (devices.putIfAbsent(deviceName, deviceSessionCtx) == null) {
SessionInfoProto deviceSessionInfo = deviceSessionCtx.getSessionInfo(); SessionInfoProto deviceSessionInfo = deviceSessionCtx.getSessionInfo();
transportService.process(deviceSessionInfo, MqttTransportHandler.getSessionEventMsg(TransportProtos.SessionEvent.OPEN), null); transportService.process(deviceSessionInfo, MqttTransportHandler.getSessionEventMsg(TransportProtos.SessionEvent.OPEN), null);
transportService.process(deviceSessionInfo, TransportProtos.SubscribeToRPCMsg.getDefaultInstance(), null);
transportService.process(deviceSessionInfo, TransportProtos.SubscribeToAttributeUpdatesMsg.getDefaultInstance(), null);
transportService.registerSession(deviceSessionInfo, deviceSessionCtx); transportService.registerSession(deviceSessionInfo, deviceSessionCtx);
} }
future.set(devices.get(deviceName)); future.set(devices.get(deviceName));
@ -203,7 +204,7 @@ public class GatewaySessionCtx {
@Override @Override
public void onFailure(Throwable t) { public void onFailure(Throwable t) {
log.debug("[{}] Failed to process device teleemtry command: {}", sessionId, deviceName, t); log.debug("[{}] Failed to process device attributes command: {}", sessionId, deviceName, t);
} }
}, context.getExecutor()); }, context.getExecutor());
} }
@ -264,8 +265,8 @@ public class GatewaySessionCtx {
} else { } else {
result.addAllSharedAttributeNames(keys); result.addAllSharedAttributeNames(keys);
} }
int msgId = msg.variableHeader().packetId();
TransportProtos.GetAttributeRequestMsg requestMsg = result.build(); TransportProtos.GetAttributeRequestMsg requestMsg = result.build();
int msgId = msg.variableHeader().packetId();
Futures.addCallback(checkDeviceConnected(deviceName), Futures.addCallback(checkDeviceConnected(deviceName),
new FutureCallback<GatewayDeviceSessionCtx>() { new FutureCallback<GatewayDeviceSessionCtx>() {
@Override @Override
@ -275,7 +276,7 @@ public class GatewaySessionCtx {
@Override @Override
public void onFailure(Throwable t) { public void onFailure(Throwable t) {
log.debug("[{}] Failed to process device teleemtry command: {}", sessionId, deviceName, t); log.debug("[{}] Failed to process device attributes request command: {}", sessionId, deviceName, t);
} }
}, context.getExecutor()); }, context.getExecutor());
ack(msg); ack(msg);

2
transport/pom.xml
View File

@ -35,7 +35,7 @@
</properties> </properties>
<modules> <modules>
<!--<module>http</module>--> <module>http</module>
<!--<module>coap</module>--> <!--<module>coap</module>-->
<module>mqtt-common</module> <module>mqtt-common</module>
<module>mqtt-transport</module> <module>mqtt-transport</module>