简柏林 / thingskit · Commits

Commit f9810f43f7e8c5be806d447022e1b573e61dd068

Authored by xp.Huang
2 parents e76d082b f29fa66e

Merge branch 'ljl0209' into 'master' 

Ljl0209

See merge request huang/thingsboard3.3.2!43
Inline Side-by-side
Showing 70 changed files with 140 additions and 196 deletions
@@ -66,7 +66,6 @@ public class AppActor extends ContextAwareActor { @@ -66,7 +66,6 @@ public class AppActor extends ContextAwareActor {
66 66
67 @Override 67 @Override
68 protected boolean doProcess(TbActorMsg msg) { 68 protected boolean doProcess(TbActorMsg msg) {
69 - log.info("演员【AppActor】消息消费: {}", msg);  
70 if (!ruleChainsInitialized) { 69 if (!ruleChainsInitialized) {
71 initTenantActors(); 70 initTenantActors();
72 ruleChainsInitialized = true; 71 ruleChainsInitialized = true;
@@ -146,7 +145,6 @@ public class AppActor extends ContextAwareActor { @@ -146,7 +145,6 @@ public class AppActor extends ContextAwareActor {
146 msg.getMsg().getCallback().onFailure(new RuleEngineException("Message has system tenant id!")); 145 msg.getMsg().getCallback().onFailure(new RuleEngineException("Message has system tenant id!"));
147 } else { 146 } else {
148 if (!deletedTenants.contains(msg.getTenantId())) { 147 if (!deletedTenants.contains(msg.getTenantId())) {
149 - log.info("演员消息生产——普通消息内容：【{}】",msg);  
150 getOrCreateTenantActor(msg.getTenantId()).tell(msg); 148 getOrCreateTenantActor(msg.getTenantId()).tell(msg);
151 } else { 149 } else {
152 msg.getMsg().getCallback().onSuccess(); 150 msg.getMsg().getCallback().onSuccess();
@@ -175,7 +173,6 @@ public class AppActor extends ContextAwareActor { @@ -175,7 +173,6 @@ public class AppActor extends ContextAwareActor {
175 } 173 }
176 } 174 }
177 if (target != null) { 175 if (target != null) {
178 - log.info("演员消息生产——高级消息内容：【{}】", msg);  
179 target.tellWithHighPriority(msg); 176 target.tellWithHighPriority(msg);
180 } else { 177 } else {
181 log.debug("[{}] Invalid component lifecycle msg: {}", msg.getTenantId(), msg); 178 log.debug("[{}] Invalid component lifecycle msg: {}", msg.getTenantId(), msg);
@@ -186,10 +183,8 @@ public class AppActor extends ContextAwareActor { @@ -186,10 +183,8 @@ public class AppActor extends ContextAwareActor {
186 if (!deletedTenants.contains(msg.getTenantId())) { 183 if (!deletedTenants.contains(msg.getTenantId())) {
187 TbActorRef tenantActor = getOrCreateTenantActor(msg.getTenantId()); 184 TbActorRef tenantActor = getOrCreateTenantActor(msg.getTenantId());
188 if (priority) { 185 if (priority) {
189 - log.info("演员消息生产——高级消息内容：【{}】", msg);  
190 tenantActor.tellWithHighPriority(msg); 186 tenantActor.tellWithHighPriority(msg);
191 } else { 187 } else {
192 - log.info("演员消息生产——普通消息内容：【{}】",msg);  
193 tenantActor.tell(msg); 188 tenantActor.tell(msg);
194 } 189 }
195 } else { 190 } else {
@@ -213,7 +208,6 @@ public class AppActor extends ContextAwareActor { @@ -213,7 +208,6 @@ public class AppActor extends ContextAwareActor {
213 target = getOrCreateTenantActor(msg.getTenantId()); 208 target = getOrCreateTenantActor(msg.getTenantId());
214 } 209 }
215 if (target != null) { 210 if (target != null) {
216 - log.info("演员消息生产——高级消息内容：【{}】", msg);  
217 target.tellWithHighPriority(msg); 211 target.tellWithHighPriority(msg);
218 } else { 212 } else {
219 log.debug("[{}] Invalid edge event update msg: {}", msg.getTenantId(), msg); 213 log.debug("[{}] Invalid edge event update msg: {}", msg.getTenantId(), msg);
@@ -57,7 +57,6 @@ public class DeviceActor extends ContextAwareActor { @@ -57,7 +57,6 @@ public class DeviceActor extends ContextAwareActor {
57 57
58 @Override 58 @Override
59 protected boolean doProcess(TbActorMsg msg) { 59 protected boolean doProcess(TbActorMsg msg) {
60 - log.info("演员【DeviceActor】消息消费: {}", msg);  
61 switch (msg.getMsgType()) { 60 switch (msg.getMsgType()) {
62 case TRANSPORT_TO_DEVICE_ACTOR_MSG: 61 case TRANSPORT_TO_DEVICE_ACTOR_MSG:
63 processor.process(ctx, (TransportToDeviceActorMsgWrapper) msg); 62 processor.process(ctx, (TransportToDeviceActorMsgWrapper) msg);
@@ -302,7 +302,6 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor { @@ -302,7 +302,6 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor {
302 private void registerPendingRpcRequest(TbActorCtx context, ToDeviceRpcRequestActorMsg msg, boolean sent, ToDeviceRpcRequestMsg rpcRequest, long timeout) { 302 private void registerPendingRpcRequest(TbActorCtx context, ToDeviceRpcRequestActorMsg msg, boolean sent, ToDeviceRpcRequestMsg rpcRequest, long timeout) {
303 toDeviceRpcPendingMap.put(rpcRequest.getRequestId(), new ToDeviceRpcRequestMetadata(msg, sent)); 303 toDeviceRpcPendingMap.put(rpcRequest.getRequestId(), new ToDeviceRpcRequestMetadata(msg, sent));
304 DeviceActorServerSideRpcTimeoutMsg timeoutMsg = new DeviceActorServerSideRpcTimeoutMsg(rpcRequest.getRequestId(), timeout); 304 DeviceActorServerSideRpcTimeoutMsg timeoutMsg = new DeviceActorServerSideRpcTimeoutMsg(rpcRequest.getRequestId(), timeout);
305 - log.info("演员消息生产——高级消息内容：【{}】", timeoutMsg);  
306 scheduleMsgWithDelay(context, timeoutMsg, timeoutMsg.getTimeout()); 305 scheduleMsgWithDelay(context, timeoutMsg, timeoutMsg.getTimeout());
307 } 306 }
308 307
@@ -933,7 +932,6 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor { @@ -933,7 +932,6 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor {
933 } 932 }
934 933
935 void init(TbActorCtx ctx) { 934 void init(TbActorCtx ctx) {
936 - log.info("演员消息生产——高级消息内容：【{}】", SessionTimeoutCheckMsg.instance());  
937 schedulePeriodicMsgWithDelay(ctx, SessionTimeoutCheckMsg.instance(), systemContext.getSessionReportTimeout(), systemContext.getSessionReportTimeout()); 935 schedulePeriodicMsgWithDelay(ctx, SessionTimeoutCheckMsg.instance(), systemContext.getSessionReportTimeout(), systemContext.getSessionReportTimeout());
938 PageLink pageLink = new PageLink(1024, 0, null, new SortOrder("createdTime")); 936 PageLink pageLink = new PageLink(1024, 0, null, new SortOrder("createdTime"));
939 PageData<Rpc> pageData; 937 PageData<Rpc> pageData;
@@ -254,7 +254,6 @@ class DefaultTbContext implements TbContext { @@ -254,7 +254,6 @@ class DefaultTbContext implements TbContext {
254 254
255 @Override 255 @Override
256 public void tellFailure(TbMsg msg, Throwable th) { 256 public void tellFailure(TbMsg msg, Throwable th) {
257 - log.info("演员消息生产——普通消息内容：tellFailure【{}】",msg);  
258 if (nodeCtx.getSelf().isDebugMode()) { 257 if (nodeCtx.getSelf().isDebugMode()) {
259 mainCtx.persistDebugOutput(nodeCtx.getTenantId(), nodeCtx.getSelf().getId(), msg, TbRelationTypes.FAILURE, th); 258 mainCtx.persistDebugOutput(nodeCtx.getTenantId(), nodeCtx.getSelf().getId(), msg, TbRelationTypes.FAILURE, th);
260 } 259 }
@@ -48,7 +48,6 @@ public class RuleChainActor extends ComponentActor<RuleChainId, RuleChainActorMe @@ -48,7 +48,6 @@ public class RuleChainActor extends ComponentActor<RuleChainId, RuleChainActorMe
48 48
49 @Override 49 @Override
50 protected boolean doProcess(TbActorMsg msg) { 50 protected boolean doProcess(TbActorMsg msg) {
51 - log.info("演员【RuleChainActor】消息消费: {}", msg);  
52 switch (msg.getMsgType()) { 51 switch (msg.getMsgType()) {
53 case COMPONENT_LIFE_CYCLE_MSG: 52 case COMPONENT_LIFE_CYCLE_MSG:
54 onComponentLifecycleMsg((ComponentLifecycleMsg) msg); 53 onComponentLifecycleMsg((ComponentLifecycleMsg) msg);
@@ -133,7 +133,6 @@ public class RuleChainActorMessageProcessor extends ComponentMsgProcessor<RuleCh @@ -133,7 +133,6 @@ public class RuleChainActorMessageProcessor extends ComponentMsgProcessor<RuleCh
133 } else { 133 } else {
134 log.trace("[{}][{}] Updating rule node [{}]: {}", entityId, ruleNode.getId(), ruleNode.getName(), ruleNode); 134 log.trace("[{}][{}] Updating rule node [{}]: {}", entityId, ruleNode.getId(), ruleNode.getName(), ruleNode);
135 existing.setSelf(ruleNode); 135 existing.setSelf(ruleNode);
136 - log.info("演员消息生产——高级消息内容：RuleNodeUpdatedMsg【{}】", new RuleNodeUpdatedMsg(tenantId, existing.getSelf().getId()));  
137 existing.getSelfActor().tellWithHighPriority(new RuleNodeUpdatedMsg(tenantId, existing.getSelf().getId())); 136 existing.getSelfActor().tellWithHighPriority(new RuleNodeUpdatedMsg(tenantId, existing.getSelf().getId()));
138 } 137 }
139 } 138 }
@@ -143,7 +142,6 @@ public class RuleChainActorMessageProcessor extends ComponentMsgProcessor<RuleCh @@ -143,7 +142,6 @@ public class RuleChainActorMessageProcessor extends ComponentMsgProcessor<RuleCh
143 removedRules.forEach(ruleNodeId -> { 142 removedRules.forEach(ruleNodeId -> {
144 log.trace("[{}][{}] Removing rule node [{}]", tenantId, entityId, ruleNodeId); 143 log.trace("[{}][{}] Removing rule node [{}]", tenantId, entityId, ruleNodeId);
145 RuleNodeCtx removed = nodeActors.remove(ruleNodeId); 144 RuleNodeCtx removed = nodeActors.remove(ruleNodeId);
146 - log.info("演员消息生产——高级消息内容：ComponentLifecycleMsg【{}】", new ComponentLifecycleMsg(tenantId, removed.getSelf().getId(), ComponentLifecycleEvent.DELETED));  
147 removed.getSelfActor().tellWithHighPriority(new ComponentLifecycleMsg(tenantId, removed.getSelf().getId(), ComponentLifecycleEvent.DELETED)); 145 removed.getSelfActor().tellWithHighPriority(new ComponentLifecycleMsg(tenantId, removed.getSelf().getId(), ComponentLifecycleEvent.DELETED));
148 }); 146 });
149 147
@@ -163,7 +161,6 @@ public class RuleChainActorMessageProcessor extends ComponentMsgProcessor<RuleCh @@ -163,7 +161,6 @@ public class RuleChainActorMessageProcessor extends ComponentMsgProcessor<RuleCh
163 @Override 161 @Override
164 public void onPartitionChangeMsg(PartitionChangeMsg msg) { 162 public void onPartitionChangeMsg(PartitionChangeMsg msg) {
165 nodeActors.values().stream().map(RuleNodeCtx::getSelfActor).forEach(actorRef -> { 163 nodeActors.values().stream().map(RuleNodeCtx::getSelfActor).forEach(actorRef -> {
166 - log.info("演员消息生产——高级消息内容：【{}】", msg);  
167 actorRef.tellWithHighPriority(msg); 164 actorRef.tellWithHighPriority(msg);
168 }); 165 });
169 } 166 }
@@ -318,7 +315,6 @@ public class RuleChainActorMessageProcessor extends ComponentMsgProcessor<RuleCh @@ -318,7 +315,6 @@ public class RuleChainActorMessageProcessor extends ComponentMsgProcessor<RuleCh
318 pushMsgToNode(nodeActors.get(new RuleNodeId(target.getId())), msg, fromRelationType); 315 pushMsgToNode(nodeActors.get(new RuleNodeId(target.getId())), msg, fromRelationType);
319 break; 316 break;
320 case RULE_CHAIN: 317 case RULE_CHAIN:
321 - log.info("演员消息生产——普通消息内容：RuleChainToRuleChainMsg【{}】",msg);  
322 parent.tell(new RuleChainToRuleChainMsg(new RuleChainId(target.getId()), entityId, msg, fromRelationType)); 318 parent.tell(new RuleChainToRuleChainMsg(new RuleChainId(target.getId()), entityId, msg, fromRelationType));
323 break; 319 break;
324 } 320 }
@@ -350,7 +346,6 @@ public class RuleChainActorMessageProcessor extends ComponentMsgProcessor<RuleCh @@ -350,7 +346,6 @@ public class RuleChainActorMessageProcessor extends ComponentMsgProcessor<RuleCh
350 346
351 private void pushMsgToNode(RuleNodeCtx nodeCtx, TbMsg msg, String fromRelationType) { 347 private void pushMsgToNode(RuleNodeCtx nodeCtx, TbMsg msg, String fromRelationType) {
352 if (nodeCtx != null) { 348 if (nodeCtx != null) {
353 - log.info("演员消息生产——普通消息内容：RuleChainToRuleNodeMsg【{}】",msg);  
354 nodeCtx.getSelfActor().tell(new RuleChainToRuleNodeMsg(new DefaultTbContext(systemContext, ruleChainName, nodeCtx), msg, fromRelationType)); 349 nodeCtx.getSelfActor().tell(new RuleChainToRuleNodeMsg(new DefaultTbContext(systemContext, ruleChainName, nodeCtx), msg, fromRelationType));
355 } else { 350 } else {
356 log.error("[{}][{}] RuleNodeCtx is empty", entityId, ruleChainName); 351 log.error("[{}][{}] RuleNodeCtx is empty", entityId, ruleChainName);
@@ -51,7 +51,6 @@ public class RuleNodeActor extends ComponentActor<RuleNodeId, RuleNodeActorMessa @@ -51,7 +51,6 @@ public class RuleNodeActor extends ComponentActor<RuleNodeId, RuleNodeActorMessa
51 51
52 @Override 52 @Override
53 protected boolean doProcess(TbActorMsg msg) { 53 protected boolean doProcess(TbActorMsg msg) {
54 - log.info("演员【RuleNodeActor】消息消费: {}", msg);  
55 switch (msg.getMsgType()) { 54 switch (msg.getMsgType()) {
56 case COMPONENT_LIFE_CYCLE_MSG: 55 case COMPONENT_LIFE_CYCLE_MSG:
57 case RULE_NODE_UPDATED_MSG: 56 case RULE_NODE_UPDATED_MSG:
@@ -140,7 +140,6 @@ public abstract class ComponentActor<T extends EntityId, P extends ComponentMsgP @@ -140,7 +140,6 @@ public abstract class ComponentActor<T extends EntityId, P extends ComponentMsgP
140 140
141 protected void onStatsPersistTick(EntityId entityId) { 141 protected void onStatsPersistTick(EntityId entityId) {
142 try { 142 try {
143 - log.info("演员消息生产——普通消息内容：StatsPersistMsg【{}】",new StatsPersistMsg(messagesProcessed, errorsOccurred, tenantId, entityId));  
144 systemContext.getStatsActor().tell(new StatsPersistMsg(messagesProcessed, errorsOccurred, tenantId, entityId)); 143 systemContext.getStatsActor().tell(new StatsPersistMsg(messagesProcessed, errorsOccurred, tenantId, entityId));
145 resetStatsCounters(); 144 resetStatsCounters();
146 } catch (Exception e) { 145 } catch (Exception e) {
@@ -118,14 +118,12 @@ public class DefaultActorService extends TbApplicationEventListener<PartitionCha @@ -118,14 +118,12 @@ public class DefaultActorService extends TbApplicationEventListener<PartitionCha
118 @Order(value = 2) 118 @Order(value = 2)
119 public void onApplicationEvent(ApplicationReadyEvent applicationReadyEvent) { 119 public void onApplicationEvent(ApplicationReadyEvent applicationReadyEvent) {
120 log.info("Received application ready event. Sending application init message to actor system"); 120 log.info("Received application ready event. Sending application init message to actor system");
121 - log.info("演员消息生产——高级消息内容：AppInitMsg【{}】", new AppInitMsg());  
122 appActor.tellWithHighPriority(new AppInitMsg()); 121 appActor.tellWithHighPriority(new AppInitMsg());
123 } 122 }
124 123
125 @Override 124 @Override
126 protected void onTbApplicationEvent(PartitionChangeEvent event) { 125 protected void onTbApplicationEvent(PartitionChangeEvent event) {
127 log.info("Received partition change event."); 126 log.info("Received partition change event.");
128 - log.info("演员消息生产——高级消息内容：PartitionChangeMsg【{}】", new PartitionChangeMsg(event.getServiceQueueKey(), event.getPartitions()));  
129 this.appActor.tellWithHighPriority(new PartitionChangeMsg(event.getServiceQueueKey(), event.getPartitions())); 127 this.appActor.tellWithHighPriority(new PartitionChangeMsg(event.getServiceQueueKey(), event.getPartitions()));
130 } 128 }
131 129
@@ -80,7 +80,6 @@ public abstract class ComponentMsgProcessor<T extends EntityId> extends Abstract @@ -80,7 +80,6 @@ public abstract class ComponentMsgProcessor<T extends EntityId> extends Abstract
80 } 80 }
81 81
82 public void scheduleStatsPersistTick(TbActorCtx context, long statsPersistFrequency) { 82 public void scheduleStatsPersistTick(TbActorCtx context, long statsPersistFrequency) {
83 - log.info("演员消息生产——高级消息内容：ComponentActor【StatsPersistTick】");  
84 schedulePeriodicMsgWithDelay(context, new StatsPersistTick(), statsPersistFrequency, statsPersistFrequency); 83 schedulePeriodicMsgWithDelay(context, new StatsPersistTick(), statsPersistFrequency, statsPersistFrequency);
85 } 84 }
86 85
@@ -41,7 +41,6 @@ public class StatsActor extends ContextAwareActor { @@ -41,7 +41,6 @@ public class StatsActor extends ContextAwareActor {
41 41
42 @Override 42 @Override
43 protected boolean doProcess(TbActorMsg msg) { 43 protected boolean doProcess(TbActorMsg msg) {
44 - log.info("演员【StatsActor】消息消费: {}", msg);  
45 if (msg.getMsgType().equals(MsgType.STATS_PERSIST_MSG)) { 44 if (msg.getMsgType().equals(MsgType.STATS_PERSIST_MSG)) {
46 onStatsPersistMsg((StatsPersistMsg) msg); 45 onStatsPersistMsg((StatsPersistMsg) msg);
47 return true; 46 return true;
@@ -194,7 +194,6 @@ public class TenantActor extends RuleChainManagerActor { @@ -194,7 +194,6 @@ public class TenantActor extends RuleChainManagerActor {
194 if (apiUsageState.isReExecEnabled()) { 194 if (apiUsageState.isReExecEnabled()) {
195 if (tbMsg.getRuleChainId() == null) { 195 if (tbMsg.getRuleChainId() == null) {
196 if (getRootChainActor() != null) { 196 if (getRootChainActor() != null) {
197 - log.info("演员消息生产——普通消息内容：【{}】",msg);  
198 getRootChainActor().tell(msg); 197 getRootChainActor().tell(msg);
199 } else { 198 } else {
200 tbMsg.getCallback().onFailure(new RuleEngineException("No Root Rule Chain available!")); 199 tbMsg.getCallback().onFailure(new RuleEngineException("No Root Rule Chain available!"));
@@ -202,7 +201,6 @@ public class TenantActor extends RuleChainManagerActor { @@ -202,7 +201,6 @@ public class TenantActor extends RuleChainManagerActor {
202 } 201 }
203 } else { 202 } else {
204 try { 203 try {
205 - log.info("演员消息生产——普通消息内容：TbEntityActorId【{}】",msg);  
206 ctx.tell(new TbEntityActorId(tbMsg.getRuleChainId()), msg); 204 ctx.tell(new TbEntityActorId(tbMsg.getRuleChainId()), msg);
207 } catch (TbActorNotRegisteredException ex) { 205 } catch (TbActorNotRegisteredException ex) {
208 log.trace("Received message for non-existing rule chain: [{}]", tbMsg.getRuleChainId()); 206 log.trace("Received message for non-existing rule chain: [{}]", tbMsg.getRuleChainId());
@@ -218,7 +216,6 @@ public class TenantActor extends RuleChainManagerActor { @@ -218,7 +216,6 @@ public class TenantActor extends RuleChainManagerActor {
218 216
219 private void onRuleChainMsg(RuleChainAwareMsg msg) { 217 private void onRuleChainMsg(RuleChainAwareMsg msg) {
220 if (apiUsageState.isReExecEnabled()) { 218 if (apiUsageState.isReExecEnabled()) {
221 - log.info("演员消息生产——普通消息内容：【{}】",msg);  
222 getOrCreateActor(msg.getRuleChainId()).tell(msg); 219 getOrCreateActor(msg.getRuleChainId()).tell(msg);
223 } 220 }
224 } 221 }
@@ -229,10 +226,8 @@ public class TenantActor extends RuleChainManagerActor { @@ -229,10 +226,8 @@ public class TenantActor extends RuleChainManagerActor {
229 } 226 }
230 TbActorRef deviceActor = getOrCreateDeviceActor(msg.getDeviceId()); 227 TbActorRef deviceActor = getOrCreateDeviceActor(msg.getDeviceId());
231 if (priority) { 228 if (priority) {
232 - log.info("演员消息生产——高级消息内容：【{}】", msg);  
233 deviceActor.tellWithHighPriority(msg); 229 deviceActor.tellWithHighPriority(msg);
234 } else { 230 } else {
235 - log.info("演员消息生产——普通消息内容：【{}】",msg);  
236 deviceActor.tell(msg); 231 deviceActor.tell(msg);
237 } 232 }
238 } 233 }
@@ -269,7 +264,6 @@ public class TenantActor extends RuleChainManagerActor { @@ -269,7 +264,6 @@ public class TenantActor extends RuleChainManagerActor {
269 visit(ruleChain, target); 264 visit(ruleChain, target);
270 } 265 }
271 } 266 }
272 - log.info("演员消息生产——高级消息内容：【{}】",msg);  
273 target.tellWithHighPriority(msg); 267 target.tellWithHighPriority(msg);
274 } else { 268 } else {
275 log.debug("[{}] Invalid component lifecycle msg: {}", tenantId, msg); 269 log.debug("[{}] Invalid component lifecycle msg: {}", tenantId, msg);
@@ -246,7 +246,6 @@ public class DeviceProvisionServiceImpl implements DeviceProvisionService { @@ -246,7 +246,6 @@ public class DeviceProvisionServiceImpl implements DeviceProvisionService {
246 .setTenantIdMSB(tenantId.getId().getMostSignificantBits()) 246 .setTenantIdMSB(tenantId.getId().getMostSignificantBits())
247 .setTenantIdLSB(tenantId.getId().getLeastSignificantBits()).build(); 247 .setTenantIdLSB(tenantId.getId().getLeastSignificantBits()).build();
248 TbProtoQueueMsg queueMsg = new TbProtoQueueMsg<>(tbMsg.getId(), msg); 248 TbProtoQueueMsg queueMsg = new TbProtoQueueMsg<>(tbMsg.getId(), msg);
249 - log.info("队列消息生产 ——主题【{}】，消息【{}】 ",tpi.getFullTopicName(),queueMsg);  
250 ruleEngineMsgProducer.send(tpi,queueMsg , callback); 249 ruleEngineMsgProducer.send(tpi,queueMsg , callback);
251 } 250 }
252 251
@@ -252,7 +252,6 @@ public class DefaultOtaPackageStateService implements OtaPackageStateService { @@ -252,7 +252,6 @@ public class DefaultOtaPackageStateService implements OtaPackageStateService {
252 252
253 TopicPartitionInfo tpi = new TopicPartitionInfo(otaPackageStateMsgProducer.getDefaultTopic(), null, null, false); 253 TopicPartitionInfo tpi = new TopicPartitionInfo(otaPackageStateMsgProducer.getDefaultTopic(), null, null, false);
254 TbProtoQueueMsg queueMsg = new TbProtoQueueMsg<>(UUID.randomUUID(), msg); 254 TbProtoQueueMsg queueMsg = new TbProtoQueueMsg<>(UUID.randomUUID(), msg);
255 - log.info("队列消息生产 ——主题【{}】，消息【{}】 ",tpi.getFullTopicName(),queueMsg);  
256 otaPackageStateMsgProducer.send(tpi, queueMsg, null); 255 otaPackageStateMsgProducer.send(tpi, queueMsg, null);
257 256
258 List<TsKvEntry> telemetry = new ArrayList<>(); 257 List<TsKvEntry> telemetry = new ArrayList<>();
@@ -96,7 +96,6 @@ public class DefaultTbClusterService implements TbClusterService { @@ -96,7 +96,6 @@ public class DefaultTbClusterService implements TbClusterService {
96 public void pushMsgToCore(TenantId tenantId, EntityId entityId, ToCoreMsg msg, TbQueueCallback callback) { 96 public void pushMsgToCore(TenantId tenantId, EntityId entityId, ToCoreMsg msg, TbQueueCallback callback) {
97 TopicPartitionInfo tpi = partitionService.resolve(ServiceType.TB_CORE, tenantId, entityId); 97 TopicPartitionInfo tpi = partitionService.resolve(ServiceType.TB_CORE, tenantId, entityId);
98 TbProtoQueueMsg queueMsg = new TbProtoQueueMsg<>(UUID.randomUUID(), msg); 98 TbProtoQueueMsg queueMsg = new TbProtoQueueMsg<>(UUID.randomUUID(), msg);
99 - log.info("队列消息生产 ——主题【{}】，消息【{}】 ",tpi.getFullTopicName(),queueMsg);  
100 producerProvider.getTbCoreMsgProducer().send(tpi, queueMsg, callback); 99 producerProvider.getTbCoreMsgProducer().send(tpi, queueMsg, callback);
101 toCoreMsgs.incrementAndGet(); 100 toCoreMsgs.incrementAndGet();
102 } 101 }
@@ -104,7 +103,6 @@ public class DefaultTbClusterService implements TbClusterService { @@ -104,7 +103,6 @@ public class DefaultTbClusterService implements TbClusterService {
104 @Override 103 @Override
105 public void pushMsgToCore(TopicPartitionInfo tpi, UUID msgId, ToCoreMsg msg, TbQueueCallback callback) { 104 public void pushMsgToCore(TopicPartitionInfo tpi, UUID msgId, ToCoreMsg msg, TbQueueCallback callback) {
106 TbProtoQueueMsg queueMsg = new TbProtoQueueMsg<>(msgId, msg); 105 TbProtoQueueMsg queueMsg = new TbProtoQueueMsg<>(msgId, msg);
107 - log.info("队列消息生产 ——主题【{}】，消息【{}】 ",tpi.getFullTopicName(),queueMsg);  
108 producerProvider.getTbCoreMsgProducer().send(tpi,queueMsg , callback); 106 producerProvider.getTbCoreMsgProducer().send(tpi,queueMsg , callback);
109 toCoreMsgs.incrementAndGet(); 107 toCoreMsgs.incrementAndGet();
110 } 108 }
@@ -115,7 +113,6 @@ public class DefaultTbClusterService implements TbClusterService { @@ -115,7 +113,6 @@ public class DefaultTbClusterService implements TbClusterService {
115 log.trace("PUSHING msg: {} to:{}", msg, tpi); 113 log.trace("PUSHING msg: {} to:{}", msg, tpi);
116 byte[] msgBytes = encodingService.encode(msg); 114 byte[] msgBytes = encodingService.encode(msg);
117 ToCoreMsg toCoreMsg = ToCoreMsg.newBuilder().setToDeviceActorNotificationMsg(ByteString.copyFrom(msgBytes)).build(); 115 ToCoreMsg toCoreMsg = ToCoreMsg.newBuilder().setToDeviceActorNotificationMsg(ByteString.copyFrom(msgBytes)).build();
118 - log.info("队列消息生产 ——主题【{}】，消息【{}】 ",tpi.getFullTopicName(),new TbProtoQueueMsg<>(msg.getDeviceId().getId(), toCoreMsg));  
119 producerProvider.getTbCoreMsgProducer().send(tpi, new TbProtoQueueMsg<>(msg.getDeviceId().getId(), toCoreMsg), callback); 116 producerProvider.getTbCoreMsgProducer().send(tpi, new TbProtoQueueMsg<>(msg.getDeviceId().getId(), toCoreMsg), callback);
120 toCoreMsgs.incrementAndGet(); 117 toCoreMsgs.incrementAndGet();
121 } 118 }
@@ -131,7 +128,6 @@ public class DefaultTbClusterService implements TbClusterService { @@ -131,7 +128,6 @@ public class DefaultTbClusterService implements TbClusterService {
131 response.getResponse().ifPresent(builder::setResponse); 128 response.getResponse().ifPresent(builder::setResponse);
132 ToCoreNotificationMsg msg = ToCoreNotificationMsg.newBuilder().setFromDeviceRpcResponse(builder).build(); 129 ToCoreNotificationMsg msg = ToCoreNotificationMsg.newBuilder().setFromDeviceRpcResponse(builder).build();
133 TbProtoQueueMsg queueMsg = new TbProtoQueueMsg<>(response.getId(), msg); 130 TbProtoQueueMsg queueMsg = new TbProtoQueueMsg<>(response.getId(), msg);
134 - log.info("队列消息生产 ——主题【{}】，消息【{}】 ",tpi.getFullTopicName(),queueMsg);  
135 producerProvider.getTbCoreNotificationsMsgProducer().send(tpi, queueMsg, callback); 131 producerProvider.getTbCoreNotificationsMsgProducer().send(tpi, queueMsg, callback);
136 toCoreNfs.incrementAndGet(); 132 toCoreNfs.incrementAndGet();
137 } 133 }
@@ -140,7 +136,6 @@ public class DefaultTbClusterService implements TbClusterService { @@ -140,7 +136,6 @@ public class DefaultTbClusterService implements TbClusterService {
140 public void pushMsgToRuleEngine(TopicPartitionInfo tpi, UUID msgId, ToRuleEngineMsg msg, TbQueueCallback callback) { 136 public void pushMsgToRuleEngine(TopicPartitionInfo tpi, UUID msgId, ToRuleEngineMsg msg, TbQueueCallback callback) {
141 log.trace("PUSHING msg: {} to:{}", msg, tpi); 137 log.trace("PUSHING msg: {} to:{}", msg, tpi);
142 TbProtoQueueMsg queueMsg = new TbProtoQueueMsg<>(msgId, msg); 138 TbProtoQueueMsg queueMsg = new TbProtoQueueMsg<>(msgId, msg);
143 - log.info("队列消息生产 ——主题【{}】，消息【{}】 ",tpi.getFullTopicName(),queueMsg);  
144 producerProvider.getRuleEngineMsgProducer().send(tpi, queueMsg, callback); 139 producerProvider.getRuleEngineMsgProducer().send(tpi, queueMsg, callback);
145 toRuleEngineMsgs.incrementAndGet(); 140 toRuleEngineMsgs.incrementAndGet();
146 } 141 }
@@ -168,7 +163,6 @@ public class DefaultTbClusterService implements TbClusterService { @@ -168,7 +163,6 @@ public class DefaultTbClusterService implements TbClusterService {
168 .setTenantIdLSB(tenantId.getId().getLeastSignificantBits()) 163 .setTenantIdLSB(tenantId.getId().getLeastSignificantBits())
169 .setTbMsg(TbMsg.toByteString(tbMsg)).build(); 164 .setTbMsg(TbMsg.toByteString(tbMsg)).build();
170 TbProtoQueueMsg queueMsg = new TbProtoQueueMsg<>(tbMsg.getId(), msg); 165 TbProtoQueueMsg queueMsg = new TbProtoQueueMsg<>(tbMsg.getId(), msg);
171 - log.info("队列消息生产 ——主题【{}】，消息【{}】 ",tpi.getFullTopicName(),queueMsg);  
172 producerProvider.getRuleEngineMsgProducer().send(tpi, queueMsg, callback); 166 producerProvider.getRuleEngineMsgProducer().send(tpi, queueMsg, callback);
173 toRuleEngineMsgs.incrementAndGet(); 167 toRuleEngineMsgs.incrementAndGet();
174 } 168 }
@@ -202,7 +196,6 @@ public class DefaultTbClusterService implements TbClusterService { @@ -202,7 +196,6 @@ public class DefaultTbClusterService implements TbClusterService {
202 response.getResponse().ifPresent(builder::setResponse); 196 response.getResponse().ifPresent(builder::setResponse);
203 ToRuleEngineNotificationMsg msg = ToRuleEngineNotificationMsg.newBuilder().setFromDeviceRpcResponse(builder).build(); 197 ToRuleEngineNotificationMsg msg = ToRuleEngineNotificationMsg.newBuilder().setFromDeviceRpcResponse(builder).build();
204 TbProtoQueueMsg queueMsg = new TbProtoQueueMsg<>(response.getId(), msg); 198 TbProtoQueueMsg queueMsg = new TbProtoQueueMsg<>(response.getId(), msg);
205 - log.info("队列消息生产 ——主题【{}】，消息【{}】 ",tpi.getFullTopicName(),queueMsg);  
206 producerProvider.getRuleEngineNotificationsMsgProducer().send(tpi, queueMsg, callback); 199 producerProvider.getRuleEngineNotificationsMsgProducer().send(tpi, queueMsg, callback);
207 toRuleEngineNfs.incrementAndGet(); 200 toRuleEngineNfs.incrementAndGet();
208 } 201 }
@@ -219,7 +212,6 @@ public class DefaultTbClusterService implements TbClusterService { @@ -219,7 +212,6 @@ public class DefaultTbClusterService implements TbClusterService {
219 TopicPartitionInfo tpi = partitionService.getNotificationsTopic(ServiceType.TB_TRANSPORT, serviceId); 212 TopicPartitionInfo tpi = partitionService.getNotificationsTopic(ServiceType.TB_TRANSPORT, serviceId);
220 log.trace("PUSHING msg: {} to:{}", response, tpi); 213 log.trace("PUSHING msg: {} to:{}", response, tpi);
221 TbProtoQueueMsg queueMsg = new TbProtoQueueMsg<>(UUID.randomUUID(), response); 214 TbProtoQueueMsg queueMsg = new TbProtoQueueMsg<>(UUID.randomUUID(), response);
222 - log.info("队列消息生产 ——主题【{}】，消息【{}】 ",tpi.getFullTopicName(),queueMsg);  
223 producerProvider.getTransportNotificationsMsgProducer().send(tpi, queueMsg, callback); 215 producerProvider.getTransportNotificationsMsgProducer().send(tpi, queueMsg, callback);
224 toTransportNfs.incrementAndGet(); 216 toTransportNfs.incrementAndGet();
225 } 217 }
@@ -342,7 +334,6 @@ public class DefaultTbClusterService implements TbClusterService { @@ -342,7 +334,6 @@ public class DefaultTbClusterService implements TbClusterService {
342 for (String serviceId : tbCoreServices) { 334 for (String serviceId : tbCoreServices) {
343 TopicPartitionInfo tpi = partitionService.getNotificationsTopic(ServiceType.TB_CORE, serviceId); 335 TopicPartitionInfo tpi = partitionService.getNotificationsTopic(ServiceType.TB_CORE, serviceId);
344 ToCoreNotificationMsg toCoreMsg = ToCoreNotificationMsg.newBuilder().setEdgeEventUpdateMsg(ByteString.copyFrom(msgBytes)).build(); 336 ToCoreNotificationMsg toCoreMsg = ToCoreNotificationMsg.newBuilder().setEdgeEventUpdateMsg(ByteString.copyFrom(msgBytes)).build();
345 - log.info("队列消息生产 ——主题【{}】，消息【{}】 ",tpi.getFullTopicName(),new TbProtoQueueMsg<>(msg.getEdgeId().getId(), toCoreMsg));  
346 toCoreNfProducer.send(tpi, new TbProtoQueueMsg<>(msg.getEdgeId().getId(), toCoreMsg), null); 337 toCoreNfProducer.send(tpi, new TbProtoQueueMsg<>(msg.getEdgeId().getId(), toCoreMsg), null);
347 toCoreNfs.incrementAndGet(); 338 toCoreNfs.incrementAndGet();
348 } 339 }
@@ -364,7 +355,6 @@ public class DefaultTbClusterService implements TbClusterService { @@ -364,7 +355,6 @@ public class DefaultTbClusterService implements TbClusterService {
364 for (String serviceId : tbCoreServices) { 355 for (String serviceId : tbCoreServices) {
365 TopicPartitionInfo tpi = partitionService.getNotificationsTopic(ServiceType.TB_CORE, serviceId); 356 TopicPartitionInfo tpi = partitionService.getNotificationsTopic(ServiceType.TB_CORE, serviceId);
366 ToCoreNotificationMsg toCoreMsg = ToCoreNotificationMsg.newBuilder().setComponentLifecycleMsg(ByteString.copyFrom(msgBytes)).build(); 357 ToCoreNotificationMsg toCoreMsg = ToCoreNotificationMsg.newBuilder().setComponentLifecycleMsg(ByteString.copyFrom(msgBytes)).build();
367 - log.info("队列消息生产 ——主题【{}】，消息【{}】 ",tpi.getFullTopicName(),new TbProtoQueueMsg<>(msg.getEntityId().getId(), toCoreMsg));  
368 toCoreNfProducer.send(tpi, new TbProtoQueueMsg<>(msg.getEntityId().getId(), toCoreMsg), null); 358 toCoreNfProducer.send(tpi, new TbProtoQueueMsg<>(msg.getEntityId().getId(), toCoreMsg), null);
369 toCoreNfs.incrementAndGet(); 359 toCoreNfs.incrementAndGet();
370 } 360 }
@@ -374,7 +364,7 @@ public class DefaultTbClusterService implements TbClusterService { @@ -374,7 +364,7 @@ public class DefaultTbClusterService implements TbClusterService {
374 for (String serviceId : tbRuleEngineServices) { 364 for (String serviceId : tbRuleEngineServices) {
375 TopicPartitionInfo tpi = partitionService.getNotificationsTopic(ServiceType.TB_RULE_ENGINE, serviceId); 365 TopicPartitionInfo tpi = partitionService.getNotificationsTopic(ServiceType.TB_RULE_ENGINE, serviceId);
376 ToRuleEngineNotificationMsg toRuleEngineMsg = ToRuleEngineNotificationMsg.newBuilder().setComponentLifecycleMsg(ByteString.copyFrom(msgBytes)).build(); 366 ToRuleEngineNotificationMsg toRuleEngineMsg = ToRuleEngineNotificationMsg.newBuilder().setComponentLifecycleMsg(ByteString.copyFrom(msgBytes)).build();
377 - log.info("队列消息生产 ——主题【{}】，消息【{}】 ",tpi.getFullTopicName(),new TbProtoQueueMsg<>(msg.getEntityId().getId(), toRuleEngineMsg)); 367 +
378 toRuleEngineProducer.send(tpi, new TbProtoQueueMsg<>(msg.getEntityId().getId(), toRuleEngineMsg), null); 368 toRuleEngineProducer.send(tpi, new TbProtoQueueMsg<>(msg.getEntityId().getId(), toRuleEngineMsg), null);
379 toRuleEngineNfs.incrementAndGet(); 369 toRuleEngineNfs.incrementAndGet();
380 } 370 }
@@ -199,7 +199,6 @@ public class DefaultTbCoreConsumerService extends AbstractConsumerService<ToCore @@ -199,7 +199,6 @@ public class DefaultTbCoreConsumerService extends AbstractConsumerService<ToCore
199 if (msgs.isEmpty()) { 199 if (msgs.isEmpty()) {
200 continue; 200 continue;
201 } 201 }
202 - log.info("队列消息消费 ：launchMainConsumers【{}】 ", msgs);  
203 List<IdMsgPair<ToCoreMsg>> orderedMsgList = msgs.stream().map(msg -> new IdMsgPair<>(UUID.randomUUID(), msg)).collect(Collectors.toList()); 202 List<IdMsgPair<ToCoreMsg>> orderedMsgList = msgs.stream().map(msg -> new IdMsgPair<>(UUID.randomUUID(), msg)).collect(Collectors.toList());
204 ConcurrentMap<UUID, TbProtoQueueMsg<ToCoreMsg>> pendingMap = orderedMsgList.stream().collect( 203 ConcurrentMap<UUID, TbProtoQueueMsg<ToCoreMsg>> pendingMap = orderedMsgList.stream().collect(
205 Collectors.toConcurrentMap(IdMsgPair::getUuid, IdMsgPair::getMsg)); 204 Collectors.toConcurrentMap(IdMsgPair::getUuid, IdMsgPair::getMsg));
@@ -236,7 +235,6 @@ public class DefaultTbCoreConsumerService extends AbstractConsumerService<ToCore @@ -236,7 +235,6 @@ public class DefaultTbCoreConsumerService extends AbstractConsumerService<ToCore
236 tbCoreDeviceRpcService.forwardRpcRequestToDeviceActor((ToDeviceRpcRequestActorMsg) tbActorMsg); 235 tbCoreDeviceRpcService.forwardRpcRequestToDeviceActor((ToDeviceRpcRequestActorMsg) tbActorMsg);
237 } else { 236 } else {
238 log.trace("[{}] Forwarding message to App Actor {}", id, actorMsg.get()); 237 log.trace("[{}] Forwarding message to App Actor {}", id, actorMsg.get());
239 - log.info("演员消息生产——普通消息内容：【{}】",msg);  
240 actorContext.tell(actorMsg.get()); 238 actorContext.tell(actorMsg.get());
241 } 239 }
242 } 240 }
@@ -310,7 +308,6 @@ public class DefaultTbCoreConsumerService extends AbstractConsumerService<ToCore @@ -310,7 +308,6 @@ public class DefaultTbCoreConsumerService extends AbstractConsumerService<ToCore
310 Optional<TbActorMsg> actorMsg = encodingService.decode(toCoreNotification.getEdgeEventUpdateMsg().toByteArray()); 308 Optional<TbActorMsg> actorMsg = encodingService.decode(toCoreNotification.getEdgeEventUpdateMsg().toByteArray());
311 if (actorMsg.isPresent()) { 309 if (actorMsg.isPresent()) {
312 log.trace("[{}] Forwarding message to App Actor {}", id, actorMsg.get()); 310 log.trace("[{}] Forwarding message to App Actor {}", id, actorMsg.get());
313 - log.info("演员消息生产——高级消息内容：【{}】", actorMsg.get());  
314 actorContext.tellWithHighPriority(actorMsg.get()); 311 actorContext.tellWithHighPriority(actorMsg.get());
315 } 312 }
316 callback.onSuccess(); 313 callback.onSuccess();
@@ -328,7 +325,6 @@ public class DefaultTbCoreConsumerService extends AbstractConsumerService<ToCore @@ -328,7 +325,6 @@ public class DefaultTbCoreConsumerService extends AbstractConsumerService<ToCore
328 if (msgs.isEmpty()) { 325 if (msgs.isEmpty()) {
329 continue; 326 continue;
330 } 327 }
331 - log.info("队列消息消费 ：launchUsageStatsConsumer【{}】 ", msgs);  
332 ConcurrentMap<UUID, TbProtoQueueMsg<ToUsageStatsServiceMsg>> pendingMap = msgs.stream().collect( 328 ConcurrentMap<UUID, TbProtoQueueMsg<ToUsageStatsServiceMsg>> pendingMap = msgs.stream().collect(
333 Collectors.toConcurrentMap(s -> UUID.randomUUID(), Function.identity())); 329 Collectors.toConcurrentMap(s -> UUID.randomUUID(), Function.identity()));
334 CountDownLatch processingTimeoutLatch = new CountDownLatch(1); 330 CountDownLatch processingTimeoutLatch = new CountDownLatch(1);
@@ -373,7 +369,6 @@ public class DefaultTbCoreConsumerService extends AbstractConsumerService<ToCore @@ -373,7 +369,6 @@ public class DefaultTbCoreConsumerService extends AbstractConsumerService<ToCore
373 if (msgs.isEmpty()) { 369 if (msgs.isEmpty()) {
374 continue; 370 continue;
375 } 371 }
376 - log.info("队列消息消费:launchOtaPackageUpdateNotificationConsumer【{}】 ", msgs);  
377 long timeToSleep = maxProcessingTimeoutPerRecord; 372 long timeToSleep = maxProcessingTimeoutPerRecord;
378 for (TbProtoQueueMsg<ToOtaPackageStateServiceMsg> msg : msgs) { 373 for (TbProtoQueueMsg<ToOtaPackageStateServiceMsg> msg : msgs) {
379 try { 374 try {
@@ -510,7 +505,6 @@ public class DefaultTbCoreConsumerService extends AbstractConsumerService<ToCore @@ -510,7 +505,6 @@ public class DefaultTbCoreConsumerService extends AbstractConsumerService<ToCore
510 if (statsEnabled) { 505 if (statsEnabled) {
511 stats.log(toDeviceActorMsg); 506 stats.log(toDeviceActorMsg);
512 } 507 }
513 - log.info("演员消息生产——普通消息内容：TransportToDeviceActorMsgWrapper【{}】",new TransportToDeviceActorMsgWrapper(toDeviceActorMsg, callback));  
514 actorContext.tell(new TransportToDeviceActorMsgWrapper(toDeviceActorMsg, callback)); 508 actorContext.tell(new TransportToDeviceActorMsgWrapper(toDeviceActorMsg, callback));
515 } 509 }
516 510
@@ -253,7 +253,6 @@ public class DefaultTbRuleEngineConsumerService extends AbstractConsumerService< @@ -253,7 +253,6 @@ public class DefaultTbRuleEngineConsumerService extends AbstractConsumerService<
253 if (msgs.isEmpty()) { 253 if (msgs.isEmpty()) {
254 continue; 254 continue;
255 } 255 }
256 - log.info("队列消息消费 {} ", msgs);  
257 final TbRuleEngineSubmitStrategy submitStrategy = getSubmitStrategy(configuration); 256 final TbRuleEngineSubmitStrategy submitStrategy = getSubmitStrategy(configuration);
258 final TbRuleEngineProcessingStrategy ackStrategy = getAckStrategy(configuration); 257 final TbRuleEngineProcessingStrategy ackStrategy = getAckStrategy(configuration);
259 submitStrategy.init(msgs); 258 submitStrategy.init(msgs);
@@ -400,7 +399,6 @@ public class DefaultTbRuleEngineConsumerService extends AbstractConsumerService< @@ -400,7 +399,6 @@ public class DefaultTbRuleEngineConsumerService extends AbstractConsumerService<
400 } 399 }
401 } 400 }
402 msg = new QueueToRuleEngineMsg(tenantId, tbMsg, relationTypes, toRuleEngineMsg.getFailureMessage()); 401 msg = new QueueToRuleEngineMsg(tenantId, tbMsg, relationTypes, toRuleEngineMsg.getFailureMessage());
403 - log.info("演员消息生产——普通消息内容：【{}】",msg);  
404 actorContext.tell(msg); 402 actorContext.tell(msg);
405 } 403 }
406 404
@@ -115,7 +115,6 @@ public abstract class AbstractConsumerService<N extends com.google.protobuf.Gene @@ -115,7 +115,6 @@ public abstract class AbstractConsumerService<N extends com.google.protobuf.Gene
115 if (msgs.isEmpty()) { 115 if (msgs.isEmpty()) {
116 continue; 116 continue;
117 } 117 }
118 - log.info("队列消息消费 {} ", msgs);  
119 ConcurrentMap<UUID, TbProtoQueueMsg<N>> pendingMap = msgs.stream().collect( 118 ConcurrentMap<UUID, TbProtoQueueMsg<N>> pendingMap = msgs.stream().collect(
120 Collectors.toConcurrentMap(s -> UUID.randomUUID(), Function.identity())); 119 Collectors.toConcurrentMap(s -> UUID.randomUUID(), Function.identity()));
121 CountDownLatch processingTimeoutLatch = new CountDownLatch(1); 120 CountDownLatch processingTimeoutLatch = new CountDownLatch(1);
@@ -185,7 +184,6 @@ public abstract class AbstractConsumerService<N extends com.google.protobuf.Gene @@ -185,7 +184,6 @@ public abstract class AbstractConsumerService<N extends com.google.protobuf.Gene
185 } 184 }
186 } 185 }
187 log.trace("[{}] Forwarding message to App Actor {}", id, actorMsg); 186 log.trace("[{}] Forwarding message to App Actor {}", id, actorMsg);
188 - log.info("演员消息生产——高级消息内容：【{}】", actorMsg);  
189 actorContext.tellWithHighPriority(actorMsg); 187 actorContext.tellWithHighPriority(actorMsg);
190 } 188 }
191 } 189 }
@@ -123,7 +123,6 @@ public class DefaultTbCoreDeviceRpcService implements TbCoreDeviceRpcService { @@ -123,7 +123,6 @@ public class DefaultTbCoreDeviceRpcService implements TbCoreDeviceRpcService {
123 log.trace("[{}][{}] Processing local rpc call to device actor [{}]", request.getTenantId(), request.getId(), request.getDeviceId()); 123 log.trace("[{}][{}] Processing local rpc call to device actor [{}]", request.getTenantId(), request.getId(), request.getDeviceId());
124 UUID requestId = request.getId(); 124 UUID requestId = request.getId();
125 localToDeviceRpcRequests.put(requestId, rpcMsg); 125 localToDeviceRpcRequests.put(requestId, rpcMsg);
126 - log.info("演员消息生产——高级消息内容：【{}】", rpcMsg);  
127 actorContext.tellWithHighPriority(rpcMsg); 126 actorContext.tellWithHighPriority(rpcMsg);
128 scheduleToDeviceTimeout(request, requestId); 127 scheduleToDeviceTimeout(request, requestId);
129 } 128 }
@@ -143,7 +142,6 @@ public class DefaultTbCoreDeviceRpcService implements TbCoreDeviceRpcService { @@ -143,7 +142,6 @@ public class DefaultTbCoreDeviceRpcService implements TbCoreDeviceRpcService {
143 @Override 142 @Override
144 public void processRemoveRpc(RemoveRpcActorMsg removeRpcMsg) { 143 public void processRemoveRpc(RemoveRpcActorMsg removeRpcMsg) {
145 log.trace("[{}][{}] Processing remove RPC [{}]", removeRpcMsg.getTenantId(), removeRpcMsg.getRequestId(), removeRpcMsg.getDeviceId()); 144 log.trace("[{}][{}] Processing remove RPC [{}]", removeRpcMsg.getTenantId(), removeRpcMsg.getRequestId(), removeRpcMsg.getDeviceId());
146 - log.info("演员消息生产——高级消息内容：【{}】", removeRpcMsg);  
147 actorContext.tellWithHighPriority(removeRpcMsg); 145 actorContext.tellWithHighPriority(removeRpcMsg);
148 } 146 }
149 147
@@ -353,7 +353,6 @@ public class DefaultSubscriptionManagerService extends TbApplicationEventListene @@ -353,7 +353,6 @@ public class DefaultSubscriptionManagerService extends TbApplicationEventListene
353 } else { 353 } else {
354 TopicPartitionInfo tpi = partitionService.getNotificationsTopic(ServiceType.TB_CORE, s.getServiceId()); 354 TopicPartitionInfo tpi = partitionService.getNotificationsTopic(ServiceType.TB_CORE, s.getServiceId());
355 TbProtoQueueMsg queueMsg =toProto(s, subscriptionUpdate, ignoreEmptyUpdates); 355 TbProtoQueueMsg queueMsg =toProto(s, subscriptionUpdate, ignoreEmptyUpdates);
356 - log.info("队列消息生产 ——主题【{}】，消息【{}】 ",tpi.getFullTopicName(),queueMsg);  
357 toCoreNotificationsProducer.send(tpi, queueMsg, null); 356 toCoreNotificationsProducer.send(tpi, queueMsg, null);
358 } 357 }
359 } 358 }
@@ -378,7 +377,6 @@ public class DefaultSubscriptionManagerService extends TbApplicationEventListene @@ -378,7 +377,6 @@ public class DefaultSubscriptionManagerService extends TbApplicationEventListene
378 } else { 377 } else {
379 TopicPartitionInfo tpi = partitionService.getNotificationsTopic(ServiceType.TB_CORE, s.getServiceId()); 378 TopicPartitionInfo tpi = partitionService.getNotificationsTopic(ServiceType.TB_CORE, s.getServiceId());
380 TbProtoQueueMsg queueMsg =toProto(s, alarm, deleted); 379 TbProtoQueueMsg queueMsg =toProto(s, alarm, deleted);
381 - log.info("队列消息生产 ——主题【{}】，消息【{}】 ",tpi.getFullTopicName(),queueMsg);  
382 toCoreNotificationsProducer.send(tpi, queueMsg, null); 380 toCoreNotificationsProducer.send(tpi, queueMsg, null);
383 } 381 }
384 } 382 }
@@ -421,7 +419,6 @@ public class DefaultSubscriptionManagerService extends TbApplicationEventListene @@ -421,7 +419,6 @@ public class DefaultSubscriptionManagerService extends TbApplicationEventListene
421 if (!missedUpdates.isEmpty()) { 419 if (!missedUpdates.isEmpty()) {
422 TopicPartitionInfo tpi = partitionService.getNotificationsTopic(ServiceType.TB_CORE, subscription.getServiceId()); 420 TopicPartitionInfo tpi = partitionService.getNotificationsTopic(ServiceType.TB_CORE, subscription.getServiceId());
423 TbProtoQueueMsg queueMsg =toProto(subscription, missedUpdates); 421 TbProtoQueueMsg queueMsg =toProto(subscription, missedUpdates);
424 - log.info("队列消息生产 ——主题【{}】，消息【{}】 ",tpi.getFullTopicName(),queueMsg);  
425 toCoreNotificationsProducer.send(tpi, queueMsg, null); 422 toCoreNotificationsProducer.send(tpi, queueMsg, null);
426 } 423 }
427 }, 424 },
@@ -446,7 +443,6 @@ public class DefaultSubscriptionManagerService extends TbApplicationEventListene @@ -446,7 +443,6 @@ public class DefaultSubscriptionManagerService extends TbApplicationEventListene
446 if (missedUpdates != null && !missedUpdates.isEmpty()) { 443 if (missedUpdates != null && !missedUpdates.isEmpty()) {
447 TopicPartitionInfo tpi = partitionService.getNotificationsTopic(ServiceType.TB_CORE, subscription.getServiceId()); 444 TopicPartitionInfo tpi = partitionService.getNotificationsTopic(ServiceType.TB_CORE, subscription.getServiceId());
448 TbProtoQueueMsg queueMsg =toProto(subscription, missedUpdates); 445 TbProtoQueueMsg queueMsg =toProto(subscription, missedUpdates);
449 - log.info("队列消息生产 ——主题【{}】，消息【{}】 ",tpi.getFullTopicName(),queueMsg);  
450 toCoreNotificationsProducer.send(tpi, toProto(subscription, missedUpdates), null); 446 toCoreNotificationsProducer.send(tpi, toProto(subscription, missedUpdates), null);
451 } 447 }
452 }, 448 },
@@ -467,7 +463,6 @@ public class DefaultSubscriptionManagerService extends TbApplicationEventListene @@ -467,7 +463,6 @@ public class DefaultSubscriptionManagerService extends TbApplicationEventListene
467 if (missedUpdates != null && !missedUpdates.isEmpty()) { 463 if (missedUpdates != null && !missedUpdates.isEmpty()) {
468 TopicPartitionInfo tpi = partitionService.getNotificationsTopic(ServiceType.TB_CORE, subscription.getServiceId()); 464 TopicPartitionInfo tpi = partitionService.getNotificationsTopic(ServiceType.TB_CORE, subscription.getServiceId());
469 TbProtoQueueMsg queueMsg =toProto(subscription, missedUpdates); 465 TbProtoQueueMsg queueMsg =toProto(subscription, missedUpdates);
470 - log.info("队列消息生产 ——主题【{}】，消息【{}】 ",tpi.getFullTopicName(),queueMsg);  
471 toCoreNotificationsProducer.send(tpi, queueMsg, null); 466 toCoreNotificationsProducer.send(tpi, queueMsg, null);
472 } 467 }
473 }, 468 },
@@ -65,7 +65,6 @@ public class DefaultTbCoreToTransportService implements TbCoreToTransportService @@ -65,7 +65,6 @@ public class DefaultTbCoreToTransportService implements TbCoreToTransportService
65 UUID sessionId = new UUID(msg.getSessionIdMSB(), msg.getSessionIdLSB()); 65 UUID sessionId = new UUID(msg.getSessionIdMSB(), msg.getSessionIdLSB());
66 log.trace("[{}][{}] Pushing session data to topic: {}", tpi.getFullTopicName(), sessionId, msg); 66 log.trace("[{}][{}] Pushing session data to topic: {}", tpi.getFullTopicName(), sessionId, msg);
67 TbProtoQueueMsg<ToTransportMsg> queueMsg = new TbProtoQueueMsg<>(NULL_UUID, msg); 67 TbProtoQueueMsg<ToTransportMsg> queueMsg = new TbProtoQueueMsg<>(NULL_UUID, msg);
68 - log.info("队列消息生产 ——主题【{}】，消息【{}】 ",tpi.getFullTopicName(),queueMsg);  
69 tbTransportProducer.send(tpi, queueMsg, new QueueCallbackAdaptor(onSuccess, onFailure)); 68 tbTransportProducer.send(tpi, queueMsg, new QueueCallbackAdaptor(onSuccess, onFailure));
70 } 69 }
71 70
@@ -133,7 +133,6 @@ public class DefaultTbActorSystem implements TbActorSystem { @@ -133,7 +133,6 @@ public class DefaultTbActorSystem implements TbActorSystem {
133 133
134 @Override 134 @Override
135 public void tellWithHighPriority(TbActorId target, TbActorMsg actorMsg) { 135 public void tellWithHighPriority(TbActorId target, TbActorMsg actorMsg) {
136 - log.info("演员消息生产——高级消息内容：【{}】",actorMsg);  
137 tell(target, actorMsg, true); 136 tell(target, actorMsg, true);
138 } 137 }
139 138
@@ -165,7 +164,7 @@ public class DefaultTbActorSystem implements TbActorSystem { @@ -165,7 +164,7 @@ public class DefaultTbActorSystem implements TbActorSystem {
165 public void broadcastToChildren(TbActorId parent, Predicate<TbActorId> childFilter, TbActorMsg msg) { 164 public void broadcastToChildren(TbActorId parent, Predicate<TbActorId> childFilter, TbActorMsg msg) {
166 Set<TbActorId> children = parentChildMap.get(parent); 165 Set<TbActorId> children = parentChildMap.get(parent);
167 if (children != null) { 166 if (children != null) {
168 - children.stream().filter(childFilter).forEach(id -> {log.info("演员消息生产——普通消息内容：broadcastToChildren【{}】",msg);tell(id, msg);}); 167 + children.stream().filter(childFilter).forEach(id -> {tell(id, msg);});
169 } 168 }
170 } 169 }
171 170
@@ -60,7 +60,7 @@ public abstract class AbstractTbQueueConsumerTemplate<R, T extends TbQueueMsg> i @@ -60,7 +60,7 @@ public abstract class AbstractTbQueueConsumerTemplate<R, T extends TbQueueMsg> i
60 return; 60 return;
61 } 61 }
62 Set<TopicPartitionInfo> partitions =Collections.singleton(new TopicPartitionInfo(topic, null, null, true)); 62 Set<TopicPartitionInfo> partitions =Collections.singleton(new TopicPartitionInfo(topic, null, null, true));
63 - log.info("队列消息订阅 {} ", partitions); 63 +
64 subscribeQueue.add(partitions); 64 subscribeQueue.add(partitions);
65 } 65 }
66 66
@@ -71,7 +71,6 @@ public abstract class AbstractTbQueueConsumerTemplate<R, T extends TbQueueMsg> i @@ -71,7 +71,6 @@ public abstract class AbstractTbQueueConsumerTemplate<R, T extends TbQueueMsg> i
71 log.error("trying subscribe, but consumer stopped for topic {}", topic); 71 log.error("trying subscribe, but consumer stopped for topic {}", topic);
72 return; 72 return;
73 } 73 }
74 - log.info("队列消息订阅 {} ", partitions);  
75 subscribeQueue.add(partitions); 74 subscribeQueue.add(partitions);
76 } 75 }
77 76
@@ -97,7 +96,6 @@ public abstract class AbstractTbQueueConsumerTemplate<R, T extends TbQueueMsg> i @@ -97,7 +96,6 @@ public abstract class AbstractTbQueueConsumerTemplate<R, T extends TbQueueMsg> i
97 partitions = subscribeQueue.poll(); 96 partitions = subscribeQueue.poll();
98 } 97 }
99 if (!subscribed) { 98 if (!subscribed) {
100 - log.info("队列消息订阅 {} ", partitions);  
101 List<String> topicNames = partitions.stream().map(TopicPartitionInfo::getFullTopicName).collect(Collectors.toList()); 99 List<String> topicNames = partitions.stream().map(TopicPartitionInfo::getFullTopicName).collect(Collectors.toList());
102 doSubscribe(topicNames); 100 doSubscribe(topicNames);
103 subscribed = true; 101 subscribed = true;
@@ -113,7 +113,6 @@ public class DefaultTbQueueRequestTemplate<Request extends TbQueueMsg, Response @@ -113,7 +113,6 @@ public class DefaultTbQueueRequestTemplate<Request extends TbQueueMsg, Response
113 List<Response> responses = doPoll(); //poll js responses 113 List<Response> responses = doPoll(); //poll js responses
114 log.trace("Completed template poll topic {}, for pendingRequests [{}], received [{}] responses", responseTemplate.getTopic(), pendingRequestsCount, responses.size()); 114 log.trace("Completed template poll topic {}, for pendingRequests [{}], received [{}] responses", responseTemplate.getTopic(), pendingRequestsCount, responses.size());
115 if(!responses.isEmpty()){ 115 if(!responses.isEmpty()){
116 - log.info("队列消息消费 fetchAndProcessResponses【{}】 ", responses);  
117 } 116 }
118 responses.forEach(this::processResponse); //this can take a long time 117 responses.forEach(this::processResponse); //this can take a long time
119 responseTemplate.commit(); 118 responseTemplate.commit();
@@ -254,7 +253,6 @@ public class DefaultTbQueueRequestTemplate<Request extends TbQueueMsg, Response @@ -254,7 +253,6 @@ public class DefaultTbQueueRequestTemplate<Request extends TbQueueMsg, Response
254 } 253 }
255 254
256 TopicPartitionInfo topicInfo = TopicPartitionInfo.builder().topic(requestTemplate.getDefaultTopic()).build(); 255 TopicPartitionInfo topicInfo = TopicPartitionInfo.builder().topic(requestTemplate.getDefaultTopic()).build();
257 - log.info("队列消息生产 ——主题【{}】，消息【{}】 ",topicInfo.getFullTopicName(),request);  
258 requestTemplate.send(topicInfo, request, new TbQueueCallback() { 256 requestTemplate.send(topicInfo, request, new TbQueueCallback() {
259 @Override 257 @Override
260 public void onSuccess(TbQueueMsgMetadata metadata) { 258 public void onSuccess(TbQueueMsgMetadata metadata) {
@@ -94,7 +94,6 @@ public class DefaultTbQueueResponseTemplate<Request extends TbQueueMsg, Response @@ -94,7 +94,6 @@ public class DefaultTbQueueResponseTemplate<Request extends TbQueueMsg, Response
94 if (requests.isEmpty()) { 94 if (requests.isEmpty()) {
95 continue; 95 continue;
96 } 96 }
97 - log.info("队列消息消费 {} ", requests);  
98 97
99 requests.forEach(request -> { 98 requests.forEach(request -> {
100 long currentTime = System.currentTimeMillis(); 99 long currentTime = System.currentTimeMillis();
@@ -120,7 +119,6 @@ public class DefaultTbQueueResponseTemplate<Request extends TbQueueMsg, Response @@ -120,7 +119,6 @@ public class DefaultTbQueueResponseTemplate<Request extends TbQueueMsg, Response
120 pendingRequestCount.decrementAndGet(); 119 pendingRequestCount.decrementAndGet();
121 response.getHeaders().put(REQUEST_ID_HEADER, uuidToBytes(requestId)); 120 response.getHeaders().put(REQUEST_ID_HEADER, uuidToBytes(requestId));
122 TopicPartitionInfo topicInfo = TopicPartitionInfo.builder().topic(responseTopic).build(); 121 TopicPartitionInfo topicInfo = TopicPartitionInfo.builder().topic(responseTopic).build();
123 - log.info("队列消息生产 ——主题【{}】，消息【{}】 ",topicInfo.getFullTopicName(),response);  
124 responseTemplate.send(topicInfo, response, null); 122 responseTemplate.send(topicInfo, response, null);
125 stats.incrementSuccessful(); 123 stats.incrementSuccessful();
126 }, 124 },
@@ -81,7 +81,6 @@ public class TbKafkaProducerTemplate<T extends TbQueueMsg> implements TbQueuePro @@ -81,7 +81,6 @@ public class TbKafkaProducerTemplate<T extends TbQueueMsg> implements TbQueuePro
81 ProducerRecord<String, byte[]> record; 81 ProducerRecord<String, byte[]> record;
82 Iterable<Header> headers = msg.getHeaders().getData().entrySet().stream().map(e -> new RecordHeader(e.getKey(), e.getValue())).collect(Collectors.toList()); 82 Iterable<Header> headers = msg.getHeaders().getData().entrySet().stream().map(e -> new RecordHeader(e.getKey(), e.getValue())).collect(Collectors.toList());
83 record = new ProducerRecord<>(tpi.getFullTopicName(), null, key, data, headers); 83 record = new ProducerRecord<>(tpi.getFullTopicName(), null, key, data, headers);
84 - log.info("队列消息生产 ——主题【{}】，消息【{}】 ",tpi.getFullTopicName(),record);  
85 producer.send(record, (metadata, exception) -> { 84 producer.send(record, (metadata, exception) -> {
86 if (exception == null) { 85 if (exception == null) {
87 if (callback != null) { 86 if (callback != null) {
@@ -124,7 +124,6 @@ public class DefaultTbApiUsageClient implements TbApiUsageClient { @@ -124,7 +124,6 @@ public class DefaultTbApiUsageClient implements TbApiUsageClient {
124 124
125 125
126 TbProtoQueueMsg queueMsg = new TbProtoQueueMsg<>(UUID.randomUUID(), statsMsg.build()); 126 TbProtoQueueMsg queueMsg = new TbProtoQueueMsg<>(UUID.randomUUID(), statsMsg.build());
127 - log.info("队列消息生产 ——主题【{}】，消息【{}】 ",tpi.getFullTopicName(),queueMsg);  
128 msgProducer.send(tpi, queueMsg, null); 127 msgProducer.send(tpi, queueMsg, null);
129 })); 128 }));
130 129
@@ -225,7 +225,6 @@ public class DefaultTransportService implements TransportService { @@ -225,7 +225,6 @@ public class DefaultTransportService implements TransportService {
225 if (records.size() == 0) { 225 if (records.size() == 0) {
226 continue; 226 continue;
227 } 227 }
228 - log.info("队列消息消费 {} ", records);  
229 records.forEach(record -> { 228 records.forEach(record -> {
230 try { 229 try {
231 processToTransportMsg(record.getValue()); 230 processToTransportMsg(record.getValue());
@@ -1034,7 +1033,6 @@ public class DefaultTransportService implements TransportService { @@ -1034,7 +1033,6 @@ public class DefaultTransportService implements TransportService {
1034 new TransportTbQueueCallback(callback) : null; 1033 new TransportTbQueueCallback(callback) : null;
1035 tbCoreProducerStats.incrementTotal(); 1034 tbCoreProducerStats.incrementTotal();
1036 StatsCallback wrappedCallback = new StatsCallback(transportTbQueueCallback, tbCoreProducerStats); 1035 StatsCallback wrappedCallback = new StatsCallback(transportTbQueueCallback, tbCoreProducerStats);
1037 - log.info("队列消息生产 ——主题【{}】，消息【{}】 ",tpi.getFullTopicName(),new TbProtoQueueMsg<>(getRoutingKey(sessionInfo),ToCoreMsg.newBuilder().setToDeviceActorMsg(toDeviceActorMsg).build()));  
1038 tbCoreMsgProducer.send(tpi 1036 tbCoreMsgProducer.send(tpi
1039 ,new TbProtoQueueMsg<>(getRoutingKey(sessionInfo),ToCoreMsg.newBuilder().setToDeviceActorMsg(toDeviceActorMsg).build()) 1037 ,new TbProtoQueueMsg<>(getRoutingKey(sessionInfo),ToCoreMsg.newBuilder().setToDeviceActorMsg(toDeviceActorMsg).build())
1040 ,wrappedCallback); 1038 ,wrappedCallback);
@@ -1052,7 +1050,6 @@ public class DefaultTransportService implements TransportService { @@ -1052,7 +1050,6 @@ public class DefaultTransportService implements TransportService {
1052 StatsCallback wrappedCallback = new StatsCallback(callback, ruleEngineProducerStats); 1050 StatsCallback wrappedCallback = new StatsCallback(callback, ruleEngineProducerStats);
1053 1051
1054 TbProtoQueueMsg queueMsg = new TbProtoQueueMsg<>(tbMsg.getId(), msg); 1052 TbProtoQueueMsg queueMsg = new TbProtoQueueMsg<>(tbMsg.getId(), msg);
1055 - log.info("队列消息生产 ——主题【{}】，消息【{}】 ",tpi.getFullTopicName(),queueMsg);  
1056 ruleEngineMsgProducer.send(tpi, queueMsg, wrappedCallback); 1053 ruleEngineMsgProducer.send(tpi, queueMsg, wrappedCallback);
1057 } 1054 }
1058 1055
@@ -94,6 +94,8 @@ public class YtNoticeServiceImpl implements YtNoticeService { @@ -94,6 +94,8 @@ public class YtNoticeServiceImpl implements YtNoticeService {
94 */ 94 */
95 private void noticeAll(String messageCode, List<AlarmContact> alarmContactList, AlarmInfoDTO alarmInfo, Organization organization,String tenantId) { 95 private void noticeAll(String messageCode, List<AlarmContact> alarmContactList, AlarmInfoDTO alarmInfo, Organization organization,String tenantId) {
96 Optional.ofNullable(alarmContactList).ifPresent(contacts -> { 96 Optional.ofNullable(alarmContactList).ifPresent(contacts -> {
  97 +
  98 + /**可用的告警通知模板*/
97 QueryWrapper<MessageTemplate> messageTemplateQueryWrapper = new QueryWrapper<MessageTemplate>(); 99 QueryWrapper<MessageTemplate> messageTemplateQueryWrapper = new QueryWrapper<MessageTemplate>();
98 messageTemplateQueryWrapper.lambda() 100 messageTemplateQueryWrapper.lambda()
99 .eq(MessageTemplate::getTenantId, tenantId) 101 .eq(MessageTemplate::getTenantId, tenantId)
@@ -106,62 +108,143 @@ public class YtNoticeServiceImpl implements YtNoticeService { @@ -106,62 +108,143 @@ public class YtNoticeServiceImpl implements YtNoticeService {
106 templatesMap.put(item.getMessageType(), item.getId()); 108 templatesMap.put(item.getMessageType(), item.getId());
107 } 109 }
108 }); 110 });
109 - List<String> emailReceivers = new ArrayList<>();  
110 - contacts.stream().parallel().forEach(item -> {  
111 - if (messageCode.contains(MessageTypeEnum.PHONE_MESSAGE.name())  
112 - && templatesMap.containsKey(MessageTypeEnum.PHONE_MESSAGE.name())  
113 - && !item.getPhone().isEmpty()) {  
114 -  
115 -  
116 - SmsReqDTO info = new SmsReqDTO();  
117 - info.setId(templatesMap.get(MessageTypeEnum.PHONE_MESSAGE.name()));  
118 - info.setPhoneNumbers(item.getPhone());  
119 - LinkedHashMap<String, String> params = new LinkedHashMap<>();  
120 - //name-其他；device_name-其他；level-其他；location-其他；alarm_value-其他；  
121 - params.put("type", alarmInfo.getType());  
122 - params.put("device_name", alarmInfo.getDeviceName());  
123 - params.put("severity", alarmInfo.getSeverity());  
124 - params.put("organization", organization != null ? organization.getName() : "");  
125 - params.put("createtime", YtDateTimeUtils.formate(alarmInfo.getCreateTs()));  
126 - info.setParams(params);  
127 - smsService.sendSms(info);  
128 - }  
129 111
130 - if (messageCode.contains(MessageTypeEnum.EMAIL_MESSAGE.name())  
131 - && templatesMap.containsKey(MessageTypeEnum.EMAIL_MESSAGE.name())  
132 - && !item.getEmail().isEmpty()) {  
133 - emailReceivers.add(item.getEmail());  
134 - }  
135 112
  113 + if (messageCode.contains(MessageTypeEnum.PHONE_MESSAGE.name())
  114 + && templatesMap.containsKey(MessageTypeEnum.PHONE_MESSAGE.name())) {
  115 + sms4Alarm(alarmInfo,templatesMap.get(MessageTypeEnum.PHONE_MESSAGE.name()),organization,contacts);
  116 + }
136 117
137 - if (messageCode.contains(MessageTypeEnum.DING_TALK_MESSAGE.name())  
138 - && templatesMap.containsKey(MessageTypeEnum.DING_TALK_MESSAGE.name())  
139 - && !item.getDingtalk().isEmpty()) { 118 + if (messageCode.contains(MessageTypeEnum.EMAIL_MESSAGE.name())
  119 + && templatesMap.containsKey(MessageTypeEnum.EMAIL_MESSAGE.name())) {
  120 + email4Alarm(alarmInfo,templatesMap.get(MessageTypeEnum.EMAIL_MESSAGE.name()),organization,contacts);
  121 + }
140 122
141 - }  
142 123
143 - if (messageCode.contains(MessageTypeEnum.WECHAT_MESSAGE.name())  
144 - && templatesMap.containsKey(MessageTypeEnum.WECHAT_MESSAGE.name())  
145 - && !item.getWechat().isEmpty()) { 124 + if (messageCode.contains(MessageTypeEnum.DING_TALK_MESSAGE.name())
  125 + && templatesMap.containsKey(MessageTypeEnum.DING_TALK_MESSAGE.name())) {
  126 + dingTalk4Alarm(alarmInfo,templatesMap.get(MessageTypeEnum.DING_TALK_MESSAGE.name()),organization,contacts);
  127 + }
146 128
147 - }  
148 - });  
149 - if (!emailReceivers.isEmpty()) {  
150 - EmailReqDTO info = new EmailReqDTO();  
151 - info.setTo(emailReceivers.toArray(new String[emailReceivers.size()]));  
152 - info.setSubject(String.format("【%s】告警通知",alarmInfo.getDeviceName()));  
153 - String body =String.format("%s位于【%s】的设备【%s】触发【%s】级的【%s】，请尽快处理！"  
154 - ,YtDateTimeUtils.formate(alarmInfo.getCreateTs())  
155 - , organization != null ? organization.getName() : ""  
156 - ,alarmInfo.getDeviceName()  
157 - ,alarmInfo.getSeverity()  
158 - , alarmInfo.getType());  
159 - info.setBody(body);  
160 - info.setEmailFormatEnum(EmailFormatEnum.TEXT.name());  
161 - info.setId(templatesMap.get(MessageTypeEnum.EMAIL_MESSAGE.name()));  
162 -  
163 - mailService.sendEmail(info); 129 + if (messageCode.contains(MessageTypeEnum.WECHAT_MESSAGE.name())
  130 + && templatesMap.containsKey(MessageTypeEnum.WECHAT_MESSAGE.name())) {
  131 + weChat4Alarm(alarmInfo,templatesMap.get(MessageTypeEnum.WECHAT_MESSAGE.name()),organization,contacts);
164 } 132 }
165 }); 133 });
166 } 134 }
  135 +
  136 + /**
  137 + * 短信通知设备告警信息
  138 + * @param alarmInfo 告警信息
  139 + * @param templateId 告警模板主键
  140 + * @param organization 设备所属组织
  141 + * @param contacts 设备告警联系人
  142 + */
  143 + private void sms4Alarm(AlarmInfoDTO alarmInfo,String templateId, Organization organization,List<AlarmContact> contacts){
  144 + SmsReqDTO info = new SmsReqDTO();
  145 + info.setId(templateId);
  146 + LinkedHashMap<String, String> params = new LinkedHashMap<>();
  147 + //name-其他；device_name-其他；level-其他；location-其他；alarm_value-其他；
  148 + params.put("type", alarmInfo.getType());
  149 + params.put("device_name", alarmInfo.getDeviceName());
  150 + params.put("severity", alarmInfo.getSeverity());
  151 + params.put("organization", organization != null ? organization.getName() : "");
  152 + params.put("createtime", YtDateTimeUtils.formate(alarmInfo.getCreateTs()));
  153 + info.setParams(params);
  154 + contacts.stream().parallel().forEach(item -> {
  155 + if (!item.getPhone().isEmpty()) {
  156 + info.setPhoneNumbers(item.getPhone());
  157 + smsService.sendSms(info);
  158 + }
  159 + });
  160 + }
  161 +
  162 + /**
  163 + * 邮件通知设备告警信息
  164 + * @param alarmInfo 告警信息
  165 + * @param templateId 告警模板主键
  166 + * @param organization 设备所属组织
  167 + * @param contacts 设备告警联系人
  168 + */
  169 + private void email4Alarm(AlarmInfoDTO alarmInfo,String templateId, Organization organization,List<AlarmContact> contacts){
  170 + List<String> emailReceivers = new ArrayList<>();
  171 + contacts.stream().parallel().forEach(item -> {
  172 + if (!item.getEmail().isEmpty()) {
  173 + emailReceivers.add(item.getEmail());
  174 + }
  175 + });
  176 + if (!emailReceivers.isEmpty()) {
  177 + EmailReqDTO info = new EmailReqDTO();
  178 + info.setTo(emailReceivers.toArray(new String[emailReceivers.size()]));
  179 + info.setSubject(String.format("【%s】告警通知",alarmInfo.getDeviceName()));
  180 + String body =String.format("%s位于【%s】的设备【%s】触发【%s】级的【%s】，请尽快处理！"
  181 + ,YtDateTimeUtils.formate(alarmInfo.getCreateTs())
  182 + , organization != null ? organization.getName() : ""
  183 + ,alarmInfo.getDeviceName()
  184 + ,alarmInfo.getSeverity()
  185 + , alarmInfo.getType());
  186 + info.setBody(body);
  187 + info.setEmailFormatEnum(EmailFormatEnum.TEXT.name());
  188 + info.setId(templateId);
  189 +
  190 + mailService.sendEmail(info);
  191 + }
  192 + }
  193 +
  194 + /**
  195 + * 钉钉通知设备告警信息
  196 + * @param alarmInfo 告警信息
  197 + * @param templateId 告警模板主键
  198 + * @param organization 设备所属组织
  199 + * @param contacts 设备告警联系人
  200 + */
  201 + private void dingTalk4Alarm(AlarmInfoDTO alarmInfo,String templateId, Organization organization,List<AlarmContact> contacts){
  202 + // TODO 推送钉钉消息
  203 + SmsReqDTO info = new SmsReqDTO();
  204 + info.setId(templateId);
  205 + LinkedHashMap<String, String> params = new LinkedHashMap<>();
  206 + //name-其他；device_name-其他；level-其他；location-其他；alarm_value-其他；
  207 + params.put("type", alarmInfo.getType());
  208 + params.put("device_name", alarmInfo.getDeviceName());
  209 + params.put("severity", alarmInfo.getSeverity());
  210 + params.put("organization", organization != null ? organization.getName() : "");
  211 + params.put("createtime", YtDateTimeUtils.formate(alarmInfo.getCreateTs()));
  212 + info.setParams(params);
  213 + contacts.stream().parallel().forEach(item -> {
  214 + if (!item.getDingtalk().isEmpty()) {
  215 + info.setPhoneNumbers(item.getDingtalk());
  216 + smsService.sendSms(info);
  217 + }
  218 + });
  219 +
  220 + }
  221 +
  222 + /**
  223 + * 微信通知设备告警信息
  224 + * @param alarmInfo 告警信息
  225 + * @param templateId 告警模板主键
  226 + * @param organization 设备所属组织
  227 + * @param contacts 设备告警联系人
  228 + */
  229 + private void weChat4Alarm(AlarmInfoDTO alarmInfo,String templateId, Organization organization,List<AlarmContact> contacts){
  230 + // TODO 推送微信通知
  231 + SmsReqDTO info = new SmsReqDTO();
  232 + info.setId(templateId);
  233 + LinkedHashMap<String, String> params = new LinkedHashMap<>();
  234 + //name-其他；device_name-其他；level-其他；location-其他；alarm_value-其他；
  235 + params.put("type", alarmInfo.getType());
  236 + params.put("device_name", alarmInfo.getDeviceName());
  237 + params.put("severity", alarmInfo.getSeverity());
  238 + params.put("organization", organization != null ? organization.getName() : "");
  239 + params.put("createtime", YtDateTimeUtils.formate(alarmInfo.getCreateTs()));
  240 + info.setParams(params);
  241 + contacts.stream().parallel().forEach(item -> {
  242 + if (!item.getWechat().isEmpty()) {
  243 + info.setPhoneNumbers(item.getWechat());
  244 + smsService.sendSms(info);
  245 + }
  246 + });
  247 + }
  248 +
  249 +
167 } 250 }
@@ -53,10 +53,8 @@ public abstract class TbAbstractAlarmNode<C extends TbAbstractAlarmNodeConfigura @@ -53,10 +53,8 @@ public abstract class TbAbstractAlarmNode<C extends TbAbstractAlarmNodeConfigura
53 53
54 @Override 54 @Override
55 public void onMsg(TbContext ctx, TbMsg msg) { 55 public void onMsg(TbContext ctx, TbMsg msg) {
56 - log.info("节点【TbAbstractAlarmNode】处理消费: {}", msg);  
57 withCallback(processAlarm(ctx, msg), 56 withCallback(processAlarm(ctx, msg),
58 alarmResult -> { 57 alarmResult -> {
59 - log.info("演员消息生产——普通消息内容：【{}】",msg);  
60 if (alarmResult.alarm == null) { 58 if (alarmResult.alarm == null) {
61 ctx.tellNext(msg, "False"); 59 ctx.tellNext(msg, "False");
62 } else if (alarmResult.isCreated) { 60 } else if (alarmResult.isCreated) {
@@ -111,7 +109,7 @@ public abstract class TbAbstractAlarmNode<C extends TbAbstractAlarmNodeConfigura @@ -111,7 +109,7 @@ public abstract class TbAbstractAlarmNode<C extends TbAbstractAlarmNodeConfigura
111 109
112 private void tellNext(TbContext ctx, TbMsg msg, TbAlarmResult alarmResult, String entityAction, String alarmAction) { 110 private void tellNext(TbContext ctx, TbMsg msg, TbAlarmResult alarmResult, String entityAction, String alarmAction) {
113 ctx.enqueue(ctx.alarmActionMsg(alarmResult.alarm, ctx.getSelfId(), entityAction), 111 ctx.enqueue(ctx.alarmActionMsg(alarmResult.alarm, ctx.getSelfId(), entityAction),
114 - () -> {log.info("演员消息生产——普通消息内容：transformSuccess【{}】",toAlarmMsg(ctx, alarmResult, msg));ctx.tellNext(toAlarmMsg(ctx, alarmResult, msg), alarmAction);}, 112 + () -> {ctx.tellNext(toAlarmMsg(ctx, alarmResult, msg), alarmAction);},
115 throwable -> ctx.tellFailure(toAlarmMsg(ctx, alarmResult, msg), throwable)); 113 throwable -> ctx.tellFailure(toAlarmMsg(ctx, alarmResult, msg), throwable));
116 } 114 }
117 } 115 }
@@ -64,7 +64,6 @@ public abstract class TbAbstractCustomerActionNode<C extends TbAbstractCustomerA @@ -64,7 +64,6 @@ public abstract class TbAbstractCustomerActionNode<C extends TbAbstractCustomerA
64 public void onMsg(TbContext ctx, TbMsg msg) { 64 public void onMsg(TbContext ctx, TbMsg msg) {
65 withCallback(processCustomerAction(ctx, msg), 65 withCallback(processCustomerAction(ctx, msg),
66 m -> { 66 m -> {
67 - log.info("演员消息生产——普通消息内容：tellFailure【{}】",msg);  
68 ctx.tellSuccess(msg);} 67 ctx.tellSuccess(msg);}
69 ,t -> ctx.tellFailure(msg, t), ctx.getDbCallbackExecutor()); 68 ,t -> ctx.tellFailure(msg, t), ctx.getDbCallbackExecutor());
70 } 69 }
@@ -84,7 +84,6 @@ public abstract class TbAbstractRelationActionNode<C extends TbAbstractRelationA @@ -84,7 +84,6 @@ public abstract class TbAbstractRelationActionNode<C extends TbAbstractRelationA
84 String relationType = processPattern(msg, config.getRelationType()); 84 String relationType = processPattern(msg, config.getRelationType());
85 withCallback(processEntityRelationAction(ctx, msg, relationType), 85 withCallback(processEntityRelationAction(ctx, msg, relationType),
86 filterResult -> { 86 filterResult -> {
87 - log.info("演员消息生产——普通消息内容：tellFailure【{}】",filterResult.getMsg());  
88 ctx.tellNext(filterResult.getMsg(), filterResult.isResult() ? SUCCESS : FAILURE);} 87 ctx.tellNext(filterResult.getMsg(), filterResult.isResult() ? SUCCESS : FAILURE);}
89 , t -> ctx.tellFailure(msg, t), ctx.getDbCallbackExecutor()); 88 , t -> ctx.tellFailure(msg, t), ctx.getDbCallbackExecutor());
90 } 89 }
@@ -56,7 +56,6 @@ public class TbClearAlarmNode extends TbAbstractAlarmNode<TbClearAlarmNodeConfig @@ -56,7 +56,6 @@ public class TbClearAlarmNode extends TbAbstractAlarmNode<TbClearAlarmNodeConfig
56 56
57 @Override 57 @Override
58 protected ListenableFuture<TbAlarmResult> processAlarm(TbContext ctx, TbMsg msg) { 58 protected ListenableFuture<TbAlarmResult> processAlarm(TbContext ctx, TbMsg msg) {
59 - log.info("节点【TbClearAlarmNode】处理消费: {}", msg);  
60 String alarmType = TbNodeUtils.processPattern(this.config.getAlarmType(), msg); 59 String alarmType = TbNodeUtils.processPattern(this.config.getAlarmType(), msg);
61 ListenableFuture<Alarm> alarmFuture; 60 ListenableFuture<Alarm> alarmFuture;
62 if (msg.getOriginator().getEntityType().equals(EntityType.ALARM)) { 61 if (msg.getOriginator().getEntityType().equals(EntityType.ALARM)) {
@@ -81,7 +81,6 @@ public class TbCreateAlarmNode extends TbAbstractAlarmNode<TbCreateAlarmNodeConf @@ -81,7 +81,6 @@ public class TbCreateAlarmNode extends TbAbstractAlarmNode<TbCreateAlarmNodeConf
81 81
82 @Override 82 @Override
83 protected ListenableFuture<TbAlarmResult> processAlarm(TbContext ctx, TbMsg msg) { 83 protected ListenableFuture<TbAlarmResult> processAlarm(TbContext ctx, TbMsg msg) {
84 - log.info("节点【TbCreateAlarmNode】处理消费: {}", msg);  
85 String alarmType; 84 String alarmType;
86 final Alarm msgAlarm; 85 final Alarm msgAlarm;
87 86
@@ -64,7 +64,6 @@ public class TbLogNode implements TbNode { @@ -64,7 +64,6 @@ public class TbLogNode implements TbNode {
64 public void onSuccess(@Nullable String result) { 64 public void onSuccess(@Nullable String result) {
65 ctx.logJsEvalResponse(); 65 ctx.logJsEvalResponse();
66 log.info(result); 66 log.info(result);
67 - log.info("演员消息生产——普通消息内容：【{}】", msg);  
68 ctx.tellSuccess(msg); 67 ctx.tellSuccess(msg);
69 } 68 }
70 69
@@ -96,7 +96,6 @@ public class TbMsgCountNode implements TbNode { @@ -96,7 +96,6 @@ public class TbMsgCountNode implements TbNode {
96 long curDelay = Math.max(0L, (lastScheduledTs - curTs)); 96 long curDelay = Math.max(0L, (lastScheduledTs - curTs));
97 TbMsg tickMsg = ctx.newMsg(ServiceQueue.MAIN, TB_MSG_COUNT_NODE_MSG, ctx.getSelfId(), msg != null ? msg.getCustomerId() : null, new TbMsgMetaData(), ""); 97 TbMsg tickMsg = ctx.newMsg(ServiceQueue.MAIN, TB_MSG_COUNT_NODE_MSG, ctx.getSelfId(), msg != null ? msg.getCustomerId() : null, new TbMsgMetaData(), "");
98 nextTickId = tickMsg.getId(); 98 nextTickId = tickMsg.getId();
99 - log.info("演员消息生产——高级消息内容：【{}】", tickMsg);  
100 ctx.tellSelf(tickMsg, curDelay); 99 ctx.tellSelf(tickMsg, curDelay);
101 } 100 }
102 101
@@ -105,7 +105,6 @@ public class TbMsgGeneratorNode implements TbNode { @@ -105,7 +105,6 @@ public class TbMsgGeneratorNode implements TbNode {
105 @Override 105 @Override
106 public void onMsg(TbContext ctx, TbMsg msg) { 106 public void onMsg(TbContext ctx, TbMsg msg) {
107 log.trace("onMsg, config {}, msg {}", config, msg); 107 log.trace("onMsg, config {}, msg {}", config, msg);
108 - log.info("节点【TbMsgGeneratorNode】处理消费: {}", msg);  
109 if (initialized.get() && msg.getType().equals(TB_MSG_GENERATOR_NODE_MSG) && msg.getId().equals(nextTickId)) { 108 if (initialized.get() && msg.getType().equals(TB_MSG_GENERATOR_NODE_MSG) && msg.getId().equals(nextTickId)) {
110 TbStopWatch sw = TbStopWatch.startNew(); 109 TbStopWatch sw = TbStopWatch.startNew();
111 withCallback(generate(ctx, msg), 110 withCallback(generate(ctx, msg),
@@ -138,7 +137,6 @@ public class TbMsgGeneratorNode implements TbNode { @@ -138,7 +137,6 @@ public class TbMsgGeneratorNode implements TbNode {
138 long curDelay = Math.max(0L, (lastScheduledTs - curTs)); 137 long curDelay = Math.max(0L, (lastScheduledTs - curTs));
139 TbMsg tickMsg = ctx.newMsg(ServiceQueue.MAIN, TB_MSG_GENERATOR_NODE_MSG, ctx.getSelfId(), new TbMsgMetaData(), ""); 138 TbMsg tickMsg = ctx.newMsg(ServiceQueue.MAIN, TB_MSG_GENERATOR_NODE_MSG, ctx.getSelfId(), new TbMsgMetaData(), "");
140 nextTickId = tickMsg.getId(); 139 nextTickId = tickMsg.getId();
141 - log.info("演员消息生产——高级消息内容：【{}】", tickMsg);  
142 ctx.tellSelf(tickMsg, curDelay); 140 ctx.tellSelf(tickMsg, curDelay);
143 } 141 }
144 142
@@ -74,7 +74,6 @@ public class TbMsgDelayNode implements TbNode { @@ -74,7 +74,6 @@ public class TbMsgDelayNode implements TbNode {
74 if (pendingMsgs.size() < config.getMaxPendingMsgs()) { 74 if (pendingMsgs.size() < config.getMaxPendingMsgs()) {
75 pendingMsgs.put(msg.getId(), msg); 75 pendingMsgs.put(msg.getId(), msg);
76 TbMsg tickMsg = ctx.newMsg(ServiceQueue.MAIN, TB_MSG_DELAY_NODE_MSG, ctx.getSelfId(), msg.getCustomerId(), new TbMsgMetaData(), msg.getId().toString()); 76 TbMsg tickMsg = ctx.newMsg(ServiceQueue.MAIN, TB_MSG_DELAY_NODE_MSG, ctx.getSelfId(), msg.getCustomerId(), new TbMsgMetaData(), msg.getId().toString());
77 - log.info("演员消息生产——高级消息内容：【{}】", msg);  
78 ctx.tellSelf(tickMsg, getDelay(msg)); 77 ctx.tellSelf(tickMsg, getDelay(msg));
79 ctx.ack(msg); 78 ctx.ack(msg);
80 } else { 79 } else {
@@ -150,7 +150,6 @@ public class TbMsgPushToEdgeNode implements TbNode { @@ -150,7 +150,6 @@ public class TbMsgPushToEdgeNode implements TbNode {
150 private void notifyEdge(TbContext ctx, TbMsg msg, EdgeEvent edgeEvent, EdgeId edgeId) { 150 private void notifyEdge(TbContext ctx, TbMsg msg, EdgeEvent edgeEvent, EdgeId edgeId) {
151 edgeEvent.setEdgeId(edgeId); 151 edgeEvent.setEdgeId(edgeId);
152 ctx.getEdgeEventService().save(edgeEvent); 152 ctx.getEdgeEventService().save(edgeEvent);
153 - log.info("演员消息生产——普通消息内容：【{}】",msg);  
154 ctx.tellNext(msg, SUCCESS); 153 ctx.tellNext(msg, SUCCESS);
155 ctx.onEdgeEventUpdate(ctx.getTenantId(), edgeId); 154 ctx.onEdgeEventUpdate(ctx.getTenantId(), edgeId);
156 } 155 }
@@ -56,7 +56,6 @@ public class TbCheckAlarmStatusNode implements TbNode { @@ -56,7 +56,6 @@ public class TbCheckAlarmStatusNode implements TbNode {
56 56
57 @Override 57 @Override
58 public void onMsg(TbContext ctx, TbMsg msg) throws TbNodeException { 58 public void onMsg(TbContext ctx, TbMsg msg) throws TbNodeException {
59 - log.info("节点【TbCheckAlarmStatusNode】处理消费: {}", msg);  
60 try { 59 try {
61 Alarm alarm = mapper.readValue(msg.getData(), Alarm.class); 60 Alarm alarm = mapper.readValue(msg.getData(), Alarm.class);
62 61
@@ -73,7 +72,6 @@ public class TbCheckAlarmStatusNode implements TbNode { @@ -73,7 +72,6 @@ public class TbCheckAlarmStatusNode implements TbNode {
73 break; 72 break;
74 } 73 }
75 } 74 }
76 - log.info("演员消息生产——普通消息内容：【{}】", msg);  
77 if (isPresent) { 75 if (isPresent) {
78 ctx.tellNext(msg, "True"); 76 ctx.tellNext(msg, "True");
79 } else { 77 } else {
@@ -58,7 +58,6 @@ public class TbCheckMessageNode implements TbNode { @@ -58,7 +58,6 @@ public class TbCheckMessageNode implements TbNode {
58 @Override 58 @Override
59 public void onMsg(TbContext ctx, TbMsg msg) { 59 public void onMsg(TbContext ctx, TbMsg msg) {
60 try { 60 try {
61 - log.info("演员消息生产——普通消息内容：【{}】", msg);  
62 if (config.isCheckAllKeys()) { 61 if (config.isCheckAllKeys()) {
63 ctx.tellNext(msg, allKeysData(msg) && allKeysMetadata(msg) ? "True" : "False"); 62 ctx.tellNext(msg, allKeysData(msg) && allKeysMetadata(msg) ? "True" : "False");
64 } else { 63 } else {
@@ -69,7 +69,7 @@ public class TbCheckRelationNode implements TbNode { @@ -69,7 +69,7 @@ public class TbCheckRelationNode implements TbNode {
69 } else { 69 } else {
70 checkRelationFuture = processList(ctx, msg); 70 checkRelationFuture = processList(ctx, msg);
71 } 71 }
72 - withCallback(checkRelationFuture, filterResult -> {log.info("演员消息生产——普通消息内容：【{}】", msg);ctx.tellNext(msg, filterResult ? "True" : "False");}, t -> ctx.tellFailure(msg, t), ctx.getDbCallbackExecutor()); 72 + withCallback(checkRelationFuture, filterResult -> {ctx.tellNext(msg, filterResult ? "True" : "False");}, t -> ctx.tellFailure(msg, t), ctx.getDbCallbackExecutor());
73 } 73 }
74 74
75 private ListenableFuture<Boolean> processSingle(TbContext ctx, TbMsg msg) { 75 private ListenableFuture<Boolean> processSingle(TbContext ctx, TbMsg msg) {
@@ -59,7 +59,6 @@ public class TbJsFilterNode implements TbNode { @@ -59,7 +59,6 @@ public class TbJsFilterNode implements TbNode {
59 withCallback(jsEngine.executeFilterAsync(msg), 59 withCallback(jsEngine.executeFilterAsync(msg),
60 filterResult -> { 60 filterResult -> {
61 ctx.logJsEvalResponse(); 61 ctx.logJsEvalResponse();
62 - log.info("演员消息生产——普通消息内容：【{}】", msg);  
63 ctx.tellNext(msg, filterResult ? "True" : "False"); 62 ctx.tellNext(msg, filterResult ? "True" : "False");
64 }, 63 },
65 t -> { 64 t -> {
@@ -77,7 +77,6 @@ public class TbJsSwitchNode implements TbNode { @@ -77,7 +77,6 @@ public class TbJsSwitchNode implements TbNode {
77 } 77 }
78 78
79 private void processSwitch(TbContext ctx, TbMsg msg, Set<String> nextRelations) { 79 private void processSwitch(TbContext ctx, TbMsg msg, Set<String> nextRelations) {
80 - log.info("演员消息生产——普通消息内容：processSwitch【{}】",msg);  
81 ctx.tellNext(msg, nextRelations); 80 ctx.tellNext(msg, nextRelations);
82 } 81 }
83 82
@@ -49,7 +49,6 @@ public class TbMsgTypeFilterNode implements TbNode { @@ -49,7 +49,6 @@ public class TbMsgTypeFilterNode implements TbNode {
49 49
50 @Override 50 @Override
51 public void onMsg(TbContext ctx, TbMsg msg) { 51 public void onMsg(TbContext ctx, TbMsg msg) {
52 - log.info("演员消息生产——普通消息内容：【{}】", msg);  
53 ctx.tellNext(msg, config.getMessageTypes().contains(msg.getType()) ? "True" : "False"); 52 ctx.tellNext(msg, config.getMessageTypes().contains(msg.getType()) ? "True" : "False");
54 } 53 }
55 54
@@ -114,7 +114,6 @@ public class TbMsgTypeSwitchNode implements TbNode { @@ -114,7 +114,6 @@ public class TbMsgTypeSwitchNode implements TbNode {
114 } else { 114 } else {
115 relationType = "Other"; 115 relationType = "Other";
116 } 116 }
117 - log.info("演员消息生产——普通消息内容：类型{}【{}】",msg.getType(), msg);  
118 ctx.tellNext(msg, relationType); 117 ctx.tellNext(msg, relationType);
119 } 118 }
120 119
@@ -48,7 +48,6 @@ public class TbOriginatorTypeFilterNode implements TbNode { @@ -48,7 +48,6 @@ public class TbOriginatorTypeFilterNode implements TbNode {
48 @Override 48 @Override
49 public void onMsg(TbContext ctx, TbMsg msg) { 49 public void onMsg(TbContext ctx, TbMsg msg) {
50 EntityType originatorType = msg.getOriginator().getEntityType(); 50 EntityType originatorType = msg.getOriginator().getEntityType();
51 - log.info("演员消息生产——普通消息内容：【{}】", msg);  
52 ctx.tellNext(msg, config.getOriginatorTypes().contains(originatorType) ? "True" : "False"); 51 ctx.tellNext(msg, config.getOriginatorTypes().contains(originatorType) ? "True" : "False");
53 } 52 }
54 53
@@ -87,7 +87,6 @@ public class TbOriginatorTypeSwitchNode implements TbNode { @@ -87,7 +87,6 @@ public class TbOriginatorTypeSwitchNode implements TbNode {
87 default: 87 default:
88 throw new TbNodeException("Unsupported originator type: " + originatorType); 88 throw new TbNodeException("Unsupported originator type: " + originatorType);
89 } 89 }
90 - log.info("演员消息生产——普通消息内容：【{}】", msg);  
91 ctx.tellNext(msg, relationType); 90 ctx.tellNext(msg, relationType);
92 } 91 }
93 92
@@ -48,7 +48,6 @@ public class TbAckNode implements TbNode { @@ -48,7 +48,6 @@ public class TbAckNode implements TbNode {
48 @Override 48 @Override
49 public void onMsg(TbContext ctx, TbMsg msg) { 49 public void onMsg(TbContext ctx, TbMsg msg) {
50 ctx.ack(msg); 50 ctx.ack(msg);
51 - log.info("演员消息生产——普通消息内容：【{}】", msg);  
52 ctx.tellSuccess(msg); 51 ctx.tellSuccess(msg);
53 } 52 }
54 53
@@ -101,13 +101,11 @@ public class TbPubSubNode implements TbNode { @@ -101,13 +101,11 @@ public class TbPubSubNode implements TbNode {
101 ApiFutures.addCallback(messageIdFuture, new ApiFutureCallback<String>() { 101 ApiFutures.addCallback(messageIdFuture, new ApiFutureCallback<String>() {
102 public void onSuccess(String messageId) { 102 public void onSuccess(String messageId) {
103 TbMsg next = processPublishResult(ctx, msg, messageId); 103 TbMsg next = processPublishResult(ctx, msg, messageId);
104 - log.info("演员消息生产——普通消息内容：【{}】", msg);  
105 ctx.tellSuccess(next); 104 ctx.tellSuccess(next);
106 } 105 }
107 106
108 public void onFailure(Throwable t) { 107 public void onFailure(Throwable t) {
109 TbMsg next = processException(ctx, msg, t); 108 TbMsg next = processException(ctx, msg, t);
110 - log.info("演员消息生产——普通消息内容：【{}】", next);  
111 ctx.tellFailure(next, t); 109 ctx.tellFailure(next, t);
112 } 110 }
113 }, 111 },
@@ -97,7 +97,6 @@ public class TbGpsGeofencingActionNode extends AbstractGeofencingNode<TbGpsGeofe @@ -97,7 +97,6 @@ public class TbGpsGeofencingActionNode extends AbstractGeofencingNode<TbGpsGeofe
97 } 97 }
98 } 98 }
99 if (!told) { 99 if (!told) {
100 - log.info("演员消息生产——普通消息内容：【{}】", msg);  
101 ctx.tellSuccess(msg); 100 ctx.tellSuccess(msg);
102 } 101 }
103 } 102 }
@@ -77,7 +77,7 @@ public class TbSendEmailNode implements TbNode { @@ -77,7 +77,7 @@ public class TbSendEmailNode implements TbNode {
77 sendEmail(ctx, msg, email); 77 sendEmail(ctx, msg, email);
78 return null; 78 return null;
79 }), 79 }),
80 - ok -> {log.info("演员消息生产——普通消息内容：【{}】", msg);ctx.tellSuccess(msg);}, 80 + ok -> {ctx.tellSuccess(msg);},
81 fail -> ctx.tellFailure(msg, fail)); 81 fail -> ctx.tellFailure(msg, fail));
82 } catch (Exception ex) { 82 } catch (Exception ex) {
83 ctx.tellFailure(msg, ex); 83 ctx.tellFailure(msg, ex);
@@ -91,7 +91,6 @@ public class CalculateDeltaNode implements TbNode { @@ -91,7 +91,6 @@ public class CalculateDeltaNode implements TbNode {
91 BigDecimal delta = BigDecimal.valueOf(previousData != null ? currentValue - previousData.value : 0.0); 91 BigDecimal delta = BigDecimal.valueOf(previousData != null ? currentValue - previousData.value : 0.0);
92 92
93 if (config.isTellFailureIfDeltaIsNegative() && delta.doubleValue() < 0) { 93 if (config.isTellFailureIfDeltaIsNegative() && delta.doubleValue() < 0) {
94 - log.info("演员消息生产——普通消息内容：【{}】", msg);  
95 ctx.tellNext(msg, TbRelationTypes.FAILURE); 94 ctx.tellNext(msg, TbRelationTypes.FAILURE);
96 return; 95 return;
97 } 96 }
@@ -112,16 +111,13 @@ public class CalculateDeltaNode implements TbNode { @@ -112,16 +111,13 @@ public class CalculateDeltaNode implements TbNode {
112 long period = previousData != null ? currentTs - previousData.ts : 0; 111 long period = previousData != null ? currentTs - previousData.ts : 0;
113 result.put(config.getPeriodValueKey(), period); 112 result.put(config.getPeriodValueKey(), period);
114 } 113 }
115 - log.info("演员消息生产——普通消息内容：【{}】", TbMsg.transformMsg(msg, msg.getType(), msg.getOriginator(), msg.getMetaData(), JacksonUtil.toString(result)));  
116 ctx.tellSuccess(TbMsg.transformMsg(msg, msg.getType(), msg.getOriginator(), msg.getMetaData(), JacksonUtil.toString(result))); 114 ctx.tellSuccess(TbMsg.transformMsg(msg, msg.getType(), msg.getOriginator(), msg.getMetaData(), JacksonUtil.toString(result)));
117 }, 115 },
118 t -> ctx.tellFailure(msg, t), ctx.getDbCallbackExecutor()); 116 t -> ctx.tellFailure(msg, t), ctx.getDbCallbackExecutor());
119 } else { 117 } else {
120 - log.info("演员消息生产——普通消息内容：【{}】", msg);  
121 ctx.tellNext(msg, "Other"); 118 ctx.tellNext(msg, "Other");
122 } 119 }
123 } else { 120 } else {
124 - log.info("演员消息生产——普通消息内容：【{}】", msg);  
125 ctx.tellNext(msg, "Other"); 121 ctx.tellNext(msg, "Other");
126 } 122 }
127 } 123 }
@@ -71,7 +71,6 @@ public abstract class TbAbstractGetAttributesNode<C extends TbGetAttributesNodeC @@ -71,7 +71,6 @@ public abstract class TbAbstractGetAttributesNode<C extends TbGetAttributesNodeC
71 71
72 @Override 72 @Override
73 public void onMsg(TbContext ctx, TbMsg msg) throws TbNodeException { 73 public void onMsg(TbContext ctx, TbMsg msg) throws TbNodeException {
74 - log.info("节点【TbAbstractGetAttributesNode】处理消费: {}", msg);  
75 try { 74 try {
76 withCallback( 75 withCallback(
77 findEntityIdAsync(ctx, msg), 76 findEntityIdAsync(ctx, msg),
@@ -90,7 +89,6 @@ public abstract class TbAbstractGetAttributesNode<C extends TbGetAttributesNodeC @@ -90,7 +89,6 @@ public abstract class TbAbstractGetAttributesNode<C extends TbGetAttributesNodeC
90 89
91 private void safePutAttributes(TbContext ctx, TbMsg msg, T entityId) { 90 private void safePutAttributes(TbContext ctx, TbMsg msg, T entityId) {
92 if (entityId == null || entityId.isNullUid()) { 91 if (entityId == null || entityId.isNullUid()) {
93 - log.info("演员消息生产——普通消息内容：【{}】", msg);  
94 ctx.tellNext(msg, FAILURE); 92 ctx.tellNext(msg, FAILURE);
95 return; 93 return;
96 } 94 }
@@ -105,7 +103,6 @@ public abstract class TbAbstractGetAttributesNode<C extends TbGetAttributesNodeC @@ -105,7 +103,6 @@ public abstract class TbAbstractGetAttributesNode<C extends TbGetAttributesNodeC
105 if (!failuresMap.isEmpty()) { 103 if (!failuresMap.isEmpty()) {
106 throw reportFailures(failuresMap); 104 throw reportFailures(failuresMap);
107 } 105 }
108 - log.info("演员消息生产——普通消息内容：【{}】", msg);  
109 ctx.tellSuccess(msg); 106 ctx.tellSuccess(msg);
110 }, t -> ctx.tellFailure(msg, t), ctx.getDbCallbackExecutor()); 107 }, t -> ctx.tellFailure(msg, t), ctx.getDbCallbackExecutor());
111 } 108 }
@@ -61,7 +61,6 @@ public abstract class TbEntityGetAttrNode<T extends EntityId> implements TbNode @@ -61,7 +61,6 @@ public abstract class TbEntityGetAttrNode<T extends EntityId> implements TbNode
61 61
62 private void safeGetAttributes(TbContext ctx, TbMsg msg, T entityId) { 62 private void safeGetAttributes(TbContext ctx, TbMsg msg, T entityId) {
63 if (entityId == null || entityId.isNullUid()) { 63 if (entityId == null || entityId.isNullUid()) {
64 - log.info("演员消息生产——普通消息内容：transformSuccess【{}】",msg);  
65 ctx.tellNext(msg, FAILURE); 64 ctx.tellNext(msg, FAILURE);
66 return; 65 return;
67 } 66 }
@@ -89,7 +88,6 @@ public abstract class TbEntityGetAttrNode<T extends EntityId> implements TbNode @@ -89,7 +88,6 @@ public abstract class TbEntityGetAttrNode<T extends EntityId> implements TbNode
89 String attrName = config.getAttrMapping().get(r.getKey()); 88 String attrName = config.getAttrMapping().get(r.getKey());
90 msg.getMetaData().putValue(attrName, r.getValueAsString()); 89 msg.getMetaData().putValue(attrName, r.getValueAsString());
91 }); 90 });
92 - log.info("演员消息生产——普通消息内容：putAttributesAndTell【{}】", msg);  
93 ctx.tellSuccess(msg); 91 ctx.tellSuccess(msg);
94 } 92 }
95 93
@@ -56,7 +56,7 @@ public class TbGetOriginatorFieldsNode implements TbNode { @@ -56,7 +56,7 @@ public class TbGetOriginatorFieldsNode implements TbNode {
56 public void onMsg(TbContext ctx, TbMsg msg) { 56 public void onMsg(TbContext ctx, TbMsg msg) {
57 try { 57 try {
58 withCallback(putEntityFields(ctx, msg.getOriginator(), msg), 58 withCallback(putEntityFields(ctx, msg.getOriginator(), msg),
59 - i ->{ log.info("演员消息生产——普通消息内容：【{}】", msg);ctx.tellSuccess(msg);}, t -> ctx.tellFailure(msg, t), ctx.getDbCallbackExecutor()); 59 + i ->{ctx.tellSuccess(msg);}, t -> ctx.tellFailure(msg, t), ctx.getDbCallbackExecutor());
60 } catch (Throwable th) { 60 } catch (Throwable th) {
61 ctx.tellFailure(msg, th); 61 ctx.tellFailure(msg, th);
62 } 62 }
@@ -106,7 +106,6 @@ public class TbGetTelemetryNode implements TbNode { @@ -106,7 +106,6 @@ public class TbGetTelemetryNode implements TbNode {
106 106
107 @Override 107 @Override
108 public void onMsg(TbContext ctx, TbMsg msg) throws ExecutionException, InterruptedException, TbNodeException { 108 public void onMsg(TbContext ctx, TbMsg msg) throws ExecutionException, InterruptedException, TbNodeException {
109 - log.info("节点【TbGetTelemetryNode】处理消费: {}", msg);  
110 if (tsKeyNames.isEmpty()) { 109 if (tsKeyNames.isEmpty()) {
111 ctx.tellFailure(msg, new IllegalStateException("Telemetry is not selected!")); 110 ctx.tellFailure(msg, new IllegalStateException("Telemetry is not selected!"));
112 } else { 111 } else {
@@ -118,7 +117,6 @@ public class TbGetTelemetryNode implements TbNode { @@ -118,7 +117,6 @@ public class TbGetTelemetryNode implements TbNode {
118 ListenableFuture<List<TsKvEntry>> list = ctx.getTimeseriesService().findAll(ctx.getTenantId(), msg.getOriginator(), buildQueries(msg, keys)); 117 ListenableFuture<List<TsKvEntry>> list = ctx.getTimeseriesService().findAll(ctx.getTenantId(), msg.getOriginator(), buildQueries(msg, keys));
119 DonAsynchron.withCallback(list, data -> { 118 DonAsynchron.withCallback(list, data -> {
120 process(data, msg, keys); 119 process(data, msg, keys);
121 - log.info("演员消息生产——普通消息内容：【{}】", ctx.transformMsg(msg, msg.getType(), msg.getOriginator(), msg.getMetaData(), msg.getData()));  
122 ctx.tellSuccess(ctx.transformMsg(msg, msg.getType(), msg.getOriginator(), msg.getMetaData(), msg.getData())); 120 ctx.tellSuccess(ctx.transformMsg(msg, msg.getType(), msg.getOriginator(), msg.getMetaData(), msg.getData()));
123 }, error -> ctx.tellFailure(msg, error), ctx.getDbCallbackExecutor()); 121 }, error -> ctx.tellFailure(msg, error), ctx.getDbCallbackExecutor());
124 } catch (Exception e) { 122 } catch (Exception e) {
@@ -75,12 +75,10 @@ public class TbMqttNode implements TbNode { @@ -75,12 +75,10 @@ public class TbMqttNode implements TbNode {
75 75
76 @Override 76 @Override
77 public void onMsg(TbContext ctx, TbMsg msg) { 77 public void onMsg(TbContext ctx, TbMsg msg) {
78 - log.info("节点【TbMqttNode】处理消费: {}", msg);  
79 String topic = TbNodeUtils.processPattern(this.mqttNodeConfiguration.getTopicPattern(), msg); 78 String topic = TbNodeUtils.processPattern(this.mqttNodeConfiguration.getTopicPattern(), msg);
80 this.mqttClient.publish(topic, Unpooled.wrappedBuffer(msg.getData().getBytes(UTF8)), MqttQoS.AT_LEAST_ONCE) 79 this.mqttClient.publish(topic, Unpooled.wrappedBuffer(msg.getData().getBytes(UTF8)), MqttQoS.AT_LEAST_ONCE)
81 .addListener(future -> { 80 .addListener(future -> {
82 if (future.isSuccess()) { 81 if (future.isSuccess()) {
83 - log.info("演员消息生产——普通消息内容：【{}】", msg);  
84 ctx.tellSuccess(msg); 82 ctx.tellSuccess(msg);
85 } else { 83 } else {
86 TbMsg next = processException(ctx, msg, future.cause()); 84 TbMsg next = processException(ctx, msg, future.cause());
@@ -67,7 +67,7 @@ public class AlarmNoticeNode implements TbNode { @@ -67,7 +67,7 @@ public class AlarmNoticeNode implements TbNode {
67 posetApi(ctx, msg); 67 posetApi(ctx, msg);
68 return null; 68 return null;
69 }), 69 }),
70 - ok -> {log.info("演员消息生产——普通消息内容：【{}】", msg);ctx.tellSuccess(msg);}, 70 + ok -> {ctx.tellSuccess(msg);},
71 fail -> ctx.tellFailure(msg, fail)); 71 fail -> ctx.tellFailure(msg, fail));
72 } catch (Exception ex) { 72 } catch (Exception ex) {
73 ctx.tellFailure(msg, ex); 73 ctx.tellFailure(msg, ex);
@@ -157,7 +157,6 @@ class DeviceState { @@ -157,7 +157,6 @@ class DeviceState {
157 if (msg.getType().equals(DataConstants.ENTITY_ASSIGNED) || msg.getType().equals(DataConstants.ENTITY_UNASSIGNED)) { 157 if (msg.getType().equals(DataConstants.ENTITY_ASSIGNED) || msg.getType().equals(DataConstants.ENTITY_UNASSIGNED)) {
158 dynamicPredicateValueCtx.resetCustomer(); 158 dynamicPredicateValueCtx.resetCustomer();
159 } 159 }
160 - log.info("演员消息生产——普通消息内容：{}【{}】",msg.getType(), msg);  
161 ctx.tellSuccess(msg); 160 ctx.tellSuccess(msg);
162 } 161 }
163 if (persistState && stateChanged) { 162 if (persistState && stateChanged) {
@@ -183,7 +182,6 @@ class DeviceState { @@ -183,7 +182,6 @@ class DeviceState {
183 a -> new AlarmState(this.deviceProfile, deviceId, alarm, getOrInitPersistedAlarmState(alarm), dynamicPredicateValueCtx)); 182 a -> new AlarmState(this.deviceProfile, deviceId, alarm, getOrInitPersistedAlarmState(alarm), dynamicPredicateValueCtx));
184 stateChanged |= alarmState.processAlarmClear(ctx, alarmNf); 183 stateChanged |= alarmState.processAlarmClear(ctx, alarmNf);
185 } 184 }
186 - log.info("演员消息生产——普通消息内容：processAlarmClearNotification【{}】", msg);  
187 ctx.tellSuccess(msg); 185 ctx.tellSuccess(msg);
188 return stateChanged; 186 return stateChanged;
189 } 187 }
@@ -195,14 +193,12 @@ class DeviceState { @@ -195,14 +193,12 @@ class DeviceState {
195 a -> new AlarmState(this.deviceProfile, deviceId, alarm, getOrInitPersistedAlarmState(alarm), dynamicPredicateValueCtx)); 193 a -> new AlarmState(this.deviceProfile, deviceId, alarm, getOrInitPersistedAlarmState(alarm), dynamicPredicateValueCtx));
196 alarmState.processAckAlarm(alarmNf); 194 alarmState.processAckAlarm(alarmNf);
197 } 195 }
198 - log.info("演员消息生产——普通消息内容：processAlarmAckNotification【{}】", msg);  
199 ctx.tellSuccess(msg); 196 ctx.tellSuccess(msg);
200 } 197 }
201 198
202 private void processAlarmDeleteNotification(TbContext ctx, TbMsg msg) { 199 private void processAlarmDeleteNotification(TbContext ctx, TbMsg msg) {
203 Alarm alarm = JacksonUtil.fromString(msg.getData(), Alarm.class); 200 Alarm alarm = JacksonUtil.fromString(msg.getData(), Alarm.class);
204 alarmStates.values().removeIf(alarmState -> alarmState.getCurrentAlarm().getId().equals(alarm.getId())); 201 alarmStates.values().removeIf(alarmState -> alarmState.getCurrentAlarm().getId().equals(alarm.getId()));
205 - log.info("演员消息生产——普通消息内容：processAlarmDeleteNotification【{}】", msg);  
206 ctx.tellSuccess(msg); 202 ctx.tellSuccess(msg);
207 } 203 }
208 204
@@ -231,7 +227,6 @@ class DeviceState { @@ -231,7 +227,6 @@ class DeviceState {
231 stateChanged |= alarmState.process(ctx, msg, latestValues, null); 227 stateChanged |= alarmState.process(ctx, msg, latestValues, null);
232 } 228 }
233 } 229 }
234 - log.info("演员消息生产——普通消息内容：processAttributesDeleteNotification【{}】", msg);  
235 ctx.tellSuccess(msg); 230 ctx.tellSuccess(msg);
236 return stateChanged; 231 return stateChanged;
237 } 232 }
@@ -251,7 +246,6 @@ class DeviceState { @@ -251,7 +246,6 @@ class DeviceState {
251 stateChanged |= alarmState.process(ctx, msg, latestValues, update); 246 stateChanged |= alarmState.process(ctx, msg, latestValues, update);
252 } 247 }
253 } 248 }
254 - log.info("演员消息生产——普通消息内容：processAttributes【{}】", msg);  
255 ctx.tellSuccess(msg); 249 ctx.tellSuccess(msg);
256 return stateChanged; 250 return stateChanged;
257 } 251 }
@@ -277,7 +271,6 @@ class DeviceState { @@ -277,7 +271,6 @@ class DeviceState {
277 } 271 }
278 } 272 }
279 } 273 }
280 - log.info("演员消息生产——普通消息内容：processTelemetry【{}】", msg);  
281 ctx.tellSuccess(msg); 274 ctx.tellSuccess(msg);
282 return stateChanged; 275 return stateChanged;
283 } 276 }
@@ -106,7 +106,6 @@ public class TbDeviceProfileNode implements TbNode { @@ -106,7 +106,6 @@ public class TbDeviceProfileNode implements TbNode {
106 106
107 @Override 107 @Override
108 public void onMsg(TbContext ctx, TbMsg msg) throws ExecutionException, InterruptedException { 108 public void onMsg(TbContext ctx, TbMsg msg) throws ExecutionException, InterruptedException {
109 - log.info("节点【TbDeviceProfileNode】处理消费: {}", msg);  
110 EntityType originatorType = msg.getOriginator().getEntityType(); 109 EntityType originatorType = msg.getOriginator().getEntityType();
111 if (msg.getType().equals(PERIODIC_MSG_TYPE)) { 110 if (msg.getType().equals(PERIODIC_MSG_TYPE)) {
112 scheduleAlarmHarvesting(ctx, msg); 111 scheduleAlarmHarvesting(ctx, msg);
@@ -126,11 +125,9 @@ public class TbDeviceProfileNode implements TbNode { @@ -126,11 +125,9 @@ public class TbDeviceProfileNode implements TbNode {
126 DeviceId deviceId = new DeviceId(msg.getOriginator().getId()); 125 DeviceId deviceId = new DeviceId(msg.getOriginator().getId());
127 if (msg.getType().equals(DataConstants.ENTITY_UPDATED)) { 126 if (msg.getType().equals(DataConstants.ENTITY_UPDATED)) {
128 invalidateDeviceProfileCache(deviceId, msg.getData()); 127 invalidateDeviceProfileCache(deviceId, msg.getData());
129 - log.info("演员消息生产——普通消息内容：ENTITY_UPDATED【{}】", msg);  
130 ctx.tellSuccess(msg); 128 ctx.tellSuccess(msg);
131 } else if (msg.getType().equals(DataConstants.ENTITY_DELETED)) { 129 } else if (msg.getType().equals(DataConstants.ENTITY_DELETED)) {
132 removeDeviceState(deviceId); 130 removeDeviceState(deviceId);
133 - log.info("演员消息生产——普通消息内容：ENTITY_DELETED【{}】", msg);  
134 ctx.tellSuccess(msg); 131 ctx.tellSuccess(msg);
135 } else { 132 } else {
136 DeviceState deviceState = getOrCreateDeviceState(ctx, deviceId, null); 133 DeviceState deviceState = getOrCreateDeviceState(ctx, deviceId, null);
@@ -142,7 +139,6 @@ public class TbDeviceProfileNode implements TbNode { @@ -142,7 +139,6 @@ public class TbDeviceProfileNode implements TbNode {
142 } 139 }
143 } 140 }
144 } else { 141 } else {
145 - log.info("演员消息生产——普通消息内容：{}【{}】", msg.getType(),msg);  
146 ctx.tellSuccess(msg); 142 ctx.tellSuccess(msg);
147 } 143 }
148 } 144 }
@@ -174,7 +170,6 @@ public class TbDeviceProfileNode implements TbNode { @@ -174,7 +170,6 @@ public class TbDeviceProfileNode implements TbNode {
174 170
175 protected void scheduleAlarmHarvesting(TbContext ctx, TbMsg msg) { 171 protected void scheduleAlarmHarvesting(TbContext ctx, TbMsg msg) {
176 TbMsg periodicCheck = TbMsg.newMsg(PERIODIC_MSG_TYPE, ctx.getTenantId(), msg != null ? msg.getCustomerId() : null, TbMsgMetaData.EMPTY, "{}"); 172 TbMsg periodicCheck = TbMsg.newMsg(PERIODIC_MSG_TYPE, ctx.getTenantId(), msg != null ? msg.getCustomerId() : null, TbMsgMetaData.EMPTY, "{}");
177 - log.info("演员消息生产——高级消息内容：scheduleAlarmHarvesting【{}】", periodicCheck);  
178 ctx.tellSelf(periodicCheck, TimeUnit.MINUTES.toMillis(1)); 173 ctx.tellSelf(periodicCheck, TimeUnit.MINUTES.toMillis(1));
179 } 174 }
180 175
@@ -199,7 +194,6 @@ public class TbDeviceProfileNode implements TbNode { @@ -199,7 +194,6 @@ public class TbDeviceProfileNode implements TbNode {
199 } 194 }
200 195
201 protected void onProfileUpdate(DeviceProfile profile) { 196 protected void onProfileUpdate(DeviceProfile profile) {
202 - log.info("演员消息生产——高级消息内容：onProfileUpdate【{}】", TbMsg.newMsg(PROFILE_UPDATE_MSG_TYPE, ctx.getTenantId(), TbMsgMetaData.EMPTY, profile.getId().getId().toString()));  
203 ctx.tellSelf(TbMsg.newMsg(PROFILE_UPDATE_MSG_TYPE, ctx.getTenantId(), TbMsgMetaData.EMPTY, profile.getId().getId().toString()), 0L); 197 ctx.tellSelf(TbMsg.newMsg(PROFILE_UPDATE_MSG_TYPE, ctx.getTenantId(), TbMsgMetaData.EMPTY, profile.getId().getId().toString()), 0L);
204 } 198 }
205 199
@@ -209,7 +203,6 @@ public class TbDeviceProfileNode implements TbNode { @@ -209,7 +203,6 @@ public class TbDeviceProfileNode implements TbNode {
209 if (deviceProfile != null) { 203 if (deviceProfile != null) {
210 msgData.put("deviceProfileId", deviceProfile.getId().getId().toString()); 204 msgData.put("deviceProfileId", deviceProfile.getId().getId().toString());
211 } 205 }
212 - log.info("演员消息生产——高级消息内容：onDeviceUpdate【{}】", TbMsg.newMsg(DEVICE_UPDATE_MSG_TYPE, ctx.getTenantId(), TbMsgMetaData.EMPTY, JacksonUtil.toString(msgData)));  
213 ctx.tellSelf(TbMsg.newMsg(DEVICE_UPDATE_MSG_TYPE, ctx.getTenantId(), TbMsgMetaData.EMPTY, JacksonUtil.toString(msgData)), 0L); 206 ctx.tellSelf(TbMsg.newMsg(DEVICE_UPDATE_MSG_TYPE, ctx.getTenantId(), TbMsgMetaData.EMPTY, JacksonUtil.toString(msgData)), 0L);
214 } 207 }
215 208
@@ -202,11 +202,9 @@ public class TbHttpClient { @@ -202,11 +202,9 @@ public class TbHttpClient {
202 public void onSuccess(ResponseEntity<String> responseEntity) { 202 public void onSuccess(ResponseEntity<String> responseEntity) {
203 if (responseEntity.getStatusCode().is2xxSuccessful()) { 203 if (responseEntity.getStatusCode().is2xxSuccessful()) {
204 TbMsg next = processResponse(ctx, msg, responseEntity); 204 TbMsg next = processResponse(ctx, msg, responseEntity);
205 - log.info("演员消息生产——普通消息内容：transformSuccess【{}】",next);  
206 ctx.tellSuccess(next); 205 ctx.tellSuccess(next);
207 } else { 206 } else {
208 TbMsg next = processFailureResponse(ctx, msg, responseEntity); 207 TbMsg next = processFailureResponse(ctx, msg, responseEntity);
209 - log.info("演员消息生产——普通消息内容：transformSuccess【{}】",next);  
210 ctx.tellNext(next, TbRelationTypes.FAILURE); 208 ctx.tellNext(next, TbRelationTypes.FAILURE);
211 } 209 }
212 } 210 }
@@ -66,8 +66,6 @@ public class TbSendRPCReplyNode implements TbNode { @@ -66,8 +66,6 @@ public class TbSendRPCReplyNode implements TbNode {
66 ctx.tellFailure(msg, new RuntimeException("Request body is empty!")); 66 ctx.tellFailure(msg, new RuntimeException("Request body is empty!"));
67 } else { 67 } else {
68 ctx.getRpcService().sendRpcReplyToDevice(serviceIdStr, UUID.fromString(sessionIdStr), Integer.parseInt(requestIdStr), msg.getData()); 68 ctx.getRpcService().sendRpcReplyToDevice(serviceIdStr, UUID.fromString(sessionIdStr), Integer.parseInt(requestIdStr), msg.getData());
69 -  
70 - log.info("演员消息生产——普通消息内容：【{}】", msg);  
71 ctx.tellSuccess(msg); 69 ctx.tellSuccess(msg);
72 } 70 }
73 } 71 }
@@ -58,7 +58,6 @@ public class TbMsgAttributesNode implements TbNode { @@ -58,7 +58,6 @@ public class TbMsgAttributesNode implements TbNode {
58 58
59 @Override 59 @Override
60 public void onMsg(TbContext ctx, TbMsg msg) { 60 public void onMsg(TbContext ctx, TbMsg msg) {
61 - log.info("节点【TbMsgAttributesNode】处理消费: {}", msg);  
62 if (!msg.getType().equals(SessionMsgType.POST_ATTRIBUTES_REQUEST.name())) { 61 if (!msg.getType().equals(SessionMsgType.POST_ATTRIBUTES_REQUEST.name())) {
63 ctx.tellFailure(msg, new IllegalArgumentException("Unsupported msg type: " + msg.getType())); 62 ctx.tellFailure(msg, new IllegalArgumentException("Unsupported msg type: " + msg.getType()));
64 return; 63 return;
@@ -72,7 +72,6 @@ public class TbMsgTimeseriesNode implements TbNode { @@ -72,7 +72,6 @@ public class TbMsgTimeseriesNode implements TbNode {
72 72
73 @Override 73 @Override
74 public void onMsg(TbContext ctx, TbMsg msg) { 74 public void onMsg(TbContext ctx, TbMsg msg) {
75 - log.info("节点【TbMsgTimeseriesNode】处理消费: {}", msg);  
76 if (!msg.getType().equals(SessionMsgType.POST_TELEMETRY_REQUEST.name())) { 75 if (!msg.getType().equals(SessionMsgType.POST_TELEMETRY_REQUEST.name())) {
77 ctx.tellFailure(msg, new IllegalArgumentException("Unsupported msg type: " + msg.getType())); 76 ctx.tellFailure(msg, new IllegalArgumentException("Unsupported msg type: " + msg.getType()));
78 return; 77 return;
@@ -34,7 +34,6 @@ class TelemetryNodeCallback implements FutureCallback<Void> { @@ -34,7 +34,6 @@ class TelemetryNodeCallback implements FutureCallback<Void> {
34 34
35 @Override 35 @Override
36 public void onSuccess(@Nullable Void result) { 36 public void onSuccess(@Nullable Void result) {
37 - log.info("演员消息生产——普通消息内容：【{}】", msg);  
38 ctx.tellSuccess(msg); 37 ctx.tellSuccess(msg);
39 } 38 }
40 39
@@ -46,7 +46,6 @@ public class TbSynchronizationBeginNode implements TbNode { @@ -46,7 +46,6 @@ public class TbSynchronizationBeginNode implements TbNode {
46 @Override 46 @Override
47 public void onMsg(TbContext ctx, TbMsg msg) { 47 public void onMsg(TbContext ctx, TbMsg msg) {
48 log.warn("Synchronization Start/End nodes are deprecated since TB 2.5. Use queue with submit strategy SEQUENTIAL_BY_ORIGINATOR instead."); 48 log.warn("Synchronization Start/End nodes are deprecated since TB 2.5. Use queue with submit strategy SEQUENTIAL_BY_ORIGINATOR instead.");
49 - log.info("演员消息生产——普通消息内容：【{}】", msg);  
50 ctx.tellSuccess(msg); 49 ctx.tellSuccess(msg);
51 } 50 }
52 51
@@ -45,8 +45,6 @@ public class TbSynchronizationEndNode implements TbNode { @@ -45,8 +45,6 @@ public class TbSynchronizationEndNode implements TbNode {
45 @Override 45 @Override
46 public void onMsg(TbContext ctx, TbMsg msg) { 46 public void onMsg(TbContext ctx, TbMsg msg) {
47 log.warn("Synchronization Start/End nodes are deprecated since TB 2.5. Use queue with submit strategy SEQUENTIAL_BY_ORIGINATOR instead."); 47 log.warn("Synchronization Start/End nodes are deprecated since TB 2.5. Use queue with submit strategy SEQUENTIAL_BY_ORIGINATOR instead.");
48 -  
49 - log.info("演员消息生产——普通消息内容：【{}】", msg);  
50 ctx.tellSuccess(msg); 48 ctx.tellSuccess(msg);
51 } 49 }
52 50
@@ -46,7 +46,6 @@ public abstract class TbAbstractTransformNode implements TbNode { @@ -46,7 +46,6 @@ public abstract class TbAbstractTransformNode implements TbNode {
46 46
47 @Override 47 @Override
48 public void onMsg(TbContext ctx, TbMsg msg) { 48 public void onMsg(TbContext ctx, TbMsg msg) {
49 - log.info("节点【TbAbstractTransformNode】处理消费: {}", msg);  
50 withCallback(transform(ctx, msg), 49 withCallback(transform(ctx, msg),
51 m -> transformSuccess(ctx, msg, m), 50 m -> transformSuccess(ctx, msg, m),
52 t -> transformFailure(ctx, msg, t), 51 t -> transformFailure(ctx, msg, t),
@@ -58,7 +57,6 @@ public abstract class TbAbstractTransformNode implements TbNode { @@ -58,7 +57,6 @@ public abstract class TbAbstractTransformNode implements TbNode {
58 } 57 }
59 58
60 protected void transformSuccess(TbContext ctx, TbMsg msg, TbMsg m) { 59 protected void transformSuccess(TbContext ctx, TbMsg msg, TbMsg m) {
61 - log.info("演员消息生产——普通消息内容：transformSuccess【{}】",msg);  
62 if (m != null) { 60 if (m != null) {
63 ctx.tellSuccess(m); 61 ctx.tellSuccess(m);
64 } else { 62 } else {
@@ -69,7 +67,6 @@ public abstract class TbAbstractTransformNode implements TbNode { @@ -69,7 +67,6 @@ public abstract class TbAbstractTransformNode implements TbNode {
69 protected void transformSuccess(TbContext ctx, TbMsg msg, List<TbMsg> msgs) { 67 protected void transformSuccess(TbContext ctx, TbMsg msg, List<TbMsg> msgs) {
70 if (msgs != null && !msgs.isEmpty()) { 68 if (msgs != null && !msgs.isEmpty()) {
71 if (msgs.size() == 1) { 69 if (msgs.size() == 1) {
72 - log.info("演员消息生产——普通消息内容：transformSuccess【{}】",msgs.get(0));  
73 ctx.tellSuccess(msgs.get(0)); 70 ctx.tellSuccess(msgs.get(0));
74 } else { 71 } else {
75 TbMsgCallbackWrapper wrapper = new MultipleTbMsgsCallbackWrapper(msgs.size(), new TbMsgCallback() { 72 TbMsgCallbackWrapper wrapper = new MultipleTbMsgsCallbackWrapper(msgs.size(), new TbMsgCallback() {
@@ -86,7 +83,6 @@ public abstract class TbAbstractTransformNode implements TbNode { @@ -86,7 +83,6 @@ public abstract class TbAbstractTransformNode implements TbNode {
86 msgs.forEach(newMsg -> ctx.enqueueForTellNext(newMsg, "Success", wrapper::onSuccess, wrapper::onFailure)); 83 msgs.forEach(newMsg -> ctx.enqueueForTellNext(newMsg, "Success", wrapper::onSuccess, wrapper::onFailure));
87 } 84 }
88 } else { 85 } else {
89 - log.info("演员消息生产——普通消息内容：transformSuccess【{}】",msg);  
90 ctx.tellNext(msg, FAILURE); 86 ctx.tellNext(msg, FAILURE);
91 } 87 }
92 } 88 }
@@ -169,7 +169,6 @@ public class TbGetCustomerAttributeNodeTest { @@ -169,7 +169,6 @@ public class TbGetCustomerAttributeNodeTest {
169 169
170 170
171 node.onMsg(ctx, msg); 171 node.onMsg(ctx, msg);
172 - log.info("演员消息生产——普通消息内容：tellFailure【{}】",msg);  
173 verify(ctx).tellNext(msg, FAILURE); 172 verify(ctx).tellNext(msg, FAILURE);
174 assertTrue(msg.getMetaData().getData().isEmpty()); 173 assertTrue(msg.getMetaData().getData().isEmpty());
175 } 174 }
@@ -257,7 +256,6 @@ public class TbGetCustomerAttributeNodeTest { @@ -257,7 +256,6 @@ public class TbGetCustomerAttributeNodeTest {
257 .thenReturn(Futures.immediateFuture(timeseries)); 256 .thenReturn(Futures.immediateFuture(timeseries));
258 257
259 node.onMsg(ctx, msg); 258 node.onMsg(ctx, msg);
260 - log.info("演员消息生产——普通消息内容：deviceCustomerTelemetryFetched【{}】", msg);  
261 verify(ctx).tellSuccess(msg); 259 verify(ctx).tellSuccess(msg);
262 assertEquals(msg.getMetaData().getValue("tempo"), "highest"); 260 assertEquals(msg.getMetaData().getValue("tempo"), "highest");
263 } 261 }
@@ -270,7 +268,6 @@ public class TbGetCustomerAttributeNodeTest { @@ -270,7 +268,6 @@ public class TbGetCustomerAttributeNodeTest {
270 .thenReturn(Futures.immediateFuture(attributes)); 268 .thenReturn(Futures.immediateFuture(attributes));
271 269
272 node.onMsg(ctx, msg); 270 node.onMsg(ctx, msg);
273 - log.info("演员消息生产——普通消息内容：entityAttributeFetched【{}】", msg);  
274 verify(ctx).tellSuccess(msg); 271 verify(ctx).tellSuccess(msg);
275 assertEquals(msg.getMetaData().getValue("tempo"), "high"); 272 assertEquals(msg.getMetaData().getValue("tempo"), "high");
276 } 273 }