diff --git a/application/src/main/java/org/thingsboard/server/actors/device/DeviceActorMessageProcessor.java b/application/src/main/java/org/thingsboard/server/actors/device/DeviceActorMessageProcessor.java
index c590104665..1cfdcd202a 100644
--- a/application/src/main/java/org/thingsboard/server/actors/device/DeviceActorMessageProcessor.java
+++ b/application/src/main/java/org/thingsboard/server/actors/device/DeviceActorMessageProcessor.java
@@ -411,29 +411,6 @@ public class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcesso
}
}
-// private void processRpcResponses(ActorContext context, DeviceToDeviceActorMsg msg) {
-// SessionId sessionId = msg.getSessionId();
-// FromDeviceMsg inMsg = msg.getPayload();
-// if (inMsg.getMsgType() == SessionMsgType.TO_DEVICE_RPC_RESPONSE) {
-// logger.debug("[{}] Processing rpc command response [{}]", deviceId, sessionId);
-// ToDeviceRpcResponseMsg responseMsg = (ToDeviceRpcResponseMsg) inMsg;
-// ToDeviceRpcRequestMetadata requestMd = toDeviceRpcPendingMap.remove(responseMsg.getRequestId());
-// boolean success = requestMd != null;
-// if (success) {
-// systemContext.getDeviceRpcService().processRpcResponseFromDevice(new FromDeviceRpcResponse(requestMd.getMsg().getMsg().getId(),
-// requestMd.getMsg().getServerAddress(), responseMsg.getData(), null));
-// } else {
-// logger.debug("[{}] Rpc command response [{}] is stale!", deviceId, responseMsg.getRequestId());
-// }
-// if (msg.getSessionType() == SessionType.SYNC) {
-// BasicCommandAckResponse response = success
-// ? BasicCommandAckResponse.onSuccess(SessionMsgType.TO_DEVICE_RPC_REQUEST, responseMsg.getRequestId())
-// : BasicCommandAckResponse.onError(SessionMsgType.TO_DEVICE_RPC_REQUEST, responseMsg.getRequestId(), new TimeoutException());
-// sendMsgToSessionActor(new BasicActorSystemToDeviceSessionActorMsg(response, msg.getSessionId()), msg.getServerAddress());
-// }
-// }
-// }
-
void processClusterEventMsg(ClusterEventMsg msg) {
// if (!msg.isAdded()) {
// logger.debug("[{}] Clearing attributes/rpc subscription for server [{}]", deviceId, msg.getServerAddress());
diff --git a/application/src/main/java/org/thingsboard/server/actors/shared/SessionTimeoutMsg.java b/application/src/main/java/org/thingsboard/server/actors/shared/SessionTimeoutMsg.java
index d015fe01de..d89848b8a8 100644
--- a/application/src/main/java/org/thingsboard/server/actors/shared/SessionTimeoutMsg.java
+++ b/application/src/main/java/org/thingsboard/server/actors/shared/SessionTimeoutMsg.java
@@ -16,7 +16,6 @@
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;
diff --git a/application/src/main/java/org/thingsboard/server/service/transport/RemoteTransportApiService.java b/application/src/main/java/org/thingsboard/server/service/transport/RemoteTransportApiService.java
index 6e9b2efa0a..5bcd8ac888 100644
--- a/application/src/main/java/org/thingsboard/server/service/transport/RemoteTransportApiService.java
+++ b/application/src/main/java/org/thingsboard/server/service/transport/RemoteTransportApiService.java
@@ -1,12 +1,12 @@
/**
* 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
- *
+ *
+ * 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.
@@ -24,11 +24,16 @@ 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.GetOrCreateDeviceFromGatewayRequestMsg;
+import org.thingsboard.server.gen.transport.TransportProtos.GetOrCreateDeviceFromGatewayResponseMsg;
import org.thingsboard.server.gen.transport.TransportProtos.DeviceInfoProto;
import org.thingsboard.server.gen.transport.TransportProtos.TransportApiRequestMsg;
import org.thingsboard.server.gen.transport.TransportProtos.TransportApiResponseMsg;
@@ -40,8 +45,10 @@ 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.state.DeviceStateService;
import javax.annotation.PostConstruct;
+import java.util.UUID;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
@@ -77,9 +84,15 @@ public class RemoteTransportApiService implements TransportApiService {
@Autowired
private DeviceService deviceService;
+ @Autowired
+ private RelationService relationService;
+
@Autowired
private DeviceCredentialsService deviceCredentialsService;
+ @Autowired
+ private DeviceStateService deviceStateService;
+
private ExecutorService transportCallbackExecutor;
private TbKafkaResponseTemplate transportApiTemplate;
@@ -123,6 +136,8 @@ public class RemoteTransportApiService implements TransportApiService {
} 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();
}
@@ -137,6 +152,32 @@ public class RemoteTransportApiService implements TransportApiService {
}
}
+ private ListenableFuture handle(GetOrCreateDeviceFromGatewayRequestMsg requestMsg) {
+ DeviceId gatewayId = new DeviceId(new UUID(requestMsg.getGatewayIdMSB(), requestMsg.getGatewayIdLSB()));
+ ListenableFuture gatewayFuture = deviceService.findDeviceByIdAsync(gatewayId);
+ return Futures.transform(gatewayFuture, gateway -> {
+ Device device = deviceService.findDeviceByTenantIdAndName(gateway.getTenantId(), gateway.getName());
+ 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 {
+ 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);
+ }
+ }, transportCallbackExecutor);
+ }
+
+
private ListenableFuture getDeviceInfo(DeviceId deviceId) {
return Futures.transform(deviceService.findDeviceByIdAsync(deviceId), device -> {
if (device == null) {
@@ -144,17 +185,8 @@ public class RemoteTransportApiService implements TransportApiService {
return getEmptyTransportApiResponse();
}
try {
- DeviceInfoProto deviceInfoProto = 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();
return TransportApiResponseMsg.newBuilder()
- .setValidateTokenResponseMsg(ValidateDeviceCredentialsResponseMsg.newBuilder().setDeviceInfo(deviceInfoProto).build()).build();
+ .setValidateTokenResponseMsg(ValidateDeviceCredentialsResponseMsg.newBuilder().setDeviceInfo(getDeviceInfoProto(device)).build()).build();
} catch (JsonProcessingException e) {
log.warn("[{}] Failed to lookup device by id", deviceId, e);
return getEmptyTransportApiResponse();
@@ -162,6 +194,18 @@ public class RemoteTransportApiService implements TransportApiService {
});
}
+ 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 getEmptyTransportApiResponseFuture() {
return Futures.immediateFuture(getEmptyTransportApiResponse());
}
diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/id/SessionId.java b/common/data/src/main/java/org/thingsboard/server/common/data/id/SessionId.java
deleted file mode 100644
index 1b4772768f..0000000000
--- a/common/data/src/main/java/org/thingsboard/server/common/data/id/SessionId.java
+++ /dev/null
@@ -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.data.id;
-
-import java.io.Serializable;
-
-public interface SessionId extends Serializable {
-
- String toUidStr();
-
-}
diff --git a/common/message/src/main/java/org/thingsboard/server/common/msg/aware/SessionAwareMsg.java b/common/message/src/main/java/org/thingsboard/server/common/msg/aware/SessionAwareMsg.java
index db6ff6f632..1c258239d4 100644
--- a/common/message/src/main/java/org/thingsboard/server/common/msg/aware/SessionAwareMsg.java
+++ b/common/message/src/main/java/org/thingsboard/server/common/msg/aware/SessionAwareMsg.java
@@ -15,8 +15,6 @@
*/
package org.thingsboard.server.common.msg.aware;
-import org.thingsboard.server.common.data.id.SessionId;
-
public interface SessionAwareMsg {
SessionId getSessionId();
diff --git a/common/message/src/main/java/org/thingsboard/server/common/msg/core/SessionOpenMsg.java b/common/message/src/main/java/org/thingsboard/server/common/msg/core/SessionOpenMsg.java
index 28e319e525..d8af5a953c 100644
--- a/common/message/src/main/java/org/thingsboard/server/common/msg/core/SessionOpenMsg.java
+++ b/common/message/src/main/java/org/thingsboard/server/common/msg/core/SessionOpenMsg.java
@@ -16,11 +16,8 @@
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
diff --git a/common/message/src/main/java/org/thingsboard/server/common/msg/session/BasicSessionMsg.java b/common/message/src/main/java/org/thingsboard/server/common/msg/session/BasicSessionMsg.java
index 855cce53d7..84c5052662 100644
--- a/common/message/src/main/java/org/thingsboard/server/common/msg/session/BasicSessionMsg.java
+++ b/common/message/src/main/java/org/thingsboard/server/common/msg/session/BasicSessionMsg.java
@@ -15,8 +15,6 @@
*/
package org.thingsboard.server.common.msg.session;
-import org.thingsboard.server.common.data.id.SessionId;
-
public class BasicSessionMsg implements SessionMsg {
private final SessionContext ctx;
diff --git a/common/message/src/main/java/org/thingsboard/server/common/msg/session/BasicTransportToDeviceSessionActorMsg.java b/common/message/src/main/java/org/thingsboard/server/common/msg/session/BasicTransportToDeviceSessionActorMsg.java
index 0704e6e1bf..4a81fd4578 100644
--- a/common/message/src/main/java/org/thingsboard/server/common/msg/session/BasicTransportToDeviceSessionActorMsg.java
+++ b/common/message/src/main/java/org/thingsboard/server/common/msg/session/BasicTransportToDeviceSessionActorMsg.java
@@ -18,7 +18,6 @@ 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;
diff --git a/common/message/src/main/java/org/thingsboard/server/common/msg/session/SessionContext.java b/common/message/src/main/java/org/thingsboard/server/common/msg/session/SessionContext.java
index 2f2d88047e..6607295ee3 100644
--- a/common/message/src/main/java/org/thingsboard/server/common/msg/session/SessionContext.java
+++ b/common/message/src/main/java/org/thingsboard/server/common/msg/session/SessionContext.java
@@ -1,12 +1,12 @@
/**
* 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
- *
+ *
+ * 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.
@@ -15,21 +15,11 @@
*/
package org.thingsboard.server.common.msg.session;
-import org.thingsboard.server.common.data.id.SessionId;
-import org.thingsboard.server.common.msg.session.ex.SessionException;
+import java.util.UUID;
public interface SessionContext {
- SessionId getSessionId();
-
- SessionType getSessionType();
-
- void onMsg(SessionActorToAdaptorMsg msg) throws SessionException;
-
- void onMsg(SessionCtrlMsg msg) throws SessionException;
-
- boolean isClosed();
-
- long getTimeout();
+ UUID getSessionId();
+ int nextMsgId();
}
diff --git a/common/message/src/main/java/org/thingsboard/server/common/msg/session/ctrl/SessionCloseMsg.java b/common/message/src/main/java/org/thingsboard/server/common/msg/session/ctrl/SessionCloseMsg.java
index 49fad13236..22a13d03b5 100644
--- a/common/message/src/main/java/org/thingsboard/server/common/msg/session/ctrl/SessionCloseMsg.java
+++ b/common/message/src/main/java/org/thingsboard/server/common/msg/session/ctrl/SessionCloseMsg.java
@@ -15,7 +15,6 @@
*/
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;
diff --git a/common/transport/src/main/java/org/thingsboard/server/common/transport/TransportAdaptor.java b/common/transport/src/main/java/org/thingsboard/server/common/transport/TransportAdaptor.java
index 86067d536e..b59d49362b 100644
--- a/common/transport/src/main/java/org/thingsboard/server/common/transport/TransportAdaptor.java
+++ b/common/transport/src/main/java/org/thingsboard/server/common/transport/TransportAdaptor.java
@@ -15,19 +15,8 @@
*/
package org.thingsboard.server.common.transport;
-import org.thingsboard.server.common.msg.session.AdaptorToSessionActorMsg;
-import org.thingsboard.server.common.msg.session.SessionMsgType;
-import org.thingsboard.server.common.msg.session.SessionActorToAdaptorMsg;
import org.thingsboard.server.common.msg.session.SessionContext;
-import org.thingsboard.server.common.transport.adaptor.AdaptorException;
-import org.thingsboard.server.gen.transport.TransportProtos;
-
-import java.util.Optional;
public interface TransportAdaptor {
- AdaptorToSessionActorMsg convertToActorMsg(C ctx, SessionMsgType type, T inbound) throws AdaptorException;
-
- Optional convertToAdaptorMsg(C ctx, SessionActorToAdaptorMsg msg) throws AdaptorException;
-
}
diff --git a/common/transport/src/main/java/org/thingsboard/server/common/transport/TransportService.java b/common/transport/src/main/java/org/thingsboard/server/common/transport/TransportService.java
index cd3041939b..c9a04c4daf 100644
--- a/common/transport/src/main/java/org/thingsboard/server/common/transport/TransportService.java
+++ b/common/transport/src/main/java/org/thingsboard/server/common/transport/TransportService.java
@@ -15,6 +15,7 @@
*/
package org.thingsboard.server.common.transport;
+import org.thingsboard.server.gen.transport.TransportProtos;
import org.thingsboard.server.gen.transport.TransportProtos.ToServerRpcRequestMsg;
import org.thingsboard.server.gen.transport.TransportProtos.ToDeviceRpcResponseMsg;
import org.thingsboard.server.gen.transport.TransportProtos.SubscribeToAttributeUpdatesMsg;
@@ -39,6 +40,9 @@ public interface TransportService {
void process(ValidateDeviceX509CertRequestMsg msg,
TransportServiceCallback callback);
+ void process(TransportProtos.GetOrCreateDeviceFromGatewayRequestMsg msg,
+ TransportServiceCallback callback);
+
void process(SessionInfoProto sessionInfo, SessionEventMsg msg, TransportServiceCallback callback);
void process(SessionInfoProto sessionInfo, PostTelemetryMsg msg, TransportServiceCallback callback);
diff --git a/common/transport/src/main/java/org/thingsboard/server/common/transport/adaptor/JsonConverter.java b/common/transport/src/main/java/org/thingsboard/server/common/transport/adaptor/JsonConverter.java
index db7ee884dc..e02bb4bf34 100644
--- a/common/transport/src/main/java/org/thingsboard/server/common/transport/adaptor/JsonConverter.java
+++ b/common/transport/src/main/java/org/thingsboard/server/common/transport/adaptor/JsonConverter.java
@@ -1,12 +1,12 @@
/**
* 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
- *
+ *
+ * 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.
@@ -52,7 +52,8 @@ import java.util.stream.Collectors;
public class JsonConverter {
private static final Gson GSON = new Gson();
- public static final String CAN_T_PARSE_VALUE = "Can't parse value: ";
+ private static final String CAN_T_PARSE_VALUE = "Can't parse value: ";
+ private static final String DEVICE_PROPERTY = "device";
public static TelemetryUploadRequest convertToTelemetry(JsonElement jsonObject) throws JsonSyntaxException {
return convertToTelemetry(jsonObject, BasicRequest.DEFAULT_REQUEST_ID);
@@ -318,6 +319,57 @@ public class JsonConverter {
return result;
}
+ public static JsonObject getJsonObjectForGateway(TransportProtos.GetAttributeResponseMsg responseMsg) {
+ JsonObject result = new JsonObject();
+ result.addProperty("id", responseMsg.getRequestId());
+ if (responseMsg.getClientAttributeListCount() > 0) {
+ addValues(result, responseMsg.getClientAttributeListList());
+ }
+ if (responseMsg.getSharedAttributeListCount() > 0) {
+ addValues(result, responseMsg.getSharedAttributeListList());
+ }
+ return result;
+ }
+
+ public static JsonObject getJsonObjectForGateway(String deviceName, AttributeUpdateNotificationMsg notificationMsg) {
+ JsonObject result = new JsonObject();
+ result.addProperty(DEVICE_PROPERTY, deviceName);
+ result.add("data", toJson(notificationMsg));
+ return result;
+ }
+
+ private static void addValues(JsonObject result, List kvList) {
+ if (kvList.size() == 1) {
+ addValueToJson(result, "value", kvList.get(0).getKv());
+ } else {
+ JsonObject values;
+ if (result.has("values")) {
+ values = result.get("values").getAsJsonObject();
+ } else {
+ values = new JsonObject();
+ result.add("values", values);
+ }
+ kvList.forEach(value -> addValueToJson(values, value.getKv().getKey(), value.getKv()));
+ }
+ }
+
+ private static void addValueToJson(JsonObject json, String name, TransportProtos.KeyValueProto entry) {
+ switch (entry.getType()) {
+ case BOOLEAN_V:
+ json.addProperty(name, entry.getBoolV());
+ break;
+ case STRING_V:
+ json.addProperty(name, entry.getStringV());
+ break;
+ case DOUBLE_V:
+ json.addProperty(name, entry.getDoubleV());
+ break;
+ case LONG_V:
+ json.addProperty(name, entry.getLongV());
+ break;
+ }
+ }
+
private static Consumer addToObject(JsonArray result) {
return key -> result.add(key.getAttributeKey());
}
diff --git a/common/transport/src/main/java/org/thingsboard/server/common/transport/session/DeviceAwareSessionContext.java b/common/transport/src/main/java/org/thingsboard/server/common/transport/session/DeviceAwareSessionContext.java
index b46cfb81df..93921d347f 100644
--- a/common/transport/src/main/java/org/thingsboard/server/common/transport/session/DeviceAwareSessionContext.java
+++ b/common/transport/src/main/java/org/thingsboard/server/common/transport/session/DeviceAwareSessionContext.java
@@ -31,16 +31,9 @@ public abstract class DeviceAwareSessionContext implements SessionContext {
@Getter
private volatile DeviceId deviceId;
+ @Getter
private volatile DeviceInfoProto deviceInfo;
- public long getDeviceIdMSB() {
- return deviceInfo.getDeviceIdMSB();
- }
-
- public long getDeviceIdLSB() {
- return deviceInfo.getDeviceIdLSB();
- }
-
public DeviceId getDeviceId() {
return deviceId;
}
diff --git a/common/transport/src/main/proto/transport.proto b/common/transport/src/main/proto/transport.proto
index bcf5f529b9..3099c5c7f6 100644
--- a/common/transport/src/main/proto/transport.proto
+++ b/common/transport/src/main/proto/transport.proto
@@ -125,6 +125,17 @@ message ValidateDeviceCredentialsResponseMsg {
DeviceInfoProto deviceInfo = 1;
}
+message GetOrCreateDeviceFromGatewayRequestMsg {
+ int64 gatewayIdMSB = 1;
+ int64 gatewayIdLSB = 2;
+ string deviceName = 3;
+ string deviceType = 4;
+}
+
+message GetOrCreateDeviceFromGatewayResponseMsg {
+ DeviceInfoProto deviceInfo = 1;
+}
+
message SessionCloseNotificationProto {
string message = 1;
}
@@ -196,8 +207,10 @@ message ToTransportMsg {
message TransportApiRequestMsg {
ValidateDeviceTokenRequestMsg validateTokenRequestMsg = 1;
ValidateDeviceX509CertRequestMsg validateX509CertRequestMsg = 2;
+ GetOrCreateDeviceFromGatewayRequestMsg getOrCreateDeviceRequestMsg = 3;
}
message TransportApiResponseMsg {
ValidateDeviceCredentialsResponseMsg validateTokenResponseMsg = 1;
+ GetOrCreateDeviceFromGatewayResponseMsg getOrCreateDeviceResponseMsg = 2;
}
\ No newline at end of file
diff --git a/transport/coap/src/main/java/org/thingsboard/server/transport/coap/CoapTransportResource.java b/transport/coap/src/main/java/org/thingsboard/server/transport/coap/CoapTransportResource.java
index 0853124acb..88fb69ecd0 100644
--- a/transport/coap/src/main/java/org/thingsboard/server/transport/coap/CoapTransportResource.java
+++ b/transport/coap/src/main/java/org/thingsboard/server/transport/coap/CoapTransportResource.java
@@ -27,7 +27,6 @@ import org.eclipse.californium.core.network.Exchange;
import org.eclipse.californium.core.network.ExchangeObserver;
import org.eclipse.californium.core.server.resources.CoapExchange;
import org.eclipse.californium.core.server.resources.Resource;
-import org.thingsboard.server.common.data.id.SessionId;
import org.thingsboard.server.common.data.security.DeviceCredentialsFilter;
import org.thingsboard.server.common.data.security.DeviceTokenCredentials;
import org.thingsboard.server.common.msg.session.*;
diff --git a/transport/coap/src/main/java/org/thingsboard/server/transport/coap/session/CoapSessionCtx.java b/transport/coap/src/main/java/org/thingsboard/server/transport/coap/session/CoapSessionCtx.java
index c43a4ded7a..efabc4eac2 100644
--- a/transport/coap/src/main/java/org/thingsboard/server/transport/coap/session/CoapSessionCtx.java
+++ b/transport/coap/src/main/java/org/thingsboard/server/transport/coap/session/CoapSessionCtx.java
@@ -20,20 +20,16 @@ import org.eclipse.californium.core.coap.CoAP.ResponseCode;
import org.eclipse.californium.core.coap.Request;
import org.eclipse.californium.core.coap.Response;
import org.eclipse.californium.core.server.resources.CoapExchange;
-import org.thingsboard.server.common.data.id.SessionId;
import org.thingsboard.server.common.msg.session.SessionActorToAdaptorMsg;
import org.thingsboard.server.common.msg.session.SessionCtrlMsg;
import org.thingsboard.server.common.msg.session.SessionType;
import org.thingsboard.server.common.msg.session.ctrl.SessionCloseMsg;
-import org.thingsboard.server.common.msg.session.ex.SessionAuthException;
import org.thingsboard.server.common.msg.session.ex.SessionException;
import org.thingsboard.server.common.transport.SessionMsgProcessor;
import org.thingsboard.server.common.transport.adaptor.AdaptorException;
import org.thingsboard.server.common.transport.auth.DeviceAuthService;
import org.thingsboard.server.common.transport.session.DeviceAwareSessionContext;
import org.thingsboard.server.transport.coap.adaptors.CoapTransportAdaptor;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
import java.util.concurrent.atomic.AtomicInteger;
diff --git a/transport/coap/src/main/java/org/thingsboard/server/transport/coap/session/CoapSessionId.java b/transport/coap/src/main/java/org/thingsboard/server/transport/coap/session/CoapSessionId.java
index 888fb2fc30..2bfcf1d2c6 100644
--- a/transport/coap/src/main/java/org/thingsboard/server/transport/coap/session/CoapSessionId.java
+++ b/transport/coap/src/main/java/org/thingsboard/server/transport/coap/session/CoapSessionId.java
@@ -15,8 +15,6 @@
*/
package org.thingsboard.server.transport.coap.session;
-import org.thingsboard.server.common.data.id.SessionId;
-
public final class CoapSessionId implements SessionId {
private final String clientAddress;
diff --git a/transport/http/src/main/java/org/thingsboard/server/transport/http/session/HttpSessionCtx.java b/transport/http/src/main/java/org/thingsboard/server/transport/http/session/HttpSessionCtx.java
index 5269285d25..62999cc64b 100644
--- a/transport/http/src/main/java/org/thingsboard/server/transport/http/session/HttpSessionCtx.java
+++ b/transport/http/src/main/java/org/thingsboard/server/transport/http/session/HttpSessionCtx.java
@@ -20,7 +20,6 @@ import lombok.extern.slf4j.Slf4j;
import org.springframework.http.HttpStatus;
import org.springframework.http.ResponseEntity;
import org.springframework.web.context.request.async.DeferredResult;
-import org.thingsboard.server.common.data.id.SessionId;
import org.thingsboard.server.common.msg.core.*;
import org.thingsboard.server.common.msg.session.*;
import org.thingsboard.server.common.msg.session.ex.SessionException;
diff --git a/transport/http/src/main/java/org/thingsboard/server/transport/http/session/HttpSessionId.java b/transport/http/src/main/java/org/thingsboard/server/transport/http/session/HttpSessionId.java
index 3211765334..a5426ca051 100644
--- a/transport/http/src/main/java/org/thingsboard/server/transport/http/session/HttpSessionId.java
+++ b/transport/http/src/main/java/org/thingsboard/server/transport/http/session/HttpSessionId.java
@@ -15,8 +15,6 @@
*/
package org.thingsboard.server.transport.http.session;
-import org.thingsboard.server.common.data.id.SessionId;
-
import java.util.UUID;
/**
diff --git a/transport/mqtt-common/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportContext.java b/transport/mqtt-common/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportContext.java
index b500a31e69..2c8ff59f1f 100644
--- a/transport/mqtt-common/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportContext.java
+++ b/transport/mqtt-common/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportContext.java
@@ -1,12 +1,12 @@
/**
* 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
- *
+ *
+ * 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.
@@ -18,6 +18,7 @@ package org.thingsboard.server.transport.mqtt;
import com.fasterxml.jackson.databind.ObjectMapper;
import io.netty.handler.ssl.SslHandler;
import lombok.Data;
+import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.RandomStringUtils;
import org.springframework.beans.factory.annotation.Autowired;
@@ -30,8 +31,11 @@ import org.thingsboard.server.common.transport.quota.host.HostRequestsQuotaServi
import org.thingsboard.server.transport.mqtt.adaptors.MqttTransportAdaptor;
import javax.annotation.PostConstruct;
+import javax.annotation.PreDestroy;
import java.net.InetAddress;
import java.net.UnknownHostException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
/**
* Created by ashvayka on 04.10.18.
@@ -64,6 +68,9 @@ public class MqttTransportContext {
private SslHandler sslHandler;
+ @Getter
+ private ExecutorService executor;
+
@PostConstruct
public void init() {
if (StringUtils.isEmpty(nodeId)) {
@@ -74,6 +81,14 @@ public class MqttTransportContext {
}
}
log.info("Current NodeId: {}", nodeId);
+ executor = Executors.newCachedThreadPool();
+ }
+
+ @PreDestroy
+ public void stop() {
+ if (executor != null) {
+ executor.shutdownNow();
+ }
}
}
diff --git a/transport/mqtt-common/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java b/transport/mqtt-common/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java
index 3099b24b87..21d33354cd 100644
--- a/transport/mqtt-common/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java
+++ b/transport/mqtt-common/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java
@@ -1,12 +1,12 @@
/**
* 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
- *
+ *
+ * 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.
@@ -108,7 +108,7 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement
this.quotaService = context.getQuotaService();
this.sslHandler = context.getSslHandler();
this.mqttQoSMap = new ConcurrentHashMap<>();
- this.deviceSessionCtx = new DeviceSessionCtx(mqttQoSMap);
+ this.deviceSessionCtx = new DeviceSessionCtx(sessionId, mqttQoSMap);
}
@Override
@@ -176,42 +176,40 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement
if (topicName.startsWith(MqttTopics.BASE_GATEWAY_API_TOPIC)) {
if (gatewaySessionCtx != null) {
- gatewaySessionCtx.setChannel(ctx);
-// handleMqttPublishMsg(topicName, msgId, mqttMsg);
+ handleGatewayPublishMsg(topicName, msgId, mqttMsg);
}
} else {
processDevicePublish(ctx, mqttMsg, topicName, msgId);
}
}
- //
-// private void handleMqttPublishMsg(String topicName, int msgId, MqttPublishMessage mqttMsg) {
-// try {
-// switch (topicName) {
-// case GATEWAY_TELEMETRY_TOPIC:
-// gatewaySessionCtx.onDeviceTelemetry(mqttMsg);
-// break;
-// case GATEWAY_ATTRIBUTES_TOPIC:
-// gatewaySessionCtx.onDeviceAttributes(mqttMsg);
-// break;
-// case GATEWAY_ATTRIBUTES_REQUEST_TOPIC:
-// gatewaySessionCtx.onDeviceAttributesRequest(mqttMsg);
-// break;
-// case GATEWAY_RPC_TOPIC:
-// gatewaySessionCtx.onDeviceRpcResponse(mqttMsg);
-// break;
-// case GATEWAY_CONNECT_TOPIC:
-// gatewaySessionCtx.onDeviceConnect(mqttMsg);
-// break;
-// case GATEWAY_DISCONNECT_TOPIC:
-// gatewaySessionCtx.onDeviceDisconnect(mqttMsg);
-// break;
-// }
-// } catch (RuntimeException | AdaptorException e) {
-// log.warn("[{}] Failed to process publish msg [{}][{}]", sessionId, topicName, msgId, e);
-// }
-// }
-//
+ private void handleGatewayPublishMsg(String topicName, int msgId, MqttPublishMessage mqttMsg) {
+ try {
+ switch (topicName) {
+ case MqttTopics.GATEWAY_TELEMETRY_TOPIC:
+ gatewaySessionCtx.onDeviceTelemetry(mqttMsg);
+ break;
+ case MqttTopics.GATEWAY_ATTRIBUTES_TOPIC:
+ gatewaySessionCtx.onDeviceAttributes(mqttMsg);
+ break;
+ case MqttTopics.GATEWAY_ATTRIBUTES_REQUEST_TOPIC:
+ gatewaySessionCtx.onDeviceAttributesRequest(mqttMsg);
+ break;
+ case MqttTopics.GATEWAY_RPC_TOPIC:
+ gatewaySessionCtx.onDeviceRpcResponse(mqttMsg);
+ break;
+ case MqttTopics.GATEWAY_CONNECT_TOPIC:
+ gatewaySessionCtx.onDeviceConnect(mqttMsg);
+ break;
+ case MqttTopics.GATEWAY_DISCONNECT_TOPIC:
+ gatewaySessionCtx.onDeviceDisconnect(mqttMsg);
+ break;
+ }
+ } catch (RuntimeException | AdaptorException e) {
+ log.warn("[{}] Failed to process publish msg [{}][{}]", sessionId, topicName, msgId, e);
+ }
+ }
+
private void processDevicePublish(ChannelHandlerContext ctx, MqttPublishMessage mqttMsg, String topicName, int msgId) {
try {
if (topicName.equals(MqttTopics.DEVICE_TELEMETRY_TOPIC)) {
@@ -223,10 +221,10 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement
} else if (topicName.startsWith(MqttTopics.DEVICE_ATTRIBUTES_REQUEST_TOPIC_PREFIX)) {
TransportProtos.GetAttributeRequestMsg getAttributeMsg = adaptor.convertToGetAttributes(deviceSessionCtx, mqttMsg);
transportService.process(sessionInfo, getAttributeMsg, getPubAckCallback(ctx, msgId, getAttributeMsg));
- } else if (topicName.startsWith(MqttTopics.DEVICE_RPC_RESPONSE_TOPIC)){
+ } else if (topicName.startsWith(MqttTopics.DEVICE_RPC_RESPONSE_TOPIC)) {
TransportProtos.ToDeviceRpcResponseMsg rpcResponseMsg = adaptor.convertToDeviceRpcResponse(deviceSessionCtx, mqttMsg);
transportService.process(sessionInfo, rpcResponseMsg, getPubAckCallback(ctx, msgId, rpcResponseMsg));
- } else if (topicName.startsWith(MqttTopics.DEVICE_RPC_REQUESTS_TOPIC)){
+ } else if (topicName.startsWith(MqttTopics.DEVICE_RPC_REQUESTS_TOPIC)) {
TransportProtos.ToServerRpcRequestMsg rpcRequestMsg = adaptor.convertToServerRpcRequest(deviceSessionCtx, mqttMsg);
transportService.process(sessionInfo, rpcRequestMsg, getPubAckCallback(ctx, msgId, rpcRequestMsg));
}
@@ -469,7 +467,7 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement
if (infoNode != null) {
JsonNode gatewayNode = infoNode.get("gateway");
if (gatewayNode != null && gatewayNode.asBoolean()) {
- gatewaySessionCtx = new GatewaySessionCtx(deviceSessionCtx);
+ gatewaySessionCtx = new GatewaySessionCtx(context, deviceSessionCtx, sessionId);
}
}
} catch (IOException e) {
@@ -477,7 +475,7 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement
}
}
- private SessionEventMsg getSessionEventMsg(SessionEvent event) {
+ public static SessionEventMsg getSessionEventMsg(SessionEvent event) {
return SessionEventMsg.newBuilder()
.setSessionType(TransportProtos.SessionType.ASYNC)
.setEvent(event).build();
diff --git a/transport/mqtt-common/src/main/java/org/thingsboard/server/transport/mqtt/adaptors/JsonMqttAdaptor.java b/transport/mqtt-common/src/main/java/org/thingsboard/server/transport/mqtt/adaptors/JsonMqttAdaptor.java
index a2be169656..85dc813428 100644
--- a/transport/mqtt-common/src/main/java/org/thingsboard/server/transport/mqtt/adaptors/JsonMqttAdaptor.java
+++ b/transport/mqtt-common/src/main/java/org/thingsboard/server/transport/mqtt/adaptors/JsonMqttAdaptor.java
@@ -1,12 +1,12 @@
/**
* 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
- *
+ *
+ * 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.
@@ -17,6 +17,7 @@ package org.thingsboard.server.transport.mqtt.adaptors;
import com.google.gson.Gson;
import com.google.gson.JsonElement;
+import com.google.gson.JsonObject;
import com.google.gson.JsonParser;
import com.google.gson.JsonSyntaxException;
import io.netty.buffer.ByteBuf;
@@ -26,7 +27,8 @@ import io.netty.handler.codec.mqtt.*;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;
-import org.thingsboard.server.common.data.id.SessionId;
+import org.thingsboard.server.common.data.kv.AttributeKvEntry;
+import org.thingsboard.server.common.data.kv.KvEntry;
import org.thingsboard.server.common.msg.core.*;
import org.thingsboard.server.common.msg.kv.AttributesKVMsg;
import org.thingsboard.server.common.msg.session.*;
@@ -36,12 +38,15 @@ import org.thingsboard.server.gen.transport.TransportProtos;
import org.thingsboard.server.transport.mqtt.MqttTopics;
import org.thingsboard.server.transport.mqtt.session.DeviceSessionCtx;
import org.thingsboard.server.transport.mqtt.MqttTransportHandler;
+import org.thingsboard.server.transport.mqtt.session.MqttDeviceAwareSessionContext;
import java.nio.charset.Charset;
import java.util.Arrays;
import java.util.HashSet;
+import java.util.List;
import java.util.Optional;
import java.util.Set;
+import java.util.UUID;
/**
* @author Andrew Shvayka
@@ -55,7 +60,7 @@ public class JsonMqttAdaptor implements MqttTransportAdaptor {
private static final ByteBufAllocator ALLOCATOR = new UnpooledByteBufAllocator(false);
@Override
- public TransportProtos.PostTelemetryMsg convertToPostTelemetry(DeviceSessionCtx ctx, MqttPublishMessage inbound) throws AdaptorException {
+ public TransportProtos.PostTelemetryMsg convertToPostTelemetry(MqttDeviceAwareSessionContext ctx, MqttPublishMessage inbound) throws AdaptorException {
String payload = validatePayload(ctx.getSessionId(), inbound.payload());
try {
return JsonConverter.convertToTelemetryProto(new JsonParser().parse(payload));
@@ -65,7 +70,7 @@ public class JsonMqttAdaptor implements MqttTransportAdaptor {
}
@Override
- public TransportProtos.PostAttributeMsg convertToPostAttributes(DeviceSessionCtx ctx, MqttPublishMessage inbound) throws AdaptorException {
+ public TransportProtos.PostAttributeMsg convertToPostAttributes(MqttDeviceAwareSessionContext ctx, MqttPublishMessage inbound) throws AdaptorException {
String payload = validatePayload(ctx.getSessionId(), inbound.payload());
try {
return JsonConverter.convertToAttributesProto(new JsonParser().parse(payload));
@@ -75,7 +80,7 @@ public class JsonMqttAdaptor implements MqttTransportAdaptor {
}
@Override
- public TransportProtos.GetAttributeRequestMsg convertToGetAttributes(DeviceSessionCtx ctx, MqttPublishMessage inbound) throws AdaptorException {
+ public TransportProtos.GetAttributeRequestMsg convertToGetAttributes(MqttDeviceAwareSessionContext ctx, MqttPublishMessage inbound) throws AdaptorException {
String topicName = inbound.variableHeader().topicName();
try {
TransportProtos.GetAttributeRequestMsg.Builder result = TransportProtos.GetAttributeRequestMsg.newBuilder();
@@ -98,7 +103,7 @@ public class JsonMqttAdaptor implements MqttTransportAdaptor {
}
@Override
- public TransportProtos.ToDeviceRpcResponseMsg convertToDeviceRpcResponse(DeviceSessionCtx ctx, MqttPublishMessage inbound) throws AdaptorException {
+ public TransportProtos.ToDeviceRpcResponseMsg convertToDeviceRpcResponse(MqttDeviceAwareSessionContext ctx, MqttPublishMessage inbound) throws AdaptorException {
String topicName = inbound.variableHeader().topicName();
try {
Integer requestId = Integer.valueOf(topicName.substring(MqttTopics.DEVICE_RPC_RESPONSE_TOPIC.length()));
@@ -111,7 +116,7 @@ public class JsonMqttAdaptor implements MqttTransportAdaptor {
}
@Override
- public TransportProtos.ToServerRpcRequestMsg convertToServerRpcRequest(DeviceSessionCtx ctx, MqttPublishMessage inbound) throws AdaptorException {
+ public TransportProtos.ToServerRpcRequestMsg convertToServerRpcRequest(MqttDeviceAwareSessionContext ctx, MqttPublishMessage inbound) throws AdaptorException {
String topicName = inbound.variableHeader().topicName();
String payload = validatePayload(ctx.getSessionId(), inbound.payload());
try {
@@ -123,7 +128,7 @@ public class JsonMqttAdaptor implements MqttTransportAdaptor {
}
@Override
- public Optional convertToPublish(DeviceSessionCtx ctx, TransportProtos.GetAttributeResponseMsg responseMsg) throws AdaptorException {
+ public Optional convertToPublish(MqttDeviceAwareSessionContext ctx, TransportProtos.GetAttributeResponseMsg responseMsg) throws AdaptorException {
if (!StringUtils.isEmpty(responseMsg.getError())) {
throw new AdaptorException(responseMsg.getError());
} else {
@@ -138,99 +143,37 @@ public class JsonMqttAdaptor implements MqttTransportAdaptor {
}
@Override
- public Optional convertToPublish(DeviceSessionCtx ctx, TransportProtos.AttributeUpdateNotificationMsg notificationMsg) throws AdaptorException {
- return Optional.of(createMqttPublishMsg(ctx, MqttTopics.DEVICE_ATTRIBUTES_TOPIC, JsonConverter.toJson(notificationMsg)));
+ public Optional convertToGatewayPublish(MqttDeviceAwareSessionContext ctx, TransportProtos.GetAttributeResponseMsg responseMsg) throws AdaptorException {
+ if (!StringUtils.isEmpty(responseMsg.getError())) {
+ throw new AdaptorException(responseMsg.getError());
+ } else {
+ JsonObject result = JsonConverter.getJsonObjectForGateway(responseMsg);
+ return Optional.of(createMqttPublishMsg(ctx, MqttTopics.GATEWAY_ATTRIBUTES_RESPONSE_TOPIC, result));
+ }
}
@Override
- public Optional convertToPublish(DeviceSessionCtx ctx, TransportProtos.ToDeviceRpcRequestMsg rpcRequest) throws AdaptorException {
- return Optional.of(createMqttPublishMsg(ctx, MqttTopics.DEVICE_RPC_REQUESTS_TOPIC + rpcRequest.getRequestId(), JsonConverter.toJson(rpcRequest, false)));
+ public Optional convertToPublish(MqttDeviceAwareSessionContext ctx, TransportProtos.AttributeUpdateNotificationMsg notificationMsg) throws AdaptorException {
+ return Optional.of(createMqttPublishMsg(ctx, MqttTopics.DEVICE_ATTRIBUTES_TOPIC, JsonConverter.toJson(notificationMsg)));
}
@Override
- public Optional convertToPublish(DeviceSessionCtx ctx, TransportProtos.ToServerRpcResponseMsg rpcResponse) {
- return Optional.of(createMqttPublishMsg(ctx, MqttTopics.DEVICE_RPC_RESPONSE_TOPIC + rpcResponse.getRequestId(), JsonConverter.toJson(rpcResponse)));
+ public Optional convertToGatewayPublish(MqttDeviceAwareSessionContext ctx, String deviceName, TransportProtos.AttributeUpdateNotificationMsg notificationMsg) throws AdaptorException {
+ JsonObject result = JsonConverter.getJsonObjectForGateway(deviceName, notificationMsg);
+ return Optional.of(createMqttPublishMsg(ctx, MqttTopics.GATEWAY_ATTRIBUTES_TOPIC, result));
}
@Override
- public AdaptorToSessionActorMsg convertToActorMsg(DeviceSessionCtx ctx, SessionMsgType type, MqttMessage inbound) throws AdaptorException {
- FromDeviceMsg msg;
- switch (type) {
- case POST_TELEMETRY_REQUEST:
- msg = convertToTelemetryUploadRequest(ctx, (MqttPublishMessage) inbound);
- break;
- case POST_ATTRIBUTES_REQUEST:
- msg = convertToUpdateAttributesRequest(ctx, (MqttPublishMessage) inbound);
- break;
- case SUBSCRIBE_ATTRIBUTES_REQUEST:
- msg = new AttributesSubscribeMsg();
- break;
- case UNSUBSCRIBE_ATTRIBUTES_REQUEST:
- msg = new AttributesUnsubscribeMsg();
- break;
- case SUBSCRIBE_RPC_COMMANDS_REQUEST:
- msg = new RpcSubscribeMsg();
- break;
- case UNSUBSCRIBE_RPC_COMMANDS_REQUEST:
- msg = new RpcUnsubscribeMsg();
- break;
- case GET_ATTRIBUTES_REQUEST:
- msg = convertToGetAttributesRequest(ctx, (MqttPublishMessage) inbound);
- break;
- case TO_DEVICE_RPC_RESPONSE:
- msg = convertToRpcCommandResponse(ctx, (MqttPublishMessage) inbound);
- break;
- case TO_SERVER_RPC_REQUEST:
- msg = null;//convertToServerRpcRequest(ctx, (MqttPublishMessage) inbound);
- break;
- default:
- log.warn("[{}] Unsupported msg type: {}!", ctx.getSessionId(), type);
- throw new AdaptorException(new IllegalArgumentException("Unsupported msg type: " + type + "!"));
- }
- return new BasicAdaptorToSessionActorMsg(ctx, msg);
+ public Optional convertToPublish(MqttDeviceAwareSessionContext ctx, TransportProtos.ToDeviceRpcRequestMsg rpcRequest) throws AdaptorException {
+ return Optional.of(createMqttPublishMsg(ctx, MqttTopics.DEVICE_RPC_REQUESTS_TOPIC + rpcRequest.getRequestId(), JsonConverter.toJson(rpcRequest, false)));
}
@Override
- public Optional convertToAdaptorMsg(DeviceSessionCtx ctx, SessionActorToAdaptorMsg sessionMsg) throws AdaptorException {
- MqttMessage result = null;
- ToDeviceMsg msg = sessionMsg.getMsg();
- switch (msg.getSessionMsgType()) {
- case STATUS_CODE_RESPONSE:
- case GET_ATTRIBUTES_RESPONSE:
- ResponseMsg responseMsg = (ResponseMsg) msg;
- Optional responseError = responseMsg.getError();
- if (responseMsg.isSuccess()) {
- result = convertResponseMsg(ctx, msg, responseMsg, responseError);
- } else {
- if (responseError.isPresent()) {
- throw new AdaptorException(responseError.get());
- }
- }
- break;
- case ATTRIBUTES_UPDATE_NOTIFICATION:
- AttributesUpdateNotification notification = (AttributesUpdateNotification) msg;
- result = createMqttPublishMsg(ctx, MqttTopics.DEVICE_ATTRIBUTES_TOPIC, notification.getData(), false);
- break;
- case TO_DEVICE_RPC_REQUEST:
- ToDeviceRpcRequestMsg rpcRequest = (ToDeviceRpcRequestMsg) msg;
- result = createMqttPublishMsg(ctx, MqttTopics.DEVICE_RPC_REQUESTS_TOPIC + rpcRequest.getRequestId(), rpcRequest);
- break;
- case TO_SERVER_RPC_RESPONSE:
-// ToServerRpcResponseMsg rpcResponse = (ToServerRpcResponseMsg) msg;
-// result = createMqttPublishMsg(ctx, MqttTopics.DEVICE_RPC_RESPONSE_TOPIC + rpcResponse.getRequestId(),
-// rpcResponse);
- break;
- case RULE_ENGINE_ERROR:
- RuleEngineErrorMsg errorMsg = (RuleEngineErrorMsg) msg;
- result = createMqttPublishMsg(ctx, "errors", JsonConverter.toErrorJson(errorMsg.getErrorMsg()));
- break;
- default:
- break;
- }
- return Optional.ofNullable(result);
+ public Optional convertToPublish(MqttDeviceAwareSessionContext ctx, TransportProtos.ToServerRpcResponseMsg rpcResponse) {
+ return Optional.of(createMqttPublishMsg(ctx, MqttTopics.DEVICE_RPC_RESPONSE_TOPIC + rpcResponse.getRequestId(), JsonConverter.toJson(rpcResponse)));
}
- private MqttMessage convertResponseMsg(DeviceSessionCtx ctx, ToDeviceMsg msg,
+ private MqttMessage convertResponseMsg(MqttDeviceAwareSessionContext ctx, ToDeviceMsg msg,
ResponseMsg responseMsg, Optional responseError) throws AdaptorException {
MqttMessage result = null;
SessionMsgType requestMsgType = responseMsg.getRequestMsgType();
@@ -255,19 +198,19 @@ public class JsonMqttAdaptor implements MqttTransportAdaptor {
return result;
}
- private MqttPublishMessage createMqttPublishMsg(DeviceSessionCtx ctx, String topic, AttributesKVMsg msg, boolean asMap) {
+ private MqttPublishMessage createMqttPublishMsg(MqttDeviceAwareSessionContext ctx, String topic, AttributesKVMsg msg, boolean asMap) {
return createMqttPublishMsg(ctx, topic, JsonConverter.toJson(msg, asMap));
}
- private MqttPublishMessage createMqttPublishMsg(DeviceSessionCtx ctx, String topic, ToDeviceRpcRequestMsg msg) {
+ private MqttPublishMessage createMqttPublishMsg(MqttDeviceAwareSessionContext ctx, String topic, ToDeviceRpcRequestMsg msg) {
return createMqttPublishMsg(ctx, topic, JsonConverter.toJson(msg, false));
}
- private MqttPublishMessage createMqttPublishMsg(DeviceSessionCtx ctx, String topic, TransportProtos.ToServerRpcResponseMsg msg) {
+ private MqttPublishMessage createMqttPublishMsg(MqttDeviceAwareSessionContext ctx, String topic, TransportProtos.ToServerRpcResponseMsg msg) {
return createMqttPublishMsg(ctx, topic, JsonConverter.toJson(msg));
}
- private MqttPublishMessage createMqttPublishMsg(DeviceSessionCtx ctx, String topic, JsonElement json) {
+ private MqttPublishMessage createMqttPublishMsg(MqttDeviceAwareSessionContext ctx, String topic, JsonElement json) {
MqttFixedHeader mqttFixedHeader =
new MqttFixedHeader(MqttMessageType.PUBLISH, false, ctx.getQoSForTopic(topic), false, 0);
MqttPublishVariableHeader header = new MqttPublishVariableHeader(topic, ctx.nextMsgId());
@@ -276,7 +219,7 @@ public class JsonMqttAdaptor implements MqttTransportAdaptor {
return new MqttPublishMessage(mqttFixedHeader, header, payload);
}
- private FromDeviceMsg convertToGetAttributesRequest(DeviceSessionCtx ctx, MqttPublishMessage inbound) throws AdaptorException {
+ private FromDeviceMsg convertToGetAttributesRequest(MqttDeviceAwareSessionContext ctx, MqttPublishMessage inbound) throws AdaptorException {
String topicName = inbound.variableHeader().topicName();
try {
Integer requestId = Integer.valueOf(topicName.substring(MqttTopics.DEVICE_ATTRIBUTES_REQUEST_TOPIC_PREFIX.length()));
@@ -295,7 +238,7 @@ public class JsonMqttAdaptor implements MqttTransportAdaptor {
}
}
- private FromDeviceMsg convertToRpcCommandResponse(DeviceSessionCtx ctx, MqttPublishMessage inbound) throws AdaptorException {
+ private FromDeviceMsg convertToRpcCommandResponse(MqttDeviceAwareSessionContext ctx, MqttPublishMessage inbound) throws AdaptorException {
String topicName = inbound.variableHeader().topicName();
try {
Integer requestId = Integer.valueOf(topicName.substring(MqttTopics.DEVICE_RPC_RESPONSE_TOPIC.length()));
@@ -336,7 +279,7 @@ public class JsonMqttAdaptor implements MqttTransportAdaptor {
}
}
- public static JsonElement validateJsonPayload(SessionId sessionId, ByteBuf payloadData) throws AdaptorException {
+ public static JsonElement validateJsonPayload(UUID sessionId, ByteBuf payloadData) throws AdaptorException {
String payload = validatePayload(sessionId, payloadData);
try {
return new JsonParser().parse(payload);
@@ -345,11 +288,11 @@ public class JsonMqttAdaptor implements MqttTransportAdaptor {
}
}
- public static String validatePayload(SessionId sessionId, ByteBuf payloadData) throws AdaptorException {
+ public static String validatePayload(UUID sessionId, ByteBuf payloadData) throws AdaptorException {
try {
String payload = payloadData.toString(UTF8);
if (payload == null) {
- log.warn("[{}] Payload is empty!", sessionId.toUidStr());
+ log.warn("[{}] Payload is empty!", sessionId);
throw new AdaptorException(new IllegalArgumentException("Payload is empty!"));
}
return payload;
diff --git a/transport/mqtt-common/src/main/java/org/thingsboard/server/transport/mqtt/adaptors/MqttTransportAdaptor.java b/transport/mqtt-common/src/main/java/org/thingsboard/server/transport/mqtt/adaptors/MqttTransportAdaptor.java
index 48173f968a..37b5395b69 100644
--- a/transport/mqtt-common/src/main/java/org/thingsboard/server/transport/mqtt/adaptors/MqttTransportAdaptor.java
+++ b/transport/mqtt-common/src/main/java/org/thingsboard/server/transport/mqtt/adaptors/MqttTransportAdaptor.java
@@ -17,41 +17,44 @@ package org.thingsboard.server.transport.mqtt.adaptors;
import io.netty.handler.codec.mqtt.MqttMessage;
import io.netty.handler.codec.mqtt.MqttPublishMessage;
-import org.thingsboard.server.common.transport.TransportAdaptor;
import org.thingsboard.server.common.transport.adaptor.AdaptorException;
-import org.thingsboard.server.gen.transport.TransportProtos.ToServerRpcResponseMsg;
-import org.thingsboard.server.gen.transport.TransportProtos.ToServerRpcRequestMsg;
-import org.thingsboard.server.gen.transport.TransportProtos.ToDeviceRpcResponseMsg;
import org.thingsboard.server.gen.transport.TransportProtos.AttributeUpdateNotificationMsg;
import org.thingsboard.server.gen.transport.TransportProtos.GetAttributeRequestMsg;
import org.thingsboard.server.gen.transport.TransportProtos.GetAttributeResponseMsg;
import org.thingsboard.server.gen.transport.TransportProtos.PostAttributeMsg;
import org.thingsboard.server.gen.transport.TransportProtos.PostTelemetryMsg;
import org.thingsboard.server.gen.transport.TransportProtos.ToDeviceRpcRequestMsg;
-import org.thingsboard.server.transport.mqtt.session.DeviceSessionCtx;
+import org.thingsboard.server.gen.transport.TransportProtos.ToDeviceRpcResponseMsg;
+import org.thingsboard.server.gen.transport.TransportProtos.ToServerRpcRequestMsg;
+import org.thingsboard.server.gen.transport.TransportProtos.ToServerRpcResponseMsg;
+import org.thingsboard.server.transport.mqtt.session.MqttDeviceAwareSessionContext;
import java.util.Optional;
/**
* @author Andrew Shvayka
*/
-public interface MqttTransportAdaptor extends TransportAdaptor {
+public interface MqttTransportAdaptor {
+
+ PostTelemetryMsg convertToPostTelemetry(MqttDeviceAwareSessionContext ctx, MqttPublishMessage inbound) throws AdaptorException;
+
+ PostAttributeMsg convertToPostAttributes(MqttDeviceAwareSessionContext ctx, MqttPublishMessage inbound) throws AdaptorException;
- PostTelemetryMsg convertToPostTelemetry(DeviceSessionCtx ctx, MqttPublishMessage inbound) throws AdaptorException;
+ GetAttributeRequestMsg convertToGetAttributes(MqttDeviceAwareSessionContext ctx, MqttPublishMessage inbound) throws AdaptorException;
- PostAttributeMsg convertToPostAttributes(DeviceSessionCtx ctx, MqttPublishMessage inbound) throws AdaptorException;
+ ToDeviceRpcResponseMsg convertToDeviceRpcResponse(MqttDeviceAwareSessionContext ctx, MqttPublishMessage mqttMsg) throws AdaptorException;
- GetAttributeRequestMsg convertToGetAttributes(DeviceSessionCtx ctx, MqttPublishMessage inbound) throws AdaptorException;
+ ToServerRpcRequestMsg convertToServerRpcRequest(MqttDeviceAwareSessionContext ctx, MqttPublishMessage mqttMsg) throws AdaptorException;
- ToDeviceRpcResponseMsg convertToDeviceRpcResponse(DeviceSessionCtx ctx, MqttPublishMessage mqttMsg) throws AdaptorException;
+ Optional convertToPublish(MqttDeviceAwareSessionContext ctx, GetAttributeResponseMsg responseMsg) throws AdaptorException;
- ToServerRpcRequestMsg convertToServerRpcRequest(DeviceSessionCtx ctx, MqttPublishMessage mqttMsg) throws AdaptorException;
+ Optional convertToGatewayPublish(MqttDeviceAwareSessionContext ctx, GetAttributeResponseMsg responseMsg) throws AdaptorException;
- Optional convertToPublish(DeviceSessionCtx ctx, GetAttributeResponseMsg responseMsg) throws AdaptorException;
+ Optional convertToPublish(MqttDeviceAwareSessionContext ctx, AttributeUpdateNotificationMsg notificationMsg) throws AdaptorException;
- Optional convertToPublish(DeviceSessionCtx ctx, AttributeUpdateNotificationMsg notificationMsg) throws AdaptorException;
+ Optional convertToGatewayPublish(MqttDeviceAwareSessionContext ctx, String deviceName, AttributeUpdateNotificationMsg notificationMsg) throws AdaptorException;
- Optional convertToPublish(DeviceSessionCtx ctx, ToDeviceRpcRequestMsg rpcRequest) throws AdaptorException;
+ Optional convertToPublish(MqttDeviceAwareSessionContext ctx, ToDeviceRpcRequestMsg rpcRequest) throws AdaptorException;
- Optional convertToPublish(DeviceSessionCtx ctx, ToServerRpcResponseMsg rpcResponse);
+ Optional convertToPublish(MqttDeviceAwareSessionContext ctx, ToServerRpcResponseMsg rpcResponse) throws AdaptorException;
}
diff --git a/transport/mqtt-common/src/main/java/org/thingsboard/server/transport/mqtt/session/DeviceSessionCtx.java b/transport/mqtt-common/src/main/java/org/thingsboard/server/transport/mqtt/session/DeviceSessionCtx.java
index 6bbb9fa6fd..6195fdd440 100644
--- a/transport/mqtt-common/src/main/java/org/thingsboard/server/transport/mqtt/session/DeviceSessionCtx.java
+++ b/transport/mqtt-common/src/main/java/org/thingsboard/server/transport/mqtt/session/DeviceSessionCtx.java
@@ -1,12 +1,12 @@
/**
* 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
- *
+ *
+ * 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.
@@ -19,19 +19,14 @@ import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.mqtt.*;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
-import org.thingsboard.server.common.data.id.SessionId;
import org.thingsboard.server.common.msg.session.SessionActorToAdaptorMsg;
import org.thingsboard.server.common.msg.session.SessionCtrlMsg;
import org.thingsboard.server.common.msg.session.SessionType;
import org.thingsboard.server.common.msg.session.ctrl.SessionCloseMsg;
import org.thingsboard.server.common.msg.session.ex.SessionException;
-import org.thingsboard.server.common.transport.SessionMsgProcessor;
import org.thingsboard.server.common.transport.adaptor.AdaptorException;
-import org.thingsboard.server.common.transport.auth.DeviceAuthService;
-import org.thingsboard.server.common.transport.session.DeviceAwareSessionContext;
-import org.thingsboard.server.transport.mqtt.adaptors.MqttTransportAdaptor;
-import java.util.Map;
+import java.util.UUID;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicInteger;
@@ -41,14 +36,14 @@ import java.util.concurrent.atomic.AtomicInteger;
@Slf4j
public class DeviceSessionCtx extends MqttDeviceAwareSessionContext {
- private final MqttSessionId sessionId;
+ private final UUID sessionId;
@Getter
private ChannelHandlerContext channel;
private AtomicInteger msgIdSeq = new AtomicInteger(0);
- public DeviceSessionCtx(ConcurrentMap mqttQoSMap) {
+ public DeviceSessionCtx(UUID sessionId, ConcurrentMap mqttQoSMap) {
super(null, null, mqttQoSMap);
- this.sessionId = new MqttSessionId();
+ this.sessionId = sessionId;
}
@Override
@@ -94,7 +89,7 @@ public class DeviceSessionCtx extends MqttDeviceAwareSessionContext {
}
@Override
- public SessionId getSessionId() {
+ public UUID getSessionId() {
return sessionId;
}
diff --git a/transport/mqtt-common/src/main/java/org/thingsboard/server/transport/mqtt/session/GatewayDeviceSessionCtx.java b/transport/mqtt-common/src/main/java/org/thingsboard/server/transport/mqtt/session/GatewayDeviceSessionCtx.java
index 793836d793..387d9cf439 100644
--- a/transport/mqtt-common/src/main/java/org/thingsboard/server/transport/mqtt/session/GatewayDeviceSessionCtx.java
+++ b/transport/mqtt-common/src/main/java/org/thingsboard/server/transport/mqtt/session/GatewayDeviceSessionCtx.java
@@ -1,12 +1,12 @@
/**
* 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
- *
+ *
+ * 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.
@@ -16,70 +16,80 @@
package org.thingsboard.server.transport.mqtt.session;
import com.google.gson.Gson;
-import com.google.gson.JsonArray;
import com.google.gson.JsonElement;
import com.google.gson.JsonObject;
+import com.sun.xml.internal.bind.v2.TODO;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufAllocator;
import io.netty.buffer.UnpooledByteBufAllocator;
-import io.netty.handler.codec.mqtt.*;
-import org.thingsboard.server.common.data.Device;
-import org.thingsboard.server.common.data.id.SessionId;
+import io.netty.handler.codec.mqtt.MqttFixedHeader;
+import io.netty.handler.codec.mqtt.MqttMessage;
+import io.netty.handler.codec.mqtt.MqttMessageType;
+import io.netty.handler.codec.mqtt.MqttPublishMessage;
+import io.netty.handler.codec.mqtt.MqttPublishVariableHeader;
+import lombok.extern.slf4j.Slf4j;
import org.thingsboard.server.common.data.kv.AttributeKvEntry;
import org.thingsboard.server.common.data.kv.KvEntry;
-import org.thingsboard.server.common.msg.core.*;
+import org.thingsboard.server.common.msg.core.AttributesUpdateNotification;
+import org.thingsboard.server.common.msg.core.GetAttributesResponse;
+import org.thingsboard.server.common.msg.core.ResponseMsg;
+import org.thingsboard.server.common.msg.core.ToDeviceRpcRequestMsg;
import org.thingsboard.server.common.msg.kv.AttributesKVMsg;
-import org.thingsboard.server.common.msg.session.*;
-import org.thingsboard.server.common.msg.session.ex.SessionException;
+import org.thingsboard.server.common.msg.session.SessionActorToAdaptorMsg;
+import org.thingsboard.server.common.msg.session.SessionMsgType;
+import org.thingsboard.server.common.msg.session.ToDeviceMsg;
+import org.thingsboard.server.common.transport.SessionMsgListener;
import org.thingsboard.server.common.transport.adaptor.JsonConverter;
-import org.thingsboard.server.common.transport.session.DeviceAwareSessionContext;
+import org.thingsboard.server.gen.transport.TransportProtos;
+import org.thingsboard.server.gen.transport.TransportProtos.DeviceInfoProto;
+import org.thingsboard.server.gen.transport.TransportProtos.SessionInfoProto;
import org.thingsboard.server.transport.mqtt.MqttTopics;
import org.thingsboard.server.transport.mqtt.MqttTransportHandler;
import java.nio.charset.Charset;
import java.util.List;
-import java.util.Map;
import java.util.Optional;
+import java.util.UUID;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicInteger;
/**
* Created by ashvayka on 19.01.17.
*/
-public class GatewayDeviceSessionCtx extends MqttDeviceAwareSessionContext {
+@Slf4j
+public class GatewayDeviceSessionCtx extends MqttDeviceAwareSessionContext implements SessionMsgListener {
private static final Gson GSON = new Gson();
private static final Charset UTF8 = Charset.forName("UTF-8");
private static final ByteBufAllocator ALLOCATOR = new UnpooledByteBufAllocator(false);
- public static final String DEVICE_PROPERTY = "device";
- private GatewaySessionCtx parent;
- private final MqttSessionId sessionId;
+ private final GatewaySessionCtx parent;
+ private final UUID sessionId;
+ private final SessionInfoProto sessionInfo;
private volatile boolean closed;
private AtomicInteger msgIdSeq = new AtomicInteger(0);
- public GatewayDeviceSessionCtx(GatewaySessionCtx parent, Device device, ConcurrentMap mqttQoSMap) {
- super(parent.getProcessor(), parent.getAuthService(), device, mqttQoSMap);
+ public GatewayDeviceSessionCtx(GatewaySessionCtx parent, DeviceInfoProto deviceInfo, ConcurrentMap mqttQoSMap) {
+ super(mqttQoSMap);
this.parent = parent;
- this.sessionId = new MqttSessionId();
+ this.sessionId = UUID.randomUUID();
+ this.sessionInfo = SessionInfoProto.newBuilder()
+ .setNodeId(parent.getNodeId())
+ .setSessionIdMSB(sessionId.getMostSignificantBits())
+ .setSessionIdLSB(sessionId.getLeastSignificantBits())
+ .setDeviceIdMSB(deviceInfo.getDeviceIdMSB())
+ .setDeviceIdLSB(deviceInfo.getDeviceIdLSB())
+ .setTenantIdMSB(deviceInfo.getTenantIdMSB())
+ .setTenantIdLSB(deviceInfo.getTenantIdLSB())
+ .build();
+ setDeviceInfo(deviceInfo);
}
@Override
- public SessionId getSessionId() {
+ public UUID getSessionId() {
return sessionId;
}
- @Override
- public SessionType getSessionType() {
- return SessionType.ASYNC;
- }
-
- @Override
- public void onMsg(SessionActorToAdaptorMsg sessionMsg) throws SessionException {
- Optional message = getToDeviceMsg(sessionMsg);
- message.ifPresent(parent::writeAndFlush);
- }
-
private Optional getToDeviceMsg(SessionActorToAdaptorMsg sessionMsg) {
ToDeviceMsg msg = sessionMsg.getMsg();
switch (msg.getSessionMsgType()) {
@@ -113,25 +123,6 @@ public class GatewayDeviceSessionCtx extends MqttDeviceAwareSessionContext {
return Optional.empty();
}
- @Override
- public void onMsg(SessionCtrlMsg msg) throws SessionException {
- //Do nothing
- }
-
- @Override
- public boolean isClosed() {
- return closed;
- }
-
- public void setClosed(boolean closed) {
- this.closed = closed;
- }
-
- @Override
- public long getTimeout() {
- return 0;
- }
-
private MqttMessage createMqttPublishMsg(String topic, GetAttributesResponse response) {
JsonObject result = new JsonObject();
result.addProperty("id", response.getRequestId());
@@ -204,4 +195,40 @@ public class GatewayDeviceSessionCtx extends MqttDeviceAwareSessionContext {
return new MqttPublishMessage(mqttFixedHeader, header, payload);
}
+ SessionInfoProto getSessionInfo() {
+ return sessionInfo;
+ }
+
+ @Override
+ public void onGetAttributesResponse(TransportProtos.GetAttributeResponseMsg response) {
+ try {
+ parent.getAdaptor().convertToGatewayPublish(this, response).ifPresent(parent::writeAndFlush);
+ } catch (Exception e) {
+ log.trace("[{}] Failed to convert device attributes response to MQTT msg", sessionId, e);
+ }
+ }
+
+ @Override
+ public void onAttributeUpdate(TransportProtos.AttributeUpdateNotificationMsg notification) {
+ try {
+ parent.getAdaptor().convertToGatewayPublish(this, getDeviceInfo().getDeviceName(), notification).ifPresent(parent::writeAndFlush);
+ } catch (Exception e) {
+ log.trace("[{}] Failed to convert device attributes response to MQTT msg", sessionId, e);
+ }
+ }
+
+ @Override
+ public void onRemoteSessionCloseCommand(TransportProtos.SessionCloseNotificationProto sessionCloseNotification) {
+ parent.deregisterSession(getDeviceInfo().getDeviceName());
+ }
+
+ @Override
+ public void onToDeviceRpcRequest(TransportProtos.ToDeviceRpcRequestMsg toDeviceRequest) {
+
+ }
+
+ @Override
+ public void onToServerRpcResponse(TransportProtos.ToServerRpcResponseMsg toServerResponse) {
+ TODO
+ }
}
diff --git a/transport/mqtt-common/src/main/java/org/thingsboard/server/transport/mqtt/session/GatewaySessionCtx.java b/transport/mqtt-common/src/main/java/org/thingsboard/server/transport/mqtt/session/GatewaySessionCtx.java
index fafbc4ec01..3788d57e54 100644
--- a/transport/mqtt-common/src/main/java/org/thingsboard/server/transport/mqtt/session/GatewaySessionCtx.java
+++ b/transport/mqtt-common/src/main/java/org/thingsboard/server/transport/mqtt/session/GatewaySessionCtx.java
@@ -1,12 +1,12 @@
/**
* 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
- *
+ *
+ * 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.
@@ -15,6 +15,11 @@
*/
package org.thingsboard.server.transport.mqtt.session;
+
+import com.google.common.util.concurrent.FutureCallback;
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.ListenableFuture;
+import com.google.common.util.concurrent.SettableFuture;
import com.google.gson.JsonArray;
import com.google.gson.JsonElement;
import com.google.gson.JsonNull;
@@ -25,28 +30,28 @@ import io.netty.handler.codec.mqtt.MqttMessage;
import io.netty.handler.codec.mqtt.MqttPublishMessage;
import lombok.extern.slf4j.Slf4j;
import org.springframework.util.StringUtils;
-import org.thingsboard.server.common.data.Device;
-import org.thingsboard.server.common.data.id.SessionId;
-import org.thingsboard.server.common.data.kv.BaseAttributeKvEntry;
-import org.thingsboard.server.common.data.relation.EntityRelation;
-import org.thingsboard.server.common.msg.core.*;
-import org.thingsboard.server.common.msg.session.BasicAdaptorToSessionActorMsg;
-import org.thingsboard.server.common.msg.session.BasicTransportToDeviceSessionActorMsg;
-import org.thingsboard.server.common.msg.session.ctrl.SessionCloseMsg;
-import org.thingsboard.server.common.transport.SessionMsgProcessor;
+import org.thingsboard.server.common.transport.TransportService;
+import org.thingsboard.server.common.transport.TransportServiceCallback;
import org.thingsboard.server.common.transport.adaptor.AdaptorException;
import org.thingsboard.server.common.transport.adaptor.JsonConverter;
-import org.thingsboard.server.common.transport.auth.DeviceAuthService;
-import org.thingsboard.server.dao.device.DeviceService;
-import org.thingsboard.server.dao.relation.RelationService;
+import org.thingsboard.server.gen.transport.TransportProtos;
+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.SessionInfoProto;
+import org.thingsboard.server.transport.mqtt.MqttTransportContext;
import org.thingsboard.server.transport.mqtt.MqttTransportHandler;
import org.thingsboard.server.transport.mqtt.adaptors.JsonMqttAdaptor;
+import org.thingsboard.server.transport.mqtt.adaptors.MqttTransportAdaptor;
-import java.util.*;
+import javax.annotation.Nullable;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
-import java.util.stream.Collectors;
-
-import static org.thingsboard.server.transport.mqtt.adaptors.JsonMqttAdaptor.validateJsonPayload;
/**
* Created by ashvayka on 19.01.17.
@@ -55,184 +60,235 @@ import static org.thingsboard.server.transport.mqtt.adaptors.JsonMqttAdaptor.val
public class GatewaySessionCtx {
private static final String DEFAULT_DEVICE_TYPE = "default";
- public static final String CAN_T_PARSE_VALUE = "Can't parse value: ";
- public static final String DEVICE_PROPERTY = "device";
-// private final Device gateway;
-// private final SessionId gatewaySessionId;
-// private final SessionMsgProcessor processor;
-// private final DeviceService deviceService;
-// private final DeviceAuthService authService;
-// private final RelationService relationService;
-// private final Map devices;
-// private final ConcurrentMap mqttQoSMap;
- private ChannelHandlerContext channel;
-
-// public GatewaySessionCtx(SessionMsgProcessor processor, DeviceService deviceService, DeviceAuthService authService, RelationService relationService, DeviceSessionCtx gatewaySessionCtx) {
-// this.processor = processor;
-// this.deviceService = deviceService;
-// this.authService = authService;
-// this.relationService = relationService;
-// this.gateway = gatewaySessionCtx.getDevice();
-// this.gatewaySessionId = gatewaySessionCtx.getSessionId();
-// this.devices = new HashMap<>();
-// this.mqttQoSMap = gatewaySessionCtx.getMqttQoSMap();
-// }
-
- public GatewaySessionCtx(DeviceSessionCtx deviceSessionCtx) {
+ private static final String CAN_T_PARSE_VALUE = "Can't parse value: ";
+ private static final String DEVICE_PROPERTY = "device";
+
+ private final MqttTransportContext context;
+ private final TransportService transportService;
+ private final DeviceInfoProto gateway;
+ private final UUID sessionId;
+ private final Map devices;
+ private final ConcurrentMap mqttQoSMap;
+ private final ChannelHandlerContext channel;
+ public GatewaySessionCtx(MqttTransportContext context, DeviceSessionCtx deviceSessionCtx, UUID sessionId) {
+ this.context = context;
+ this.transportService = context.getTransportService();
+ this.gateway = deviceSessionCtx.getDeviceInfo();
+ this.sessionId = sessionId;
+ this.devices = new ConcurrentHashMap<>();
+ this.mqttQoSMap = deviceSessionCtx.getMqttQoSMap();
+ this.channel = deviceSessionCtx.getChannel();
}
public void onDeviceConnect(MqttPublishMessage msg) throws AdaptorException {
JsonElement json = getJson(msg);
String deviceName = checkDeviceName(getDeviceName(json));
String deviceType = getDeviceType(json);
- onDeviceConnect(deviceName, deviceType);
- ack(msg);
+ Futures.addCallback(onDeviceConnect(deviceName, deviceType), new FutureCallback() {
+ @Override
+ public void onSuccess(@Nullable GatewayDeviceSessionCtx result) {
+ ack(msg);
+ }
+
+ @Override
+ public void onFailure(Throwable t) {
+ log.warn("[{}] Failed to process device connect command: {}", sessionId, deviceName, t);
+
+ }
+ }, context.getExecutor());
}
- private void onDeviceConnect(String deviceName, String deviceType) {
-// if (!devices.containsKey(deviceName)) {
-// Device device = deviceService.findDeviceByTenantIdAndName(gateway.getTenantId(), deviceName);
-// if (device == null) {
-// device = new Device();
-// device.setTenantId(gateway.getTenantId());
-// device.setName(deviceName);
-// device.setType(deviceType);
-// device.setCustomerId(gateway.getCustomerId());
-// device = deviceService.saveDevice(device);
-// relationService.saveRelationAsync(new EntityRelation(gateway.getId(), device.getId(), "Created"));
-// processor.onDeviceAdded(device);
-// }
-// GatewayDeviceSessionCtx ctx = new GatewayDeviceSessionCtx(this, device, mqttQoSMap);
-// devices.put(deviceName, ctx);
-// log.debug("[{}] Added device [{}] to the gateway session", gatewaySessionId, deviceName);
-// processor.process(new BasicTransportToDeviceSessionActorMsg(device, new BasicAdaptorToSessionActorMsg(ctx, new AttributesSubscribeMsg())));
-// processor.process(new BasicTransportToDeviceSessionActorMsg(device, new BasicAdaptorToSessionActorMsg(ctx, new RpcSubscribeMsg())));
-// }
+ private ListenableFuture onDeviceConnect(String deviceName, String deviceType) {
+ SettableFuture future = SettableFuture.create();
+ GatewayDeviceSessionCtx result = devices.get(deviceName);
+ if (result == null) {
+ transportService.process(GetOrCreateDeviceFromGatewayRequestMsg.newBuilder()
+ .setDeviceName(deviceName)
+ .setDeviceType(deviceType)
+ .setGatewayIdMSB(gateway.getDeviceIdMSB())
+ .setGatewayIdLSB(gateway.getDeviceIdLSB()).build(),
+ new TransportServiceCallback() {
+ @Override
+ public void onSuccess(GetOrCreateDeviceFromGatewayResponseMsg msg) {
+ GatewayDeviceSessionCtx deviceSessionCtx = new GatewayDeviceSessionCtx(GatewaySessionCtx.this, msg.getDeviceInfo(), mqttQoSMap);
+ if (devices.putIfAbsent(deviceName, deviceSessionCtx) == null) {
+ SessionInfoProto deviceSessionInfo = deviceSessionCtx.getSessionInfo();
+ transportService.process(deviceSessionInfo, MqttTransportHandler.getSessionEventMsg(TransportProtos.SessionEvent.OPEN), null);
+ transportService.registerSession(deviceSessionInfo, deviceSessionCtx);
+ }
+ future.set(devices.get(deviceName));
+ }
+
+ @Override
+ public void onError(Throwable e) {
+ log.warn("[{}] Failed to process device connect command: {}", sessionId, deviceName, e);
+ future.setException(e);
+ }
+ });
+ } else {
+ future.set(result);
+ }
+ return future;
}
public void onDeviceDisconnect(MqttPublishMessage msg) throws AdaptorException {
-// String deviceName = checkDeviceName(getDeviceName(getJson(msg)));
-// GatewayDeviceSessionCtx deviceSessionCtx = devices.remove(deviceName);
-// if (deviceSessionCtx != null) {
-// processor.process(SessionCloseMsg.onDisconnect(deviceSessionCtx.getSessionId()));
-// deviceSessionCtx.setClosed(true);
-// log.debug("[{}] Removed device [{}] from the gateway session", gatewaySessionId, deviceName);
-// } else {
-// log.debug("[{}] Device [{}] was already removed from the gateway session", gatewaySessionId, deviceName);
-// }
-// ack(msg);
+ String deviceName = checkDeviceName(getDeviceName(getJson(msg)));
+ deregisterSession(deviceName);
+ ack(msg);
+ }
+
+ void deregisterSession(String deviceName) {
+ GatewayDeviceSessionCtx deviceSessionCtx = devices.remove(deviceName);
+ if (deviceSessionCtx != null) {
+ deregisterSession(deviceName, deviceSessionCtx);
+ } else {
+ log.debug("[{}] Device [{}] was already removed from the gateway session", sessionId, deviceName);
+ }
}
public void onGatewayDisconnect() {
-// devices.forEach((k, v) -> {
-// processor.process(SessionCloseMsg.onDisconnect(v.getSessionId()));
-// });
+ devices.forEach(this::deregisterSession);
}
public void onDeviceTelemetry(MqttPublishMessage mqttMsg) throws AdaptorException {
-// JsonElement json = validateJsonPayload(gatewaySessionId, mqttMsg.payload());
-// int requestId = mqttMsg.variableHeader().messageId();
-// if (json.isJsonObject()) {
-// JsonObject jsonObj = json.getAsJsonObject();
-// for (Map.Entry deviceEntry : jsonObj.entrySet()) {
-// String deviceName = checkDeviceConnected(deviceEntry.getKey());
-// if (!deviceEntry.getValue().isJsonArray()) {
-// throw new JsonSyntaxException(CAN_T_PARSE_VALUE + json);
-// }
-// BasicTelemetryUploadRequest request = new BasicTelemetryUploadRequest(requestId);
-// JsonArray deviceData = deviceEntry.getValue().getAsJsonArray();
-// for (JsonElement element : deviceData) {
-// JsonConverter.parseWithTs(request, element.getAsJsonObject());
-// }
-// GatewayDeviceSessionCtx deviceSessionCtx = devices.get(deviceName);
-// processor.process(new BasicTransportToDeviceSessionActorMsg(deviceSessionCtx.getDevice(),
-// new BasicAdaptorToSessionActorMsg(deviceSessionCtx, request)));
-// }
-// } else {
-// throw new JsonSyntaxException(CAN_T_PARSE_VALUE + json);
-// }
- }
+ JsonElement json = JsonMqttAdaptor.validateJsonPayload(sessionId, mqttMsg.payload());
+ int msgId = mqttMsg.variableHeader().packetId();
+ if (json.isJsonObject()) {
+ JsonObject jsonObj = json.getAsJsonObject();
+ for (Map.Entry deviceEntry : jsonObj.entrySet()) {
+ String deviceName = deviceEntry.getKey();
+ Futures.addCallback(checkDeviceConnected(deviceName),
+ new FutureCallback() {
+ @Override
+ public void onSuccess(@Nullable GatewayDeviceSessionCtx deviceCtx) {
+ if (!deviceEntry.getValue().isJsonArray()) {
+ throw new JsonSyntaxException(CAN_T_PARSE_VALUE + json);
+ }
+ TransportProtos.PostTelemetryMsg postTelemetryMsg = JsonConverter.convertToTelemetryProto(deviceEntry.getValue().getAsJsonArray());
+ transportService.process(deviceCtx.getSessionInfo(), postTelemetryMsg, getPubAckCallback(channel, deviceName, msgId, postTelemetryMsg));
+ }
- public void onDeviceRpcResponse(MqttPublishMessage mqttMsg) throws AdaptorException {
-// JsonElement json = validateJsonPayload(gatewaySessionId, mqttMsg.payload());
-// if (json.isJsonObject()) {
-// JsonObject jsonObj = json.getAsJsonObject();
-// String deviceName = checkDeviceConnected(jsonObj.get(DEVICE_PROPERTY).getAsString());
-// Integer requestId = jsonObj.get("id").getAsInt();
-// String data = jsonObj.get("data").toString();
-// GatewayDeviceSessionCtx deviceSessionCtx = devices.get(deviceName);
-// processor.process(new BasicTransportToDeviceSessionActorMsg(deviceSessionCtx.getDevice(),
-// new BasicAdaptorToSessionActorMsg(deviceSessionCtx, new ToDeviceRpcResponseMsg(requestId, data))));
-// ack(mqttMsg);
-// } else {
-// throw new JsonSyntaxException(CAN_T_PARSE_VALUE + json);
-// }
+ @Override
+ public void onFailure(Throwable t) {
+ log.debug("[{}] Failed to process device teleemtry command: {}", sessionId, deviceName, t);
+ }
+ }, context.getExecutor());
+ }
+ } else {
+ throw new JsonSyntaxException(CAN_T_PARSE_VALUE + json);
+ }
}
public void onDeviceAttributes(MqttPublishMessage mqttMsg) throws AdaptorException {
-// JsonElement json = validateJsonPayload(gatewaySessionId, mqttMsg.payload());
-// int requestId = mqttMsg.variableHeader().messageId();
-// if (json.isJsonObject()) {
-// JsonObject jsonObj = json.getAsJsonObject();
-// for (Map.Entry deviceEntry : jsonObj.entrySet()) {
-// String deviceName = checkDeviceConnected(deviceEntry.getKey());
-// if (!deviceEntry.getValue().isJsonObject()) {
-// throw new JsonSyntaxException(CAN_T_PARSE_VALUE + json);
-// }
-// long ts = System.currentTimeMillis();
-// BasicAttributesUpdateRequest request = new BasicAttributesUpdateRequest(requestId);
-// JsonObject deviceData = deviceEntry.getValue().getAsJsonObject();
-// request.add(JsonConverter.parseValues(deviceData).stream().map(kv -> new BaseAttributeKvEntry(kv, ts)).collect(Collectors.toList()));
-// GatewayDeviceSessionCtx deviceSessionCtx = devices.get(deviceName);
-// processor.process(new BasicTransportToDeviceSessionActorMsg(deviceSessionCtx.getDevice(),
-// new BasicAdaptorToSessionActorMsg(deviceSessionCtx, request)));
-// }
-// } else {
-// throw new JsonSyntaxException(CAN_T_PARSE_VALUE + json);
-// }
+ JsonElement json = JsonMqttAdaptor.validateJsonPayload(sessionId, mqttMsg.payload());
+ int msgId = mqttMsg.variableHeader().packetId();
+ if (json.isJsonObject()) {
+ JsonObject jsonObj = json.getAsJsonObject();
+ for (Map.Entry deviceEntry : jsonObj.entrySet()) {
+ String deviceName = deviceEntry.getKey();
+ Futures.addCallback(checkDeviceConnected(deviceName),
+ new FutureCallback() {
+ @Override
+ public void onSuccess(@Nullable GatewayDeviceSessionCtx deviceCtx) {
+ if (!deviceEntry.getValue().isJsonObject()) {
+ throw new JsonSyntaxException(CAN_T_PARSE_VALUE + json);
+ }
+ TransportProtos.PostAttributeMsg postAttributeMsg = JsonConverter.convertToAttributesProto(deviceEntry.getValue().getAsJsonObject());
+ transportService.process(deviceCtx.getSessionInfo(), postAttributeMsg, getPubAckCallback(channel, deviceName, msgId, postAttributeMsg));
+ }
+
+ @Override
+ public void onFailure(Throwable t) {
+ log.debug("[{}] Failed to process device teleemtry command: {}", sessionId, deviceName, t);
+ }
+ }, context.getExecutor());
+ }
+ } else {
+ throw new JsonSyntaxException(CAN_T_PARSE_VALUE + json);
+ }
+ }
+
+ public void onDeviceRpcResponse(MqttPublishMessage mqttMsg) throws AdaptorException {
+ JsonElement json = JsonMqttAdaptor.validateJsonPayload(sessionId, mqttMsg.payload());
+ int msgId = mqttMsg.variableHeader().packetId();
+ if (json.isJsonObject()) {
+ JsonObject jsonObj = json.getAsJsonObject();
+ String deviceName = jsonObj.get(DEVICE_PROPERTY).getAsString();
+ Futures.addCallback(checkDeviceConnected(deviceName),
+ new FutureCallback() {
+ @Override
+ public void onSuccess(@Nullable GatewayDeviceSessionCtx deviceCtx) {
+ Integer requestId = jsonObj.get("id").getAsInt();
+ String data = jsonObj.get("data").toString();
+ TransportProtos.ToDeviceRpcResponseMsg rpcResponseMsg = TransportProtos.ToDeviceRpcResponseMsg.newBuilder()
+ .setRequestId(requestId).setPayload(data).build();
+ transportService.process(deviceCtx.getSessionInfo(), rpcResponseMsg, getPubAckCallback(channel, deviceName, msgId, rpcResponseMsg));
+ }
+
+ @Override
+ public void onFailure(Throwable t) {
+ log.debug("[{}] Failed to process device teleemtry command: {}", sessionId, deviceName, t);
+ }
+ }, context.getExecutor());
+ } else {
+ throw new JsonSyntaxException(CAN_T_PARSE_VALUE + json);
+ }
}
public void onDeviceAttributesRequest(MqttPublishMessage msg) throws AdaptorException {
-// JsonElement json = validateJsonPayload(gatewaySessionId, msg.payload());
-// if (json.isJsonObject()) {
-// JsonObject jsonObj = json.getAsJsonObject();
-// int requestId = jsonObj.get("id").getAsInt();
-// String deviceName = jsonObj.get(DEVICE_PROPERTY).getAsString();
-// boolean clientScope = jsonObj.get("client").getAsBoolean();
-// Set keys;
-// if (jsonObj.has("key")) {
-// keys = Collections.singleton(jsonObj.get("key").getAsString());
-// } else {
-// JsonArray keysArray = jsonObj.get("keys").getAsJsonArray();
-// keys = new HashSet<>();
-// for (JsonElement keyObj : keysArray) {
-// keys.add(keyObj.getAsString());
-// }
-// }
-//
-// BasicGetAttributesRequest request;
-// if (clientScope) {
-// request = new BasicGetAttributesRequest(requestId, keys, null);
-// } else {
-// request = new BasicGetAttributesRequest(requestId, null, keys);
-// }
-// GatewayDeviceSessionCtx deviceSessionCtx = devices.get(deviceName);
-// processor.process(new BasicTransportToDeviceSessionActorMsg(deviceSessionCtx.getDevice(),
-// new BasicAdaptorToSessionActorMsg(deviceSessionCtx, request)));
-// ack(msg);
-// } else {
-// throw new JsonSyntaxException(CAN_T_PARSE_VALUE + json);
-// }
+ JsonElement json = JsonMqttAdaptor.validateJsonPayload(sessionId, msg.payload());
+ if (json.isJsonObject()) {
+ JsonObject jsonObj = json.getAsJsonObject();
+ int requestId = jsonObj.get("id").getAsInt();
+ String deviceName = jsonObj.get(DEVICE_PROPERTY).getAsString();
+ boolean clientScope = jsonObj.get("client").getAsBoolean();
+ Set keys;
+ if (jsonObj.has("key")) {
+ keys = Collections.singleton(jsonObj.get("key").getAsString());
+ } else {
+ JsonArray keysArray = jsonObj.get("keys").getAsJsonArray();
+ keys = new HashSet<>();
+ for (JsonElement keyObj : keysArray) {
+ keys.add(keyObj.getAsString());
+ }
+ }
+ TransportProtos.GetAttributeRequestMsg.Builder result = TransportProtos.GetAttributeRequestMsg.newBuilder();
+ result.setRequestId(requestId);
+
+ if (clientScope) {
+ result.addAllClientAttributeNames(keys);
+ } else {
+ result.addAllSharedAttributeNames(keys);
+ }
+ int msgId = msg.variableHeader().packetId();
+ TransportProtos.GetAttributeRequestMsg requestMsg = result.build();
+ Futures.addCallback(checkDeviceConnected(deviceName),
+ new FutureCallback() {
+ @Override
+ public void onSuccess(@Nullable GatewayDeviceSessionCtx deviceCtx) {
+ transportService.process(deviceCtx.getSessionInfo(), requestMsg, getPubAckCallback(channel, deviceName, msgId, requestMsg));
+ }
+
+ @Override
+ public void onFailure(Throwable t) {
+ log.debug("[{}] Failed to process device teleemtry command: {}", sessionId, deviceName, t);
+ }
+ }, context.getExecutor());
+ ack(msg);
+ } else {
+ throw new JsonSyntaxException(CAN_T_PARSE_VALUE + json);
+ }
}
- private String checkDeviceConnected(String deviceName) {
-// if (!devices.containsKey(deviceName)) {
-// log.debug("[{}] Missing device [{}] for the gateway session", gatewaySessionId, deviceName);
-// onDeviceConnect(deviceName, DEFAULT_DEVICE_TYPE);
-// }
-// return deviceName;
- return null;
+ private ListenableFuture checkDeviceConnected(String deviceName) {
+ GatewayDeviceSessionCtx ctx = devices.get(deviceName);
+ if (ctx == null) {
+ log.debug("[{}] Missing device [{}] for the gateway session", sessionId, deviceName);
+ return onDeviceConnect(deviceName, DEFAULT_DEVICE_TYPE);
+ } else {
+ return Futures.immediateFuture(ctx);
+ }
}
private String checkDeviceName(String deviceName) {
@@ -253,32 +309,52 @@ public class GatewaySessionCtx {
}
private JsonElement getJson(MqttPublishMessage mqttMsg) throws AdaptorException {
-// return JsonMqttAdaptor.validateJsonPayload(gatewaySessionId, mqttMsg.payload());
- return null;
+ return JsonMqttAdaptor.validateJsonPayload(sessionId, mqttMsg.payload());
+ }
+
+ private void ack(MqttPublishMessage msg) {
+ if (msg.variableHeader().packetId() > 0) {
+ writeAndFlush(MqttTransportHandler.createMqttPubAckMsg(msg.variableHeader().packetId()));
+ }
}
- protected SessionMsgProcessor getProcessor() {
-// return processor;
- return null;
+ void writeAndFlush(MqttMessage mqttMessage) {
+ channel.writeAndFlush(mqttMessage);
}
- DeviceAuthService getAuthService() {
-// return authService;
- return null;
+ public String getNodeId() {
+ return context.getNodeId();
}
- public void setChannel(ChannelHandlerContext channel) {
- this.channel = channel;
+ private void deregisterSession(String deviceName, GatewayDeviceSessionCtx deviceSessionCtx) {
+ transportService.deregisterSession(deviceSessionCtx.getSessionInfo());
+ transportService.process(deviceSessionCtx.getSessionInfo(), MqttTransportHandler.getSessionEventMsg(TransportProtos.SessionEvent.CLOSED), null);
+ log.debug("[{}] Removed device [{}] from the gateway session", sessionId, deviceName);
}
- private void ack(MqttPublishMessage msg) {
- if (msg.variableHeader().messageId() > 0) {
- writeAndFlush(MqttTransportHandler.createMqttPubAckMsg(msg.variableHeader().messageId()));
- }
+ private TransportServiceCallback getPubAckCallback(final ChannelHandlerContext ctx, final String deviceName, final int msgId, final T msg) {
+ return new TransportServiceCallback() {
+ @Override
+ public void onSuccess(Void dummy) {
+ log.trace("[{}][{}] Published msg: {}", sessionId, deviceName, msg);
+ if (msgId > 0) {
+ ctx.writeAndFlush(MqttTransportHandler.createMqttPubAckMsg(msgId));
+ }
+ }
+
+ @Override
+ public void onError(Throwable e) {
+ log.trace("[{}] Failed to publish msg: {}", sessionId, deviceName, msg, e);
+ ctx.close();
+ }
+ };
}
- void writeAndFlush(MqttMessage mqttMessage) {
- channel.writeAndFlush(mqttMessage);
+ public MqttTransportContext getContext() {
+ return context;
}
+ public MqttTransportAdaptor getAdaptor() {
+ return context.getAdaptor();
+ }
}
diff --git a/transport/mqtt-common/src/main/java/org/thingsboard/server/transport/mqtt/session/MqttDeviceAwareSessionContext.java b/transport/mqtt-common/src/main/java/org/thingsboard/server/transport/mqtt/session/MqttDeviceAwareSessionContext.java
index c01e103889..9e66566383 100644
--- a/transport/mqtt-common/src/main/java/org/thingsboard/server/transport/mqtt/session/MqttDeviceAwareSessionContext.java
+++ b/transport/mqtt-common/src/main/java/org/thingsboard/server/transport/mqtt/session/MqttDeviceAwareSessionContext.java
@@ -31,12 +31,7 @@ public abstract class MqttDeviceAwareSessionContext extends DeviceAwareSessionCo
private final ConcurrentMap mqttQoSMap;
- public MqttDeviceAwareSessionContext(SessionMsgProcessor processor, DeviceAuthService authService, ConcurrentMap mqttQoSMap) {
- super();
- this.mqttQoSMap = mqttQoSMap;
- }
-
- public MqttDeviceAwareSessionContext(SessionMsgProcessor processor, DeviceAuthService authService, Device device, ConcurrentMap mqttQoSMap) {
+ public MqttDeviceAwareSessionContext(ConcurrentMap mqttQoSMap) {
super();
this.mqttQoSMap = mqttQoSMap;
}
diff --git a/transport/mqtt-common/src/main/java/org/thingsboard/server/transport/mqtt/session/MqttSessionId.java b/transport/mqtt-common/src/main/java/org/thingsboard/server/transport/mqtt/session/MqttSessionId.java
index 03006128bf..9256e5d8c0 100644
--- a/transport/mqtt-common/src/main/java/org/thingsboard/server/transport/mqtt/session/MqttSessionId.java
+++ b/transport/mqtt-common/src/main/java/org/thingsboard/server/transport/mqtt/session/MqttSessionId.java
@@ -15,9 +15,6 @@
*/
package org.thingsboard.server.transport.mqtt.session;
-import org.thingsboard.server.common.data.id.SessionId;
-
-import java.util.UUID;
import java.util.concurrent.atomic.AtomicLong;
/**
diff --git a/transport/mqtt-transport/src/main/java/org/thingsboard/server/mqtt/service/MqttTransportService.java b/transport/mqtt-transport/src/main/java/org/thingsboard/server/mqtt/service/MqttTransportService.java
index 15392386c4..bfc69efdc4 100644
--- a/transport/mqtt-transport/src/main/java/org/thingsboard/server/mqtt/service/MqttTransportService.java
+++ b/transport/mqtt-transport/src/main/java/org/thingsboard/server/mqtt/service/MqttTransportService.java
@@ -1,12 +1,12 @@
/**
* 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
- *
+ *
+ * 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.
@@ -25,7 +25,9 @@ import org.springframework.stereotype.Service;
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.gen.transport.TransportProtos;
+import org.thingsboard.server.gen.transport.TransportProtos.*;
+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;
@@ -153,7 +155,7 @@ public class MqttTransportService implements TransportService {
try {
ToTransportMsg toTransportMsg = mainConsumer.decode(record);
if (toTransportMsg.hasToDeviceSessionMsg()) {
- TransportProtos.DeviceActorToTransportMsg toSessionMsg = toTransportMsg.getToDeviceSessionMsg();
+ DeviceActorToTransportMsg toSessionMsg = toTransportMsg.getToDeviceSessionMsg();
UUID sessionId = new UUID(toSessionMsg.getSessionIdMSB(), toSessionMsg.getSessionIdLSB());
SessionMsgListener listener = sessions.get(sessionId);
if (listener != null) {
@@ -227,10 +229,17 @@ public class MqttTransportService implements TransportService {
response -> callback.onSuccess(response.getValidateTokenResponseMsg()), callback::onError, transportCallbackExecutor);
}
+ @Override
+ public void process(GetOrCreateDeviceFromGatewayRequestMsg msg, TransportServiceCallback callback) {
+ AsyncCallbackTemplate.withCallback(transportApiTemplate.post(msg.getTenantIdMSB() + msg.getTenantIdLSB() + msg.getDeviceName(),
+ TransportApiRequestMsg.newBuilder().setGetOrCreateDeviceRequestMsg(msg).build()),
+ response -> callback.onSuccess(response.getGetOrCreateDeviceResponseMsg()), callback::onError, transportCallbackExecutor);
+ }
+
@Override
public void process(SessionInfoProto sessionInfo, SessionEventMsg msg, TransportServiceCallback callback) {
ToRuleEngineMsg toRuleEngineMsg = ToRuleEngineMsg.newBuilder().setToDeviceActorMsg(
- TransportProtos.TransportToDeviceActorMsg.newBuilder().setSessionInfo(sessionInfo)
+ TransportToDeviceActorMsg.newBuilder().setSessionInfo(sessionInfo)
.setSessionEvent(msg).build()
).build();
send(sessionInfo, toRuleEngineMsg, callback);
@@ -239,7 +248,7 @@ public class MqttTransportService implements TransportService {
@Override
public void process(SessionInfoProto sessionInfo, PostTelemetryMsg msg, TransportServiceCallback callback) {
ToRuleEngineMsg toRuleEngineMsg = ToRuleEngineMsg.newBuilder().setToDeviceActorMsg(
- TransportProtos.TransportToDeviceActorMsg.newBuilder().setSessionInfo(sessionInfo)
+ TransportToDeviceActorMsg.newBuilder().setSessionInfo(sessionInfo)
.setPostTelemetry(msg).build()
).build();
send(sessionInfo, toRuleEngineMsg, callback);
@@ -248,52 +257,52 @@ public class MqttTransportService implements TransportService {
@Override
public void process(SessionInfoProto sessionInfo, PostAttributeMsg msg, TransportServiceCallback callback) {
ToRuleEngineMsg toRuleEngineMsg = ToRuleEngineMsg.newBuilder().setToDeviceActorMsg(
- TransportProtos.TransportToDeviceActorMsg.newBuilder().setSessionInfo(sessionInfo)
+ TransportToDeviceActorMsg.newBuilder().setSessionInfo(sessionInfo)
.setPostAttributes(msg).build()
).build();
send(sessionInfo, toRuleEngineMsg, callback);
}
@Override
- public void process(SessionInfoProto sessionInfo, TransportProtos.GetAttributeRequestMsg msg, TransportServiceCallback callback) {
+ public void process(SessionInfoProto sessionInfo, GetAttributeRequestMsg msg, TransportServiceCallback callback) {
ToRuleEngineMsg toRuleEngineMsg = ToRuleEngineMsg.newBuilder().setToDeviceActorMsg(
- TransportProtos.TransportToDeviceActorMsg.newBuilder().setSessionInfo(sessionInfo)
+ TransportToDeviceActorMsg.newBuilder().setSessionInfo(sessionInfo)
.setGetAttributes(msg).build()
).build();
send(sessionInfo, toRuleEngineMsg, callback);
}
@Override
- public void process(SessionInfoProto sessionInfo, TransportProtos.SubscribeToAttributeUpdatesMsg msg, TransportServiceCallback callback) {
+ public void process(SessionInfoProto sessionInfo, SubscribeToAttributeUpdatesMsg msg, TransportServiceCallback callback) {
ToRuleEngineMsg toRuleEngineMsg = ToRuleEngineMsg.newBuilder().setToDeviceActorMsg(
- TransportProtos.TransportToDeviceActorMsg.newBuilder().setSessionInfo(sessionInfo)
+ TransportToDeviceActorMsg.newBuilder().setSessionInfo(sessionInfo)
.setSubscribeToAttributes(msg).build()
).build();
send(sessionInfo, toRuleEngineMsg, callback);
}
@Override
- public void process(SessionInfoProto sessionInfo, TransportProtos.SubscribeToRPCMsg msg, TransportServiceCallback callback) {
+ public void process(SessionInfoProto sessionInfo, SubscribeToRPCMsg msg, TransportServiceCallback callback) {
ToRuleEngineMsg toRuleEngineMsg = ToRuleEngineMsg.newBuilder().setToDeviceActorMsg(
- TransportProtos.TransportToDeviceActorMsg.newBuilder().setSessionInfo(sessionInfo)
+ TransportToDeviceActorMsg.newBuilder().setSessionInfo(sessionInfo)
.setSubscribeToRPC(msg).build()
).build();
send(sessionInfo, toRuleEngineMsg, callback);
}
@Override
- public void process(SessionInfoProto sessionInfo, TransportProtos.ToDeviceRpcResponseMsg msg, TransportServiceCallback callback) {
+ public void process(SessionInfoProto sessionInfo, ToDeviceRpcResponseMsg msg, TransportServiceCallback callback) {
ToRuleEngineMsg toRuleEngineMsg = ToRuleEngineMsg.newBuilder().setToDeviceActorMsg(
- TransportProtos.TransportToDeviceActorMsg.newBuilder().setSessionInfo(sessionInfo)
+ TransportToDeviceActorMsg.newBuilder().setSessionInfo(sessionInfo)
.setToDeviceRPCCallResponse(msg).build()
).build();
send(sessionInfo, toRuleEngineMsg, callback);
}
@Override
- public void process(SessionInfoProto sessionInfo, TransportProtos.ToServerRpcRequestMsg msg, TransportServiceCallback callback) {
+ public void process(SessionInfoProto sessionInfo, ToServerRpcRequestMsg msg, TransportServiceCallback callback) {
ToRuleEngineMsg toRuleEngineMsg = ToRuleEngineMsg.newBuilder().setToDeviceActorMsg(
- TransportProtos.TransportToDeviceActorMsg.newBuilder().setSessionInfo(sessionInfo)
+ TransportToDeviceActorMsg.newBuilder().setSessionInfo(sessionInfo)
.setToServerRPCCallRequest(msg).build()
).build();
send(sessionInfo, toRuleEngineMsg, callback);