This is an automated email from the ASF dual-hosted git repository. davsclaus pushed a commit to branch CAMEL-26184-quickfix-inout in repository https://gitbox.apache.org/repos/asf/camel.git
commit 74d2fa60e2acf3ecf751f232e5c918bca1b3ef88 Author: Claus Ibsen <[email protected]> AuthorDate: Mon Nov 2 23:54:51 2026 +0100 CAMEL-26184: camel-quickfix - an InOut consumer replies only to application messages the route answers, and an InOut producer without CorrelationCriteria fails before sending An InOut consumer answered every event with the body the route ended with, so unanswered messages were echoed back and the AdminMessageSent event of the Logon sent the Logon again until a StackOverflowError. It now replies only to an AppMessageReceived event, when the route set a body other than the received message (and not null), and the exchange did not fail. A reply that is not a FIX message goes to the exception handler instead of failing inside fromApp, where the engine asked for the request again and again. An InOut producer without the CorrelationCriteria property registered a rule with a null predicate, which failed every engine event and timed out. It now fails with an IllegalArgumentException before the message is sent. A producer body that is not a FIX message fails with an InvalidPayloadException instead of a NullPointerException. Co-Authored-By: Claude Opus 5.5 <[email protected]> Signed-off-by: Claus Ibsen <[email protected]> --- .../apache/camel/catalog/components/quickfix.json | 4 +- .../camel/catalog/docs/quickfix-component.adoc | 9 +- .../apache/camel/component/quickfixj/quickfix.json | 4 +- .../src/main/docs/quickfix-component.adoc | 9 +- .../component/quickfixj/MessageCorrelator.java | 7 + .../component/quickfixj/QuickfixjConsumer.java | 31 ++- .../component/quickfixj/QuickfixjEndpoint.java | 10 +- .../component/quickfixj/QuickfixjProducer.java | 2 +- .../component/quickfixj/QuickfixjConsumerTest.java | 69 ++++++- .../component/quickfixj/QuickfixjInOutTest.java | 220 +++++++++++++++++++++ .../QuickfixjProducerCorrelationCleanupTest.java | 18 ++ .../component/quickfixj/QuickfixjProducerTest.java | 2 +- .../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 17 ++ 13 files changed, 385 insertions(+), 17 deletions(-) diff --git a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/quickfix.json b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/quickfix.json index e4a007573f98..be6f7f15f600 100644 --- a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/quickfix.json +++ b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/quickfix.json @@ -40,9 +40,9 @@ }, "bodyTypes": { "consumerOut": { "carriers": [ "pojo", "null" ], "description": "The FIX message: a quickfix.Message, such as a quickfix.fix44.Email, with the kind of event in the EventCategory header and the MsgType in the MessageType header. SessionCreated, SessionLogon and SessionLogoff events have no body" }, - "consumerReply": { "carriers": [ "pojo", "text", "bytes" ], "description": "With exchangePattern=InOut, the reply sent back on the session of the request: a quickfix.Message, or the FIX message as text or bytes. Every event is answered with the body the route ends with: a route that sets no reply sends back the message it received" }, + "consumerReply": { "carriers": [ "pojo", "text", "bytes" ], "description": "With exchangePattern=InOut, the reply sent back on the session of the request: a quickfix.Message, or the FIX message as text or bytes. Only an application message received (AppMessageReceived) is answered, and only when the route sets a reply: a route that ends with the message it received or a null body sends none. A reply that is not a FIX message, such as a Map, goes to the exception handler" }, "producerIn": { "carriers": [ "pojo", "text", "bytes" ], "description": "The FIX message to send: a quickfix.Message, or the FIX message as text or bytes, parsed with the DataDictionary exchange property or the data dictionary of the session. A stream, a Map or no body fails" }, - "producerOut": { "carriers": [ "unchanged", "pojo" ], "description": "The body is left as it was. An InOut exchange waits for the reply that matches the CorrelationCriteria exchange property, and the reply quickfix.Message is the body" } + "producerOut": { "carriers": [ "unchanged", "pojo" ], "description": "The body is left as it was. An InOut exchange waits for the reply that matches the CorrelationCriteria exchange property, and the reply quickfix.Message is the body. An InOut exchange without the property fails before the message is sent" } }, "properties": { "configurationName": { "index": 0, "kind": "path", "displayName": "Configuration Name", "group": "common", "label": "", "required": true, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "supportFileReference": true, "description": "Path to the quickfix configuration file. You can prefix with: classpath, file, http, ref, or bean. classpath, file and http loads the configuration file using these protocol [...] diff --git a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/quickfix-component.adoc b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/quickfix-component.adoc index 61d92dbca9fa..fadd6176ead0 100644 --- a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/quickfix-component.adoc +++ b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/quickfix-component.adoc @@ -278,6 +278,12 @@ YAML:: In YAML, the headers are written by name: `MessageType` is `QuickfixjEndpoint.MESSAGE_TYPE_KEY`, and the type of an OrderStatusRequest is `H`. +A reply is sent only for an application message received (the `AppMessageReceived` event), and only when the +route set a reply: the body the route ends with must be a FIX message other than the one received. A route that leaves +the received message as the body, or sets a null body, sends no reply, and the other events (such as the Logon being +sent) are never answered. A reply that is not a FIX message, such as a `Map`, is not sent: the error goes to the +exception handler of the consumer (logged at WARN by default). No reply is sent when the exchange failed. + The reply is always sent on the session the request was received on. Changing the `SessionID` header while routing does not change where the reply goes. To send a message to another session, use a QuickFIX/J @@ -311,7 +317,8 @@ matches our request. The SenderCompID and TargetCompID of the session ID are matched as given against the header of the reply, so use the session ID as the __replier__ sees it: for a request sent on `FIX.4.2:TRADER->MARKET`, use `FIX.4.2:MARKET->TRADER`. The correlation criteria is required for an InOut -exchange. +exchange: without it, the exchange fails with an `IllegalArgumentException` before the message is sent. Use the +`InOnly` exchange pattern to send a message without waiting for a reply. ._Java-only: programmatic correlation criteria setup using QuickFIX/J API_ [source,java] diff --git a/components/camel-quickfix/src/generated/resources/META-INF/org/apache/camel/component/quickfixj/quickfix.json b/components/camel-quickfix/src/generated/resources/META-INF/org/apache/camel/component/quickfixj/quickfix.json index e4a007573f98..be6f7f15f600 100644 --- a/components/camel-quickfix/src/generated/resources/META-INF/org/apache/camel/component/quickfixj/quickfix.json +++ b/components/camel-quickfix/src/generated/resources/META-INF/org/apache/camel/component/quickfixj/quickfix.json @@ -40,9 +40,9 @@ }, "bodyTypes": { "consumerOut": { "carriers": [ "pojo", "null" ], "description": "The FIX message: a quickfix.Message, such as a quickfix.fix44.Email, with the kind of event in the EventCategory header and the MsgType in the MessageType header. SessionCreated, SessionLogon and SessionLogoff events have no body" }, - "consumerReply": { "carriers": [ "pojo", "text", "bytes" ], "description": "With exchangePattern=InOut, the reply sent back on the session of the request: a quickfix.Message, or the FIX message as text or bytes. Every event is answered with the body the route ends with: a route that sets no reply sends back the message it received" }, + "consumerReply": { "carriers": [ "pojo", "text", "bytes" ], "description": "With exchangePattern=InOut, the reply sent back on the session of the request: a quickfix.Message, or the FIX message as text or bytes. Only an application message received (AppMessageReceived) is answered, and only when the route sets a reply: a route that ends with the message it received or a null body sends none. A reply that is not a FIX message, such as a Map, goes to the exception handler" }, "producerIn": { "carriers": [ "pojo", "text", "bytes" ], "description": "The FIX message to send: a quickfix.Message, or the FIX message as text or bytes, parsed with the DataDictionary exchange property or the data dictionary of the session. A stream, a Map or no body fails" }, - "producerOut": { "carriers": [ "unchanged", "pojo" ], "description": "The body is left as it was. An InOut exchange waits for the reply that matches the CorrelationCriteria exchange property, and the reply quickfix.Message is the body" } + "producerOut": { "carriers": [ "unchanged", "pojo" ], "description": "The body is left as it was. An InOut exchange waits for the reply that matches the CorrelationCriteria exchange property, and the reply quickfix.Message is the body. An InOut exchange without the property fails before the message is sent" } }, "properties": { "configurationName": { "index": 0, "kind": "path", "displayName": "Configuration Name", "group": "common", "label": "", "required": true, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "supportFileReference": true, "description": "Path to the quickfix configuration file. You can prefix with: classpath, file, http, ref, or bean. classpath, file and http loads the configuration file using these protocol [...] diff --git a/components/camel-quickfix/src/main/docs/quickfix-component.adoc b/components/camel-quickfix/src/main/docs/quickfix-component.adoc index 61d92dbca9fa..fadd6176ead0 100644 --- a/components/camel-quickfix/src/main/docs/quickfix-component.adoc +++ b/components/camel-quickfix/src/main/docs/quickfix-component.adoc @@ -278,6 +278,12 @@ YAML:: In YAML, the headers are written by name: `MessageType` is `QuickfixjEndpoint.MESSAGE_TYPE_KEY`, and the type of an OrderStatusRequest is `H`. +A reply is sent only for an application message received (the `AppMessageReceived` event), and only when the +route set a reply: the body the route ends with must be a FIX message other than the one received. A route that leaves +the received message as the body, or sets a null body, sends no reply, and the other events (such as the Logon being +sent) are never answered. A reply that is not a FIX message, such as a `Map`, is not sent: the error goes to the +exception handler of the consumer (logged at WARN by default). No reply is sent when the exchange failed. + The reply is always sent on the session the request was received on. Changing the `SessionID` header while routing does not change where the reply goes. To send a message to another session, use a QuickFIX/J @@ -311,7 +317,8 @@ matches our request. The SenderCompID and TargetCompID of the session ID are matched as given against the header of the reply, so use the session ID as the __replier__ sees it: for a request sent on `FIX.4.2:TRADER->MARKET`, use `FIX.4.2:MARKET->TRADER`. The correlation criteria is required for an InOut -exchange. +exchange: without it, the exchange fails with an `IllegalArgumentException` before the message is sent. Use the +`InOnly` exchange pattern to send a message without waiting for a reply. ._Java-only: programmatic correlation criteria setup using QuickFIX/J API_ [source,java] diff --git a/components/camel-quickfix/src/main/java/org/apache/camel/component/quickfixj/MessageCorrelator.java b/components/camel-quickfix/src/main/java/org/apache/camel/component/quickfixj/MessageCorrelator.java index 8b6a9bcc276d..cca3b749d309 100644 --- a/components/camel-quickfix/src/main/java/org/apache/camel/component/quickfixj/MessageCorrelator.java +++ b/components/camel-quickfix/src/main/java/org/apache/camel/component/quickfixj/MessageCorrelator.java @@ -34,6 +34,13 @@ public class MessageCorrelator implements QuickfixjEventListener { public Callable<Message> getReply(Exchange exchange) { MessagePredicate messageCriteria = (MessagePredicate) exchange.getProperty(QuickfixjProducer.CORRELATION_CRITERIA_KEY); + if (messageCriteria == null) { + // without criteria no message can be matched as the reply + throw new IllegalArgumentException( + "An InOut exchange needs the " + QuickfixjProducer.CORRELATION_CRITERIA_KEY + + " exchange property (a MessagePredicate) to match the reply." + + " Set the property, or send the message with the InOnly exchange pattern."); + } final MessageCorrelationRule correlationRule = new MessageCorrelationRule(exchange, messageCriteria); rules.add(correlationRule); diff --git a/components/camel-quickfix/src/main/java/org/apache/camel/component/quickfixj/QuickfixjConsumer.java b/components/camel-quickfix/src/main/java/org/apache/camel/component/quickfixj/QuickfixjConsumer.java index 7f60ad11da4c..c13cc4c0d1f1 100644 --- a/components/camel-quickfix/src/main/java/org/apache/camel/component/quickfixj/QuickfixjConsumer.java +++ b/components/camel-quickfix/src/main/java/org/apache/camel/component/quickfixj/QuickfixjConsumer.java @@ -18,6 +18,7 @@ package org.apache.camel.component.quickfixj; import org.apache.camel.Endpoint; import org.apache.camel.Exchange; +import org.apache.camel.InvalidPayloadException; import org.apache.camel.Message; import org.apache.camel.Processor; import org.apache.camel.support.DefaultConsumer; @@ -59,11 +60,18 @@ public class QuickfixjConsumer extends DefaultConsumer { // the reply goes back on the session the request arrived on, so capture it before routing // can change the header SessionID messageSessionID = exchange.getIn().getHeader(QuickfixjEndpoint.SESSION_ID_KEY, SessionID.class); + // only an application message received is a request; the other events (such as the Logon being sent) + // must never be answered + QuickfixjEventCategory category + = exchange.getIn().getHeader(QuickfixjEndpoint.EVENT_CATEGORY_KEY, QuickfixjEventCategory.class); + boolean replyExpected + = exchange.getPattern().isOutCapable() && category == QuickfixjEventCategory.AppMessageReceived; + Object request = exchange.getIn().getBody(); getProcessor().process(exchange); - if (exchange.getPattern().isOutCapable() && exchange.hasOut()) { - sendOutMessage(exchange, messageSessionID); + if (replyExpected && !exchange.isFailed()) { + sendOutMessage(exchange, messageSessionID, request); } } catch (Exception e) { exchange.setException(e); @@ -71,9 +79,24 @@ public class QuickfixjConsumer extends DefaultConsumer { } } - private void sendOutMessage(Exchange exchange, SessionID messageSessionID) throws QFJException { + private void sendOutMessage(Exchange exchange, SessionID messageSessionID, Object request) throws QFJException { Message camelMessage = exchange.getMessage(); - quickfix.Message quickfixjMessage = camelMessage.getBody(quickfix.Message.class); + Object body = camelMessage.getBody(); + if (body == null || body == request) { + // the route set no reply (an InOut route ends with the message it received when it does not set one) + LOG.debug("No FIX message reply to send for exchange: {}", exchange.getExchangeId()); + return; + } + + quickfix.Message quickfixjMessage; + try { + quickfixjMessage = camelMessage.getMandatoryBody(quickfix.Message.class); + } catch (InvalidPayloadException e) { + // the request has been routed, so it must not fail inside the QuickFIX/J callback, where the engine would + // ask the counterparty to resend it and route it again, over and over + getExceptionHandler().handleException("Cannot send the reply, as it is not a FIX message", exchange, e); + return; + } LOG.debug("Sending FIX message reply: {}", quickfixjMessage); diff --git a/components/camel-quickfix/src/main/java/org/apache/camel/component/quickfixj/QuickfixjEndpoint.java b/components/camel-quickfix/src/main/java/org/apache/camel/component/quickfixj/QuickfixjEndpoint.java index a15f79350ba6..7a2b48cec166 100644 --- a/components/camel-quickfix/src/main/java/org/apache/camel/component/quickfixj/QuickfixjEndpoint.java +++ b/components/camel-quickfix/src/main/java/org/apache/camel/component/quickfixj/QuickfixjEndpoint.java @@ -57,8 +57,11 @@ import quickfix.SessionID; consumerReply = @BodyType(value = { Carrier.POJO, Carrier.TEXT, Carrier.BYTES }, description = "With exchangePattern=InOut, the reply sent back on the session of the" + " request: a quickfix.Message, or the FIX message as text or bytes." - + " Every event is answered with the body the route ends with: a route" - + " that sets no reply sends back the message it received"), + + " Only an application message received (AppMessageReceived) is" + + " answered, and only when the route sets a reply: a route that ends" + + " with the message it received or a null body sends none. A reply" + + " that is not a FIX message, such as a Map, goes to the exception" + + " handler"), producerIn = @BodyType(value = { Carrier.POJO, Carrier.TEXT, Carrier.BYTES }, description = "The FIX message to send: a quickfix.Message, or the FIX message as text or" + " bytes, parsed with the DataDictionary exchange property or the data" @@ -66,7 +69,8 @@ import quickfix.SessionID; producerOut = @BodyType(value = { Carrier.UNCHANGED, Carrier.POJO }, description = "The body is left as it was. An InOut exchange waits for the reply that" + " matches the CorrelationCriteria exchange property, and the reply" - + " quickfix.Message is the body")) + + " quickfix.Message is the body. An InOut exchange without the property" + + " fails before the message is sent")) public class QuickfixjEndpoint extends DefaultEndpoint implements QuickfixjEventListener, MultipleConsumersSupport { @Metadata(description = "The event category.", javaType = "org.apache.camel.component.quickfixj.QuickfixjEventCategory") public static final String EVENT_CATEGORY_KEY = "EventCategory"; diff --git a/components/camel-quickfix/src/main/java/org/apache/camel/component/quickfixj/QuickfixjProducer.java b/components/camel-quickfix/src/main/java/org/apache/camel/component/quickfixj/QuickfixjProducer.java index 7392e440c5f2..4caace8fee96 100644 --- a/components/camel-quickfix/src/main/java/org/apache/camel/component/quickfixj/QuickfixjProducer.java +++ b/components/camel-quickfix/src/main/java/org/apache/camel/component/quickfixj/QuickfixjProducer.java @@ -66,7 +66,7 @@ public class QuickfixjProducer extends DefaultProducer { } void sendMessage(Exchange exchange, org.apache.camel.Message camelMessage) throws Exception { - Message message = camelMessage.getBody(Message.class); + Message message = camelMessage.getMandatoryBody(Message.class); LOG.debug("Sending FIX message: {}", message); SessionID messageSessionID = getEndpoint().getSID(); diff --git a/components/camel-quickfix/src/test/java/org/apache/camel/component/quickfixj/QuickfixjConsumerTest.java b/components/camel-quickfix/src/test/java/org/apache/camel/component/quickfixj/QuickfixjConsumerTest.java index e072018e11f4..9f753710c265 100644 --- a/components/camel-quickfix/src/test/java/org/apache/camel/component/quickfixj/QuickfixjConsumerTest.java +++ b/components/camel-quickfix/src/test/java/org/apache/camel/component/quickfixj/QuickfixjConsumerTest.java @@ -16,10 +16,14 @@ */ package org.apache.camel.component.quickfixj; +import java.util.Map; + import org.apache.camel.Exchange; import org.apache.camel.ExchangePattern; +import org.apache.camel.InvalidPayloadException; import org.apache.camel.Processor; import org.apache.camel.component.quickfixj.converter.QuickfixjConverters; +import org.apache.camel.spi.ExceptionHandler; import org.apache.camel.test.junit6.CamelTestSupport; import org.junit.jupiter.api.Test; import org.mockito.Mockito; @@ -62,6 +66,63 @@ public class QuickfixjConsumerTest extends CamelTestSupport { Mockito.verify(consumer, Mockito.never()).getSession(otherSessionID); } + @Test + public void otherEventsAreNotAnswered() throws Exception { + Message reply = new Message(); + Session requestSession = Mockito.mock(Session.class); + + QuickfixjConsumer consumer = startConsumer(replyWith(reply, null), requestSession); + for (QuickfixjEventCategory category : QuickfixjEventCategory.values()) { + if (category != QuickfixjEventCategory.AppMessageReceived) { + Exchange exchange = receive(consumer, category, new Message()); + assertThat(exchange.getException(), nullValue()); + } + } + + Mockito.verify(requestSession, Mockito.never()).send(Mockito.any()); + } + + @Test + public void receivedMessageIsNotSentBack() throws Exception { + Session requestSession = Mockito.mock(Session.class); + + // an InOut route that sets no reply ends with the message it received + QuickfixjConsumer consumer = startConsumer(exchange -> exchange.getMessage().setBody(exchange.getIn().getBody()), + requestSession); + Exchange exchange = receive(consumer); + + assertThat(exchange.getException(), nullValue()); + Mockito.verify(requestSession, Mockito.never()).send(Mockito.any()); + } + + @Test + public void nullReplyIsNotSent() throws Exception { + Session requestSession = Mockito.mock(Session.class); + + QuickfixjConsumer consumer = startConsumer(exchange -> exchange.getMessage().setBody(null), requestSession); + Exchange exchange = receive(consumer); + + assertThat(exchange.getException(), nullValue()); + Mockito.verify(requestSession, Mockito.never()).send(Mockito.any()); + } + + @Test + public void replyThatIsNotAFixMessageGoesToTheExceptionHandler() throws Exception { + Session requestSession = Mockito.mock(Session.class); + ExceptionHandler exceptionHandler = Mockito.mock(ExceptionHandler.class); + + QuickfixjConsumer consumer = startConsumer(exchange -> exchange.getMessage().setBody(Map.of("Subject", "reply")), + requestSession); + consumer.setExceptionHandler(exceptionHandler); + Exchange exchange = receive(consumer); + + // the exchange must not fail, or the engine would not take the request as received and route it again + assertThat(exchange.getException(), nullValue()); + Mockito.verify(requestSession, Mockito.never()).send(Mockito.any()); + Mockito.verify(exceptionHandler).handleException(Mockito.anyString(), Mockito.eq(exchange), + Mockito.isA(InvalidPayloadException.class)); + } + private QuickfixjConsumer startConsumer(Processor processor, Session requestSession) throws Exception { QuickfixjEndpoint endpoint = Mockito.mock(QuickfixjEndpoint.class); Mockito.when(endpoint.getCamelContext()).thenReturn(context); @@ -73,8 +134,12 @@ public class QuickfixjConsumerTest extends CamelTestSupport { } private Exchange receive(QuickfixjConsumer consumer) { - Exchange exchange = QuickfixjConverters.toExchange(consumer, requestSessionID, new Message(), - QuickfixjEventCategory.AppMessageReceived, ExchangePattern.InOut); + return receive(consumer, QuickfixjEventCategory.AppMessageReceived, new Message()); + } + + private Exchange receive(QuickfixjConsumer consumer, QuickfixjEventCategory category, Message message) { + Exchange exchange = QuickfixjConverters.toExchange(consumer, requestSessionID, message, category, + ExchangePattern.InOut); consumer.onExchange(exchange); return exchange; } diff --git a/components/camel-quickfix/src/test/java/org/apache/camel/component/quickfixj/QuickfixjInOutTest.java b/components/camel-quickfix/src/test/java/org/apache/camel/component/quickfixj/QuickfixjInOutTest.java new file mode 100644 index 000000000000..9c17a72a20c5 --- /dev/null +++ b/components/camel-quickfix/src/test/java/org/apache/camel/component/quickfixj/QuickfixjInOutTest.java @@ -0,0 +1,220 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.camel.component.quickfixj; + +import java.util.HashMap; +import java.util.Map; +import java.util.concurrent.ThreadLocalRandom; + +import org.apache.camel.CamelContext; +import org.apache.camel.Exchange; +import org.apache.camel.ExchangePattern; +import org.apache.camel.InvalidPayloadException; +import org.apache.camel.builder.RouteBuilder; +import org.apache.camel.component.mock.MockEndpoint; +import org.apache.camel.test.junit6.CamelTestSupport; +import org.junit.jupiter.api.Test; +import quickfix.Acceptor; +import quickfix.FieldNotFound; +import quickfix.FixVersions; +import quickfix.Initiator; +import quickfix.Session; +import quickfix.SessionFactory; +import quickfix.SessionID; +import quickfix.SessionSettings; +import quickfix.field.EmailThreadID; +import quickfix.field.EmailType; +import quickfix.field.MsgType; +import quickfix.field.Subject; +import quickfix.fix44.Email; +import quickfix.mina.ProtocolFactory; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertInstanceOf; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * InOut exchanges between an in-process acceptor and initiator: the consumer replies only to the application messages + * it is asked to answer, and the producer gets the reply that matches its correlation criteria. + */ +class QuickfixjInOutTest extends CamelTestSupport { + + private static final SessionID ACCEPTOR = new SessionID(FixVersions.BEGINSTRING_FIX44, "ACCEPTOR", "INITIATOR"); + private static final SessionID INITIATOR = new SessionID(FixVersions.BEGINSTRING_FIX44, "INITIATOR", "ACCEPTOR"); + private static final String CONSUMER = "quickfix:inout?sessionID=" + ACCEPTOR + "&exchangePattern=InOut"; + private static final String PRODUCER = "quickfix:inout?sessionID=" + INITIATOR; + + @Override + protected CamelContext createCamelContext() throws Exception { + CamelContext context = super.createCamelContext(); + // the values are text, as in a settings file; the vm pipe "port" is a name inside the JVM, not a socket + long port = ThreadLocalRandom.current().nextInt(20000, 30000); + + Map<Object, Object> defaults = new HashMap<>(); + defaults.put(Acceptor.SETTING_SOCKET_ACCEPT_PROTOCOL, ProtocolFactory.getTypeString(ProtocolFactory.VM_PIPE)); + defaults.put(Initiator.SETTING_SOCKET_CONNECT_PROTOCOL, ProtocolFactory.getTypeString(ProtocolFactory.VM_PIPE)); + defaults.put(Session.SETTING_USE_DATA_DICTIONARY, "N"); + defaults.put(Session.SETTING_NON_STOP_SESSION, "Y"); + + QuickfixjConfiguration configuration = new QuickfixjConfiguration(); + configuration.setDefaultSettings(defaults); + configuration.addSessionSetting(ACCEPTOR, session(ACCEPTOR, SessionFactory.ACCEPTOR_CONNECTION_TYPE, + Acceptor.SETTING_SOCKET_ACCEPT_PORT, port)); + configuration.addSessionSetting(INITIATOR, session(INITIATOR, SessionFactory.INITIATOR_CONNECTION_TYPE, + Initiator.SETTING_SOCKET_CONNECT_PORT, port)); + + QuickfixjComponent component = new QuickfixjComponent(); + component.setConfigurations(Map.of("inout", configuration)); + context.addComponent("quickfix", component); + return context; + } + + private static Map<Object, Object> session(SessionID id, String connectionType, String portKey, long port) { + Map<Object, Object> settings = new HashMap<>(); + settings.put(SessionSettings.BEGINSTRING, id.getBeginString()); + settings.put(SessionSettings.SENDERCOMPID, id.getSenderCompID()); + settings.put(SessionSettings.TARGETCOMPID, id.getTargetCompID()); + settings.put(SessionFactory.SETTING_CONNECTION_TYPE, connectionType); + settings.put(portKey, String.valueOf(port)); + settings.put(Initiator.SETTING_RECONNECT_INTERVAL, "1"); + settings.put(Session.SETTING_HEARTBTINT, "30"); + return settings; + } + + @Test + void consumerRepliesOnlyToTheMessagesTheRouteAnswers() throws Exception { + awaitLogon(); + MockEndpoint replies = getMockEndpoint("mock:replies"); + replies.expectedBodiesReceived("re: ping"); + + // the route sets no reply for this one, so it must not be echoed back; the session keeps the order, so an + // echo would arrive before the reply to the ping + template.sendBody(PRODUCER, email("no reply")); + template.sendBody(PRODUCER, email("ping")); + + MockEndpoint.assertIsSatisfied(context); + } + + @Test + void consumerReplyThatIsNotAFixMessageIsNotSent() throws Exception { + awaitLogon(); + MockEndpoint requests = getMockEndpoint("mock:requests"); + requests.expectedBodiesReceived("map", "null", "ping"); + MockEndpoint replies = getMockEndpoint("mock:replies"); + replies.expectedBodiesReceived("re: ping"); + + template.sendBody(PRODUCER, email("map")); + template.sendBody(PRODUCER, email("null")); + template.sendBody(PRODUCER, email("ping")); + + // each request is routed once: a reply that cannot be sent does not make the engine ask for it again + MockEndpoint.assertIsSatisfied(context); + } + + @Test + void producerGetsTheReplyThatMatchesTheCriteria() throws Exception { + awaitLogon(); + Exchange exchange = template.send(PRODUCER, ExchangePattern.InOut, e -> { + e.getIn().setBody(email("ping")); + // the reply is matched with the session ID as the replier sees it + e.setProperty(QuickfixjProducer.CORRELATION_CRITERIA_KEY, + new MessagePredicate(ACCEPTOR, MsgType.EMAIL).withField(Subject.FIELD, "re: ping")); + e.setProperty(QuickfixjProducer.CORRELATION_TIMEOUT_KEY, 10000L); + }); + + assertNull(exchange.getException()); + Email reply = assertInstanceOf(Email.class, exchange.getMessage().getBody()); + assertEquals("re: ping", reply.getSubject().getValue()); + } + + @Test + void producerWithoutCriteriaFailsBeforeSending() throws Exception { + awaitLogon(); + Exchange exchange = template.send(PRODUCER, ExchangePattern.InOut, e -> e.getIn().setBody(email("lost"))); + + IllegalArgumentException e = assertInstanceOf(IllegalArgumentException.class, exchange.getException()); + assertTrue(e.getMessage().contains(QuickfixjProducer.CORRELATION_CRITERIA_KEY), e.getMessage()); + + // the request was not sent, and the session still works + MockEndpoint requests = getMockEndpoint("mock:requests"); + requests.expectedBodiesReceived("ping"); + template.sendBody(PRODUCER, email("ping")); + MockEndpoint.assertIsSatisfied(context); + } + + @Test + void producerBodyThatIsNotAFixMessageNamesItsType() throws Exception { + awaitLogon(); + Exchange exchange = template.send(PRODUCER, e -> e.getIn().setBody(Map.of("Subject", "map"))); + + InvalidPayloadException e = assertInstanceOf(InvalidPayloadException.class, exchange.getException()); + assertTrue(e.getMessage().contains("quickfix.Message"), e.getMessage()); + } + + private void awaitLogon() throws InterruptedException { + MockEndpoint logons = getMockEndpoint("mock:logon"); + logons.expectedMinimumMessageCount(2); + logons.setResultWaitTime(10000); + logons.assertIsSatisfied(); + logons.reset(); + } + + private static Email email(String subject) { + return new Email(new EmailThreadID("ID"), new EmailType(EmailType.NEW), new Subject(subject)); + } + + private static String subject(Exchange exchange) throws FieldNotFound { + return exchange.getIn().getBody(Email.class).getSubject().getValue(); + } + + @Override + protected RouteBuilder createRouteBuilder() { + return new RouteBuilder() { + @Override + public void configure() { + // the acceptor answers the pings; every event of the session reaches the InOut route + from(CONSUMER).routeId("acceptor") + .choice() + .when(header(QuickfixjEndpoint.EVENT_CATEGORY_KEY).isEqualTo(QuickfixjEventCategory.SessionLogon)) + .to("mock:logon") + .when(header(QuickfixjEndpoint.EVENT_CATEGORY_KEY).isEqualTo(QuickfixjEventCategory.AppMessageReceived)) + .setVariable("request", body()) + .process(e -> e.getMessage().setBody(subject(e))) + .to("mock:requests") + .process(e -> { + String subject = e.getMessage().getBody(String.class); + switch (subject) { + case "ping" -> e.getMessage().setBody(email("re: ping")); + case "map" -> e.getMessage().setBody(Map.of("Subject", "re: map")); + case "null" -> e.getMessage().setBody(null); + // no reply: the route ends with the message it received + default -> e.getMessage().setBody(e.getVariable("request")); + } + }); + // the initiator, which sends the requests and receives the replies + from(PRODUCER).routeId("initiator") + .choice() + .when(header(QuickfixjEndpoint.EVENT_CATEGORY_KEY).isEqualTo(QuickfixjEventCategory.SessionLogon)) + .to("mock:logon") + .when(header(QuickfixjEndpoint.EVENT_CATEGORY_KEY).isEqualTo(QuickfixjEventCategory.AppMessageReceived)) + .process(e -> e.getMessage().setBody(subject(e))) + .to("mock:replies"); + } + }; + } +} diff --git a/components/camel-quickfix/src/test/java/org/apache/camel/component/quickfixj/QuickfixjProducerCorrelationCleanupTest.java b/components/camel-quickfix/src/test/java/org/apache/camel/component/quickfixj/QuickfixjProducerCorrelationCleanupTest.java index 16a4b1915b59..21ded8a0ac84 100644 --- a/components/camel-quickfix/src/test/java/org/apache/camel/component/quickfixj/QuickfixjProducerCorrelationCleanupTest.java +++ b/components/camel-quickfix/src/test/java/org/apache/camel/component/quickfixj/QuickfixjProducerCorrelationCleanupTest.java @@ -38,6 +38,7 @@ import quickfix.field.TargetCompID; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertInstanceOf; +import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.ArgumentMatchers.isA; /** @@ -115,6 +116,23 @@ class QuickfixjProducerCorrelationCleanupTest { assertEquals(0, pendingRules(), "correlation rules left after a send that threw"); } + @Test + void requestWithoutCorrelationCriteriaFailsBeforeItIsSent() throws Exception { + Mockito.doReturn(true).when(session).send(isA(Message.class)); + + Exchange exchange = createInOutExchange(5000L); + exchange.removeProperty(QuickfixjProducer.CORRELATION_CRITERIA_KEY); + producer.process(exchange); + + IllegalArgumentException e = assertInstanceOf(IllegalArgumentException.class, exchange.getException()); + assertTrue(e.getMessage().contains(QuickfixjProducer.CORRELATION_CRITERIA_KEY), e.getMessage()); + Mockito.verify(session, Mockito.never()).send(isA(Message.class)); + assertEquals(0, pendingRules(), "correlation rules left after a request without criteria"); + + // the engine events are still dispatched to the correlator without failing + engine.getMessageCorrelator().onEvent(QuickfixjEventCategory.AppMessageReceived, responseSessionID, fixMessage); + } + private Exchange createInOutExchange(long timeout) { Exchange exchange = new DefaultExchange(context, ExchangePattern.InOut); exchange.getIn().setBody(fixMessage); diff --git a/components/camel-quickfix/src/test/java/org/apache/camel/component/quickfixj/QuickfixjProducerTest.java b/components/camel-quickfix/src/test/java/org/apache/camel/component/quickfixj/QuickfixjProducerTest.java index 9cd2f09bda2c..7028ea4b1aa2 100644 --- a/components/camel-quickfix/src/test/java/org/apache/camel/component/quickfixj/QuickfixjProducerTest.java +++ b/components/camel-quickfix/src/test/java/org/apache/camel/component/quickfixj/QuickfixjProducerTest.java @@ -68,7 +68,7 @@ public class QuickfixjProducerTest { inboundFixMessage.getHeader().setString(TargetCompID.FIELD, "TARGET"); sessionID = MessageUtils.getSessionID(inboundFixMessage); - Mockito.when(mockCamelMessage.getBody(Message.class)).thenReturn(inboundFixMessage); + Mockito.when(mockCamelMessage.getMandatoryBody(Message.class)).thenReturn(inboundFixMessage); Mockito.when(mockEndpoint.getSID()).thenReturn(sessionID); diff --git a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc index 10f5a47c1b89..6dcc4c86e0ce 100644 --- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc +++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc @@ -2480,6 +2480,23 @@ an `Integer`, `Long`, `Short`, `Float` or `Double` was written in its binary for now also written in the charset of the response `Content-Type`, as a `String` is, rather than in the charset of the exchange. To send the binary form of a number, convert the body to `byte[]` first. +=== camel-quickfix - an InOut consumer replies only to application messages, and an InOut producer needs CorrelationCriteria + +A consumer with `exchangePattern=InOut` now sends a reply only for an `AppMessageReceived` event, and only when the +route set a reply: a body that is not the FIX message received, and not null. Before, every event was answered with the +body the route ended with, so the messages the route did not answer were echoed back to the counterparty, and the +`AdminMessageSent` event of the Logon sent the Logon again until a `StackOverflowError`. A reply that is not a FIX +message (such as a `Map`) is no longer sent: the error goes to the exception handler of the consumer (logged at WARN by +default), and the request counts as received. Before, it failed inside QuickFIX/J, which then asked the counterparty to +resend the request over and over. No reply is sent when the exchange failed. + +A producer with an InOut exchange now fails before it sends the message when the `CorrelationCriteria` exchange property +is not set, with an `IllegalArgumentException` that names the property. Before, the message was sent and the exchange +failed with `ExchangeTimedOutException`, as no reply could be matched. Send with `InOnly` when no reply is expected. + +A producer body that does not convert to a FIX message (such as a stream, a `Map` or no body) now fails with an +`InvalidPayloadException` that names the body type, instead of a `NullPointerException`. + === camel-report-maven-plugin - Java routes are read with the Java DSL parser The `validate` and `route-coverage` goals read Java routes with the Java DSL parser of camel-java-io instead of
