You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by da...@apache.org on 2018/04/12 08:50:18 UTC
[camel] branch camel-2.21.x updated: CAMEL-12427: camel-netty4 -
Add SPI to plugin custom correlation state for request/reply in producer
This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch camel-2.21.x
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/camel-2.21.x by this push:
new 715a242 CAMEL-12427: camel-netty4 - Add SPI to plugin custom correlation state for request/reply in producer
715a242 is described below
commit 715a24254bbf9f1e99d4c84a337bebd6b8c352e5
Author: Claus Ibsen <cl...@gmail.com>
AuthorDate: Thu Apr 12 10:48:05 2018 +0200
CAMEL-12427: camel-netty4 - Add SPI to plugin custom correlation state for request/reply in producer
---
.../camel-hl7/src/main/docs/hl7-dataformat.adoc | 194 ++++++++-------
.../src/main/docs/netty4-http-component.adoc | 2 +-
.../src/main/docs/netty4-component.adoc | 270 ++++++++++-----------
.../netty4/DefaultClientInitializerFactory.java | 1 -
.../netty4/NettyCamelStateCorrelationManager.java | 8 +-
.../camel/component/netty4/NettyConfiguration.java | 11 +-
.../camel/component/netty4/NettyProducer.java | 2 +-
.../springboot/NettyComponentConfiguration.java | 17 +-
8 files changed, 257 insertions(+), 248 deletions(-)
diff --git a/components/camel-hl7/src/main/docs/hl7-dataformat.adoc b/components/camel-hl7/src/main/docs/hl7-dataformat.adoc
index 544a751..3ba91fc 100644
--- a/components/camel-hl7/src/main/docs/hl7-dataformat.adoc
+++ b/components/camel-hl7/src/main/docs/hl7-dataformat.adoc
@@ -20,16 +20,16 @@ Maven users will need to add the following dependency to their `pom.xml`
for this component:
[source,xml]
-------------------------------------------------------------
+----
<dependency>
<groupId>org.apache.camel</groupId>
<artifactId>camel-hl7</artifactId>
<version>x.x.x</version>
<!-- use the same version as your Camel core version -->
</dependency>
-------------------------------------------------------------
+----
-### HL7 MLLP protocol
+=== HL7 MLLP protocol
HL7 is often used with the HL7 MLLP protocol, which is a text based TCP
socket based protocol. This component ships with a Mina and Netty4 Codec
@@ -43,7 +43,7 @@ a HL7 listener service, the <<mina2-component,camel-mina2>> or
HL7 MLLP codec can be configured as follows:
[width="100%",cols="10%,10%,80%",options="header",]
-|=======================================================================
+|===
|Name |Default Value |Description
|`startByte` |`0x0b` |The start byte spanning the HL7 payload.
@@ -65,17 +65,17 @@ the HL7 message content.
|`convertLFtoCR` |`false` |Will convert `\n` to `\r` (`0x0d`, 13 decimal) as HL7 stipulates `\r` as
segment terminators. The HAPI library requires the use of `\r`.
-|=======================================================================
+|===
-#### Exposing an HL7 listener using Mina
+==== Exposing an HL7 listener using Mina
In the Spring XML file, we configure a mina2 endpoint to listen for HL7
requests using TCP on port `8888`:
[source,xml]
----------------------------------------------------------------------------------------------------
- <endpoint id="hl7MinaListener" uri="mina2:tcp://localhost:8888?sync=true&codec=#hl7codec"/>
----------------------------------------------------------------------------------------------------
+----
+<endpoint id="hl7MinaListener" uri="mina2:tcp://localhost:8888?sync=true&codec=#hl7codec"/>
+----
*sync=true* indicates that this listener is synchronous and therefore
will return a HL7 response to the caller. The HL7 codec is setup with
@@ -84,34 +84,35 @@ could be named `mygreatcodecforhl7` or whatever. The codec is also set
up in the Spring XML file:
[source,xml]
-----------------------------------------------------------------------------
- <bean id="hl7codec" class="org.apache.camel.component.hl7.HL7MLLPCodec">
- <property name="charset" value="iso-8859-1"/>
- </bean>
-----------------------------------------------------------------------------
+----
+<bean id="hl7codec" class="org.apache.camel.component.hl7.HL7MLLPCodec">
+ <property name="charset" value="iso-8859-1"/>
+</bean>
+----
The endpoint *hl7MinaLlistener* can then be used in a route as a
consumer, as this Java DSL example illustrates:
[source,java]
-------------------------------------------------------------
- from("hl7MinaListener").beanRef("patientLookupService");
-------------------------------------------------------------
+----
+from("hl7MinaListener")
+ .bean("patientLookupService");
+----
This is a very simple route that will listen for HL7 and route it to a
service named *patientLookupService*. This is also Spring bean ID,
configured in the Spring XML as:
[source,xml]
----------------------------------------------------------------------------------------------------
- <bean id="patientLookupService" class="com.mycompany.healthcare.service.PatientLookupService"/>
----------------------------------------------------------------------------------------------------
+----
+<bean id="patientLookupService" class="com.mycompany.healthcare.service.PatientLookupService"/>
+----
The business logic can be implemented in POJO classes that do not depend
on Camel, as shown here:
[source,java]
-----------------------------------------------------------------------------------------------------
+----
import ca.uhn.hl7v2.HL7Exception;
import ca.uhn.hl7v2.model.Message;
import ca.uhn.hl7v2.model.v24.segment.QRD;
@@ -125,17 +126,17 @@ public class PatientLookupService {
Message response = ... create and set response data
return response
}
-----------------------------------------------------------------------------------------------------
+----
-#### Exposing an HL7 listener using Netty (available from Camel 2.15 onwards)
+==== Exposing an HL7 listener using Netty (available from Camel 2.15 onwards)
In the Spring XML file, we configure a netty4 endpoint to listen for HL7
requests using TCP on port `8888`:
[source,xml]
----------------------------------------------------------------------------------------------------------------------------------
- <endpoint id="hl7NettyListener" uri="netty4:tcp://localhost:8888?sync=true&encoder=#hl7encoder&decoder=#hl7decoder"/>
----------------------------------------------------------------------------------------------------------------------------------
+----
+<endpoint id="hl7NettyListener" uri="netty4:tcp://localhost:8888?sync=true&encoder=#hl7encoder&decoder=#hl7decoder"/>
+----
*sync=true* indicates that this listener is synchronous and therefore
will return a HL7 response to the caller. The HL7 codec is setup with
@@ -144,20 +145,21 @@ and `hl7decoder` are just bean IDs, so they could be named differently.
The beans can be set in the Spring XML file:
[source,xml]
----------------------------------------------------------------------------------------------
- <bean id="hl7decoder" class="org.apache.camel.component.hl7.HL7MLLPNettyDecoderFactory"/>
- <bean id="hl7encoder" class="org.apache.camel.component.hl7.HL7MLLPNettyEncoderFactory"/>
----------------------------------------------------------------------------------------------
+----
+<bean id="hl7decoder" class="org.apache.camel.component.hl7.HL7MLLPNettyDecoderFactory"/>
+<bean id="hl7encoder" class="org.apache.camel.component.hl7.HL7MLLPNettyEncoderFactory"/>
+----
The endpoint *hl7NettyListener* can then be used in a route as a
consumer, as this Java DSL example illustrates:
[source,java]
--------------------------------------------------------------
- from("hl7NettyListener").beanRef("patientLookupService");
--------------------------------------------------------------
+----
+from("hl7NettyListener")
+ .bean("patientLookupService");
+----
-### HL7 Model using java.lang.String or byte[]
+=== HL7 Model using java.lang.String or byte[]
The HL7 MLLP codec uses plain String as its data format. Camel uses its
Type Converter to convert to/from strings to
@@ -169,7 +171,7 @@ plain `byte[]` as its data format by setting the `produceString`
property to false. The Type Converter is also capable of converting
the `byte[]` to/from HAPI HL7 model objects.
-### HL7v2 Model using HAPI
+=== HL7v2 Model using HAPI
The HL7v2 model uses Java objects from the HAPI library. Using this
library, you can encode and decode from the EDI format (ER7) that is
@@ -178,21 +180,21 @@ mostly used with HL7v2.
The sample below is a request to lookup a patient with the patient ID
`0101701234`.
-[source,java]
------------------------------------------------------------------------------
+[source,text]
+----
MSH|^~\\&|MYSENDER|MYRECEIVER|MYAPPLICATION||200612211200||QRY^A19|1234|P|2.4
QRD|200612211200|R|I|GetPatient|||1^RD|0101701234|DEM||
------------------------------------------------------------------------------
+----
Using the HL7 model you can work with a `ca.uhn.hl7v2.model.Message`
object, e.g. to retrieve a patient ID:
[source,java]
---------------------------------------------------------------------------------------
+----
Message msg = exchange.getIn().getBody(Message.class);
QRD qrd = (QRD)msg.get("QRD");
String patientId = qrd.getWhoSubjectFilter(0).getIDNumber().getValue(); // 0101701234
---------------------------------------------------------------------------------------
+----
This is powerful when combined with the HL7 listener, because you don't
have to work with `byte[]`, `String` or any other simple object formats.
@@ -200,12 +202,12 @@ You can just use the HAPI HL7v2 model objects. If you know the message
type in advance, you can be more type-safe:
[source,java]
---------------------------------------------------------------------------------
+----
QRY_A19 msg = exchange.getIn().getBody(QRY_A19.class);
String patientId = msg.getQRD().getWhoSubjectFilter(0).getIDNumber().getValue();
---------------------------------------------------------------------------------
+----
-### HL7 DataFormat
+=== HL7 DataFormat
The <<hl7-dataformat,HL7>> component ships with a HL7 data format that can
be used to marshal or unmarshal HL7 model objects.
@@ -232,34 +234,38 @@ To use the data format, simply instantiate an instance and invoke the
marshal or unmarshal operation in the route builder:
[source,java]
------------------------------------------------------------
+----
DataFormat hl7 = new HL7DataFormat();
- ...
- from("direct:hl7in").marshal(hl7).to("jms:queue:hl7out");
------------------------------------------------------------
+
+ from("direct:hl7in")
+ .marshal(hl7)
+ .to("jms:queue:hl7out");
+----
In the sample above, the HL7 is marshalled from a HAPI Message object to
a byte stream and put on a JMS queue. +
The next example is the opposite:
[source,java]
----------------------------------------------------------------------
+----
DataFormat hl7 = new HL7DataFormat();
- ...
- from("jms:queue:hl7out").unmarshal(hl7).to("patientLookupService");
----------------------------------------------------------------------
+
+ from("jms:queue:hl7out")
+ .unmarshal(hl7)
+ .to("patientLookupService");
+----
Here we unmarshal the byte stream into a HAPI Message object that is
passed to our patient lookup service.
-*Serializable messages*
+==== Serializable messages
As of HAPI 2.0 (used by *Camel 2.11*), the HL7v2 model classes are fully
serializable. So you can put HL7v2 messages directly into a JMS queue
(i.e. without calling `marshal()` and read them again directly from the
queue (i.e. without calling `unmarshal()`.
-*Segment separators*
+==== Segment separators
As of *Camel 2.11*, `unmarshal` does not automatically fix segment
separators anymore by converting `\n` to `\r`. If you +
@@ -268,7 +274,7 @@ separators anymore by converting `\n` to `\r`. If you +
`Expression` for this purpose.
-*Charset*
+==== Charset
As of *Camel 2.14.1*, both `marshal and unmarshal` evaluate the charset
provided in the field `MSH-18`. If this field is empty, by default the
@@ -284,18 +290,23 @@ are commonly used. Then you don't need to create an instance of the `HL7DataForm
object:
[source,java]
-------------------------------------------------------------------------
- from("direct:hl7in").marshal().hl7().to("jms:queue:hl7out");
- from("jms:queue:hl7out").unmarshal().hl7().to("patientLookupService");
-------------------------------------------------------------------------
+----
+ from("direct:hl7in")
+ .marshal().hl7()
+ .to("jms:queue:hl7out");
-### Message Headers
+ from("jms:queue:hl7out")
+ .unmarshal().hl7()
+ .to("patientLookupService");
+----
+
+=== Message Headers
The unmarshal operation adds these fields from the MSH segment as
headers on the Camel message:
[width="100%",cols="10%,10%,80%",options="header",]
-|=======================================================================
+|===
|Key |MSH field |Example
|`CamelHL7SendingApplication` |`MSH-3` |`MYSERVER`
@@ -325,17 +336,17 @@ http://hl7api.sourceforge.net/base/apidocs/ca/uhn/hl7v2/HapiContext.html[HapiCon
that was used to parse the message
|`CamelHL7Charset` |`MSH-18` |*(Camel 2.14.1)* `UNICODE UTF-8`
-|=======================================================================
+|===
All headers except `CamelHL7Context `are `String` types. If a header
value is missing, its value is `null`.
-### Options
+=== Options
The HL7 Data Format supports the following options:
[width="100%",cols="10%,10%,80%",options="header",]
-|=======================================================================
+|===
|Option |Default |Description
|`validate` |true |Whether the HAPI Parser should validate the message using the default
@@ -352,9 +363,9 @@ also allows to parse XML-encoded HL7v2 messages
|`hapiContext` |`ca.uhn.hl7v2.DefaultHapiContext` |*Camel 2.14:* Custom HAPI context that can define a custom parser,
custom ValidationContext etc. This gives you full control over the HL7
parsing and rendering process.
-|=======================================================================
+|===
-### Dependencies
+=== Dependencies
To use HL7 in your Camel routes you'll need to add a dependency on
*camel-hl7* listed above, which implements this data format.
@@ -387,7 +398,7 @@ themselves. For example, if an application works with HL7v2 message
versions 2.4 and 2.5 then the following dependencies must be added:
[source,xml]
------------------------------------------------------------
+----
<dependency>
<groupId>ca.uhn.hapi</groupId>
<artifactId>hapi-structures-v24</artifactId>
@@ -400,7 +411,7 @@ versions 2.4 and 2.5 then the following dependencies must be added:
<version>2.2</version>
<!-- use the same version as your hapi-base version -->
</dependency>
------------------------------------------------------------
+----
Alternatively, an OSGi bundle containing the base library, all
structures libraries and required dependencies (on the bundle classpath)
@@ -409,15 +420,15 @@ http://repo1.maven.org/maven2/ca/uhn/hapi/hapi-osgi-base[central Maven
repository].
[source,xml]
--------------------------------------------
+----
<dependency>
<groupId>ca.uhn.hapi</groupId>
<artifactId>hapi-osgi-base</artifactId>
<version>2.2</version>
</dependency>
--------------------------------------------
+----
-### Terser language
+=== Terser language
http://hl7api.sourceforge.net[HAPI] provides a
http://hl7api.sourceforge.net/base/apidocs/ca/uhn/hl7v2/util/Terser.html[Terser]
@@ -429,22 +440,21 @@ and predicates for filtering, content-based routing etc.
Sample:
[source,java]
---------------------------------------------------------------------------------------------------
+----
import static org.apache.camel.component.hl7.HL7.terser;
-...
// extract patient ID from field QRD-8 in the QRY_A19 message above and put into message header
from("direct:test1")
.setHeader("PATIENT_ID",terser("QRD-8(0)-1"))
.to("mock:test1");
- // continue processing if extracted field equals a message header
+ // continue processing if extracted field equals a message header
from("direct:test2")
.filter(terser("QRD-8(0)-1").isEqualTo(header("PATIENT_ID"))
.to("mock:test2");
---------------------------------------------------------------------------------------------------
+----
-### HL7 Validation predicate
+=== HL7 Validation predicate
Often it is preferable to first parse a HL7v2 message and in a separate
step validate it against a HAPI
@@ -453,10 +463,9 @@ http://hl7api.sourceforge.net/base/apidocs/ca/uhn/hl7v2/validation/ValidationCon
Sample:
[source,java]
-----------------------------------------------------------------------
+----
import static org.apache.camel.component.hl7.HL7.messageConformsTo;
import ca.uhn.hl7v2.validation.impl.DefaultValidation;
-...
// Use standard or define your own validation rules
ValidationContext defaultContext = new DefaultValidation();
@@ -465,9 +474,9 @@ import ca.uhn.hl7v2.validation.impl.DefaultValidation;
from("direct:test1")
.validate(messageConformsTo(defaultContext))
.to("mock:test1");
-----------------------------------------------------------------------
+----
-### HL7 Validation predicate using the HapiContext (Camel 2.14)
+=== HL7 Validation predicate using the HapiContext (Camel 2.14)
The HAPI Context is always configured with a
http://hl7api.sourceforge.net/base/apidocs/ca/uhn/hl7v2/validation/ValidationContext.html[ValidationContext]
@@ -479,10 +488,9 @@ the `CamelHL7Context` header, and the validation rules of this context
can be easily reused:
[source,java]
-----------------------------------------------------------------------------------------------------------------
+----
import static org.apache.camel.component.hl7.HL7.messageConformsTo;
import static org.apache.camel.component.hl7.HL7.messageConforms
-...
HapiContext hapiContext = new DefaultHapiContext();
hapiContext.getParserConfiguration().setValidating(false); // don't validate during parsing
@@ -506,11 +514,9 @@ import static org.apache.camel.component.hl7.HL7.messageConforms
.validate(messageConforms()) // uses the validation rules returned from the HapiContext
// equivalent with .validate(messageConformsTo(hapiContext))
// route continues from here
-----------------------------------------------------------------------------------------------------------------
-
-
+----
-### HL7 Acknowledgement expression
+=== HL7 Acknowledgement expression
A common task in HL7v2 processing is to generate an acknowledgement
message as response to an incoming HL7v2 message, e.g. based on a
@@ -518,11 +524,10 @@ validation result. The `ack` expression lets us accomplish this very
elegantly:
[source,java]
-------------------------------------------------------------------------------------------
+----
import static org.apache.camel.component.hl7.HL7.messageConformsTo;
import static org.apache.camel.component.hl7.HL7.ack;
import ca.uhn.hl7v2.validation.impl.DefaultValidation;
-...
// Use standard or define your own validation rules
ValidationContext defaultContext = new DefaultValidation();
@@ -534,12 +539,12 @@ import ca.uhn.hl7v2.validation.impl.DefaultValidation;
.end()
.validate(messageConformsTo(defaultContext))
// do something meaningful here
- ...
+
// acknowledgement
.transform(ack())
-------------------------------------------------------------------------------------------
+----
-### More Samples
+=== More Samples
In the following example, a plain `String` HL7 request is sent to an HL7
listener that sends back a response:
@@ -554,12 +559,3 @@ Note that by using the HL7 DataFormat the Camel message headers are
populated with the fields from the MSH segment. The headers are
particularly useful for filtering or content-based routing as shown in
the example above.
-
-
-
-### See Also
-
-* Configuring Camel
-* Component
-* Endpoint
-* Getting Started
diff --git a/components/camel-netty4-http/src/main/docs/netty4-http-component.adoc b/components/camel-netty4-http/src/main/docs/netty4-http-component.adoc
index edbe900..974400c 100644
--- a/components/camel-netty4-http/src/main/docs/netty4-http-component.adoc
+++ b/components/camel-netty4-http/src/main/docs/netty4-http-component.adoc
@@ -168,7 +168,7 @@ with the following path and query parameters:
| *clientInitializerFactory* (producer) | To use a custom ClientInitializerFactory | | ClientInitializer Factory
| *lazyChannelCreation* (producer) | Channels can be lazily created to avoid exceptions, if the remote server is not up and running when the Camel producer is started. | true | boolean
| *okStatusCodeRange* (producer) | The status codes which are considered a success response. The values are inclusive. Multiple ranges can be defined, separated by comma, e.g. 200-204,209,301-304. Each range must be a single number or from-to with the dash included. The default range is 200-299 | 200-299 | String
-| *producerPoolEnabled* (producer) | Whether producer pool is enabled or not. Important: Do not turn this off, as the pooling is needed for handling concurrency and reliable request/reply. | true | boolean
+| *producerPoolEnabled* (producer) | Whether producer pool is enabled or not. Important: If you turn this off then a single shared connection is used for the producer, also if you are doing request/reply. That means there is a potential issue with interleaved responses if replies comes back out-of-order. Therefore you need to have a correlation id in both the request and reply messages so you can properly correlate the replies to the Camel callback that is responsible for continue proces [...]
| *producerPoolMaxActive* (producer) | Sets the cap on the number of objects that can be allocated by the pool (checked out to clients, or idle awaiting checkout) at a given time. Use a negative value for no limit. | -1 | int
| *producerPoolMaxIdle* (producer) | Sets the cap on the number of idle instances in the pool. | 100 | int
| *producerPoolMinEvictable Idle* (producer) | Sets the minimum amount of time (value in millis) an object may sit idle in the pool before it is eligible for eviction by the idle object evictor. | 300000 | long
diff --git a/components/camel-netty4/src/main/docs/netty4-component.adoc b/components/camel-netty4/src/main/docs/netty4-component.adoc
index 4d8c61e..f66a582 100644
--- a/components/camel-netty4/src/main/docs/netty4-component.adoc
+++ b/components/camel-netty4/src/main/docs/netty4-component.adoc
@@ -22,24 +22,24 @@ Maven users will need to add the following dependency to their `pom.xml`
for this component:
[source,xml]
-------------------------------------------------------------
+----
<dependency>
<groupId>org.apache.camel</groupId>
<artifactId>camel-netty4</artifactId>
<version>x.x.x</version>
<!-- use the same version as your Camel core version -->
</dependency>
-------------------------------------------------------------
+----
-### URI format
+=== URI format
The URI scheme for a netty component is as follows
-[source,java]
-----------------------------------------
+[source,text]
+----
netty4:tcp://localhost:99999[?options]
netty4:udp://remotehost:99999/[?options]
-----------------------------------------
+----
This component supports producer and consumer endpoints for both TCP and
UDP.
@@ -47,11 +47,7 @@ UDP.
You can append query options to the URI in the following format,
`?option=value&option=value&...`
-### Options
-
-
-
-
+=== Options
// component options: START
The Netty4 component supports 5 options which are listed below.
@@ -70,15 +66,6 @@ The Netty4 component supports 5 options which are listed below.
// component options: END
-
-
-
-
-
-
-
-
-
// endpoint options: START
The Netty4 endpoint is configured using URI syntax:
@@ -133,9 +120,9 @@ with the following path and query parameters:
| *connectTimeout* (producer) | Time to wait for a socket connection to be available. Value is in millis. | 10000 | int
| *requestTimeout* (producer) | Allows to use a timeout for the Netty producer when calling a remote server. By default no timeout is in use. The value is in milli seconds, so eg 30000 is 30 seconds. The requestTimeout is using Netty's ReadTimeoutHandler to trigger the timeout. | | long
| *clientInitializerFactory* (producer) | To use a custom ClientInitializerFactory | | ClientInitializer Factory
-| *correlationManager* (producer) | To use a custom correlation manager to manage how request and reply messages are mapped when using request/reply with the netty producer. This should only be used if you have a way to map requests together with replies such as if there is correlation ids in both the request and reply messages. This can be used if you want to multiplex concurrent messages on the same channel (aka connection) in netty. When doing this you must have a way to correlate the [...]
+| *correlationManager* (producer) | To use a custom correlation manager to manage how request and reply messages are mapped when using request/reply with the netty producer. This should only be used if you have a way to map requests together with replies such as if there is correlation ids in both the request and reply messages. This can be used if you want to multiplex concurrent messages on the same channel (aka connection) in netty. When doing this you must have a way to correlate the [...]
| *lazyChannelCreation* (producer) | Channels can be lazily created to avoid exceptions, if the remote server is not up and running when the Camel producer is started. | true | boolean
-| *producerPoolEnabled* (producer) | Whether producer pool is enabled or not. Important: Do not turn this off, as the pooling is needed for handling concurrency and reliable request/reply. | true | boolean
+| *producerPoolEnabled* (producer) | Whether producer pool is enabled or not. Important: If you turn this off then a single shared connection is used for the producer, also if you are doing request/reply. That means there is a potential issue with interleaved responses if replies comes back out-of-order. Therefore you need to have a correlation id in both the request and reply messages so you can properly correlate the replies to the Camel callback that is responsible for continue proces [...]
| *producerPoolMaxActive* (producer) | Sets the cap on the number of objects that can be allocated by the pool (checked out to clients, or idle awaiting checkout) at a given time. Use a negative value for no limit. | -1 | int
| *producerPoolMaxIdle* (producer) | Sets the cap on the number of idle instances in the pool. | 100 | int
| *producerPoolMinEvictable Idle* (producer) | Sets the minimum amount of time (value in millis) an object may sit idle in the pool before it is eligible for eviction by the idle object evictor. | 300000 | long
@@ -182,19 +169,13 @@ with the following path and query parameters:
// endpoint options: END
+=== Registry based Options
-
-
-
-
-### Registry based Options
-
-Codec Handlers and SSL Keystores can be enlisted in the
-Registry, such as in the Spring XML file.
+Codec Handlers and SSL Keystores can be enlisted in the Registry, such as in the Spring XML file.
The values that could be passed in, are the following:
[width="100%",cols="10%,90%",options="header",]
-|=======================================================================
+|===
|Name |Description
|`passphrase` |password setting to use in order to encrypt/decrypt payloads sent using
@@ -239,11 +220,11 @@ io.netty.channel.ChannelOutboundHandlerAdapter.
separated by comma, and have the values be looked up in the
Registry. Just remember to prefix the value with #
so Camel knows it should lookup.
-|=======================================================================
+|===
-*Important:* Read below about using non shareable encoders/decoders.
+NOTE: Read below about using non shareable encoders/decoders.
-#### Using non shareable encoders or decoders
+==== Using non shareable encoders or decoders
If your encoders or decoders is not shareable (eg they have the
@Shareable class annotation), then your encoder/decoder must implement
@@ -257,18 +238,16 @@ The Netty component offers a
`org.apache.camel.component.netty.ChannelHandlerFactories` factory
class, that has a number of commonly used methods.
-### Sending Messages to/from a Netty endpoint
+=== Sending Messages to/from a Netty endpoint
-#### Netty Producer
+==== Netty Producer
In Producer mode, the component provides the ability to send payloads to
-a socket endpoint +
- using either TCP or UDP protocols (with optional SSL support).
+a socket endpoint using either TCP or UDP protocols (with optional SSL support).
-The producer mode supports both one-way and request-response based
-operations.
+The producer mode supports both one-way and request-response based operations.
-#### Netty Consumer
+==== Netty Consumer
In Consumer mode, the component provides the ability to:
@@ -281,12 +260,12 @@ object based payloads and
The consumer mode supports both one-way and request-response based
operations.
-### Usage Samples
+=== Examples
-#### A UDP Netty endpoint using Request-Reply and serialized object payload
+==== A UDP Netty endpoint using Request-Reply and serialized object payload
[source,java]
-------------------------------------------------------------------
+----
RouteBuilder builder = new RouteBuilder() {
public void configure() {
from("netty4:udp://localhost:5155?sync=true")
@@ -299,21 +278,21 @@ RouteBuilder builder = new RouteBuilder() {
}
}
};
-------------------------------------------------------------------
+----
-#### A TCP based Netty consumer endpoint using One-way communication
+==== A TCP based Netty consumer endpoint using One-way communication
[source,java]
--------------------------------------------
+----
RouteBuilder builder = new RouteBuilder() {
public void configure() {
from("netty4:tcp://localhost:5150")
.to("mock:result");
}
};
--------------------------------------------
+----
-#### An SSL/TCP based Netty consumer endpoint using Request-Reply communication
+==== An SSL/TCP based Netty consumer endpoint using Request-Reply communication
[[Netty4-UsingtheJSSEConfigurationUtility]]
Using the JSSE Configuration Utility
@@ -329,7 +308,7 @@ to use the utility with the Netty component.
Programmatic configuration of the component
[source,java]
-------------------------------------------------------------------------------------------
+----
KeyStoreParameters ksp = new KeyStoreParameters();
ksp.setResource("/users/home/server/keystore.jks");
ksp.setPassword("keystorePassword");
@@ -343,13 +322,13 @@ scp.setKeyManagers(kmp);
NettyComponent nettyComponent = getContext().getComponent("netty4", NettyComponent.class);
nettyComponent.setSslContextParameters(scp);
-------------------------------------------------------------------------------------------
+----
[[Netty4-SpringDSLbasedconfigurationofendpoint]]
Spring DSL based configuration of endpoint
[source,xml]
--------------------------------------------------------------------------------------------------------
+----
...
<camel:sslContextParameters
id="sslContextParameters">
@@ -363,13 +342,13 @@ Spring DSL based configuration of endpoint
...
<to uri="netty4:tcp://localhost:5150?sync=true&ssl=true&sslContextParameters=#sslContextParameters"/>
...
--------------------------------------------------------------------------------------------------------
+----
[[Netty4-UsingBasicSSL/TLSconfigurationontheJettyComponent]]
Using Basic SSL/TLS configuration on the Jetty Component
[source,java]
-------------------------------------------------------------------------------
+----
JndiRegistry registry = new JndiRegistry(createJndiContext());
registry.bind("password", "changeit");
registry.bind("ksf", new File("src/test/resources/keystore.jks"));
@@ -393,25 +372,23 @@ context.addRoutes(new RouteBuilder() {
}
}
});
-------------------------------------------------------------------------------
+----
[[Netty4-GettingaccesstoSSLSessionandtheclientcertificate]]
Getting access to SSLSession and the client certificate
-*Available as of Camel 2.12*
-
You can get access to the `javax.net.ssl.SSLSession` if you eg need to
get details about the client certificate. When `ssl=true` then the
<<netty4-component,Netty4>> component will store the `SSLSession` as a
header on the Camel Message as shown below:
[source,java]
-----------------------------------------------------------------------------------------------------
+----
SSLSession session = exchange.getIn().getHeader(NettyConstants.NETTY_SSL_SESSION, SSLSession.class);
// get the first certificate which is client certificate
javax.security.cert.X509Certificate cert = session.getPeerCertificateChain()[0];
Principal principal = cert.getSubjectDN();
-----------------------------------------------------------------------------------------------------
+----
Remember to set `needClientAuth=true` to authenticate the client,
otherwise `SSLSession` cannot access information about the client
@@ -425,7 +402,7 @@ enriches the Camel Message with headers having
details about the client certificate. For example the subject name is
readily available in the header `CamelNettySSLClientCertSubjectName`.
-#### Using Multiple Codecs
+==== Using Multiple Codecs
In certain cases it may be necessary to add chains of encoders and
decoders to the netty pipeline. To add multpile codecs to a camel netty
@@ -436,13 +413,13 @@ ChannelDownstreamHandlers) that should be added to the pipeline. Note
that if encoders is specified then the encoder param will be ignored,
similarly for decoders and the decoder param.
-INFO: Read further above about using non shareable encoders/decoders.
+NOTE: Read further above about using non shareable encoders/decoders.
The lists of codecs need to be added to the Camel's registry so they can
be resolved when the endpoint is created.
[source,java]
--------------------------------------------------------------------------------------------------------------------
+----
ChannelHandlerFactory lengthDecoder = ChannelHandlerFactories.newLengthFieldBasedFrameDecoder(1048576, 0, 4, 0, 4);
StringDecoder stringDecoder = new StringDecoder();
@@ -464,15 +441,13 @@ encoders.add(stringEncoder);
registry.bind("encoders", encoders);
registry.bind("decoders", decoders);
-
--------------------------------------------------------------------------------------------------------------------
+----
Spring's native collections support can be used to specify the codec
lists in an application context
-[source,java]
--------------------------------------------------------------------------------------------------------------------------------------------------
-
+[source,xml]
+----
<util:list id="decoders" list-class="java.util.LinkedList">
<bean class="org.apache.camel.component.netty4.ChannelHandlerFactories" factory-method="newLengthFieldBasedFrameDecoder">
<constructor-arg value="1048576"/>
@@ -504,35 +479,35 @@ lists in an application context
<constructor-arg value="4"/>
</bean>
<bean id="string-decoder" class="io.netty.handler.codec.string.StringDecoder"/>
--------------------------------------------------------------------------------------------------------------------------------------------------
+----
The bean names can then be used in netty endpoint definitions either as
a comma separated list or contained in a List e.g.
[source,java]
------------------------------------------------------------------------------------------------------------------------
+----
from("direct:multiple-codec").to("netty4:tcp://localhost:{{port}}?encoders=#encoders&sync=false");
from("netty4:tcp://localhost:{{port}}?decoders=#length-decoder,#string-decoder&sync=false").to("mock:multiple-codec");
------------------------------------------------------------------------------------------------------------------------
+----
-or via spring.
+or via XML.
-[source,java]
--------------------------------------------------------------------------------------------------------------
- <camelContext id="multiple-netty-codecs-context" xmlns="http://camel.apache.org/schema/spring">
- <route>
- <from uri="direct:multiple-codec"/>
- <to uri="netty4:tcp://localhost:5150?encoders=#encoders&sync=false"/>
- </route>
- <route>
- <from uri="netty4:tcp://localhost:5150?decoders=#length-decoder,#string-decoder&sync=false"/>
- <to uri="mock:multiple-codec"/>
- </route>
- </camelContext>
--------------------------------------------------------------------------------------------------------------
-
-### Closing Channel When Complete
+[source,xml]
+----
+<camelContext id="multiple-netty-codecs-context" xmlns="http://camel.apache.org/schema/spring">
+ <route>
+ <from uri="direct:multiple-codec"/>
+ <to uri="netty4:tcp://localhost:5150?encoders=#encoders&sync=false"/>
+ </route>
+ <route>
+ <from uri="netty4:tcp://localhost:5150?decoders=#length-decoder,#string-decoder&sync=false"/>
+ <to uri="mock:multiple-codec"/>
+ </route>
+</camelContext>
+----
+
+=== Closing Channel When Complete
When acting as a server you sometimes want to close the channel when,
for example, a client conversion is finished. +
@@ -547,22 +522,23 @@ value. +
written the bye message back to the client:
[source,java]
---------------------------------------------------------------------------------------------------------
- from("netty4:tcp://localhost:8080").process(new Processor() {
- public void process(Exchange exchange) throws Exception {
- String body = exchange.getIn().getBody(String.class);
- exchange.getOut().setBody("Bye " + body);
- // some condition which determines if we should close
- if (close) {
- exchange.getOut().setHeader(NettyConstants.NETTY_CLOSE_CHANNEL_WHEN_COMPLETE, true);
- }
- }
- });
---------------------------------------------------------------------------------------------------------
+----
+from("netty4:tcp://localhost:8080").process(new Processor() {
+ public void process(Exchange exchange) throws Exception {
+ String body = exchange.getIn().getBody(String.class);
+ exchange.getOut().setBody("Bye " + body);
+ // some condition which determines if we should close
+ if (close) {
+ exchange.getOut().setHeader(NettyConstants.NETTY_CLOSE_CHANNEL_WHEN_COMPLETE, true);
+ }
+ }
+});
+----
[[Netty4-Addingcustomchannelpipelinefactoriestogaincompletecontroloveracreatedpipeline]]
Adding custom channel pipeline factories to gain complete control over a
-### created pipeline
+
+=== Custom pipeline
Custom channel pipelines provide complete control to the user over the
handler/interceptor chain by inserting custom handler(s), encoder(s) &
@@ -571,7 +547,7 @@ very simple way.
In order to add a custom pipeline, a custom channel pipeline factory
must be created and registered with the context via the context registry
-(JNDIRegistry,or the camel-spring ApplicationContextRegistry etc).
+(JNDIRegistry, or the camel-spring ApplicationContextRegistry etc).
A custom pipeline factory must be constructed as follows
@@ -581,20 +557,20 @@ class `ClientPipelineFactory`.
class `ServerInitializerFactory`.
* The classes should override the initChannel() method in order to
insert custom handler(s), encoder(s) and decoder(s). Not overriding the
-initChannel() method creates a pipeline with no handlers, encoders or
+`initChannel()` method creates a pipeline with no handlers, encoders or
decoders wired to the pipeline.
The example below shows how ServerInitializerFactory factory may be
created
-*Using custom pipeline factory*
+==== Using custom pipeline factory
[source,java]
---------------------------------------------------------------------------------------------------------------------------------
+----
public class SampleServerInitializerFactory extends ServerInitializerFactory {
private int maxLineSize = 1024;
- protected void initChannel(Channel ch) throws Exception {
+ protected void initChannel(Channel ch) throws Exception {
ChannelPipeline channelPipeline = ch.pipeline();
channelPipeline.addLast("encoder-SD", new StringEncoder(CharsetUtil.UTF_8));
@@ -604,13 +580,13 @@ public class SampleServerInitializerFactory extends ServerInitializerFactory {
channelPipeline.addLast("handler", new ServerChannelHandler(consumer));
}
}
---------------------------------------------------------------------------------------------------------------------------------
+----
The custom channel pipeline factory can then be added to the registry
and instantiated/utilized on a camel route in the following way
[source,java]
-----------------------------------------------------------------------
+----
Registry registry = camelContext.getRegistry();
ServerInitializerFactory factory = new TestServerInitializerFactory();
registry.bind("spf", factory);
@@ -630,11 +606,9 @@ context.addRoutes(new RouteBuilder() {
}
}
});
-----------------------------------------------------------------------
-
-### Reusing Netty boss and worker thread pools
+----
-*Available as of Camel 2.12*
+=== Reusing Netty boss and worker thread pools
Netty has two kind of thread pools: boss and worker. By default each
Netty consumer and producer has their private thread pools. If you want
@@ -646,17 +620,17 @@ For example using Spring XML we can create a shared worker thread pool
using the `NettyWorkerPoolBuilder` with 2 worker threads as shown below:
[source,xml]
------------------------------------------------------------------------------------------
- <!-- use the worker pool builder to help create the shared thread pool -->
- <bean id="poolBuilder" class="org.apache.camel.component.netty.NettyWorkerPoolBuilder">
- <property name="workerCount" value="2"/>
- </bean>
-
- <!-- the shared worker thread pool -->
- <bean id="sharedPool" class="org.jboss.netty.channel.socket.nio.WorkerPool"
- factory-bean="poolBuilder" factory-method="build" destroy-method="shutdown">
- </bean>
------------------------------------------------------------------------------------------
+----
+<!-- use the worker pool builder to help create the shared thread pool -->
+<bean id="poolBuilder" class="org.apache.camel.component.netty.NettyWorkerPoolBuilder">
+ <property name="workerCount" value="2"/>
+</bean>
+
+<!-- the shared worker thread pool -->
+<bean id="sharedPool" class="org.jboss.netty.channel.socket.nio.WorkerPool"
+ factory-bean="poolBuilder" factory-method="build" destroy-method="shutdown">
+</bean>
+----
TIP: For boss thread pool there is a
`org.apache.camel.component.netty4.NettyServerBossPoolBuilder` builder
@@ -670,32 +644,46 @@ https://cwiki.apache.org/confluence/pages/createpage.action?spaceKey=CAMEL&title
as shown below:
[source,xml]
--------------------------------------------------------------------------------------------------------------------------------------
- <route>
- <from uri="netty4:tcp://localhost:5021?textline=true&sync=true&workerPool=#sharedPool&usingExecutorService=false"/>
- <to uri="log:result"/>
- ...
- </route>
--------------------------------------------------------------------------------------------------------------------------------------
+----
+<route>
+ <from uri="netty4:tcp://localhost:5021?textline=true&sync=true&workerPool=#sharedPool&usingExecutorService=false"/>
+ <to uri="log:result"/>
+ ...
+</route>
+----
And if we have another route we can refer to the shared worker pool:
[source,xml]
--------------------------------------------------------------------------------------------------------------------------------------
- <route>
- <from uri="netty4:tcp://localhost:5022?textline=true&sync=true&workerPool=#sharedPool&usingExecutorService=false"/>
- <to uri="log:result"/>
- ...
- </route>
--------------------------------------------------------------------------------------------------------------------------------------
+----
+<route>
+ <from uri="netty4:tcp://localhost:5022?textline=true&sync=true&workerPool=#sharedPool&usingExecutorService=false"/>
+ <to uri="log:result"/>
+ ...
+</route>
+----
+
+and so forth.
+
+=== Multiplexing concurrent messages over a single connection with request/reply
+
+When using Netty for request/reply messaging via the netty producer then by default each
+message is sent via a non-shared connection (pooled). This ensures that replies are
+automatic being able to map to the correct request thread for further routing in Camel.
+In other words correlation between request/reply messages happens out-of-the-box because
+the replies comes back on the same connection that was used for sending the request;
+and this connection is not shared with others. When the response comes back, the connection
+is returned back to the connection pool, where it can be reused by others.
-... and so forth.
+However if you want to multiplex concurrent request/responses on a single shared connection,
+then you need to turn off the connection pooling by setting `producerPoolEnabled=false`.
+Now this means there is a potential issue with interleaved responses if replies comes back out-of-order.
+Therefore you need to have a correlation id in both the request and reply messages so you can properly
+correlate the replies to the Camel callback that is responsible for continue processing the message in Camel.
+To do this you need to implement `NettyCamelStateCorrelationManager` as correlation manager and configure
+it via the `correlationManager=#myManager` option.
-### See Also
+=== See Also
-* Configuring Camel
-* Component
-* Endpoint
-* Getting Started
* <<netty-http-component,Netty HTTP>>
* <<mina2-component,MINA>>
diff --git a/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/DefaultClientInitializerFactory.java b/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/DefaultClientInitializerFactory.java
index 54c90c5..36829b1 100644
--- a/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/DefaultClientInitializerFactory.java
+++ b/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/DefaultClientInitializerFactory.java
@@ -94,7 +94,6 @@ public class DefaultClientInitializerFactory extends ClientInitializerFactory {
addToPipeline("handler", channelPipeline, new ClientChannelHandler(producer));
LOG.trace("Created ChannelPipeline: {}", channelPipeline);
-
}
private void addToPipeline(String name, ChannelPipeline pipeline, ChannelHandler handler) {
diff --git a/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/NettyCamelStateCorrelationManager.java b/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/NettyCamelStateCorrelationManager.java
index cd76196..fea5ba3 100644
--- a/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/NettyCamelStateCorrelationManager.java
+++ b/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/NettyCamelStateCorrelationManager.java
@@ -16,6 +16,8 @@
*/
package org.apache.camel.component.netty4;
+import java.util.Map;
+
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
@@ -46,6 +48,10 @@ public interface NettyCamelStateCorrelationManager {
/**
* Gets the state when a response message has been received.
+ * <p/>
+ * If the implementation stores the state temporary in for example a {@link Map} instance
+ * then this method should remove the state from the map as its no longer needed. In other
+ * words use the {@link Map#remove(Object)} to get and remove the state.
*
* @param ctx netty channel handler context
* @param channel the channel
@@ -54,7 +60,7 @@ public interface NettyCamelStateCorrelationManager {
NettyCamelState getState(ChannelHandlerContext ctx, Channel channel, Object msg);
/**
- * Gets the state when some error occurred.
+ * Gets the state when some internal error occurred.
*
* @param ctx netty channel handler context
* @param channel the channel
diff --git a/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/NettyConfiguration.java b/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/NettyConfiguration.java
index 98e9298..a2086ef 100644
--- a/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/NettyConfiguration.java
+++ b/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/NettyConfiguration.java
@@ -589,7 +589,14 @@ public class NettyConfiguration extends NettyServerBootstrapConfiguration implem
/**
* Whether producer pool is enabled or not.
- * Important: Do not turn this off, as the pooling is needed for handling concurrency and reliable request/reply.
+ *
+ * Important: If you turn this off then a single shared connection is used for the producer, also if you are doing request/reply.
+ * That means there is a potential issue with interleaved responses if replies comes back out-of-order. Therefore you need to
+ * have a correlation id in both the request and reply messages so you can properly correlate the replies to the Camel callback
+ * that is responsible for continue processing the message in Camel. To do this you need to implement {@link NettyCamelStateCorrelationManager}
+ * as correlation manager and configure it via the <tt>correlationManager</tt> option.
+ * <p/>
+ * See also the <tt>correlationManager</tt> option for more details.
*/
public void setProducerPoolEnabled(boolean producerPoolEnabled) {
this.producerPoolEnabled = producerPoolEnabled;
@@ -667,6 +674,8 @@ public class NettyConfiguration extends NettyServerBootstrapConfiguration implem
* and reply messages. This can be used if you want to multiplex concurrent messages on the same channel (aka connection) in netty. When doing
* this you must have a way to correlate the request and reply messages so you can store the right reply on the inflight Camel Exchange before
* its continued routed.
+ * <p/>
+ * See also the <tt>producerPoolEnabled</tt> option for more details.
*/
public void setCorrelationManager(NettyCamelStateCorrelationManager correlationManager) {
this.correlationManager = correlationManager;
diff --git a/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/NettyProducer.java b/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/NettyProducer.java
index 591f6fd..0da5508 100644
--- a/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/NettyProducer.java
+++ b/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/NettyProducer.java
@@ -223,7 +223,7 @@ public class NettyProducer extends DefaultAsyncProducer {
}
// get a channel from the pool
- ChannelFuture channelFuture = null;
+ ChannelFuture channelFuture;
Channel channel = null;
try {
if (getConfiguration().isReuseChannel()) {
diff --git a/platforms/spring-boot/components-starter/camel-netty4-starter/src/main/java/org/apache/camel/component/netty4/springboot/NettyComponentConfiguration.java b/platforms/spring-boot/components-starter/camel-netty4-starter/src/main/java/org/apache/camel/component/netty4/springboot/NettyComponentConfiguration.java
index 0125ee4..99d0c33 100644
--- a/platforms/spring-boot/components-starter/camel-netty4-starter/src/main/java/org/apache/camel/component/netty4/springboot/NettyComponentConfiguration.java
+++ b/platforms/spring-boot/components-starter/camel-netty4-starter/src/main/java/org/apache/camel/component/netty4/springboot/NettyComponentConfiguration.java
@@ -273,9 +273,18 @@ public class NettyComponentConfiguration
*/
private Long producerPoolMinEvictableIdle = 300000L;
/**
- * Whether producer pool is enabled or not. Important: Do not turn this
- * off, as the pooling is needed for handling concurrency and reliable
- * request/reply.
+ * Whether producer pool is enabled or not. Important: If you turn this
+ * off then a single shared connection is used for the producer, also if
+ * you are doing request/reply. That means there is a potential issue
+ * with interleaved responses if replies comes back out-of-order.
+ * Therefore you need to have a correlation id in both the request and
+ * reply messages so you can properly correlate the replies to the Camel
+ * callback that is responsible for continue processing the message in
+ * Camel. To do this you need to implement
+ * {@link NettyCamelStateCorrelationManager} as correlation manager and
+ * configure it via the <tt>correlationManager</tt> option.
+ * <p/>
+ * See also the <tt>correlationManager</tt> option for more details.
*/
private Boolean producerPoolEnabled = true;
/**
@@ -324,6 +333,8 @@ public class NettyComponentConfiguration
* When doing this you must have a way to correlate the request and
* reply messages so you can store the right reply on the inflight Camel
* Exchange before its continued routed.
+ * <p/>
+ * See also the <tt>producerPoolEnabled</tt> option for more details.
*/
private NettyCamelStateCorrelationManager correlationManager;
/**
--
To stop receiving notification emails like this one, please contact
davsclaus@apache.org.