|
|
@ -15,11 +15,20 @@ |
|
|
*/ |
|
|
*/ |
|
|
package org.thingsboard.server.common.transport.service; |
|
|
package org.thingsboard.server.common.transport.service; |
|
|
|
|
|
|
|
|
|
|
|
import com.google.gson.Gson; |
|
|
|
|
|
import com.google.gson.JsonObject; |
|
|
|
|
|
import com.google.protobuf.ByteString; |
|
|
import lombok.extern.slf4j.Slf4j; |
|
|
import lombok.extern.slf4j.Slf4j; |
|
|
import org.springframework.beans.factory.annotation.Value; |
|
|
import org.springframework.beans.factory.annotation.Value; |
|
|
import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression; |
|
|
import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression; |
|
|
import org.springframework.stereotype.Service; |
|
|
import org.springframework.stereotype.Service; |
|
|
import org.thingsboard.common.util.ThingsBoardThreadFactory; |
|
|
import org.thingsboard.common.util.ThingsBoardThreadFactory; |
|
|
|
|
|
import org.thingsboard.server.common.msg.TbMsg; |
|
|
|
|
|
import org.thingsboard.server.common.msg.TbMsgDataType; |
|
|
|
|
|
import org.thingsboard.server.common.msg.TbMsgMetaData; |
|
|
|
|
|
import org.thingsboard.server.common.msg.queue.TbMsgCallback; |
|
|
|
|
|
import org.thingsboard.server.common.msg.session.SessionMsgType; |
|
|
|
|
|
import org.thingsboard.server.common.transport.util.JsonUtils; |
|
|
import org.thingsboard.server.queue.TbQueueCallback; |
|
|
import org.thingsboard.server.queue.TbQueueCallback; |
|
|
import org.thingsboard.server.queue.TbQueueConsumer; |
|
|
import org.thingsboard.server.queue.TbQueueConsumer; |
|
|
import org.thingsboard.server.queue.TbQueueMsgMetadata; |
|
|
import org.thingsboard.server.queue.TbQueueMsgMetadata; |
|
|
@ -50,16 +59,19 @@ import org.thingsboard.server.queue.common.AsyncCallbackTemplate; |
|
|
|
|
|
|
|
|
import javax.annotation.PostConstruct; |
|
|
import javax.annotation.PostConstruct; |
|
|
import javax.annotation.PreDestroy; |
|
|
import javax.annotation.PreDestroy; |
|
|
|
|
|
import java.util.ArrayList; |
|
|
import java.util.List; |
|
|
import java.util.List; |
|
|
import java.util.Random; |
|
|
import java.util.Random; |
|
|
import java.util.UUID; |
|
|
import java.util.UUID; |
|
|
import java.util.concurrent.ConcurrentHashMap; |
|
|
import java.util.concurrent.ConcurrentHashMap; |
|
|
import java.util.concurrent.ConcurrentMap; |
|
|
import java.util.concurrent.ConcurrentMap; |
|
|
|
|
|
import java.util.concurrent.CountDownLatch; |
|
|
import java.util.concurrent.ExecutorService; |
|
|
import java.util.concurrent.ExecutorService; |
|
|
import java.util.concurrent.Executors; |
|
|
import java.util.concurrent.Executors; |
|
|
import java.util.concurrent.ScheduledExecutorService; |
|
|
import java.util.concurrent.ScheduledExecutorService; |
|
|
import java.util.concurrent.ScheduledFuture; |
|
|
import java.util.concurrent.ScheduledFuture; |
|
|
import java.util.concurrent.TimeUnit; |
|
|
import java.util.concurrent.TimeUnit; |
|
|
|
|
|
import java.util.concurrent.atomic.AtomicInteger; |
|
|
|
|
|
|
|
|
/** |
|
|
/** |
|
|
* Created by ashvayka on 17.10.18. |
|
|
* Created by ashvayka on 17.10.18. |
|
|
@ -82,6 +94,7 @@ public class DefaultTransportService implements TransportService { |
|
|
@Value("${queue.transport.poll_interval}") |
|
|
@Value("${queue.transport.poll_interval}") |
|
|
private int notificationsPollDuration; |
|
|
private int notificationsPollDuration; |
|
|
|
|
|
|
|
|
|
|
|
private final Gson gson = new Gson(); |
|
|
private final TransportQueueProvider queueProvider; |
|
|
private final TransportQueueProvider queueProvider; |
|
|
private final PartitionService partitionService; |
|
|
private final PartitionService partitionService; |
|
|
|
|
|
|
|
|
@ -221,8 +234,19 @@ public class DefaultTransportService implements TransportService { |
|
|
public void process(TransportProtos.SessionInfoProto sessionInfo, TransportProtos.PostTelemetryMsg msg, TransportServiceCallback<Void> callback) { |
|
|
public void process(TransportProtos.SessionInfoProto sessionInfo, TransportProtos.PostTelemetryMsg msg, TransportServiceCallback<Void> callback) { |
|
|
if (checkLimits(sessionInfo, msg, callback)) { |
|
|
if (checkLimits(sessionInfo, msg, callback)) { |
|
|
reportActivityInternal(sessionInfo); |
|
|
reportActivityInternal(sessionInfo); |
|
|
sendToRuleEngine(sessionInfo, TransportToRuleEngineMsg.newBuilder().setSessionInfo(sessionInfo). |
|
|
TenantId tenantId = new TenantId(new UUID(sessionInfo.getTenantIdMSB(), sessionInfo.getTenantIdLSB())); |
|
|
setPostTelemetry(msg).build(), callback); |
|
|
DeviceId deviceId = new DeviceId(new UUID(sessionInfo.getDeviceIdMSB(), sessionInfo.getDeviceIdLSB())); |
|
|
|
|
|
MsgPackCallback packCallback = new MsgPackCallback(msg.getTsKvListCount(), callback); |
|
|
|
|
|
for (TransportProtos.TsKvListProto tsKv : msg.getTsKvListList()) { |
|
|
|
|
|
TbMsgMetaData metaData = new TbMsgMetaData(); |
|
|
|
|
|
metaData.putValue("deviceName", sessionInfo.getDeviceName()); |
|
|
|
|
|
metaData.putValue("deviceType", sessionInfo.getDeviceType()); |
|
|
|
|
|
metaData.putValue("ts", tsKv.getTs() + ""); |
|
|
|
|
|
JsonObject json = JsonUtils.getJsonObject(tsKv.getKvList()); |
|
|
|
|
|
TbMsg tbMsg = new TbMsg(UUID.randomUUID(), SessionMsgType.POST_TELEMETRY_REQUEST.name(), |
|
|
|
|
|
deviceId, metaData, TbMsgDataType.JSON, gson.toJson(json), null, null, null); |
|
|
|
|
|
sendToRuleEngine(tenantId, tbMsg, packCallback); |
|
|
|
|
|
} |
|
|
} |
|
|
} |
|
|
} |
|
|
} |
|
|
|
|
|
|
|
|
@ -230,8 +254,15 @@ public class DefaultTransportService implements TransportService { |
|
|
public void process(TransportProtos.SessionInfoProto sessionInfo, TransportProtos.PostAttributeMsg msg, TransportServiceCallback<Void> callback) { |
|
|
public void process(TransportProtos.SessionInfoProto sessionInfo, TransportProtos.PostAttributeMsg msg, TransportServiceCallback<Void> callback) { |
|
|
if (checkLimits(sessionInfo, msg, callback)) { |
|
|
if (checkLimits(sessionInfo, msg, callback)) { |
|
|
reportActivityInternal(sessionInfo); |
|
|
reportActivityInternal(sessionInfo); |
|
|
sendToRuleEngine(sessionInfo, TransportToRuleEngineMsg.newBuilder().setSessionInfo(sessionInfo). |
|
|
TenantId tenantId = new TenantId(new UUID(sessionInfo.getTenantIdMSB(), sessionInfo.getTenantIdLSB())); |
|
|
setPostAttributes(msg).build(), callback); |
|
|
DeviceId deviceId = new DeviceId(new UUID(sessionInfo.getDeviceIdMSB(), sessionInfo.getDeviceIdLSB())); |
|
|
|
|
|
JsonObject json = JsonUtils.getJsonObject(msg.getKvList()); |
|
|
|
|
|
TbMsgMetaData metaData = new TbMsgMetaData(); |
|
|
|
|
|
metaData.putValue("deviceName", sessionInfo.getDeviceName()); |
|
|
|
|
|
metaData.putValue("deviceType", sessionInfo.getDeviceType()); |
|
|
|
|
|
TbMsg tbMsg = new TbMsg(UUID.randomUUID(), SessionMsgType.POST_ATTRIBUTES_REQUEST.name(), deviceId, metaData, |
|
|
|
|
|
TbMsgDataType.JSON, gson.toJson(json), null, null, null); |
|
|
|
|
|
sendToRuleEngine(tenantId, tbMsg, new TransportTbQueueCallback(callback)); |
|
|
} |
|
|
} |
|
|
} |
|
|
} |
|
|
|
|
|
|
|
|
@ -278,8 +309,8 @@ public class DefaultTransportService implements TransportService { |
|
|
public void process(TransportProtos.SessionInfoProto sessionInfo, TransportProtos.ToServerRpcRequestMsg msg, TransportServiceCallback<Void> callback) { |
|
|
public void process(TransportProtos.SessionInfoProto sessionInfo, TransportProtos.ToServerRpcRequestMsg msg, TransportServiceCallback<Void> callback) { |
|
|
if (checkLimits(sessionInfo, msg, callback)) { |
|
|
if (checkLimits(sessionInfo, msg, callback)) { |
|
|
reportActivityInternal(sessionInfo); |
|
|
reportActivityInternal(sessionInfo); |
|
|
sendToRuleEngine(sessionInfo, TransportToRuleEngineMsg.newBuilder().setSessionInfo(sessionInfo). |
|
|
// sendToRuleEngine(sessionInfo, TransportToRuleEngineMsg.newBuilder().setSessionInfo(sessionInfo).
|
|
|
setToServerRPCCallRequest(msg).build(), callback); |
|
|
// setToServerRPCCallRequest(msg).build(), new TransportTbQueueCallback(callback));
|
|
|
} |
|
|
} |
|
|
} |
|
|
} |
|
|
|
|
|
|
|
|
@ -455,12 +486,12 @@ public class DefaultTransportService implements TransportService { |
|
|
new TransportTbQueueCallback(callback) : null); |
|
|
new TransportTbQueueCallback(callback) : null); |
|
|
} |
|
|
} |
|
|
|
|
|
|
|
|
protected void sendToRuleEngine(TransportProtos.SessionInfoProto sessionInfo, TransportToRuleEngineMsg toRuleEngineMsg, TransportServiceCallback<Void> callback) { |
|
|
protected void sendToRuleEngine(TenantId tenantId, TbMsg tbMsg, TbQueueCallback callback) { |
|
|
TopicPartitionInfo tpi = partitionService.resolve(ServiceType.TB_RULE_ENGINE, getTenantId(sessionInfo), getDeviceId(sessionInfo)); |
|
|
TopicPartitionInfo tpi = partitionService.resolve(ServiceType.TB_RULE_ENGINE, tenantId, tbMsg.getOriginator()); |
|
|
ruleEngineMsgProducer.send(tpi, |
|
|
ToRuleEngineMsg msg = ToRuleEngineMsg.newBuilder().setTbMsg(ByteString.copyFrom(TbMsg.toByteArray(tbMsg))) |
|
|
new TbProtoQueueMsg<>(getRoutingKey(sessionInfo), |
|
|
.setTenantIdMSB(tenantId.getId().getMostSignificantBits()) |
|
|
ToRuleEngineMsg.newBuilder().setToRuleEngineMsg(toRuleEngineMsg).build()), callback != null ? |
|
|
.setTenantIdLSB(tenantId.getId().getLeastSignificantBits()).build(); |
|
|
new TransportTbQueueCallback(callback) : null); |
|
|
ruleEngineMsgProducer.send(tpi, new TbProtoQueueMsg<>(tbMsg.getId(), msg), callback); |
|
|
} |
|
|
} |
|
|
|
|
|
|
|
|
private class TransportTbQueueCallback implements TbQueueCallback { |
|
|
private class TransportTbQueueCallback implements TbQueueCallback { |
|
|
@ -480,4 +511,26 @@ public class DefaultTransportService implements TransportService { |
|
|
DefaultTransportService.this.transportCallbackExecutor.submit(() -> callback.onError(t)); |
|
|
DefaultTransportService.this.transportCallbackExecutor.submit(() -> callback.onError(t)); |
|
|
} |
|
|
} |
|
|
} |
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
private class MsgPackCallback implements TbQueueCallback { |
|
|
|
|
|
private final AtomicInteger msgCount; |
|
|
|
|
|
private final TransportServiceCallback<Void> callback; |
|
|
|
|
|
|
|
|
|
|
|
public MsgPackCallback(Integer msgCount, TransportServiceCallback<Void> callback) { |
|
|
|
|
|
this.msgCount = new AtomicInteger(msgCount); |
|
|
|
|
|
this.callback = callback; |
|
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
@Override |
|
|
|
|
|
public void onSuccess(TbQueueMsgMetadata metadata) { |
|
|
|
|
|
if (msgCount.decrementAndGet() <= 0) { |
|
|
|
|
|
callback.onSuccess(null); |
|
|
|
|
|
} |
|
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
@Override |
|
|
|
|
|
public void onFailure(Throwable t) { |
|
|
|
|
|
callback.onError(t); |
|
|
|
|
|
} |
|
|
|
|
|
} |
|
|
} |
|
|
} |
|
|
|
Andrii Shvaika
7 years ago