Browse Source

Improve remote transport service to process rule engine callbacks asynchronously

pull/2247/head
Igor Kulikov 7 years ago
parent
commit
1b5a8ac6fc
  1. 33
      common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/RemoteTransportService.java

33
common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/RemoteTransportService.java

@ -320,28 +320,19 @@ public class RemoteTransportService extends AbstractTransportService {
send(sessionInfo, toRuleEngineMsg, callback); send(sessionInfo, toRuleEngineMsg, callback);
} }
private static class TransportCallbackAdaptor implements Callback { private void send(SessionInfoProto sessionInfo, ToRuleEngineMsg toRuleEngineMsg, TransportServiceCallback<Void> callback) {
private final TransportServiceCallback<Void> callback; ruleEngineProducer.send(getRoutingKey(sessionInfo), toRuleEngineMsg, (metadata, exception) -> {
if (callback != null) {
TransportCallbackAdaptor(TransportServiceCallback<Void> callback) { if (exception == null) {
this.callback = callback; this.transportCallbackExecutor.submit(() -> {
} callback.onSuccess(null);
});
@Override } else {
public void onCompletion(RecordMetadata metadata, Exception exception) { this.transportCallbackExecutor.submit(() -> {
if (exception == null) { callback.onError(exception);
if (callback != null) { });
callback.onSuccess(null);
}
} else {
if (callback != null) {
callback.onError(exception);
} }
} }
} });
}
private void send(SessionInfoProto sessionInfo, ToRuleEngineMsg toRuleEngineMsg, TransportServiceCallback<Void> callback) {
ruleEngineProducer.send(getRoutingKey(sessionInfo), toRuleEngineMsg, new TransportCallbackAdaptor(callback));
} }
} }

Loading…
Cancel
Save