Debulker Client Processing

client-processing.connector {
  kafka {
    consumer {
      topic = CLIENT_PROCESSING_REQUEST
    }
  }
}

public class Pain001Processor extends ComponentProcessor<Document, Pain001Context> { (1)

    private final Pain001PaymentInstructionProcessor nextProcessor;

    public Pain001Processor(ComponentStore<NoCustomData> componentStore, Pain001PaymentInstructionProcessor nextProcessor) {
        super(componentStore, Pain001Processor::map, true); (2)
        this.nextProcessor = nextProcessor;
    }

    // Used to unmarshall the XML to an iso20022 model object
    private static Document map(String xmlString) {
        String PAIN_001_NAMESPACE = "urn:iso:std:iso:20022:tech:xsd:pain.001.001.09";
        XMLMapper xmlMapper = new XMLMapper();
        xmlMapper.registerDocumentClass(PAIN_001_NAMESPACE, Document.class);
        return xmlMapper.fromXML(xmlString, Document.class);
    }

    @Override
    protected Class<Document> getTargetClass() {
        return Document.class;
    }

    @Override
    protected CompletionStage<Done> handle(Pain001Context result) { (3)
        Document unmarshalledComponent = xmlMapper.apply(result.rawContent()); (4)
        Pain001PaymentInstructionContext pain001PaymentInstructionContext = Pain001PaymentInstructionContext.builder()
                .parentId(result.componentId())
                .processingContext(result.processingContext().withUnitOfWorkId(UnitOfWorkId.createRandom()))
                .parentMarker(result.marker())
                .build();
        return nextProcessor.process(result.processingContext(), pain001PaymentInstructionContext) (5)
                .thenApply(__ -> new Done("", "", "", Done.Result.EXECUTED)); (6)
    }

    @Override
    protected Function<Pain001Context, ComponentId> getParentComponentId() { (7)
        return null;
    }

    @Override
    public Function<Pain001Context, String> getMarker() { (8)
        return Pain001Context::marker;
    }

    @Override
    protected Function<Component<NoCustomData>, Pain001Context> transformComponent(Pain001Context additionalData) { (9)
        return component -> additionalData.toBuilder()
                .componentId(component.getId())
                .marker(component.getMarker())
                .rawContent(component.getContent())
                .build();
    }
}

public class Pain001PaymentInstructionProcessor extends ComponentProcessor<PaymentInstruction30, Pain001PaymentInstructionContext> { (1)

    private final Pain001TransactionProcessor nextProcessor;

    public Pain001PaymentInstructionProcessor(ComponentStore<NoCustomData> componentStore, Pain001TransactionProcessor nextProcessor) {
        super(componentStore, Pain001PaymentInstructionProcessor::map, false); (2)
        this.nextProcessor = nextProcessor;
    }

    // Used to unmarshall the XML to an iso20022 model object
    private static PaymentInstruction30 map(String xmlString) {
        String PAIN_001_NAMESPACE = "urn:iso:std:iso:20022:tech:xsd:pain.001.001.09";
        XMLMapper xmlMapper = new XMLMapper();
        xmlMapper.registerDocumentClass(PAIN_001_NAMESPACE, PaymentInstruction30.class);
        return xmlMapper.fromXML(xmlString, PaymentInstruction30.class);
    }

    @Override
    protected Class<PaymentInstruction30> getTargetClass() {
        return PaymentInstruction30.class;
    }

    @Override
    protected CompletionStage<Done> handle(Pain001PaymentInstructionContext result) {
        PaymentInstruction30 unmarshalledComponent = xmlMapper.apply(result.rawContent());
        Pain001TransactionContext pain001TransactionContext = Pain001TransactionContext.builder()
                .parentId(result.componentId())
                .parentMarker(result.marker()) (3)
                .processingContext(result.processingContext().withUnitOfWorkId(UnitOfWorkId.createRandom()))
                .build();
        // We are calling the next level processor directly, but you could do this via an MPS flow also
        return nextProcessor.process(result.processingContext(), pain001TransactionContext)
                .thenApply(__ -> new Done("", "", "", Done.Result.EXECUTED));

    }

    @Override
    protected Function<Pain001PaymentInstructionContext, ComponentId> getParentComponentId() { (4)
        return Pain001PaymentInstructionContext::parentId;
    }

    @Override
    public Function<Pain001PaymentInstructionContext, String> getMarker() {
        return pain001TxContext -> pain001TxContext.parentMarker() + ".CstmrCdtTrfInitn.PmtInf"; (5)
    }

    @Override
    protected Function<Component<NoCustomData>, Pain001PaymentInstructionContext> transformComponent(Pain001PaymentInstructionContext additionalData) {
        return component -> additionalData.toBuilder()
                .componentId(component.getId())
                .marker(component.getMarker())
                .rawContent(component.getContent())
                .build();
    }
}

public class Pain001TransactionProcessor extends ComponentProcessor<CreditTransferTransaction39, Pain001TransactionContext> { (1)

    public Pain001TransactionProcessor(ComponentStore<NoCustomData> componentStore) {
        super(componentStore, Pain001TransactionProcessor::map, false); (2)
    }

    // Used to unmarshall the XML to an iso20022 model object
    private static CreditTransferTransaction39 map(String xmlString) {
        String PAIN_001_NAMESPACE = "urn:iso:std:iso:20022:tech:xsd:pain.001.001.09";
        XMLMapper xmlMapper = new XMLMapper();
        xmlMapper.registerDocumentClass(PAIN_001_NAMESPACE, CreditTransferTransaction39.class);
        return xmlMapper.fromXML(xmlString, CreditTransferTransaction39.class);
    }

    @Override
    protected Class<CreditTransferTransaction39> getTargetClass() {
        return CreditTransferTransaction39.class;
    }

    @Override
    protected CompletionStage<Done> handle(Pain001TransactionContext result) { (3)
        CreditTransferTransaction39 cdtTrfTxInf = xmlMapper.apply(result.rawContent());
        String unitOfWorkId = UUID.randomUUID().toString(); // UOW Id to uniquely identify a request.
        String clientRequestId = cdtTrfTxInf.getPmtId().getEndToEndId(); // Payment related Id

        log.info("cdtTrfTxInf is instance of CreditTransferTransaction - initiate flow to process, UOW Id {}", unitOfWorkId);

        return DebulkerModelDomain.initiation().handle(new InitiateProcessDebulkedComponentsInput.Builder() (4)
                .withProcessingContext(ProcessingContext.builder()
                        .unitOfWorkId(unitOfWorkId)
                        .clientRequestId(clientRequestId)
                        .build())
                .withIsoCreditTransferComponent(cdtTrfTxInf)
                .withPaymentJourneyType("PAYMENT")
                .build()
        ).thenApply(report -> {
            log.debug("{}: Transaction Validation outcome is {}", cdtTrfTxInf.getPmtId().getEndToEndId(), report.getResult());
            return report;
        });

    }

    @Override
    protected Function<Pain001TransactionContext, ComponentId> getParentComponentId() {
        return Pain001TransactionContext::parentId;
    }

    @Override
    public Function<Pain001TransactionContext, String> getMarker() {
        return pain001TransactionContext -> pain001TransactionContext.parentMarker() + ".CdtTrfTxInf";
    }

    @Override
    protected Function<Component<NoCustomData>, Pain001TransactionContext> transformComponent(Pain001TransactionContext additionalData) {
        return component -> additionalData.toBuilder()
                .componentId(component.getId())
                .marker(component.getMarker())
                .rawContent(component.getContent())
                .build();
    }
}