How to handle [MLLP] acknowledgments using [apache-camel]

Question

I'm attempting to send a HL7 message between two MLLP apache-camel components:

@Component
public class Producer extends RouteBuilder {

    @Override
    public void configure() {
        from("direct:hl7-recipient")
                .marshal(new HL7DataFormat())
                .to(mllp("localhost:4444"))
                .end();
    }
}

@Component
public class Consumer extends RouteBuilder {

    @Override
    public void configure() {
        from(mllp("localhost:4444"))
                .log("received")
                .unmarshal(new HL7DataFormat())
                .log("unmarshalled")
                .setProperty(MLLP_ACKNOWLEDGEMENT_TYPE, ack(AcknowledgmentCode.AA))
                .log("responding")
                .end();
    }
}

I can see the received, unmarshalled and responding log messages.

The exception that is occurring:

java.net.SocketTimeoutException: Read timed out

With this extract from the stacktrace:

Failed delivery for (MessageId: 0D423F22CDC26D9-0000000000000000 on ExchangeId: 0D423F22CDC26D9-0000000000000000). Exhausted after delivery attempt: 1 caught: org.apache.camel.component.mllp.MllpAcknowledgementTimeoutException: process(0D423F22CDC26D9-0000000000000000) [Socket[addr=localhost/127.0.0.1,port=4444,localport=64357]] - timeout receiving MLLP Acknowledgment

It appears that the acknowledgment is not being returned.

I have been following the apache-camel official documentation by setting the acknowledgment property on the exchange.

Thank you for any support.

Answer 1

In the end, we managed to implement a separate integration server that used the following class to send the test responses:

@Component
public class MllpRoute extends EndpointRouteBuilder {

    @Override
    public void configure() {
        onException(HL7Exception.class)
                .log("Handling Exception")
                .transform(ack(AcknowledgmentCode.AE,"Error transforming the message",
                        ErrorCode.UNSUPPORTED_MESSAGE_TYPE))
                .handled(true)
                .end();

        // Successful ack
        from("mllp://4441")
                .log("Message Received...")
                .unmarshal().hl7()
                .log("Done unmarshalling...")
                .validate(messageConforms())
                .log("Done validation...")
                .setProperty(MLLP_ACKNOWLEDGEMENT_STRING, ack(AcknowledgmentCode.valueOf("AA")))
                .end();

        // Internal Error ack
        from("mllp://4442")
                .log("Message Received...")
                .unmarshal().hl7()
                .log("Done unmarshalling...")
                .validate(messageConforms())
                .log("Done validation...")
                .setProperty(MLLP_ACKNOWLEDGEMENT_STRING, ack(AcknowledgmentCode.valueOf("AE")))
                .end();

        // Unknown ack
        from("mllp:4443")
                .log("Message Received...")
                .unmarshal().hl7()
                .log("Done unmarshalling...")
                .validate(messageConforms())
                .log("Done validation...")
                .setProperty(MLLP_ACKNOWLEDGEMENT_STRING, () -> "UNKNOWN")
                .end();
    }
}

For additional context around messageConforms() see page: https://camel.apache.org/components/3.20.x/languages/hl7terser-language.html