Andrew Shvayka
10 years ago
committed by
GitHub
GitHub
25 changed files with 630 additions and 164 deletions
@ -0,0 +1,43 @@ |
|||
/** |
|||
* Copyright © 2016-2017 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.transport.mqtt; |
|||
|
|||
/** |
|||
* Created by ashvayka on 19.01.17. |
|||
*/ |
|||
public class MqttTopics { |
|||
|
|||
public static final String BASE_DEVICE_API_TOPIC = "v1/devices/me"; |
|||
public static final String DEVICE_RPC_RESPONSE_TOPIC = BASE_DEVICE_API_TOPIC + "/rpc/response/"; |
|||
public static final String DEVICE_RPC_RESPONSE_SUB_TOPIC = DEVICE_RPC_RESPONSE_TOPIC + "+"; |
|||
public static final String DEVICE_RPC_REQUESTS_TOPIC = BASE_DEVICE_API_TOPIC + "/rpc/request/"; |
|||
public static final String DEVICE_RPC_REQUESTS_SUB_TOPIC = DEVICE_RPC_REQUESTS_TOPIC + "+"; |
|||
public static final String DEVICE_ATTRIBUTES_RESPONSE_TOPIC_PREFIX = BASE_DEVICE_API_TOPIC + "/attributes/response/"; |
|||
public static final String DEVICE_ATTRIBUTES_RESPONSES_TOPIC = DEVICE_ATTRIBUTES_RESPONSE_TOPIC_PREFIX + "+"; |
|||
public static final String DEVICE_ATTRIBUTES_REQUEST_TOPIC_PREFIX = BASE_DEVICE_API_TOPIC + "/attributes/request/"; |
|||
public static final String DEVICE_TELEMETRY_TOPIC = BASE_DEVICE_API_TOPIC + "/telemetry"; |
|||
public static final String DEVICE_ATTRIBUTES_TOPIC = BASE_DEVICE_API_TOPIC + "/attributes"; |
|||
|
|||
public static final String BASE_GATEWAY_API_TOPIC = "v1/gateway"; |
|||
public static final String GATEWAY_CONNECT_TOPIC = "v1/gateway/connect"; |
|||
public static final String GATEWAY_DISCONNECT_TOPIC = "v1/gateway/disconnect"; |
|||
public static final String GATEWAY_ATTRIBUTES_TOPIC = "v1/gateway/attributes"; |
|||
public static final String GATEWAY_TELEMETRY_TOPIC = "v1/gateway/telemetry"; |
|||
|
|||
|
|||
private MqttTopics() { |
|||
} |
|||
} |
|||
@ -0,0 +1,36 @@ |
|||
/** |
|||
* Copyright © 2016-2017 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.transport.mqtt.adaptors; |
|||
|
|||
import io.netty.handler.codec.mqtt.MqttMessage; |
|||
import org.thingsboard.server.common.msg.session.AdaptorToSessionActorMsg; |
|||
import org.thingsboard.server.common.msg.session.MsgType; |
|||
import org.thingsboard.server.common.msg.session.SessionActorToAdaptorMsg; |
|||
import org.thingsboard.server.common.transport.adaptor.AdaptorException; |
|||
import org.thingsboard.server.transport.mqtt.session.GatewaySessionCtx; |
|||
|
|||
import java.util.Optional; |
|||
|
|||
/** |
|||
* Created by ashvayka on 19.01.17. |
|||
*/ |
|||
public interface MqttGatewayAdaptor { |
|||
|
|||
AdaptorToSessionActorMsg convertToActorMsg(GatewaySessionCtx ctx, MsgType type, MqttMessage inbound) throws AdaptorException; |
|||
|
|||
Optional<MqttMessage> convertToAdaptorMsg(GatewaySessionCtx ctx, SessionActorToAdaptorMsg msg) throws AdaptorException; |
|||
|
|||
} |
|||
@ -0,0 +1,95 @@ |
|||
/** |
|||
* Copyright © 2016-2017 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.transport.mqtt.session; |
|||
|
|||
import io.netty.handler.codec.mqtt.MqttMessage; |
|||
import org.thingsboard.server.common.data.Device; |
|||
import org.thingsboard.server.common.data.id.SessionId; |
|||
import org.thingsboard.server.common.msg.core.ResponseMsg; |
|||
import org.thingsboard.server.common.msg.session.*; |
|||
import org.thingsboard.server.common.msg.session.ex.SessionException; |
|||
import org.thingsboard.server.common.transport.session.DeviceAwareSessionContext; |
|||
import org.thingsboard.server.transport.mqtt.MqttTransportHandler; |
|||
|
|||
import java.util.Optional; |
|||
|
|||
/** |
|||
* Created by ashvayka on 19.01.17. |
|||
*/ |
|||
public class GatewayDeviceSessionCtx extends DeviceAwareSessionContext { |
|||
|
|||
private GatewaySessionCtx parent; |
|||
private final MqttSessionId sessionId; |
|||
private volatile boolean closed; |
|||
|
|||
public GatewayDeviceSessionCtx(GatewaySessionCtx parent, Device device) { |
|||
super(parent.getProcessor(), parent.getAuthService(), device); |
|||
this.parent = parent; |
|||
this.sessionId = new MqttSessionId(); |
|||
} |
|||
|
|||
@Override |
|||
public SessionId getSessionId() { |
|||
return sessionId; |
|||
} |
|||
|
|||
@Override |
|||
public SessionType getSessionType() { |
|||
return SessionType.ASYNC; |
|||
} |
|||
|
|||
@Override |
|||
public void onMsg(SessionActorToAdaptorMsg sessionMsg) throws SessionException { |
|||
Optional<MqttMessage> message = getToDeviceMsg(sessionMsg); |
|||
message.ifPresent(parent::writeAndFlush); |
|||
} |
|||
|
|||
private Optional<MqttMessage> getToDeviceMsg(SessionActorToAdaptorMsg sessionMsg) { |
|||
ToDeviceMsg msg = sessionMsg.getMsg(); |
|||
switch (msg.getMsgType()) { |
|||
case STATUS_CODE_RESPONSE: |
|||
ResponseMsg<?> responseMsg = (ResponseMsg) msg; |
|||
if (responseMsg.isSuccess()) { |
|||
MsgType requestMsgType = responseMsg.getRequestMsgType(); |
|||
Integer requestId = responseMsg.getRequestId(); |
|||
if (requestMsgType == MsgType.POST_ATTRIBUTES_REQUEST || requestMsgType == MsgType.POST_TELEMETRY_REQUEST) { |
|||
return Optional.of(MqttTransportHandler.createMqttPubAckMsg(requestId)); |
|||
} |
|||
} |
|||
break; |
|||
} |
|||
return Optional.empty(); |
|||
} |
|||
|
|||
@Override |
|||
public void onMsg(SessionCtrlMsg msg) throws SessionException { |
|||
|
|||
} |
|||
|
|||
@Override |
|||
public boolean isClosed() { |
|||
return closed; |
|||
} |
|||
|
|||
public void setClosed(boolean closed) { |
|||
this.closed = closed; |
|||
} |
|||
|
|||
@Override |
|||
public long getTimeout() { |
|||
return 0; |
|||
} |
|||
} |
|||
@ -0,0 +1,191 @@ |
|||
/** |
|||
* Copyright © 2016-2017 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.transport.mqtt.session; |
|||
|
|||
import com.google.gson.*; |
|||
import io.netty.buffer.ByteBufAllocator; |
|||
import io.netty.buffer.UnpooledByteBufAllocator; |
|||
import io.netty.channel.ChannelHandlerContext; |
|||
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.msg.core.BasicTelemetryUploadRequest; |
|||
import org.thingsboard.server.common.msg.core.BasicUpdateAttributesRequest; |
|||
import org.thingsboard.server.common.msg.core.TelemetryUploadRequest; |
|||
import org.thingsboard.server.common.msg.session.BasicAdaptorToSessionActorMsg; |
|||
import org.thingsboard.server.common.msg.session.BasicToDeviceActorSessionMsg; |
|||
import org.thingsboard.server.common.msg.session.ctrl.SessionCloseMsg; |
|||
import org.thingsboard.server.common.transport.SessionMsgProcessor; |
|||
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.transport.mqtt.MqttTransportHandler; |
|||
import org.thingsboard.server.transport.mqtt.adaptors.JsonMqttAdaptor; |
|||
|
|||
import java.nio.charset.Charset; |
|||
import java.util.HashMap; |
|||
import java.util.Map; |
|||
import java.util.Optional; |
|||
import java.util.stream.Collectors; |
|||
|
|||
import static org.thingsboard.server.transport.mqtt.adaptors.JsonMqttAdaptor.validateJsonPayload; |
|||
|
|||
/** |
|||
* Created by ashvayka on 19.01.17. |
|||
*/ |
|||
@Slf4j |
|||
public class GatewaySessionCtx { |
|||
|
|||
private static final Gson GSON = new Gson(); |
|||
private static final Charset UTF8 = Charset.forName("UTF-8"); |
|||
private static final ByteBufAllocator ALLOCATOR = new UnpooledByteBufAllocator(false); |
|||
|
|||
private final Device gateway; |
|||
private final SessionId gatewaySessionId; |
|||
private final SessionMsgProcessor processor; |
|||
private final DeviceService deviceService; |
|||
private final DeviceAuthService authService; |
|||
private final Map<String, GatewayDeviceSessionCtx> devices; |
|||
private ChannelHandlerContext channel; |
|||
|
|||
public GatewaySessionCtx(SessionMsgProcessor processor, DeviceService deviceService, DeviceAuthService authService, DeviceSessionCtx gatewaySessionCtx) { |
|||
this.processor = processor; |
|||
this.deviceService = deviceService; |
|||
this.authService = authService; |
|||
this.gateway = gatewaySessionCtx.getDevice(); |
|||
this.gatewaySessionId = gatewaySessionCtx.getSessionId(); |
|||
this.devices = new HashMap<>(); |
|||
} |
|||
|
|||
public void onDeviceConnect(MqttPublishMessage msg) throws AdaptorException { |
|||
String deviceName = checkDeviceName(getDeviceName(msg)); |
|||
Optional<Device> deviceOpt = deviceService.findDeviceByTenantIdAndName(gateway.getTenantId(), deviceName); |
|||
Device device = deviceOpt.orElseGet(() -> { |
|||
Device newDevice = new Device(); |
|||
newDevice.setTenantId(gateway.getTenantId()); |
|||
newDevice.setName(deviceName); |
|||
return deviceService.saveDevice(newDevice); |
|||
}); |
|||
devices.put(deviceName, new GatewayDeviceSessionCtx(this, device)); |
|||
ack(msg); |
|||
} |
|||
|
|||
public void onDeviceDisconnect(MqttPublishMessage msg) throws AdaptorException { |
|||
String deviceName = checkDeviceName(getDeviceName(msg)); |
|||
GatewayDeviceSessionCtx deviceSessionCtx = devices.remove(deviceName); |
|||
deviceSessionCtx.setClosed(true); |
|||
ack(msg); |
|||
} |
|||
|
|||
public void onGatewayDisconnect() { |
|||
devices.forEach((k, v) -> { |
|||
processor.process(SessionCloseMsg.onDisconnect(v.getSessionId())); |
|||
}); |
|||
} |
|||
|
|||
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<String, JsonElement> 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 BasicToDeviceActorSessionMsg(deviceSessionCtx.getDevice(), |
|||
new BasicAdaptorToSessionActorMsg(deviceSessionCtx, request))); |
|||
} |
|||
} 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<String, JsonElement> deviceEntry : jsonObj.entrySet()) { |
|||
String deviceName = checkDeviceConnected(deviceEntry.getKey()); |
|||
if (!deviceEntry.getValue().isJsonObject()) { |
|||
throw new JsonSyntaxException("Can't parse value: " + json); |
|||
} |
|||
long ts = System.currentTimeMillis(); |
|||
BasicUpdateAttributesRequest request = new BasicUpdateAttributesRequest(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 BasicToDeviceActorSessionMsg(deviceSessionCtx.getDevice(), |
|||
new BasicAdaptorToSessionActorMsg(deviceSessionCtx, request))); |
|||
} |
|||
} else { |
|||
throw new JsonSyntaxException("Can't parse value: " + json); |
|||
} |
|||
} |
|||
|
|||
private String checkDeviceConnected(String deviceName) { |
|||
if (!devices.containsKey(deviceName)) { |
|||
throw new RuntimeException("Device is not connected!"); |
|||
} else { |
|||
return deviceName; |
|||
} |
|||
} |
|||
|
|||
private String checkDeviceName(String deviceName) { |
|||
if (StringUtils.isEmpty(deviceName)) { |
|||
throw new RuntimeException("Device name is empty!"); |
|||
} else { |
|||
return deviceName; |
|||
} |
|||
} |
|||
|
|||
private String getDeviceName(MqttPublishMessage mqttMsg) throws AdaptorException { |
|||
JsonElement json = JsonMqttAdaptor.validateJsonPayload(gatewaySessionId, mqttMsg.payload()); |
|||
return json.getAsJsonObject().get("device").getAsString(); |
|||
} |
|||
|
|||
protected SessionMsgProcessor getProcessor() { |
|||
return processor; |
|||
} |
|||
|
|||
protected DeviceAuthService getAuthService() { |
|||
return authService; |
|||
} |
|||
|
|||
public void setChannel(ChannelHandlerContext channel) { |
|||
this.channel = channel; |
|||
} |
|||
|
|||
private void ack(MqttPublishMessage msg) { |
|||
writeAndFlush(MqttTransportHandler.createMqttPubAckMsg(msg.variableHeader().messageId())); |
|||
} |
|||
|
|||
protected void writeAndFlush(MqttMessage mqttMessage) { |
|||
channel.writeAndFlush(mqttMessage); |
|||
} |
|||
} |
|||
Loading…
Reference in new issue