Andrew Shvayka
8 years ago
42 changed files with 805 additions and 355 deletions
@ -0,0 +1,12 @@ |
|||
package org.thingsboard.server.kafka; |
|||
|
|||
import java.util.function.Consumer; |
|||
|
|||
/** |
|||
* Created by ashvayka on 05.10.18. |
|||
*/ |
|||
public interface TbKafkaHandler<Request, Response> { |
|||
|
|||
void handle(Request request, Consumer<Response> onSuccess, Consumer<Throwable> onFailure); |
|||
|
|||
} |
|||
@ -0,0 +1,173 @@ |
|||
/** |
|||
* Copyright © 2016-2018 The Thingsboard Authors |
|||
* <p> |
|||
* 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 |
|||
* <p> |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* <p> |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.kafka; |
|||
|
|||
import com.google.common.util.concurrent.Futures; |
|||
import com.google.common.util.concurrent.ListenableFuture; |
|||
import com.google.common.util.concurrent.SettableFuture; |
|||
import lombok.Builder; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.apache.kafka.clients.admin.CreateTopicsResult; |
|||
import org.apache.kafka.clients.admin.NewTopic; |
|||
import org.apache.kafka.clients.consumer.ConsumerRecords; |
|||
import org.apache.kafka.common.header.Header; |
|||
import org.apache.kafka.common.header.internals.RecordHeader; |
|||
|
|||
import java.io.IOException; |
|||
import java.nio.ByteBuffer; |
|||
import java.nio.charset.StandardCharsets; |
|||
import java.time.Duration; |
|||
import java.util.ArrayList; |
|||
import java.util.List; |
|||
import java.util.UUID; |
|||
import java.util.concurrent.ConcurrentHashMap; |
|||
import java.util.concurrent.ConcurrentMap; |
|||
import java.util.concurrent.ExecutorService; |
|||
import java.util.concurrent.Executors; |
|||
import java.util.concurrent.TimeoutException; |
|||
|
|||
/** |
|||
* Created by ashvayka on 25.09.18. |
|||
*/ |
|||
@Slf4j |
|||
public class TbKafkaResponseTemplate<Request, Response> { |
|||
|
|||
private final TBKafkaConsumerTemplate<Request> requestTemplate; |
|||
private final TBKafkaProducerTemplate<Response> responseTemplate; |
|||
private final TbKafkaHandler<Request, Response> handler; |
|||
private final ConcurrentMap<UUID, String> pendingRequests; |
|||
private final ExecutorService executor; |
|||
private final long maxPendingRequests; |
|||
|
|||
private final long pollInterval; |
|||
private volatile boolean stopped = false; |
|||
|
|||
@Builder |
|||
public TbKafkaResponseTemplate(TBKafkaConsumerTemplate<Request> requestTemplate, |
|||
TBKafkaProducerTemplate<Response> responseTemplate, |
|||
TbKafkaHandler<Request, Response> handler, |
|||
long pollInterval, |
|||
long maxPendingRequests, |
|||
ExecutorService executor) { |
|||
this.requestTemplate = requestTemplate; |
|||
this.responseTemplate = responseTemplate; |
|||
this.handler = handler; |
|||
this.pendingRequests = new ConcurrentHashMap<>(); |
|||
this.maxPendingRequests = maxPendingRequests; |
|||
this.pollInterval = pollInterval; |
|||
this.executor = executor; |
|||
} |
|||
|
|||
public void init() { |
|||
this.responseTemplate.init(); |
|||
requestTemplate.subscribe(); |
|||
executor.submit(() -> { |
|||
long nextCleanupMs = 0L; |
|||
while (!stopped) { |
|||
ConsumerRecords<String, byte[]> requests = requestTemplate.poll(Duration.ofMillis(pollInterval)); |
|||
requests.forEach(request -> { |
|||
Header requestIdHeader = request.headers().lastHeader(TbKafkaSettings.REQUEST_ID_HEADER); |
|||
if (requestIdHeader == null) { |
|||
log.error("[{}] Missing requestId in header", request); |
|||
return; |
|||
} |
|||
UUID requestId = bytesToUuid(requestIdHeader.value()); |
|||
if (requestId == null) { |
|||
log.error("[{}] Missing requestId in header and body", request); |
|||
return; |
|||
} |
|||
Header responseTopicHeader = request.headers().lastHeader(TbKafkaSettings.RESPONSE_TOPIC_HEADER); |
|||
if (responseTopicHeader == null) { |
|||
log.error("[{}] Missing response topic in header", request); |
|||
return; |
|||
} |
|||
String responseTopic = bytesToUuid(responseTopicHeader.value()); |
|||
if (requestId == null) { |
|||
log.error("[{}] Missing requestId in header and body", request); |
|||
return; |
|||
} |
|||
|
|||
Request decodedRequest = null; |
|||
String responseTopic = null; |
|||
|
|||
try { |
|||
if (decodedRequest == null) { |
|||
decodedRequest = requestTemplate.decode(request); |
|||
} |
|||
executor.submit(() -> { |
|||
handler.handle(decodedRequest, ); |
|||
}); |
|||
} catch (IOException e) { |
|||
expectedRequest.future.setException(e); |
|||
} |
|||
|
|||
}); |
|||
} |
|||
}); |
|||
} |
|||
|
|||
public void stop() { |
|||
stopped = true; |
|||
} |
|||
|
|||
public ListenableFuture<Response> post(String key, Request request) { |
|||
if (tickSize > maxPendingRequests) { |
|||
return Futures.immediateFailedFuture(new RuntimeException("Pending request map is full!")); |
|||
} |
|||
UUID requestId = UUID.randomUUID(); |
|||
List<Header> headers = new ArrayList<>(2); |
|||
headers.add(new RecordHeader(TbKafkaSettings.REQUEST_ID_HEADER, uuidToBytes(requestId))); |
|||
headers.add(new RecordHeader(TbKafkaSettings.RESPONSE_TOPIC_HEADER, stringToBytes(responseTemplate.getTopic()))); |
|||
SettableFuture<Response> future = SettableFuture.create(); |
|||
pendingRequests.putIfAbsent(requestId, new ResponseMetaData<>(tickTs + maxRequestTimeout, future)); |
|||
request = requestTemplate.enrich(request, responseTemplate.getTopic(), requestId); |
|||
requestTemplate.send(key, request, headers); |
|||
return future; |
|||
} |
|||
|
|||
private byte[] uuidToBytes(UUID uuid) { |
|||
ByteBuffer buf = ByteBuffer.allocate(16); |
|||
buf.putLong(uuid.getMostSignificantBits()); |
|||
buf.putLong(uuid.getLeastSignificantBits()); |
|||
return buf.array(); |
|||
} |
|||
|
|||
private static UUID bytesToUuid(byte[] bytes) { |
|||
ByteBuffer bb = ByteBuffer.wrap(bytes); |
|||
long firstLong = bb.getLong(); |
|||
long secondLong = bb.getLong(); |
|||
return new UUID(firstLong, secondLong); |
|||
} |
|||
|
|||
private byte[] stringToBytes(String string) { |
|||
return string.getBytes(StandardCharsets.UTF_8); |
|||
} |
|||
|
|||
private String bytesToString(byte[] data) { |
|||
return new String(data, StandardCharsets.UTF_8); |
|||
} |
|||
|
|||
private static class ResponseMetaData<T> { |
|||
private final long expTime; |
|||
private final SettableFuture<T> future; |
|||
|
|||
ResponseMetaData(long ts, SettableFuture<T> future) { |
|||
this.expTime = ts; |
|||
this.future = future; |
|||
} |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,284 @@ |
|||
/** |
|||
* Copyright © 2016-2018 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.transport.mqtt.session; |
|||
|
|||
import com.google.gson.JsonArray; |
|||
import com.google.gson.JsonElement; |
|||
import com.google.gson.JsonNull; |
|||
import com.google.gson.JsonObject; |
|||
import com.google.gson.JsonSyntaxException; |
|||
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.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.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.transport.mqtt.MqttTransportHandler; |
|||
import org.thingsboard.server.transport.mqtt.adaptors.JsonMqttAdaptor; |
|||
|
|||
import java.util.*; |
|||
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. |
|||
*/ |
|||
@Slf4j |
|||
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<String, GatewayDeviceSessionCtx> devices;
|
|||
// private final ConcurrentMap<String, Integer> 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) { |
|||
|
|||
} |
|||
|
|||
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); |
|||
} |
|||
|
|||
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())));
|
|||
// }
|
|||
} |
|||
|
|||
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);
|
|||
} |
|||
|
|||
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 BasicTransportToDeviceSessionActorMsg(deviceSessionCtx.getDevice(),
|
|||
// new BasicAdaptorToSessionActorMsg(deviceSessionCtx, request)));
|
|||
// }
|
|||
// } else {
|
|||
// throw new JsonSyntaxException(CAN_T_PARSE_VALUE + json);
|
|||
// }
|
|||
} |
|||
|
|||
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);
|
|||
// }
|
|||
} |
|||
|
|||
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();
|
|||
// 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);
|
|||
// }
|
|||
} |
|||
|
|||
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<String> 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);
|
|||
// }
|
|||
} |
|||
|
|||
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 String checkDeviceName(String deviceName) { |
|||
if (StringUtils.isEmpty(deviceName)) { |
|||
throw new RuntimeException("Device name is empty!"); |
|||
} else { |
|||
return deviceName; |
|||
} |
|||
} |
|||
|
|||
private String getDeviceName(JsonElement json) throws AdaptorException { |
|||
return json.getAsJsonObject().get(DEVICE_PROPERTY).getAsString(); |
|||
} |
|||
|
|||
private String getDeviceType(JsonElement json) throws AdaptorException { |
|||
JsonElement type = json.getAsJsonObject().get("type"); |
|||
return type == null || type instanceof JsonNull ? DEFAULT_DEVICE_TYPE : type.getAsString(); |
|||
} |
|||
|
|||
private JsonElement getJson(MqttPublishMessage mqttMsg) throws AdaptorException { |
|||
// return JsonMqttAdaptor.validateJsonPayload(gatewaySessionId, mqttMsg.payload());
|
|||
return null; |
|||
} |
|||
|
|||
protected SessionMsgProcessor getProcessor() { |
|||
// return processor;
|
|||
return null; |
|||
} |
|||
|
|||
DeviceAuthService getAuthService() { |
|||
// return authService;
|
|||
return null; |
|||
} |
|||
|
|||
public void setChannel(ChannelHandlerContext channel) { |
|||
this.channel = channel; |
|||
} |
|||
|
|||
private void ack(MqttPublishMessage msg) { |
|||
if (msg.variableHeader().messageId() > 0) { |
|||
writeAndFlush(MqttTransportHandler.createMqttPubAckMsg(msg.variableHeader().messageId())); |
|||
} |
|||
} |
|||
|
|||
void writeAndFlush(MqttMessage mqttMessage) { |
|||
channel.writeAndFlush(mqttMessage); |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,45 @@ |
|||
package org.thingsboard.server.mqtt.service; |
|||
|
|||
import com.google.common.util.concurrent.FutureCallback; |
|||
import com.google.common.util.concurrent.Futures; |
|||
import com.google.common.util.concurrent.ListenableFuture; |
|||
|
|||
import javax.annotation.Nullable; |
|||
import java.util.concurrent.Executor; |
|||
import java.util.function.Consumer; |
|||
|
|||
/** |
|||
* Created by ashvayka on 05.10.18. |
|||
*/ |
|||
public class AsyncCallbackTemplate { |
|||
|
|||
public static <T> void withCallback(ListenableFuture<T> future, Consumer<T> onSuccess, |
|||
Consumer<Throwable> onFailure) { |
|||
withCallback(future, onSuccess, onFailure, null); |
|||
} |
|||
|
|||
public static <T> void withCallback(ListenableFuture<T> future, Consumer<T> onSuccess, |
|||
Consumer<Throwable> onFailure, Executor executor) { |
|||
FutureCallback<T> callback = new FutureCallback<T>() { |
|||
@Override |
|||
public void onSuccess(@Nullable T result) { |
|||
try { |
|||
onSuccess.accept(result); |
|||
} catch (Throwable th) { |
|||
onFailure(th); |
|||
} |
|||
} |
|||
|
|||
@Override |
|||
public void onFailure(Throwable t) { |
|||
onFailure.accept(t); |
|||
} |
|||
}; |
|||
if (executor != null) { |
|||
Futures.addCallback(future, callback, executor); |
|||
} else { |
|||
Futures.addCallback(future, callback); |
|||
} |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,110 @@ |
|||
package org.thingsboard.server.mqtt.service; |
|||
|
|||
import org.springframework.beans.factory.annotation.Autowired; |
|||
import org.springframework.beans.factory.annotation.Value; |
|||
import org.springframework.stereotype.Service; |
|||
import org.thingsboard.server.common.transport.TransportService; |
|||
import org.thingsboard.server.common.transport.TransportServiceCallback; |
|||
import org.thingsboard.server.kafka.TBKafkaConsumerTemplate; |
|||
import org.thingsboard.server.kafka.TBKafkaProducerTemplate; |
|||
import org.thingsboard.server.kafka.TbKafkaRequestTemplate; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.*; |
|||
import org.thingsboard.server.kafka.TbKafkaSettings; |
|||
import org.thingsboard.server.transport.mqtt.MqttTransportContext; |
|||
|
|||
import javax.annotation.PostConstruct; |
|||
import javax.annotation.PreDestroy; |
|||
import java.util.concurrent.ExecutorService; |
|||
import java.util.concurrent.Executors; |
|||
|
|||
/** |
|||
* Created by ashvayka on 05.10.18. |
|||
*/ |
|||
@Service |
|||
public class MqttTransportService implements TransportService { |
|||
|
|||
@Value("${kafka.rule-engine.topic}") |
|||
private String ruleEngineTopic; |
|||
@Value("${kafka.transport-api.requests-topic}") |
|||
private String transportApiRequestsTopic; |
|||
@Value("${kafka.transport-api.responses-topic}") |
|||
private String transportApiResponsesTopic; |
|||
@Value("${kafka.transport-api.max_pending_requests}") |
|||
private long maxPendingRequests; |
|||
@Value("${kafka.transport-api.max_requests_timeout}") |
|||
private long maxRequestsTimeout; |
|||
@Value("${kafka.transport-api.response_poll_interval}") |
|||
private int responsePollDuration; |
|||
@Value("${kafka.transport-api.response_auto_commit_interval}") |
|||
private int autoCommitInterval; |
|||
|
|||
@Autowired |
|||
private TbKafkaSettings kafkaSettings; |
|||
//We use this to get the node id. We should replace this with a component that provides the node id.
|
|||
@Autowired |
|||
private MqttTransportContext transportContext; |
|||
|
|||
private ExecutorService transportCallbackExecutor; |
|||
|
|||
private TbKafkaRequestTemplate<TransportApiRequestMsg, TransportApiResponseMsg> transportApiTemplate; |
|||
|
|||
@PostConstruct |
|||
public void init() { |
|||
this.transportCallbackExecutor = Executors.newCachedThreadPool(); |
|||
|
|||
TBKafkaProducerTemplate.TBKafkaProducerTemplateBuilder<TransportApiRequestMsg> requestBuilder = TBKafkaProducerTemplate.builder(); |
|||
requestBuilder.settings(kafkaSettings); |
|||
requestBuilder.defaultTopic(transportApiRequestsTopic); |
|||
requestBuilder.encoder(new TransportApiRequestEncoder()); |
|||
|
|||
TBKafkaConsumerTemplate.TBKafkaConsumerTemplateBuilder<TransportApiResponseMsg> responseBuilder = TBKafkaConsumerTemplate.builder(); |
|||
responseBuilder.settings(kafkaSettings); |
|||
responseBuilder.topic(transportApiResponsesTopic + "." + transportContext.getNodeId()); |
|||
responseBuilder.clientId(transportContext.getNodeId()); |
|||
responseBuilder.groupId("transport-node"); |
|||
responseBuilder.autoCommit(true); |
|||
responseBuilder.autoCommitIntervalMs(autoCommitInterval); |
|||
responseBuilder.decoder(new TransportApiResponseDecoder()); |
|||
|
|||
TbKafkaRequestTemplate.TbKafkaRequestTemplateBuilder |
|||
<TransportApiRequestMsg, TransportApiResponseMsg> builder = TbKafkaRequestTemplate.builder(); |
|||
builder.requestTemplate(requestBuilder.build()); |
|||
builder.responseTemplate(responseBuilder.build()); |
|||
builder.maxPendingRequests(maxPendingRequests); |
|||
builder.maxRequestTimeout(maxRequestsTimeout); |
|||
builder.pollInterval(responsePollDuration); |
|||
transportApiTemplate = builder.build(); |
|||
transportApiTemplate.init(); |
|||
} |
|||
|
|||
@PreDestroy |
|||
public void destroy() { |
|||
if (transportApiTemplate != null) { |
|||
transportApiTemplate.stop(); |
|||
} |
|||
if (transportCallbackExecutor != null) { |
|||
transportCallbackExecutor.shutdownNow(); |
|||
} |
|||
} |
|||
|
|||
@Override |
|||
public void process(ValidateDeviceTokenRequestMsg msg, TransportServiceCallback<ValidateDeviceTokenResponseMsg> callback) { |
|||
AsyncCallbackTemplate.withCallback(transportApiTemplate.post(msg.getToken(), TransportApiRequestMsg.newBuilder().setValidateTokenRequestMsg(msg).build()), |
|||
response -> callback.onSuccess(response.getValidateTokenResponseMsg()), callback::onError, transportCallbackExecutor); |
|||
} |
|||
|
|||
@Override |
|||
public void process(SessionEventMsg msg, TransportServiceCallback<Void> callback) { |
|||
|
|||
} |
|||
|
|||
@Override |
|||
public void process(PostTelemetryMsg msg, TransportServiceCallback<Void> callback) { |
|||
|
|||
} |
|||
|
|||
@Override |
|||
public void process(PostAttributeMsg msg, TransportServiceCallback<Void> callback) { |
|||
|
|||
} |
|||
} |
|||
@ -0,0 +1,14 @@ |
|||
package org.thingsboard.server.mqtt.service; |
|||
|
|||
import org.thingsboard.server.gen.transport.TransportProtos.TransportApiRequestMsg; |
|||
import org.thingsboard.server.kafka.TbKafkaEncoder; |
|||
|
|||
/** |
|||
* Created by ashvayka on 05.10.18. |
|||
*/ |
|||
public class TransportApiRequestEncoder implements TbKafkaEncoder<TransportApiRequestMsg> { |
|||
@Override |
|||
public byte[] encode(TransportApiRequestMsg value) { |
|||
return value.toByteArray(); |
|||
} |
|||
} |
|||
@ -0,0 +1,16 @@ |
|||
package org.thingsboard.server.mqtt.service; |
|||
|
|||
import org.thingsboard.server.gen.transport.TransportProtos.TransportApiResponseMsg; |
|||
import org.thingsboard.server.kafka.TbKafkaDecoder; |
|||
|
|||
import java.io.IOException; |
|||
|
|||
/** |
|||
* Created by ashvayka on 05.10.18. |
|||
*/ |
|||
public class TransportApiResponseDecoder implements TbKafkaDecoder<TransportApiResponseMsg> { |
|||
@Override |
|||
public TransportApiResponseMsg decode(byte[] data) throws IOException { |
|||
return TransportApiResponseMsg.parseFrom(data); |
|||
} |
|||
} |
|||
@ -1,280 +0,0 @@ |
|||
/** |
|||
* Copyright © 2016-2018 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.transport.mqtt.session; |
|||
|
|||
import com.google.gson.JsonArray; |
|||
import com.google.gson.JsonElement; |
|||
import com.google.gson.JsonNull; |
|||
import com.google.gson.JsonObject; |
|||
import com.google.gson.JsonSyntaxException; |
|||
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.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.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.transport.mqtt.MqttTransportHandler; |
|||
import org.thingsboard.server.transport.mqtt.adaptors.JsonMqttAdaptor; |
|||
|
|||
import java.util.*; |
|||
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. |
|||
*/ |
|||
@Slf4j |
|||
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<String, GatewayDeviceSessionCtx> devices; |
|||
private final ConcurrentMap<String, Integer> 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) { |
|||
|
|||
} |
|||
|
|||
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); |
|||
} |
|||
|
|||
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()))); |
|||
} |
|||
} |
|||
|
|||
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); |
|||
} |
|||
|
|||
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 BasicTransportToDeviceSessionActorMsg(deviceSessionCtx.getDevice(), |
|||
new BasicAdaptorToSessionActorMsg(deviceSessionCtx, request))); |
|||
} |
|||
} else { |
|||
throw new JsonSyntaxException(CAN_T_PARSE_VALUE + json); |
|||
} |
|||
} |
|||
|
|||
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); |
|||
} |
|||
} |
|||
|
|||
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(); |
|||
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); |
|||
} |
|||
} |
|||
|
|||
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<String> 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); |
|||
} |
|||
} |
|||
|
|||
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; |
|||
} |
|||
|
|||
private String checkDeviceName(String deviceName) { |
|||
if (StringUtils.isEmpty(deviceName)) { |
|||
throw new RuntimeException("Device name is empty!"); |
|||
} else { |
|||
return deviceName; |
|||
} |
|||
} |
|||
|
|||
private String getDeviceName(JsonElement json) throws AdaptorException { |
|||
return json.getAsJsonObject().get(DEVICE_PROPERTY).getAsString(); |
|||
} |
|||
|
|||
private String getDeviceType(JsonElement json) throws AdaptorException { |
|||
JsonElement type = json.getAsJsonObject().get("type"); |
|||
return type == null || type instanceof JsonNull ? DEFAULT_DEVICE_TYPE : type.getAsString(); |
|||
} |
|||
|
|||
private JsonElement getJson(MqttPublishMessage mqttMsg) throws AdaptorException { |
|||
return JsonMqttAdaptor.validateJsonPayload(gatewaySessionId, mqttMsg.payload()); |
|||
} |
|||
|
|||
protected SessionMsgProcessor getProcessor() { |
|||
return processor; |
|||
} |
|||
|
|||
DeviceAuthService getAuthService() { |
|||
return authService; |
|||
} |
|||
|
|||
public void setChannel(ChannelHandlerContext channel) { |
|||
this.channel = channel; |
|||
} |
|||
|
|||
private void ack(MqttPublishMessage msg) { |
|||
if (msg.variableHeader().messageId() > 0) { |
|||
writeAndFlush(MqttTransportHandler.createMqttPubAckMsg(msg.variableHeader().messageId())); |
|||
} |
|||
} |
|||
|
|||
void writeAndFlush(MqttMessage mqttMessage) { |
|||
channel.writeAndFlush(mqttMessage); |
|||
} |
|||
|
|||
} |
|||
Loading…
Reference in new issue