This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new 0f9f085e8b97 CAMEL-26184: camel-quickfix - an InOut consumer replies
only to application messages the route answers, and an InOut producer without
CorrelationCriteria fails before sending (#29013)
0f9f085e8b97 is described below
commit 0f9f085e8b97aab2f45bba04fa2573fb8b6a7788
Author: Claus Ibsen <[email protected]>
AuthorDate: Tue Nov 3 07:52:34 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 (#29013)
* CAMEL-26184: camel-quickfix - an InOut consumer replies only to appli…
…cation 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.
* CAMEL-26184: camel-quickfix - an InOut consumer replies with the rece…
…ived message changed in place
The reply to an AppMessageReceived event is the body the route ends with,
also when it is the received
FIX message changed in place. A null body sends no reply.
* CAMEL-26184: camel-quickfix - the InOut consumer example answers only…
… OrderStatusRequest, and sets a null body for the other messages
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 | 41 +++-
.../apache/camel/component/quickfixj/quickfix.json | 4 +-
.../src/main/docs/quickfix-component.adoc | 41 +++-
.../component/quickfixj/MessageCorrelator.java | 7 +
.../component/quickfixj/QuickfixjConsumer.java | 26 ++-
.../component/quickfixj/QuickfixjEndpoint.java | 11 +-
.../component/quickfixj/QuickfixjProducer.java | 2 +-
.../component/quickfixj/QuickfixjConsumerTest.java | 75 ++++++-
.../component/quickfixj/QuickfixjInOutTest.java | 236 +++++++++++++++++++++
.../QuickfixjProducerCorrelationCleanupTest.java | 18 ++
.../component/quickfixj/QuickfixjProducerTest.java | 2 +-
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 18 ++
13 files changed, 450 insertions(+), 35 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..9e3e8f292c7e 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, with the body the route ends
with: a new message or the received message changed in place. A route that does
not change the body sends back the message it received, and a null body sends
no reply. A reply [...]
"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..e8956f2a410a 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
@@ -241,7 +241,8 @@ Add `exchangePattern=InOut` to the QuickFIX/J endpoint URI.
The
the example below is a bean with a synchronous service method. The
method returns the response
to the request (an ExecutionReport in this case) which is then sent back
-to the requestor session.
+to the requestor session. The `otherwise` sets a null body, so the other
+messages are not answered.
[tabs]
====
@@ -250,8 +251,11 @@ Java::
[source,java]
----
from("quickfix:examples/inprocess.qf.cfg?sessionID=FIX.4.2:MARKET->TRADER&exchangePattern=InOut")
-
.filter(header(QuickfixjEndpoint.MESSAGE_TYPE_KEY).isEqualTo(MsgType.ORDER_STATUS_REQUEST))
- .bean(new MarketOrderStatusService());
+ .choice()
+
.when(header(QuickfixjEndpoint.MESSAGE_TYPE_KEY).isEqualTo(MsgType.ORDER_STATUS_REQUEST))
+ .bean(new MarketOrderStatusService())
+ .otherwise()
+ .setBody(simple("${null}"));
----
YAML::
@@ -265,19 +269,33 @@ YAML::
sessionID: "FIX.4.2:MARKET->TRADER"
exchangePattern: InOut
steps:
- - filter:
- expression:
- simple:
- expression: "${header.MessageType} == 'H'"
- steps:
- - bean:
- beanType: com.example.MarketOrderStatusService
+ - choice:
+ when:
+ - expression:
+ simple:
+ expression: "${header.MessageType} == 'H'"
+ steps:
+ - bean:
+ beanType: com.example.MarketOrderStatusService
+ otherwise:
+ steps:
+ - setBody:
+ expression:
+ simple:
+ expression: "${null}"
----
====
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): the reply is the body
+the route ends with. This can be a new FIX message, or the received message
changed in place. A route that does not
+change the body sends the received message back, so set a null body when a
message must not be answered (for example
+in the `otherwise` of a `choice`). 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 +329,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..9e3e8f292c7e 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, with the body the route ends
with: a new message or the received message changed in place. A route that does
not change the body sends back the message it received, and a null body sends
no reply. A reply [...]
"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..e8956f2a410a 100644
--- a/components/camel-quickfix/src/main/docs/quickfix-component.adoc
+++ b/components/camel-quickfix/src/main/docs/quickfix-component.adoc
@@ -241,7 +241,8 @@ Add `exchangePattern=InOut` to the QuickFIX/J endpoint URI.
The
the example below is a bean with a synchronous service method. The
method returns the response
to the request (an ExecutionReport in this case) which is then sent back
-to the requestor session.
+to the requestor session. The `otherwise` sets a null body, so the other
+messages are not answered.
[tabs]
====
@@ -250,8 +251,11 @@ Java::
[source,java]
----
from("quickfix:examples/inprocess.qf.cfg?sessionID=FIX.4.2:MARKET->TRADER&exchangePattern=InOut")
-
.filter(header(QuickfixjEndpoint.MESSAGE_TYPE_KEY).isEqualTo(MsgType.ORDER_STATUS_REQUEST))
- .bean(new MarketOrderStatusService());
+ .choice()
+
.when(header(QuickfixjEndpoint.MESSAGE_TYPE_KEY).isEqualTo(MsgType.ORDER_STATUS_REQUEST))
+ .bean(new MarketOrderStatusService())
+ .otherwise()
+ .setBody(simple("${null}"));
----
YAML::
@@ -265,19 +269,33 @@ YAML::
sessionID: "FIX.4.2:MARKET->TRADER"
exchangePattern: InOut
steps:
- - filter:
- expression:
- simple:
- expression: "${header.MessageType} == 'H'"
- steps:
- - bean:
- beanType: com.example.MarketOrderStatusService
+ - choice:
+ when:
+ - expression:
+ simple:
+ expression: "${header.MessageType} == 'H'"
+ steps:
+ - bean:
+ beanType: com.example.MarketOrderStatusService
+ otherwise:
+ steps:
+ - setBody:
+ expression:
+ simple:
+ expression: "${null}"
----
====
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): the reply is the body
+the route ends with. This can be a new FIX message, or the received message
changed in place. A route that does not
+change the body sends the received message back, so set a null body when a
message must not be answered (for example
+in the `otherwise` of a `choice`). 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 +329,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..8056a2bb2414 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,10 +60,16 @@ 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;
getProcessor().process(exchange);
- if (exchange.getPattern().isOutCapable() && exchange.hasOut())
{
+ if (replyExpected && !exchange.isFailed()) {
sendOutMessage(exchange, messageSessionID);
}
} catch (Exception e) {
@@ -73,7 +80,22 @@ public class QuickfixjConsumer extends DefaultConsumer {
private void sendOutMessage(Exchange exchange, SessionID messageSessionID)
throws QFJException {
Message camelMessage = exchange.getMessage();
- quickfix.Message quickfixjMessage =
camelMessage.getBody(quickfix.Message.class);
+ // the reply is the body the route ends with, which can be the
received message changed in place
+ if (camelMessage.getBody() == null) {
+ // a null body is how a route says there is no reply
+ 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..f2def89a47d7 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,12 @@ 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, with the body
the route ends with: a new message or the"
+ + " received message
changed in place. A route that does not change"
+ + " the body sends back the
message it received, and a null body"
+ + " sends no reply. 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 +70,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..6c45426c2322 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,17 +16,23 @@
*/
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;
import quickfix.Message;
import quickfix.Session;
import quickfix.SessionID;
+import quickfix.field.Subject;
+import static org.hamcrest.CoreMatchers.is;
import static org.hamcrest.CoreMatchers.nullValue;
import static org.hamcrest.MatcherAssert.assertThat;
@@ -62,6 +68,67 @@ 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 receivedMessageChangedInPlaceIsSentBack() throws Exception {
+ Message request = new Message();
+ Session requestSession = Mockito.mock(Session.class);
+ Mockito.when(requestSession.send(request)).thenReturn(true);
+
+ // the route changes the received message and keeps it as the body
+ QuickfixjConsumer consumer = startConsumer(
+ exchange ->
exchange.getMessage().getBody(Message.class).setString(Subject.FIELD,
"changed"),
+ requestSession);
+ Exchange exchange = receive(consumer,
QuickfixjEventCategory.AppMessageReceived, request);
+
+ assertThat(exchange.getException(), nullValue());
+ Mockito.verify(requestSession).send(request);
+ assertThat(request.getString(Subject.FIELD), is("changed"));
+ }
+
+ @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 +140,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..2b5fcf6f533a
--- /dev/null
+++
b/components/camel-quickfix/src/test/java/org/apache/camel/component/quickfixj/QuickfixjInOutTest.java
@@ -0,0 +1,236 @@
+/*
+ * 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 a null body for this one, so it must not be
answered; the session keeps the order, so a
+ // reply would arrive before the reply to the ping
+ template.sendBody(PRODUCER, email("no reply"));
+ template.sendBody(PRODUCER, email("ping"));
+
+ MockEndpoint.assertIsSatisfied(context);
+ }
+
+ @Test
+ void consumerRepliesWithTheReceivedMessageChangedInPlace() throws
Exception {
+ awaitLogon();
+ MockEndpoint replies = getMockEndpoint("mock:replies");
+ replies.expectedBodiesReceived("re: edit");
+
+ template.sendBody(PRODUCER, email("edit"));
+
+ 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"));
+ // the received message changed in place is
the reply
+ case "edit" -> {
+ Email request = e.getVariable("request",
Email.class);
+ request.set(new Subject("re: edit"));
+ e.getMessage().setBody(request);
+ }
+ // a null body: no reply
+ default -> e.getMessage().setBody(null);
+ }
+ });
+ // 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 91b130caf3d0..cebd7ff9196c 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
@@ -2543,6 +2543,24 @@ 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, with the body the
+route ends with (a new FIX message, or the received message changed in place),
and a null body sends no reply. Before,
+every event was answered with the body the route ended with, so the
`AdminMessageSent` event of the Logon sent the Logon
+again until a `StackOverflowError`. A route that does not change the body
still sends the received message back, so
+set a null body for a message that must not be answered. 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
