ExternalCommunicatorServiceImpl.java 文件源码

@Async
@Override
public void handleExternalOutputMessage(final ExternalCommunicatorMessage message) {
  LOG.debug("Received request for external out message [{}]", message);

  final Long messageFlowId = getMessageFlowId(message.getTransferId());
  final Long configId = getActiveConfig(messageFlowId);

  final Map<String, InternalObject> businessObjects = new HashMap<>();
  message.getBusinessObjects().stream().forEachOrdered(bo -> {
    final Map<String, InternalField> fields = new HashMap<>();
    bo.getFields().stream().forEachOrdered(field -> {
      fields.put(field.getName(), new InternalField(field.getName(),
          DataType.valueOf(field.getType()), field.getValue()));
    });
    businessObjects.put(bo.getName(), new InternalObject(bo.getName(), fields));
  });

  composeSupervisorActor.tell(new ComposeMessageCreateCommand(message.getTransferId(),
      new InternalData(businessObjects), configId), ActorRef.noSender());
}