How to Setup/Locate Camel Core Log?

[SteveR]

I'm new to Camel and I'm using Camel 2.14.0 to develop a stand-alone Java
application.  So far, so good, but I can't seem to locate Camel's core log?
I'm using slf4j on top of log4j2 with its log4j2.xml file.

I see a few references to the Camel core log in the *Camel in Action* book,
but googling hasn't helped me much in figuring this out.

I must have had it working at one point with Netbeans on my laptop, because
I see a file called *camelLog*, here's a snippet:

*2015-01-23T15:39:45,490 DEBUG 
Adding routes from builder: Routes: []

2015-01-23T15:39:45,545  INFO 
Apache Camel 2.14.0 (CamelContext: MPLR) is starting

2015-01-23T15:39:45,545 DEBUG 
Using ClassResolver=org.apache.camel.impl.DefaultClassResolver@d9326df,
PackageScanClassResolver=org.apache.camel.impl.DefaultPackageScanClassResolver@37dd016f,
ApplicationContextClassLoader=sun.misc.Launcher$AppClassLoader@baf1915*

However, when building the jar file and deploying my application on a Linux
server, I can't locate the Camel core log file.

I also see one web post where someone mentions *
org.apache.camel.camel-core.log
<http://camel.465427.n5.nabble.com/Logging-into-the-bundle-log-file-via-to-quot-log-quot-td5738205.html>
 
*

Thanks in advance for any help,
   SteveR



--
View this message in context: 
http://camel.465427.n5.nabble.com/How-to-Setup-Locate-Camel-Core-Log-tp5762150.html
Sent from the Camel - Users mailing list archive at Nabble.com.

Camel 2.14/Netty: How to add ByteArrayDecoder to ServerChannelPipeline?

[SteveR]

I have a linux Java7 stand-alone application using Camel 2.14 and the
camel-netty component. I have a route that receives via netty:udp and then
attempts to mirror the received packets also via netty:udp. I'm also using
the *LengthFieldBasedFrameDecoder *to decode the UDP packets into message
frames based on a 2-byte length field within each UDP message.

The UDP messages I'm receiving contain certain characters that don't decode
in UTF-8 (e.g. 0xa1 0xb2, 0xc3 ), so I've been trying to use the
*iso-8859-1* charset. I'm thinking that what I want in my
ServerPipelineFactory subclass is the
*io.netty.handler.codec.bytes.ByteArrayDecoder*, but I'm unable to get it to
compile with the *addlast()* method.

I've been using the approach outlined  here
<https://github.com/apache/camel/blob/master/components/camel-netty/src/test/java/org/apache/camel/component/netty/NettyCustomPipelineFactorySynchTest.java>
 
thus far, but then I run into this compile issue with *io.netty *versus
*org.jboss.netty*.

Which leads me to believe that I'm confused wrt imports for *io.netty*
versus *org.jboss.netty*?  Below is my extended *ServerPipelineFactory
*class which I'd like to switch the StringDecoder/StringEncoder for
ByteArrayDecoder/ByteArrayEncoder.

Any help is greatly appreciated!

  Thanks, SteveR


package multiprotocollistenerrouter;

//import io.netty.handler.codec.bytes.ByteArrayDecoder;
//import io.netty.handler.codec.bytes.ByteArrayEncoder;
//import io.netty.channel.ChannelPipeline;
//import io.netty.handler.codec.LengthFieldBasedFrameDecoder;

import io.netty.util.CharsetUtil;
import org.apache.camel.component.netty.NettyConsumer;
import org.apache.camel.component.netty.ServerPipelineFactory;
import org.apache.camel.component.netty.handlers.ServerChannelHandler;
import org.jboss.netty.channel.ChannelHandler;
import org.jboss.netty.channel.ChannelPipeline;
import org.jboss.netty.channel.Channels;
import org.jboss.netty.handler.codec.frame.LengthFieldBasedFrameDecoder;
import org.jboss.netty.handler.codec.string.StringDecoder;
import org.jboss.netty.handler.codec.string.StringEncoder;

//import org.jboss.netty.handler.logging.LoggingHandler;
//import org.jboss.netty.logging.InternalLogLevel;

import org.slf4j.Logger;// The org.slf4j.Logger interface is the
main user entry point of SLF4J API.
import org.slf4j.LoggerFactory; // Utility class producing Loggers for
various logging APIs, most notably for log4j.

/**
 * Consumer linked channel pipeline factory for the MCQ source provider.
 * Provides custom support for reception/parsing/processing of MCQ.
 *
 * @author steve
 *
 * @see http://camel.apache.org/netty.html
 * @see
http://opensourceknowledge.blogspot.com/2010/08/customizing-netty-endpoints-using.html#
 * @see
http://seeallhearall.blogspot.com/2012/06/netty-tutorial-part-15-on-channel.html
 * @see
https://github.com/apache/camel/blob/master/components/camel-netty/src/test/
 * 
java/org/apache/camel/component/netty/NettyCustomPipelineFactorySynchTest.java
 */
public class McqServerPipelineFactory extends ServerPipelineFactory  {
private final static Logger logger   =
LoggerFactory.getLogger(McqServerPipelineFactory.class);
private static final String NEW_LINE =
System.getProperty("line.separator");

// ---
// Stateless, singleton handler instances...re-used across connections
// ---
private static final ChannelHandler STR_ENCODER = new
StringEncoder(CharsetUtil.ISO_8859_1);
private static final ChannelHandler STR_DECODER = new
StringDecoder(CharsetUtil.ISO_8859_1);

//private static final ChannelHandler LOG_HANDLER = new
LoggingHandler(InternalLogLevel.INFO);

//private static final ByteArrayDecoder BYTES_DECODER = new
ByteArrayDecoder();
//private static final ByteArrayEncoder BYTES_ENCODER = new
ByteArrayEncoder();

private final NettyConsumer consumer;

private final int maxFrameLength;
private final int lengthFieldOffset;
private final int lengthFieldLength;
private final int lengthAdjustment;
private final int initialBytesToStrip;
private boolean   invoked;
private StringrouteId;

@Override
public ChannelPipeline getPipeline() throws Exception {  
logger.trace("getPipeline(): ENTER");

invoked = true;  
ChannelPipeline channelPipeline = Channels.pipeline();  

String theRouteId = consumer.getRoute().getId();
logger.info("getPipeline(): {}, routeId = {}", consumer.toString(),
theRouteId);

// ---
// Add logger to print incoming and outgoing data.
// This is both an upstream/downstream handler.
// ---
//String loggerName = "MCQ_LOG

Re: Camel 2.14/Netty: How to add ByteArrayDecoder to ServerChannelPipeline?

[SteveR]

Hi Willem:

Actually, I'm currently using Camel 2.14 and Netty 3.x (org.jboss.netty). 
I'm wondering if there is a sensible way to migrate from netty3 to netty4
(i.e. other than just brute-force trying it) and what it would do with
respect to my existing pipeline factory code, etc?

Here's what I currently have in my pom.xml file:


org.apache.camel
camel-netty
2.14.0



io.netty
netty-all
4.0.2.Final
compile



 Thanks, Steve



--
View this message in context: 
http://camel.465427.n5.nabble.com/Camel-2-14-Netty-How-to-add-ByteArrayDecoder-to-ServerChannelPipeline-tp5768638p5768749.html
Sent from the Camel - Users mailing list archive at Nabble.com.

Logging Camel Exchanges in Hexadecimal?

[SteveR]

Is there any way to configure Camel to do Exchange logging in hexadecimal,
rather than being rendered in the associated character set?  I have a route
that I configure to use *iso-8859-1* (i.e. Latin-1) via
.*setProperty(Exchange.CHARSET_NAME, "iso-8859-1")* and the Exchange logging
displays the message body as iso-8859-1 characters.

So, for example,  I'd like to see something like this:

*[2015-06-30 19:45:58,483] DEBUG [New I/O worker #33] >>>>
Endpoint[udp://devserver-09.dev.s.mission.net:62268] Exchange[Message:
7200a1b2c3d45552659907edc201002a4f512e5063337153373239504b3232454876454d6c356a656151<<<
snip >>>00]*


... instead of this:

*[2015-06-30 19:45:58,483] DEBUG [New I/O worker #33] >>>>
Endpoint[udp://devserver-09.dev.s.mission.net:62268] Exchange[Message:
^@^@^@^@^@^@r^@¡²��URe<99>^@^@^YP�^A^@*OQ.Pc3qS729PK22EHvEMl5jeaQxxSIY.1431463279�^B^@c+17926775404<<<
snip >>>^@^@^@^@^@^@^@^@^@^@^@]*


   Thanks, SteveR








--
View this message in context: 
http://camel.465427.n5.nabble.com/Logging-Camel-Exchanges-in-Hexadecimal-tp5768750.html
Sent from the Camel - Users mailing list archive at Nabble.com.

Logging Camel route info ...

[SteveR]

I have a linux Java7 stand-alone application using Camel 2.14. In my Camel
*rootLog.txt* file with DEBUG enabled, I see that Camel prints a nice
succinct one-liner showing the details associated with a given route.

Can anyone point me to the Camel core java code that generates these kind of
trace prints? I'm assuming that it's some chain of DSL methods that are
employed to produce this kind of route info.  I'd like to use a similar
approach to incorporate these kind of trace prints in my application code,
since the Camel *rootLog.txt* will typically not be set to DEBUG level in
production.

For example:

*[2015-07-01 21:04:56,976] DEBUG [main] Route: ROUTE_ID_RAW_MCQEVENTS >>>
EventDrivenConsumerRoute[Endpoint[udp://dv-09.dev.s.mission.net:2050] ->
Pipeline[[Channel[setProperty(CamelCharsetName, iso-8859-1)],
Channel[Threads[[Choice[[When[{bean{MCQ_FILTER_BEAN_ROUTE_ID_RAW_MCQEVENTS,
method=isAuthorized}} ->
[To[log:com.mission.mplr.multiprotocollistenerrouter.BACKUP_FILE1?level=INFO],
To[log:com.mission.mplr.multiprotocollistenerrouter.THROUGHPUT_FILE1?level=INFO&groupInterval=3&groupDelay=1000&groupActiveOnly=false],
To[netty:udp://dv.dev.s.mission.net:62268?clientPipelineFactory=#MCQ_CLIENT_PIPELINE_FACTORY_ROUTE_ID_RAW_MCQEVENTS&sync=false&sendBufferSize=4194304&allowDefaultCodec=false&disconnectOnNoReply=false],
To[seda:SEDA_300], SetProperty[CamelCharsetName, {iso-8859-1}
Otherwise[[To[log:MCQ_NOT_AUTHORIZED_LOGGER?level=ERROR], Stop]*

   Thanks, SteveR



--
View this message in context: 
http://camel.465427.n5.nabble.com/Logging-Camel-route-info-tp5768807.html
Sent from the Camel - Users mailing list archive at Nabble.com.

Re: Wrong charset when using FTP2 component, locale issue?

[SteveR]

Hi Gustav:

I'm sure you already know this, but *0xefbfbd* is the  Unicode replacement
character" � (U+FFFD)
   which a
program may decide to insert for any character it couldn't decode correctly
when trying to handle Unicode.  So it looks like there are two adjacent
octets that were attempted to be converted to UTF-8 that were invalid.

  Steve



--
View this message in context: 
http://camel.465427.n5.nabble.com/Wrong-charset-when-using-FTP2-component-locale-issue-tp5768795p5768827.html
Sent from the Camel - Users mailing list archive at Nabble.com.

Re: Logging Camel route info ...

[SteveR]

Thanks Claus, that worked great!!!

Here's what I ended up with, the latter method lifted from
*/camel-core/src/main/java/org/apache/camel/management/mbean/ManagedCamelContext.java*.

  Thanks, Steve


/**
 * Dump all Camel routes for the specified CamelContext as a string.
 *
 * @param context
 * @return
 * @throws Exception 
 *
 * @see
http://camel.465427.n5.nabble.com/Logging-Camel-route-info-tt5768807.html
 */
public static String dumpRoutes(CamelContext context) throws Exception {
assert(context != null) : "dumpRoutes(): ERROR: CamelContext object
is null";
return ((ModelCamelContext)
context).getRouteDefinitions().toString();
}

/**
 * Dump all Camel routes for the specified CamelContext as XML document.
 *
 * @param context
 * @return
 * @throws Exception 
 *
 * @see
http://camel.465427.n5.nabble.com/Logging-Camel-route-info-tt5768807.html
 */
public static String dumpRoutesAsXml(CamelContext context) throws
Exception {
assert(context != null) : "dumpRoutesAsXml(): ERROR: CamelContext
object is null";
List routes =((ModelCamelContext)
context).getRouteDefinitions();
if(routes.isEmpty()) {
return null;
}
RoutesDefinition def = new RoutesDefinition();
def.setRoutes(routes);
return ModelHelper.dumpModelAsXml(def);
}





--
View this message in context: 
http://camel.465427.n5.nabble.com/Logging-Camel-route-info-tp5768807p5768826.html
Sent from the Camel - Users mailing list archive at Nabble.com.

Camel 2.14/Netty: Separate Exchange for each UDP packet?

[SteveR]

I have a Linux Java7 stand-alone application using Camel 2.14 and the
*camel-netty* component. I have several routes that each receive a UDP feed
via netty:udp.

I've already implemented a few routes that do this, but have used either a
*DelimiterBasedFrameDecoder* or a *LengthFieldBasedFrameDecoder*, depending
on whether the packets/messages in the UDP feed have a delimiter and/or a
way for me to know the total length of the payload.

Now I have another UDP feed that I need to create a route for, but this one
has payload messages each containing a fixed 16-byte header and the rest of
the message is encrypted.  Therefore, I have no message delimiter or message
length field to exploit.

Here lies my confusion: shouldn't camel-netty be able to just produce a
separate Exchange for each received UDP packet, regardless of its size or
contents?  Camel-netty interfaces to the UDP protocol and knows the size of
each received packet. I'm starting to feel like maybe I've over-complicated
things by using a *DelimiterBasedFrameDecoder * and/or
*LengthFieldBasedFrameDecoder* in my ServerPipeline.

Any help is greatly appreciated!

  Thanks, Steve




--
View this message in context: 
http://camel.465427.n5.nabble.com/Camel-2-14-Netty-Separate-Exchange-for-each-UDP-packet-tp5769372.html
Sent from the Camel - Users mailing list archive at Nabble.com.

Camel Exchange Properties when using Dead Letter Channel

[SteveR]

I have a Linux Java7 stand-alone application using Camel 2.14.  I'm using the
DLC pattern to capture exchanges to a local file when re-delivery attempts
have been exhausted, based on my *delayPattern*.  I'm also specifying the
*useOriginalMessage *option.

My goal is to compose another route that I can manually start at a later
time (e.g.via JConsole) to attempt to re-deliver the exchanges that are
orphaned in the DLC file. To get the route to be "manual" I'm using the
*noAutoStartup()* DSL method: 

*from("log:com.mission.mplr.DLC1?level=ERROR")
.routeId("DTC_DLC_REPLAY_ROUTE")
.noAutoStartup()
.process(dlcProcessor)
.to(sedaMainURI);< The route with this SEDA
queue does the delivery to kafka*

The messages do get logged in the DLC file, but all I see in the log file
for each exchange entry is the contents of the exchange inMsg body.  Is
there any way to also capture and log the Exchange properties/headers?

The ones I'm interested in are:

*NettyConstants.NETTY_REMOTE_ADDRESS* and
*NettyConstants.NETTY_LOCAL_ADDRESS*

because I typically populate these values into an application-specific
message header that accompanies each exchange body that will be delivered to
the intended destination (e.g. a kafka topic).

Btw, I'm using SLF4J, so I've been specifying the DLC file URI something
like this:

   *log:com.mission.mplr.DLC1?level=ERROR*

The log entries look similar to this, so I assume that I'll need a processor
in the manual route that attempts to strip off everything except the actual
exchange body contents?

*[2015-07-24 19:19:50,545] DEBUG [Camel (MPLR_DTC) thread #4 -
threads_ROUTE_ID_RAW_DTCEVENTS_SEDA_MAIN] Exchange[ExchangePattern: InOnly,
BodyType: org.jboss.netty.buffer.BigEndianHeapChannelBuffer, Body: [ <---
body contents ---> ]*

  Thanks, Steve




--
View this message in context: 
http://camel.465427.n5.nabble.com/Camel-Exchange-Properties-when-using-Dead-Letter-Channel-tp5769891.html
Sent from the Camel - Users mailing list archive at Nabble.com.

How to name an EventNotifier so it shows up in JConsole ...

[SteveR]

I'm using Camel 2.14 and I have an AuditEventNotifier to notify about slow
exchanges:

//
--
// Add the custom AuditEventNotifier to the CamelContext
// See
http://camel.apache.org/eventnotifier-to-log-details-about-all-sent-exchanges.html
//
--
Long maxElapsedMillis =
mainConfig.getCamelConfig().getCamelAuditEventMaxElapsedMillis();
*camelContext.getManagementStrategy().addEventNotifier(new
AuditEventNotifier(maxElapsedMillis));*
logger.info("Added AuditEventNotifier management strategy to
Camel context = {} " +
"with maxElapsedMillis = {}",
camelContext.getName(), maxElapsedMillis);


Is there any way to give an EventNotifer a name so that it shows up that way
in JConsole or VisualVM?

Currently, VisualVM displays it as *EventNotifier(0xbd8c9ab)*

   Thanks, Steve



--
View this message in context: 
http://camel.465427.n5.nabble.com/How-to-name-an-EventNotifier-so-it-shows-up-in-JConsole-tp5770092.html
Sent from the Camel - Users mailing list archive at Nabble.com.

Re: Conditionally omitting a portion of a route ...

[SteveR]

Looks like I spoke too soon, it doesn't look like it works when
*mirrorEnabled = false*.  I get:

org.apache.camel.FailedToCreateRouteException: Failed to create route
MIRROR_ROUTE_ID at: >>>
Threads[[Filter[bean{com.mission.RouteManager$MirrorBean,
method=isMirrorEnabled} -> [WireTap[no uri or ref supplied!],
To[log:com.mission.BACKUP_FILE1?level=INFO],
To[log:com.mission.THROUGHPUT_FILE1?level=INFO&groupInterval=3&groupDelay=1000&groupActiveOnly=false],
To[seda:SEDA_MAIN_100], To[seda:SEDA_ACK_100], SetProperty[CamelCharsetName,
{iso-8859-1}] <<< in route:
Route(MIRROR_ROUTE_ID)[[From[netty:udp://missioin-es-... *because of Either
'uri' or 'ref' must be specified on:
org.apache.camel.impl.DefaultRouteContext@25439d6d*
at
org.apache.camel.model.RouteDefinition.addRoutes(RouteDefinition.java:945)
~[camel-core-2.14.0.jar:2.14.0]
at
org.apache.camel.model.RouteDefinition.addRoutes(RouteDefinition.java:187)
~[camel-core-2.14.0.jar:2.14.0]



--
View this message in context: 
http://camel.465427.n5.nabble.com/Conditionally-omitting-a-portion-of-a-route-tp5770127p5770130.html
Sent from the Camel - Users mailing list archive at Nabble.com.

Re: Conditionally omitting a portion of a route ...

[SteveR]

I dumped the route as XML from within VisualVM and it appears that my
*filter() * is encompassing the remainder of the route, instead of just the
*wireTap()* portion. So when I set *mirrorEnabled = false*, nothing gets
routed to the intended destinations.


http://camel.apache.org/schema/spring";>


iso-8859-1


*   
com.mission.RouteManager$MirrorBean







iso-8859-1

*





--
View this message in context: 
http://camel.465427.n5.nabble.com/Conditionally-omitting-a-portion-of-a-route-tp5770127p5770132.html
Sent from the Camel - Users mailing list archive at Nabble.com.

Conditionally omitting a portion of a route ...

[SteveR]

A have a route similar to the below, whereby I also have a Java configuration
variable (e.g. *mirrorEnabled=true/false*) that tells me whether or not to
do the wiretap mirror.  My question is; what is the best way to implement
the route filtering to skip the wireTap() portion of the route if
mirrorEnabled= false?

I'm hoping it's doable with the simple language, but haven't yet figured out
how to cleanly do it.  The most brute-force way, would be to wrap two sets
of the below route in a Java if-else block, but that's just silly ;-)

  Thanks, SteveR

from(fromURI)
.routeId(sourceRouteId)
.routePolicy(routePolicy)
.startupOrder(routeStartupOrder)
.setProperty(Exchange.CHARSET_NAME,
ExpressionBuilder.constantExpression(charsetName))
//
---
// This wiretap implements the mirror functionality.
// A traditional wiretap whereby Camel will copy the
original
// Exchange and set its Exchange Pattern to InOnly, as
we want
// the tapped Exchange to be sent in a fire and forget
style.
//
---
*.wireTap(mirrorToURI)
.id(sourceRouteId + "_MIRROR")*
// -
// to(backupFileToURI, throughputFileToURI, sedaMainURI)
// -
.to(firstToURIs)
.id(sourceRouteId + "_TO_FIRST_URIS");



--
View this message in context: 
http://camel.465427.n5.nabble.com/Conditionally-omitting-a-portion-of-a-route-tp5770127.html
Sent from the Camel - Users mailing list archive at Nabble.com.

Re: Conditionally omitting a portion of a route ...

[SteveR]

So far, I have this solution which uses Camel Bean Language/Message Filter. 
I always want the route to send to the *firstToURIs*, but only send to the
mirror wiretap if the *mirrorEnabled* configuration value is set to true.

*  // 
  // This static variable and bean class allows for
  // conditional DSL wrt mirror functionality.
  // @see https://camel.apache.org/message-filter.html
  // @see http://camel.apache.org/bean-language.html
  // 
  private static Boolean mirrorEnabled = null; // Set based on the
mirror configuration. 
  public static class MirrorBean {
  public boolean isMirrorEnabled() {
  return mirrorEnabled.booleanValue();
  }
  }*

...

from(fromURI)
.routeId(sourceRouteId)
.routePolicy(routePolicy)
.startupOrder(routeStartupOrder)
.setProperty(Exchange.CHARSET_NAME,
ExpressionBuilder.constantExpression(charsetName))
//
---
// This wiretap implements the mirror functionality.
// A traditional wiretap whereby Camel will copy the
original
// Exchange and set its Exchange Pattern to InOnly, as
we want
// the tapped Exchange to be sent in a fire and forget
style.
//
---
*.filter().method(MirrorBean.class, "isMirrorEnabled")
.wireTap(mirrorToURI).id(sourceRouteId + "_MIRROR")
.end()*
// -
// to(backupFileToURI, throughputFileToURI, sedaMainURI)
// -
.to(firstToURIs)
.id(sourceRouteId + "_TO_FIRST_URIS"); 






--
View this message in context: 
http://camel.465427.n5.nabble.com/Conditionally-omitting-a-portion-of-a-route-tp5770127p5770129.html
Sent from the Camel - Users mailing list archive at Nabble.com.

Re: Conditionally omitting a portion of a route ...

[SteveR]

I also tried using the PredicateBuilder.constant() method to create a
predicate that resolves to true/false, based on my Java configuration
boolean:

*private static Boolean mirrorEnabled = null; // Set based on the
"enabled" MIRROR configuration value. 
Predicate mirrorWireTap = PredicateBuilder.constant(mirrorEnabled);*

 and then added this DSL:

*.choice()
 .when(mirrorWireTap).wireTap(mirrorToURI).id(sourceRouteId +
"_MIRROR")
.end()*

... but dumping the roue as XML from within VisualVM shows that the choice()
block encompasses the wireTap() and the remainder of the route.  I just want
something that lets me conditionally run the wireTap() method, or not.

  Thanks, Steve





--
View this message in context: 
http://camel.465427.n5.nabble.com/Conditionally-omitting-a-portion-of-a-route-tp5770127p5770165.html
Sent from the Camel - Users mailing list archive at Nabble.com.

Re: Conditionally omitting a portion of a route ...

[SteveR]

Thanks so much Claus!!! Just a few questions:

- I assume the last piece is:

// continue routing here
part = part.xxx 

And that there is nothing to do to finalize the ProcessorDefinition object,
just set it as you show?

Anyways, here's what I ended up with, based on your directoins:

ProcessorDefinition initialRoute =
from(fromURI)
.routeId(sourceRouteId)
.routePolicy(routePolicy)
.startupOrder(routeStartupOrder)
// See
http://grokbase.com/t/camel/users/1056thepe4/need-create-a-route-which-can-handle-several-encoding
.setProperty(Exchange.CHARSET_NAME,
ExpressionBuilder.constantExpression(charsetName))
.threads(threadPoolSize /*poolSize*/, threadPoolSize
* 2 /*maxPoolSize*/)
.threadName("threads_" + sourceRouteId)
//
---
// When a task is rejected (e.g. due to worker queue
is full), let the current caller thread
// execute the task (i.e. it will become
synchronous), also gives time for the thread pool
// to process its current tasks,without adding more
tasks (i.e. a sort of self throttling).
//
---
.callerRunsWhenRejected(true); // Hard-coded since
we always want this behavior!

if(mirrorEnabled) { // ProcessorDefinition allows us to use
Java to conditionalize the route
//
--
// Wire Tap (from the EIP patterns) allows you to route
messages to a separate location
// while they are being forwarded to the ultimate
destination. This wiretap implements
// the MPLR mirror functionality. It's a traditional
wiretap whereby Camel will copy the
// original Exchange and set its Exchange Pattern to
InOnly, as we want the tapped
// Exchange to be sent in a fire-and-forget style. Note
that this mirror wiretap is
// conditional, based on the "enabled" flag in the
MIRROR destination configuration.
// See http://camel.apache.org/wire-tap.html
//
--
initialRoute =
initialRoute.wireTap(mirrorToURI).id(sourceRouteId + "_MIRROR");
}

//
-
// to(backupFileToURI, throughputFileToURI, sedaMainURI,
sedaAckURI)
//
-
initialRoute = initialRoute.to(firstToURIs)
.id(sourceRouteId + "_TO_FIRST_URIS")
.setProperty(Exchange.CHARSET_NAME,
ExpressionBuilder.constantExpression(charsetName));*



--
View this message in context: 
http://camel.465427.n5.nabble.com/Conditionally-omitting-a-portion-of-a-route-tp5770127p5770186.html
Sent from the Camel - Users mailing list archive at Nabble.com.

Please help critique my route ...

[SteveR]

I want to get better at Camel route design and would welcome any thoughts on
the below.

  Thanks, Steve

Here's what I need to implement:

- *FROM:* receive UDP datagrams in ISO-8859-1 format.  For this I use the
camel-netty component.

- *ACK:* Each received UDP datagram must be acknowledged by extracting a
cookie from the received datagram
  and formulating the ACK using this cookie. For this function, I have an
AckBackProcessor that
  implements org.apache.camel.Processor.

  To sent the acknowledgement, the AckBackProcessor is basically doing:

exchange.getOut().setBody(ackBack.toString());
exchange.getOut().setHeaders(exchange.getIn().getHeaders()); // Copy
headers from IN to OUT to propagate them.

-* TO:* Each received datagram must be optionally forwarded (i.e. mirrored)
to a secondary remote host via UDP.
  For this function, I'm currently using the EIP WireTap pattern with a
destination of netty:udp://host:port etc.
  Maybe I should be using SEDA, not sure?

- *TO: *Log each received UDP datagram to a local backup file using the
Camel log component and the underlying
  SLF4J/Log4j2 logging system.
  
- *TO:* Log each received UDP datagram to the local file system for
throughput logging using the Camel log component
  and the underlying SLF4J/Log4j2 logging system.  This is for the Camel
Throughput logger capability.
  
- *TO:* Implement the Camel DeadLetterChannel (use original message), with
the DLC queue being the local
  file system, and using the underlying SLF4J/Log4j2 logging system.  

- *TO:* Wrap each received UDP datagram in a custom JSON wrapper and send to
a Kafka topic using a synchronous producer.
  I use the camel-kafka component for this and I have an KafkaProcessor that
implements org.apache.camel.Processor.
  This processor handles the Kafka partitioning. I have another
PayloadWrapperProcessor that adds the customn JSON wrapper.
  
  
Here's basically what my route looks like.  Just want to be sure I'm on the
right track. Any feedback and/or
improvements is greatly appreciated!!!

from(fromURI)
.routeId(sourceRouteId)
.startupOrder(routeStartupOrder)
.setProperty(Exchange.CHARSET_NAME,
ExpressionBuilder.constantExpression("ISO-8859-1"))
.wireTap(mirrorToURI)
.id(sourceRouteId + "_MIRROR")
.to(firstToURIs) // to(backupFileToURI, throughputFileToURI, 
sedaMainURI)
.id(sourceRouteId + "_TO_FIRST_URIS");


// ---
// The second route (i.e. sedaMainURI --> secondToURI) has a route
// scoped error handler. It’s a Dead Letter Channel that will
// send failed messages to a log for subsequent re-processing.
// ---
from(sedaMainURI)
.routeId(sedaMainRouteId)
// -
// Add the route-scoped DeadLetterChannel error handler.
// -
.errorHandler(deadLetterChannelBuilder)
.threads(threadPoolSize /*poolSize*/, threadPoolSize * 2 
/*maxPoolSize*/)
.threadName("threads_" + sedaMainRouteId)
.callerRunsWhenRejected(true) // Hard-coded since we always 
want this
behavior!
// --
// This processor adds an MPLR header along with the
// received exchange body and converts to wrapper format.
// --
.process(payloadWrapperProcessor)
.id(payloadWrapperProcessorId)
// 
// This processor handles Kafka related processing.
// For example, determining the Kafka partitioning.
// 
.process(kafkaProcessor)
.id(kafkaProcessorId)
// ---
// Here we route to the final destination (e.g. Kafka)
// ---
.to(secondToURI)



--
View this message in context: 
http://camel.465427.n5.nabble.com/Please-help-critique-my-route-tp5770248.html
Sent from the Camel - Users mailing list archive at Nabble.com.

Camel multicast() with aggregationStrategy() ...

[SteveR]

I have a Camel route that consumes from *netty:udp* and uses the *multicast()
*DSL method to multicast to several different destinations (i.e. Camel
throughput logger, local backup file, a SEDA queue that produces to Kafka,
and another SEDA queue on a route that contains a UdpAckBack processor that
composes an ack for each received UDP datagram).   I'm having trouble
getting the UDP acks to be sent back to the UDP client. Yes, I know that you
don't typically acknowledge received UDP datagrams, but I have a requirement
to optionally do so. 

I'm trying to use an *AggregationStrategy* so that the processed exchange
from the multicast destination containing the UdpAckBack processor is sent
back as the ack to each of the received UDP datagrams.

Additionally, I have to honor an application configuration setting that says
whether or not to send UDP acks, so in some cases I'll need to multicast to
the above destinations, but with no UDP acks sent back to the UDP client.

So my question is, can an *AggregationStrategy* be configured so that it
either sends back the result of a particular multicast destination as the
UDP ack, and also be setup so that, when configured, no multicast response
is sent?

I'm kind of stuck with how to implement this kind of aggregation strategy. 
Maybe I could have the route that contains the UdpAckBack processor set a
custom Exchange header that could be examined in the *aggregate(Exchange
exchange1, Exchange exchange2) *method?

Any help is greatly appreciated!!!

  Thanks, Steve





--
View this message in context: 
http://camel.465427.n5.nabble.com/Camel-multicast-with-aggregationStrategy-tp5770505.html
Sent from the Camel - Users mailing list archive at Nabble.com.

camel:netty udpConnectionlessSending option and "Too many open files"

[SteveR]

I'm using Camel 2.15.2 and the *camel-netty* component and I have a Camel
route that mirrors UDP datagrams to a remote server via *netty:udp*. 

I saw that *camel-netty *supports the *udpConnectionlessSending *option, and
I thought it would be a good idea as it will not result in failures when the
remote server is not listening.

However, when I set *udpConnectionlessSendin=true* and then send lots of UDP
packets into the route, my Linux box quickly exhausts file descriptors and I
get the below java exception.

  Any thoughts appreciated,
  Thanks, Steve


*org.jboss.netty.channel.ChannelException: Failed to open a
DatagramChannel.*
at
org.jboss.netty.channel.socket.nio.NioDatagramChannel.openNonBlockingChannel(NioDatagramChannel.java:94)
~[netty-3.9.6.Final.jar:?]
at
org.jboss.netty.channel.socket.nio.NioDatagramChannel.(NioDatagramChannel.java:58)
~[netty-3.9.6.Final.jar:?]
at
org.jboss.netty.channel.socket.nio.NioDatagramChannelFactory.newChannel(NioDatagramChannelFactory.java:207)
~[netty-3.9.6.Final.jar:?]
at
org.jboss.netty.channel.socket.nio.NioDatagramChannelFactory.newChannel(NioDatagramChannelFactory.java:79)
~[netty-3.9.6.Final.jar:?]
at
org.jboss.netty.bootstrap.ConnectionlessBootstrap.bind(ConnectionlessBootstrap.java:184)
~[netty-3.9.6.Final.jar:?]
at
org.apache.camel.component.netty.NettyProducer.openConnection(NettyProducer.java:433)
~[camel-netty-2.15.2.jar:2.15.2]
at
org.apache.camel.component.netty.NettyProducer$NettyProducerPoolableObjectFactory.makeObject(NettyProducer.java:543)
~[camel-netty-2.15.2.jar:2.15.2]
at
org.apache.camel.component.netty.NettyProducer$NettyProducerPoolableObjectFactory.makeObject(NettyProducer.java:539)
~[camel-netty-2.15.2.jar:2.15.2]
at
org.apache.commons.pool.impl.GenericObjectPool.borrowObject(GenericObjectPool.java:1188)
~[commons-pool-1.6.jar:1.6]
at
org.apache.camel.component.netty.NettyProducer.process(NettyProducer.java:232)
~[camel-netty-2.15.2.jar:2.15.2]
at
org.apache.camel.processor.RedeliveryErrorHandler.process(RedeliveryErrorHandler.java:448)
[camel-core-2.15.2.jar:2.15.2]
at
org.apache.camel.processor.CamelInternalProcessor.process(CamelInternalProcessor.java:191)
[camel-core-2.15.2.jar:2.15.2]
at
org.apache.camel.util.AsyncProcessorHelper.process(AsyncProcessorHelper.java:109)
[camel-core-2.15.2.jar:2.15.2]
at
org.apache.camel.processor.DelegateAsyncProcessor.process(DelegateAsyncProcessor.java:87)
[camel-core-2.15.2.jar:2.15.2]
at
org.apache.camel.processor.WireTapProcessor$1.call(WireTapProcessor.java:119)
[camel-core-2.15.2.jar:2.15.2]
at
org.apache.camel.processor.WireTapProcessor$1.call(WireTapProcessor.java:113)
[camel-core-2.15.2.jar:2.15.2]
at 
java.util.concurrent.FutureTask.run(FutureTask.java:262) [?:1.7.0_75]
at
java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
[?:1.7.0_75]
at
java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
[?:1.7.0_75]
at java.lang.Thread.run(Thread.java:745) [?:1.7.0_75]
*Caused by: java.net.SocketException: Too many open files*
at sun.nio.ch.Net.socket0(Native Method) ~[?:1.7.0_75]
at sun.nio.ch.Net.socket(Net.java:423) ~[?:1.7.0_75]
at 
sun.nio.ch.DatagramChannelImpl.(DatagramChannelImpl.java:115)
~[?:1.7.0_75]
at
sun.nio.ch.SelectorProviderImpl.openDatagramChannel(SelectorProviderImpl.java:42)
~[?:1.7.0_75]
at 
java.nio.channels.DatagramChannel.open(DatagramChannel.java:146)
~[?:1.7.0_75]
at
org.jboss.netty.channel.socket.nio.NioDatagramChannel.openNonBlockingChannel(NioDatagramChannel.java:70)
~[netty-3.9.6.Final.jar:?]
... 19 more



--
View this message in context: 
http://camel.465427.n5.nabble.com/camel-netty-udpConnectionlessSending-option-and-Too-many-open-files-tp5770768.html
Sent from the Camel - Users mailing list archive at Nabble.com.

CamelCamel 2.15.2: ExchangeException - Cannot write response

[SteveR]

Hi:

I have a Camel route that consumes from *netty:udp*, optionally sends a UDP
response back to client (if requested), and produces to a Kafka topic.  I'm
getting the following exception which is causing netty to stop listening on
the configured UDP port.  Any thoughts greatly appreciated!!!

  Thanks, Steve

*The route:*


http://camel.apache.org/schema/spring";>


iso-8859-1








iso-8859-1







*The stack trace:*

[2015-09-04 21:45:00,098] ERROR [Camel (MPLR_DTC) thread #94 - WireTap]
Failed delivery for (MessageId:
ID-talend-es-edge-01-kiad0-s-mission-net-35868-1441312070176-0-196157897 on
ExchangeId:
ID-talend-es-edge-01-kiad0-s-mission-net-35868-1441312070176-0-196157898).
Exhausted after delivery attempt: 1 caught:
java.nio.channels.ClosedChannelException

Message History
   
---
RouteId  ProcessorId  Processor 
  
Elapsed (ms)
[ROUTE_ID_RAW_DTC_E] [ROUTE_ID_RAW_DTC_E]
[udp://talend-es-edge-01.kiad0.s.mission.net:62265  ] [ 
   
1]
[ROUTE_ID_RAW_DTC_E] [setProperty1  ] [setProperty[CamelCharsetName]

] [ 0]
[ROUTE_ID_RAW_DTC_E] [threads1  ] [threads  

] [ 1]
[ROUTE_ID_RAW_DTC_E] [ROUTE_ID_RAW_DTC_E]
[wireTap[netty:udp://abinitio-edge-02.kiad0.s.mission.net:62265?clientPi] [ 
   
0]

Exchange
   
---
Exchange[
Id 
ID-talend-es-edge-01-kiad0-s-mission-net-35868-1441312070176-0-196157898
ExchangePattern InOnly
Headers
{breadcrumbId=ID-talend-es-edge-01-kiad0-s-mission-net-35868-1441312070176-0-196157886,
CamelNettyChannelHandlerContext=org.jboss.netty.channel.DefaultChannelPipeline$DefaultChannelHandlerContext@3c6eb35b,
CamelNettyLocalAddress=/69.59.232.69:62265, CamelNettyMessageEvent=[id:
0xc66f8e56, /69.59.232.69:62265] RECEIVED: ÀË8mÄ- ½Uê ª from
/197.237.41.57:1036, CamelNettyRemoteAddress=/197.237.41.57:1036,
CamelRedelivered=false, CamelRedeliveryCounter=0}
BodyTypeString
Body   ÀË8mÄ-½Uêª
]

Stacktrace
   
---

java.nio.channels.ClosedChannelException
at
org.jboss.netty.channel.socket.nio.AbstractNioWorker.cleanUpWriteBuffer(AbstractNioWorker.java:433)
~[netty-3.9.6.Final.jar:?]
at
org.jboss.netty.channel.socket.nio.NioDatagramWorker.writeFromUserCode(NioDatagramWorker.java:212)
~[netty-3.9.6.Final.jar:?]
at
org.jboss.netty.channel.socket.nio.NioDatagramPipelineSink.eventSunk(NioDatagramPipelineSink.java:97)
~[netty-3.9.6.Final.jar:?]
at
org.jboss.netty.channel.DefaultChannelPipeline$DefaultChannelHandlerContext.sendDownstream(DefaultChannelPipeline.java:779)
~[netty-3.9.6.Final.jar:?]
at org.jboss.netty.channel.Channels.write(Channels.java:725)
~[netty-3.9.6.Final.jar:?]
at
org.jboss.netty.handler.codec.oneone.OneToOneEncoder.doEncode(OneToOneEncoder.java:71)
~[netty-3.9.6.Final.jar:?]
at
org.jboss.netty.handler.codec.oneone.OneToOneEncoder.handleDownstream(OneToOneEncoder.java:59)
~[netty-3.9.6.Final.jar:?]
at
org.jboss.netty.channel.DefaultChannelPipeline.sendDownstream(DefaultChannelPipeline.java:591)
~[netty-3.9.6.Final.jar:?]
at
org.jboss.netty.channel.DefaultChannelPipeline.sendDownstream(DefaultChannelPipeline.java:582)
~[netty-3.9.6.Final.jar:?]
at org.jboss.netty.channel.Channels.write(Channels.java:704)
~[netty-3.9.6.Final.jar:?]
at
org.jboss.netty.channel.AbstractChannel.write(AbstractChannel.java:252)
~[netty-3.9.6.Final.jar:?]
at
org.jboss.netty.channel.socket.nio.NioDatagramChannel.write(NioDatagramChannel.java:299)
~[netty-3.9.6.Final.jar:?]
at
org.apache.camel.component.netty.NettyHelper.writeBodyAsync(NettyHelper.java:93)
[camel-netty-2.15.2.jar:2.15.2]
at
org.apache.camel.component.netty.NettyProducer.process(NettyProducer.java:263)
[camel-netty-2.15.2.jar:2.15.2]
at
org.apache.camel.processor.RedeliveryErrorHandler.process(RedeliveryErrorHandler.java:448)
[camel-core-2.15.2.ja

Re: CamelCamel 2.15.2: ExchangeException - Cannot write response

[SteveR]

I've seen several stackoverflow posts about *java.net.SocketException:
Invalid argument* being related to and/or fixed by executing the JVM with
the *-Djava.net.preferIPv4Stack=true* option. I'm currently testing with
this to see if it helps.

I wonder if this is a known issue that would be fixed if I updated to Camel 
netty4    component?

This is a production issue, so it's critical that I get it addressed.

  Thanks, Steve



--
View this message in context: 
http://camel.465427.n5.nabble.com/CamelCamel-2-15-2-ExchangeException-Cannot-write-response-tp5771341p5771357.html
Sent from the Camel - Users mailing list archive at Nabble.com.

Camel Load Balancer EIP with TCP Endpoints

[SteveR]

I have a stand-alone Java application using Camel 2.15 and camel-netty4 which
is running on Linux RHEL6. I see that Camel supports a Load Balancer EIP and
I'm wondering if it would a good fit based on the following requirements for
my application:

*REQUIREMENTS:*

- Accept various incoming TCP connections (each TCP feed consists of
messages that start with  and end with ). So I'll be using a
DelimiterBasedFrameDecoder in my TcpServerPipelineFactory.
  
- No policing/rejection of incoming TCP connection requests is required,
just accept each one that attempts to connect to the well-known listen port
for this application.

- Filter out uninteresting messages. I'll be using a filter bean for this.

- Uniformly distribute, via TCP, the interesting messages to a set of 4 TCP
listeners
  while failing over in round-robin fashion if a particular server is
unreachable for any reason.

- Maintain the TCP connections to the reachable servers (i.e. don't
  establish/tear-down a TCP connection for each transmitted message).

I was thinking that the Camel Load Balancer EIP with a failover policy and
the roundRobin option would be a good choice.  I'm hoping that the 4 routes
to the TCP listeners will be established by netty4 and stay connected!

Below is my Java DSL psuedo-code for the routes that I've initially come up
with.

Any thoughts about whether or not this approach would work is greatly
appreciated.
I know there are lots of URI parameters to consider and I want to be sure
I'm not missing
any important ones. Also any ideas for making this application more robust,
fault tolerant,
and better-performing would be great!

Thanks, SteveR



from("netty4:tcp://host:port?serverInitializerFactory=#TCP_SERVER_PIPELINE_FACTORY")
.to("seda:FILTER_QUEUE");


from("seda:FILTER_QUEUE?size=10&concurrentConsumers=10&waitForTaskToComplete=Never&failIfNoConsumers=true&timeout=1)
.threads(threadPoolSize /*poolSize*/, threadPoolSize * 2 
/*maxPoolSize*/)
.filter()
.method(filterBeanName, "isInterestingMessage")
.to("seda:BALANCER_QUEUE");
.end();


from("seda:BALANCER_QUEUE?size=10&concurrentConsumers=10&waitForTaskToComplete=IfReplyExpected&failIfNoConsumers=true&timeout=1)
.loadBalance()
.failover(-1,/* Never give up (i.e. continuously try to 
failover) */,
  false, /* Do not inherit route's ErrorHandler 
*/,
  true   /* Operate in round-robin mode */)
.to("seda:TCP_LISTENER1", "seda:TCP_LISTENER2", 
"seda:TCP_LISTENER3",
"seda:TCP_LISTENER4")
.end();


from("seda:TCP_LISTENER1?size=10&concurrentConsumers=10&waitForTaskToComplete=Never&failIfNoConsumers=true&timeout=1")
.threads(threadPoolSize /*poolSize*/, threadPoolSize * 2 
/*maxPoolSize*/)
.to("netty4:tcp://host:port&options ...");


from("seda:TCP_LISTENER2?size=10&concurrentConsumers=10&waitForTaskToComplete=Never&failIfNoConsumers=true&timeout=1")
.threads(threadPoolSize /*poolSize*/, threadPoolSize * 2 
/*maxPoolSize*/)
.to("netty4:tcp://host:port&options ...");


from("seda:TCP_LISTENER3?size=10&concurrentConsumers=10&waitForTaskToComplete=Never&failIfNoConsumers=true&timeout=1")
.threads(threadPoolSize /*poolSize*/, threadPoolSize * 2 
/*maxPoolSize*/)
.to("netty4:tcp://host:port&options ...");


from("seda:TCP_LISTENER4?size=10&concurrentConsumers=10&waitForTaskToComplete=Never&failIfNoConsumers=true&timeout=1")
.threads(threadPoolSize /*poolSize*/, threadPoolSize * 2 
/*maxPoolSize*/)
.to("netty4:tcp://host:port&options ...");



--
View this message in context: 
http://camel.465427.n5.nabble.com/Camel-Load-Balancer-EIP-with-TCP-Endpoints-tp5772104.html
Sent from the Camel - Users mailing list archive at Nabble.com.

How to stop ERROR logging of ClosedChannelException

[SteveR]

I have a stand-alone Java 1.7 application using Camel 2.14 and camel-netty
which is running on Linux RHEL6.

For a route that consumes from *netty:udp* and also wireTaps to *netty:udp*
on a remote server, I get lots of *java.nio.channels.ClosedChannelException*
entries at ERROR level in my Camel *rootLog.txt*.

>From googling, I see that this issue was discussed and worked for netty-http
(i.e. * CAMEL-8289 <https://issues.apache.org/jira/browse/CAMEL-8289>  *). 
My question is for *netty3:udp* is this still an issue, and would it be
resolved by upgrading to *camel-netty4* and/or Java 1.8?

Any thoughts on how I can prevent logging of these exceptions?  Below is a
sample entry from my Camel *rootLog.txt*?


  Thanks, SteveR


[2015-09-29 00:03:51,869] ERROR [New I/O worker #15] Failed delivery for
(MessageId: ID-es-edge-01-klga1-s-mission-net-49458-1443469983294-0-23256320
on ExchangeId:
ID-es-edge-01-klga1-s-mission-net-49458-1443469983294-0-23256321). Exhausted
after delivery attempt: 1 caught: java.nio.channels.ClosedChannelException

Message History

---
RouteId  ProcessorId  Processor 
  
Elapsed (ms)
[ROUTE_ID_RAW_DTC_E] [ROUTE_ID_RAW_DTC_E]
[udp://es-edge-01.klga1.s.mission.net:62265  ] [
1]
[ROUTE_ID_RAW_DTC_E] [setProperty1  ] 
[setProperty[CamelCharsetName]
] [ 0]
[ROUTE_ID_RAW_DTC_E] [threads1  ] [threads  

] [ 0]
[ROUTE_ID_RAW_DTC_E] [ROUTE_ID_RAW_DTC_E]
[wireTap[netty:udp://edge-01.klga1.s.mission.net:62265?clientPi] [
0]

Exchange

---
Exchange[
Id 
ID-es-edge-01-klga1-s-mission-net-49458-1443469983294-0-23256321
ExchangePattern InOnly
Headers
{breadcrumbId=ID-es-edge-01-klga1-s-mission-net-49458-1443469983294-0-23256317,
CamelNettyChannelHandlerContext=org.jboss.netty.channel.DefaultChannelPipeline$DefaultChannelHandlerContext@56d172ef,
CamelNettyLocalAddress=/216.115.20.22:62265, CamelNettyMessageEvent=[id:
0x451aa7fb, /216.115.20.22:62265] RECEIVED: &p Ò[ V ÕR from
/98.117.35.216:1039, CamelNettyRemoteAddress=/98.117.35.216:1039,
CamelRedelivered=false, CamelRedeliveryCounter=0}
BodyTypeString
Body
]

Stacktrace

---

java.nio.channels.ClosedChannelException
at
org.jboss.netty.channel.socket.nio.AbstractNioWorker.cleanUpWriteBuffer(AbstractNioWorker.java:433)
[netty-3.9.6.Final.jar:?]
at
org.jboss.netty.channel.socket.nio.AbstractNioWorker.close(AbstractNioWorker.java:373)
[netty-3.9.6.Final.jar:?]
at
org.jboss.netty.channel.socket.nio.NioDatagramPipelineSink.eventSunk(NioDatagramPipelineSink.java:72)
[netty-3.9.6.Final.jar:?]
at
org.jboss.netty.channel.DefaultChannelPipeline$DefaultChannelHandlerContext.sendDownstream(DefaultChannelPipeline.java:779)
[netty-3.9.6.Final.jar:?]
at
org.jboss.netty.handler.codec.oneone.OneToOneEncoder.handleDownstream(OneToOneEncoder.java:54)
[netty-3.9.6.Final.jar:?]
at
org.jboss.netty.channel.DefaultChannelPipeline.sendDownstream(DefaultChannelPipeline.java:591)
[netty-3.9.6.Final.jar:?]
at
org.jboss.netty.channel.DefaultChannelPipeline.sendDownstream(DefaultChannelPipeline.java:582)
[netty-3.9.6.Final.jar:?]
at 
org.jboss.netty.channel.Channels.close(Channels.java:812)
[netty-3.9.6.Final.jar:?]
at
org.jboss.netty.channel.AbstractChannel.close(AbstractChannel.java:197)
[netty-3.9.6.Final.jar:?]
at
org.apache.camel.component.netty.NettyHelper.close(NettyHelper.java:114)
[camel-netty-2.15.2.jar:2.15.2]
at
org.apache.camel.component.netty.NettyProducer$NettyProducerPoolableObjectFactory.destroyObject(NettyProducer.java:553)
[camel-netty-2.15.2.jar:2.15.2]
at
org.apache.camel.component.netty.NettyProducer$NettyProducerPoolableObjectFactory.destroyObject(NettyProducer.java:539)
[camel-netty-2.15.2.jar:2.15.2]
at
org.apache.commons.pool.impl.GenericObjectPool.addObjectToPool(GenericObjectPool.java:1466)
[commons-poo

camel:netty4: How to control the number of outgoing TCP connections

[SteveR]

I'm using *Camel 2.15.2* and *camel:netty4* on RHEL6 and I have a Camel route
that reads from a *SEDA *queue and sends via *netty4:tcp *to a remote host. 
When doing a load test in dev, I see that exactly 50 TCP connections get
established to the remote host. I would like to understand what controls how
many TCP connections that *camel-netty4 *attempts to establish to the remote
host.  I wonder if the remote host's *backlog *parameter is what limits it
to 50? But I'm more interested in the *camel-netty4* behavior, since it's
initiating the TCP connection attempts.

  Thanks, SteveR

Here's what the route looks like:


http://camel.apache.org/schema/spring";>




iso-8859-1




... and here's, what I see with *netstat *after starting the load test:

Proto Recv-Q Send-Q Local Address   Foreign Address 
   
State  
tcp0  0 0.0.0.0:7993 0.0.0.0:*  

LISTEN  
tcp0  0 10.262.134.26:8093  0.0.0.0:*   
   
LISTEN  
tcp0  0 10.262.134.26:9006  0.0.0.0:*   
   
LISTEN  
tcp0  0 10.262.172.26:58731 10.262.153.73:9006  
   
ESTABLISHED 
tcp0  0 10.262.172.26:58865 10.262.153.73:9006  
   
ESTABLISHED 
tcp0  0 10.262.172.26:58882 10.262.153.73:9006  
   
ESTABLISHED 
tcp0  0 10.262.172.26:58875 10.262.153.73:9006  
   
ESTABLISHED 
tcp0  0 10.262.134.26:9006  10.262.160.24:44516 
   
ESTABLISHED 
tcp0  0 10.262.172.26:58877 10.262.153.73:9006  
   
ESTABLISHED 
tcp0  0 10.262.172.26:58871 10.262.153.73:9006  
   
ESTABLISHED 
tcp0  0 10.262.172.26:58894 10.262.153.73:9006  
   
ESTABLISHED 
tcp0  0 10.262.172.26:58908 10.262.153.73:9006  
   
ESTABLISHED 
tcp0  0 10.262.172.26:58906 10.262.153.73:9006  
   
ESTABLISHED 
tcp0  0 10.262.172.26:58876 10.262.153.73:9006  
   
ESTABLISHED 
tcp0  0 10.262.172.26:58904 10.262.153.73:9006  
   
ESTABLISHED 
tcp0  0 10.262.172.26:58862 10.262.153.73:9006  
   
ESTABLISHED 
tcp0  0 10.262.172.26:58896 10.262.153.73:9006  
   
ESTABLISHED 
tcp0  0 10.262.172.26:58863 10.262.153.73:9006  
   
ESTABLISHED 
tcp0  0 10.262.172.26:58737 10.262.153.73:9006  
   
ESTABLISHED 
tcp0  0 10.262.172.26:58881 10.262.153.73:9006  
   
ESTABLISHED 
tcp0  0 10.262.172.26:58867 10.262.153.73:9006  
   
ESTABLISHED 
tcp0  0 10.262.134.26:9006  10.262.160.24:44514 
   
ESTABLISHED 
tcp0  0 10.262.172.26:58902 10.262.153.73:9006  
   
ESTABLISHED 
tcp0  0 10.262.172.26:5 10.262.153.73:9006  
   
ESTABLISHED 
tcp0  0 10.262.172.26:58878 10.262.153.73:9006  
   
ESTABLISHED 
tcp0  0 10.262.172.26:58870 10.262.153.73:9006  
   
ESTABLISHED 
tcp0  0 10.262.172.26:58884 10.262.153.73:9006  
   
ESTABLISHED 
tcp0  0 10.262.172.26:58864 10.262.153.73:9006  
   
ESTABLISHED 
tcp0  0 10.262.134.26:9006  10.262.160.24:44515 
   
ESTABLISHED 
tcp0  0 10.262.172.26:58879 10.262.153.73:9006  
   
ESTABLISHED 
tcp0  0 10.262.172.26:58869 10.262.153.73:9006  
   
ESTABLISHED 
tcp0  0 10.262.172.26:58868 10.262.153.73:9006  
   
ESTABLISHED 
tcp0  0 10.262.172.26:58909 10.262.153.73:9006  
   
ESTABLISHED 
tcp0  0 10.262.172.26:58889 10.262.153.73:9006  
   
ESTABLISHED 
tcp0  0 10.262.172.26:58861 10.262.153.73:9006  
   
ESTABLISHED 
tcp0  0 10.262.172.26:58886 10.262.153.73:9006  
   
ESTABLISHED 
tcp0  0 10.262.172.26:58905 10.262.153.73:9006  
   
ESTABLISHED 
tcp0  0 10.262.172.26:58735 10.262.153.73:9006  
   
ESTABLISHED 
tcp0  0 10.262.172.26:58887 10.262.153.73:9006  
   
ESTABLISHED 
tcp0  0 10.262.172.26:58901 10.262.153.73:9006  
   
ESTABLISHED 
tcp0  0 10.262.1

Re: Cannot Auto Create Component - Netty4

[SteveR]

Roger:

I'm new to Camel as well.  I use a Maven Java project, so I just ensure that
the Maven dependencies have the correct version of camel-netty4.  I also use
the Java DSL rather that spring.

However, did you see this stackoverflow post entitled: *Camel netty
component example doesn't work, within Spring MVC*?  It looks like an issue
similar to yours.

  SteveR





--
View this message in context: 
http://camel.465427.n5.nabble.com/Cannot-Auto-Create-Component-Netty4-tp5772149p5772252.html
Sent from the Camel - Users mailing list archive at Nabble.com.

Re: Cannot Auto Create Component - Netty4

[SteveR]

Roger:

Yes, I do see camel-netty4-2.15.2.jar in my Java classpath.

My Maven *pom.xml* dependency looks like this, where *camelCoreVersion *is
2.15.2:



org.apache.camel
camel-netty4
${camelCoreVersion}



SteveR



--
View this message in context: 
http://camel.465427.n5.nabble.com/Cannot-Auto-Create-Component-Netty4-tp5772149p5772286.html
Sent from the Camel - Users mailing list archive at Nabble.com.

Re: Camel Load Balancer EIP with TCP Endpoints

[SteveR]

CleoCleo:

Thanks for the reply.  These are long-standing TCP connections, meaning that
the client that connects to my Camel application establishes the connection
and in the "sunny-day" case continues to send a continual stream of messages
each of which are delimited by the  ...  tags. The message rate
is subject to busy-hour patterns, so it's high at certain peaks hours and
could slow to a trickle at non-peak hours.

For the load-balancing portion in the out-going direction, my Camel
application would also typically have long-standing TCP connections to each
recipient host.

  Thanks, SteveR





--
View this message in context: 
http://camel.465427.n5.nabble.com/Camel-Load-Balancer-EIP-with-TCP-Endpoints-tp5772104p5772314.html
Sent from the Camel - Users mailing list archive at Nabble.com.

camel-netty: How to set the netty closeChannelTimeMillis option?

[SteveR]

I have a stand-alone Java 1.7 application using *Camel 2.15.2* and the
*camel-netty* component which is running on Linux RHEL6.

In production, with a camel route that consumes via *netty:udp*, we appear
to be facing an issue similar to this old one:  ChannelPool closing channels
prematurely <https://linkedin.jira.com/plugins/servlet/mobile#issue/NOR-19>  

*QUESTION:* When using *camel:netty*, is there a way to set netty's
*closeChannelTimeMillis* option?, I don't see it listed on the  camel-netty
<http://camel.apache.org/netty.html>  page. I'm hoping that there might be
an undocumented or back-door way to set it.

Thanks, SteveR







--
View this message in context: 
http://camel.465427.n5.nabble.com/camel-netty-How-to-set-the-netty-closeChannelTimeMillis-option-tp5772342.html
Sent from the Camel - Users mailing list archive at Nabble.com.

Re: camel-netty: How to set the netty closeChannelTimeMillis option?

[SteveR]

On further investigation, I see that the *closeChannelTimeMillis* option is
not a *netty *option, but rather an option for 
com.linkedin.norbert.javacompat.network.NetworkClientConfig
<http://grepcode.com/file/repo1.maven.org/maven2/com.linkedin/norbert_2.8.1/0.6.33/com/linkedin/norbert/javacompat/network/NetworkClientConfig.java>
 
.

The  Norbert ChannelPool closing channels prematurely
<https://linkedin.jira.com/browse/NOR-19>   JIRA issue is still in the
unresolved state.

We are are facing a similar issue in production (with very similar stack
traces for Norbert scenarios 2 and 3, as shown in the JIRA issue) and were
hoping for a resolution to this similar issue.

Any thoughts on this are appreciated.

  Thanks, SteveR



--
View this message in context: 
http://camel.465427.n5.nabble.com/camel-netty-How-to-set-the-netty-closeChannelTimeMillis-option-tp5772342p5772347.html
Sent from the Camel - Users mailing list archive at Nabble.com.

Re: camel-netty: How to set the netty closeChannelTimeMillis option?

[SteveR]

Hi Willem:

Thanks for taking the time to reply.  Actually, we have not been specifying
the camel-netty *udpConnectionlessSending=true *option, and I see that the
default is *false* if not explicitly specified. 

We were actually contemplating to set *udpConnectionlessSending=true*,
thinking that it might help.

The Camel route in question is shown below. The issue we are having in
production, is that we see a flooding of the below ERROR messages in our
Camel *rootLog.txt* which indicates a problem with *ClosedChannelException
*when trying to mirror the received UDP packets to a remote host via the
Camel wireTap() shown in the route below.  Note also, that the remote host
attempts to ACKknowledge each received UDP packet that this Camel route
mirrors to it.  So I hope those ACKs do not cause any strange side effects
to the Camel route.  We're stuck on exactly how to go about debugging this
issue.  So the wireTap portion of the route seems to mostly be failing. 
However, some UDP packets do get through to the remote host.

Any thoughts are greatly appreciated.

  Thanks, SteveR


http://camel.apache.org/schema/spring";>


iso-8859-1








iso-8859-1





[2015-10-07 17:47:26,958] ERROR [Camel (MPLR_CQMS) thread #55 - WireTap]
Failed delivery for (MessageId:
ID-talend-es-edge-01-kewr0-s-missionnetworks-net-45813-1444160612354-0-21342281
on ExchangeId:
ID-talend-es-edge-01-kewr0-s-missionnetworks-net-45813-1444160612354-0-21342282).
Exhausted after delivery attempt: 1 caught:
java.nio.channels.ClosedChannelException

Message History

---
RouteId  ProcessorId  Processor 
  
Elapsed (ms)
[ROUTE_ID_RAW_CQMS_] [ROUTE_ID_RAW_CQMS_]
[udp://talend-es-edge-01.kewr0.s.missionnetworks.net:1 
] [ 1]
[ROUTE_ID_RAW_CQMS_] [setProperty1  ] 
[setProperty[CamelCharsetName]
] [ 0]
[ROUTE_ID_RAW_CQMS_] [threads1  ] [threads  

] [ 1]
[ROUTE_ID_RAW_CQMS_] [ROUTE_ID_RAW_CQMS_]
[wireTap[netty:udp://abinitio-edge-01.kewr0.s.missionnetworks.net:1?clientPi]
[ 0]

Exchange

---
Exchange[
Id 
ID-talend-es-edge-01-kewr0-s-missionnetworks-net-45813-1444160612354-0-21342282
ExchangePattern InOnly
Headers
{breadcrumbId=ID-talend-es-edge-01-kewr0-s-missionnetworks-net-45813-1444160612354-0-21342277,
CamelNettyChannelHandlerContext=org.jboss.netty.channel.DefaultChannelPipeline$DefaultChannelHandlerContext@5efa3b3c,
CamelNettyLocalAddress=/69.59.252.44:1, CamelNettyMessageEvent=[id:
0x5612eec3, /69.59.252.44:1] RECEIVED: {   "protocol":"MQDP-2",  
"CQM_STREAM_INFO":  "P_RTCP_XRVM_COUNT":10}   ]   } } from
/69.59.248.171:, CamelNettyRemoteAddress=/69.59.248.171:,
CamelRedelivered=false, CamelRedeliveryCounter=0}
BodyTypeString
Body{  "protocol":"MQDP-2",  
"CQM_STREAM_INFO": 
"P_RTCP_XRVM_COUNT":10}  ]  }}
]

Stacktrace

---

java.nio.channels.ClosedChannelException
at
org.jboss.netty.channel.socket.nio.AbstractNioWorker.cleanUpWriteBuffer(AbstractNioWorker.java:433)
~[netty-3.9.6.Final.jar:?]
at
org.jboss.netty.channel.socket.nio.NioDatagramWorker.writeFromUserCode(NioDatagramWorker.java:212)
~[netty-3.9.6.Final.jar:?]
at
org.jboss.netty.channel.socket.nio.NioDatagramPipelineSink.eventSunk(NioDatagramPipelineSink.java:97)
~[netty-3.9.6.Final.jar:?]
at
org.jboss.netty.channel.DefaultChannelPipeline$DefaultChannelHandlerContext.sendDownstream(DefaultChannelPipeline.java:779)
~[netty-3.9.6.Final.jar:?]
at 
org.jboss.netty.channel.Channels.write(Channels.java:725)
~[netty-3.9.6.Final.jar:?]
at
org.jboss.netty.h

Upgrading to camel-netty4: How to migrate my UdpPacketDecoder?

[SteveR]

I'm upgrading from *camel-netty* to *camel-netty4* and I see there are lots
of API changes.  Currently, with *camel-netty*, I have a route that consumes
from *netty:udp* and a *ServerPipelineFactory *that has a *UdpPacketDecoder
*that is working fine.  The idea is to detect and fire a message upstream
when netty has received each UDP datagram.  My *camel-netty* version of the
*UdpPacketDecoder* is shown below.  The idea for this *UdpPacketDecoder*
came from  Nicholas Hagen's Netty: Using Handlers
<http://www.znetdevelopment.com/blogs/2009/04/21/netty-using-handlers/>  
blog post.

I'm trying to understand how to migrate my *UdpPacketDecoder* to
*camel-netty4*?  I've just purchased and downloaded *Netty in Action Version
11* and will certainly be reading Chapter 6 first!  Also shown below is
where I'm currently at wrt my *UdpPacketDecoder* migration to
*camel-netty4*.

The basic idea of my *camel-netty* version of *UdpPacketDecoder* is to
invoke *Channels.fireMessageReceived(ctx, message, me.getRemoteAddress())*
when the received *MessageEvent* is an instance of a *ChannelBuffer* to send
it upstream to be treated as a separate Camel Exchange, else invoke
*ctx.sendUpstream(ce)*.

Any thoughts on how to finish this *UdpPacketDecoder* migration to
*camel-netty4* is much appreciated. 

  Thanks, SteveR

*camel-netty3 version:*

package com.mission.mplr.multiprotocollistenerrouter;

import org.jboss.netty.buffer.ChannelBuffer;
import org.jboss.netty.buffer.ChannelBuffers;
import org.jboss.netty.channel.ChannelEvent;
import org.jboss.netty.channel.ChannelHandlerContext;
import org.jboss.netty.channel.ChannelUpstreamHandler;
import org.jboss.netty.channel.Channels;
import org.jboss.netty.channel.MessageEvent;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Upstream handler to grab each UDP datagram and fire it
 * upstream to be treated as a separate Camel Exchange.
 * @see
http://www.znetdevelopment.com/blogs/2009/04/21/netty-using-handlers/
 */
public class UdpPacketDecoder_ORIG implements ChannelUpstreamHandler {
private final static Logger logger =
LoggerFactory.getLogger(UdpPacketDecoder_ORIG.class);

// -
// Upstream is incoming (i.e. from client to server)
// -
@Override
public void handleUpstream(ChannelHandlerContext ctx, 
ChannelEvent ce)
throws Exception {

if(!(ce instanceof MessageEvent))
{
// 
---
// We're only interested in a MessageEvent that 
represents
// a received UDP packet. Send this 
ChannelEvent to the
// ChannelUpstreamHandler which is placed in 
the closest
// upstream from the handler associated with 
this context.
// 
---
ctx.sendUpstream(ce);
logger.trace("handleUpstream(): EXIT: 
ChannelEvent is not a
MessageEvent");
return;
}

final MessageEvent me = (MessageEvent) ce;
if(!(me.getMessage() instanceof ChannelBuffer))
{
// 
---
// We're only interested in a MessageEvent that 
represents
// a received UDP packet. Send this 
ChannelEvent to the
// ChannelUpstreamHandler which is placed in 
the closest
// upstream from the handler associated with 
this context.
// 
---
ctx.sendUpstream(ce);
logger.trace("handleUpstream(): EXIT: 
MessageEvent is not a
ChannelBuffer");
return;
}

// -
// Process this MessageEvent as a ChannelBuffer and fire
// it upstream to next handler in the server pipeline
// -
final Object message = me.getMessage();
final ChannelBuffer buffer = (ChannelBuffer) message;

   

camel-netty4: MessageToMessageDecoder gives only first 2048 octets

[SteveR]

I'm using *Camel 2.15.3* and *camel-netty4*, and since upgrading from
*camel-netty3*, I'm having problems receiving full JSON messages via UDP. 
Each JSON message is about 3 to 5 kbytes, but my
*MessageToMessageDecoder* implementation is only giving me
the first 2048 (i.e. 2k bytes).  From a test program, I send in one UDP
message, and from my debug prints within my
*MessageToMessageDecoder* it shows that the *decode()*
method is only called once.

I'm currently reading through *Netty In Action*, but i see this in my log
file: *UnpooledUnsafeDirectByteBuf(ridx: 0, widx: 2048, cap: 2048))*

I desperately need to get this fixed in production, and just need to be able
to 

My route looks like the below. It consumes from *netty4:udp* and produces to
a SEDA queue, just for now while testing:


http://camel.apache.org/schema/spring";>


iso-8859-1





iso-8859-1





I print out the received *DatagramPacket*, which shows this: 
*UnpooledUnsafeDirectByteBuf(ridx: 0, widx: 2048, cap: 2048))*

*Here is my *MessageToMessageDecoder* implementation:*

package com.mission.mplr.multiprotocollistenerrouter;

import com.vonage.mplr.utils.MiscUtils;
import io.netty.channel.ChannelHandlerContext; // Represents the 
"binding"
between a ChannelHandler and the ChannelPipeline.
import io.netty.channel.socket.DatagramPacket;
import io.netty.handler.codec.MessageToMessageDecoder;
import java.nio.charset.Charset;
import java.util.List;
import org.slf4j.Logger;// The org.slf4j.Logger interface is the
main user entry point of SLF4J API.
import org.slf4j.LoggerFactory; // Utility class producing Loggers for
various logging APIs, most notably for log4j.


public class UdpDatagramDecoder extends
MessageToMessageDecoder {
private static final Logger logger  =
LoggerFactory.getLogger(UdpDatagramDecoder.class);
private static final Logger errorLogger =
LoggerFactory.getLogger("ERROR_LOGGER");
private final String CHARSET_NAME;

UdpDatagramDecoder(String charsetName) {
this.CHARSET_NAME = charsetName;
}

@Override

public boolean acceptInboundMessage(Object msg) throws 
Exception {
return true;
}

@Override
protected void decode(ChannelHandlerContext chc, DatagramPacket 
packet,
List out) throws Exception {
logger.info("decode(): ENTER");

logger.info("decode(): Received datagram = {}", packet);

String packetAsString =
packet.content().toString(Charset.forName(CHARSET_NAME));

if(packetAsString == null) {
return; // Nothing to do
} else {
out.add(packetAsString);
packet.retain();
}

logger.info("decode(): bodyBytesAsString[size={}] = {}",
packetAsString.length(), packetAsString);

String bodyBytesAsHex = 
MiscUtils.stringAsHex(packetAsString,
CHARSET_NAME);
logger.info("decode(): bodyBytesAsHex[size={}] = {}",
bodyBytesAsHex.length(), bodyBytesAsHex);

logger.info("decode(): EXIT");
}
}
// - end --

*My server pipeline has this initChannel() implementation:*

@Override
protected void initChannel(Channel ch) throws Exception {
logger.trace("initChannel(): ENTER");

ChannelPipeline channelPipeline = ch.pipeline();
serverInvoked = true;  

String theSourceRouteId = consumer.getRoute().getId();
logger.debug("initChannel(): consumer = {}, theSourceRouteId = {}",
consumer.toString(), theSourceRouteId);

//
---
// Here we add the custom UDP datagram decoder. Decoders are
typically
// stateful, thus we create a new instance with every pipeline.
//
---
String udpPacketDecoderName = "CQMS_UDP_DATAGRAM_DECODER_" +
theSourceRouteId;
logger.debug("initChannel(): Adding {}", udpPacketDecoderName);
channelPipeline.addLast(udpPacketDecoderName, new
UdpDatagramDecoder(CHARSET_NAME));

//
-
// Default Camel ServerChannelHandler for the consumer, to allow
Camel to route the message.
//
---

Camel 2.15.3: multicast with aggregationStrategy() not working ...

[SteveR]

I'm using Camel 2.15.3 with a route that reads from *netty4:udp* and
multicasts to two SEDA queues and uses *aggregationStrategy()* to choose the
correct response to send back to the client as the UDP reply.

I can see the correct reply coming into the aggregate() method, but setting
the *Exchange.AGGREGATION_COMPLETE_CURRENT_GROUP* property on the returned
exchange does not stop the aggregation.  Am I doing something wrong, or do
we know of any bugs in the area?


Thanks, SteveR

@Override
public Exchange aggregate(Exchange oldExchange, Exchange newExchange) {

String newBody = newExchange.getIn().getBody(String.class);
if(newBody.contains(CQMS_ACK_SUBSTRING)) {
   
newExchange.getOut().setBody(newExchange.getIn().getBody(String.class));
   
newExchange.getOut().setHeaders(newExchange.getIn().getHeaders());
   
newExchange.setProperty(Exchange.AGGREGATION_COMPLETE_CURRENT_GROUP, true);
return newExchange;
}
return newExchange;
}


from(fromURI)
.setExchangePattern(ExchangePattern.InOut)
.routeId(sourceRouteId)
.setProperty(Exchange.CHARSET_NAME,
ExpressionBuilder.constantExpression(charsetName))
.threads(threadPoolSize /*poolSize*/, threadPoolSize * 2 
/*maxPoolSize*/)
.threadName("threads_" + sourceRouteId)
.callerRunsWhenRejected(true); // Hard-coded since we always 
want this
behavior!
.multicast()
.parallelProcessing()
.aggregationStrategy(new 
CqmsAckBackAggregationStrategy(ackBackRequired,
charsetName))
.to(firstToURIs);  



--
View this message in context: 
http://camel.465427.n5.nabble.com/Camel-2-15-3-multicast-with-aggregationStrategy-not-working-tp5772978.html
Sent from the Camel - Users mailing list archive at Nabble.com.

Camel 2.15.3: multicast (default is take the latest reply?)

[SteveR]

>From  Apache Camel Multicast Examples
<http://javarticles.com/2015/05/apache-camel-multicast-examples.html>   it
states: /"The final output from the multicast is the the latest reply
message and it discards any earlier replies. If you want a different
aggregation strategy to include even other reply messages, you need to
create your own AggregationStrategy."/

I have a route with a multicast() to two SEDA queues and I am currently
using an aggregationStrategy() to choose the correct response to send back
to the client as the UDP reply.  But I'm unable to get it to work!

My question is, in the above statement, does *latest reply* mean the
chronologically last reply received, or that the multicast will wait for the
reply from the last multicast destination specified in the associated route?

If I know, a priori, which SEDA queue will produce the correct reply, what
is the easiest way to configure a multicast with respect to aggregation? 
I'm hoping, in thissimple case of multicasting to only two SEDA queues that
I can find a way that doesn't require a custom aggregationStrategy.

  Thanks, SteveR

Maybe something like this:

from(netty4:udp ...)
.setExchangePattern(ExchangePattern.InOut)
.multicast()
.parallelProcessing()
.to(firstSEDAUri)
.end()
.to(secondSEDAUri)  // <--- This route will return the reply 
that needs to
be sent back to the client via netty4:udp.
.end();



--
View this message in context: 
http://camel.465427.n5.nabble.com/Camel-2-15-3-multicast-default-is-take-the-latest-reply-tp5772979.html
Sent from the Camel - Users mailing list archive at Nabble.com.

Camel netty: How to send UDP reply based on NETTY_REMOTE_ADDRESS?

[SteveR]

Hi:

I have a Linux RedHat-6 application using *Camel 2.15.2* and *netty3* and I
need to reply to various UDP clients that send UDP requests into my Camel
route.  However, I can't do a blind-ACK. That is, I have to first process
each received UDP request, synchronously produce it to a remote Kafka
cluster, and only when the Exchange is successfully produced to Kafka, then
formulate the associated UDP reply and send it back to the associated UDP
client.

To do this my input route consumes via *netty:udp* on a well-known listen
port, and I then place all received exchanges on a *SEDA_KAFKA_QUEUE* that
consumes them and produces to Kafka.  I'm using the Camel
*onCompletion.onCompleteOnly()* mechanism in this Kafka route to trigger
sending the successfully processed Exchanges to a *SEDA_UDP_ACK_QUEUE* that
will use a processor to formulate the ACK body and then send it as the UDP
reply via the Camel netty component.

I can only get this to work and actually send the ACK if I use a hard-coded
port number in the netty URI. But in production, I'll be receiving from many
different UDP clients and I need to send back to their ephemeral port
number.  I'm looking for a way to parameterize the sending of the UDP
replies back using the *host:port* combo that is specified in the
*CamelNettyRemoteAddress* header of the Exchange.

Any thoughts greatly appreciated.  Below are my routes.

  Thanks, SteveR

 
from("netty:udp://dloco.m.mission.net:1?serverPipelineFactory=#MY_SERVER_PIPELINE_FACTORY&keepAlive=true&sync=true&orderedThreadPoolExecutor=false&receiveBufferSize=26214400&sendBufferSize=26214400&allowDefaultCodec=false&disconnectOnNoReply=false&receiveBufferSizePredictor=8192")
.setExchangePattern(ExchangePattern.InOnly)
.routeId(sourceRouteId)
.startupOrder(routeStartupOrder)
.setProperty(Exchange.CHARSET_NAME,
ExpressionBuilder.constantExpression(charsetName))
.threads(threadPoolSize /*poolSize*/, threadPoolSize * 2
/*maxPoolSize*/).threadName("threads_" + sourceRouteId);
.to(kafkaQueueURI).id(sourceRouteId + 
"_TO_KAFKA_QUEUE");

// 
// SEDA_KAFKA_QUEUE
// 
from(kafkaQueueURI)
.errorHandler(deadLetterChannelBuilder) // Add 
route-scoped
DeadLetterChannel error handler
.onCompletion()
.onCompleteOnly() // Synchronize only after 
Exchange completes
successfully with no errors
.parallelProcessing() // Tells Camel to use a 
thread pool for
onCompletion route
.to(ackQueueURI).id(sourceRouteId + 
"_ON_COMPLETION_ONLY_TO_ACK_QUEUE")
.end() // Must use end() to denote the end of the 
onCompletion route
.setExchangePattern(ExchangePattern.InOnly)
.routeId(kafkaQueueRouteId)
.startupOrder(kafkaQueueRouteStartupOrder)
.setProperty(Exchange.CHARSET_NAME,
ExpressionBuilder.constantExpression(charsetName))
.threads(threadPoolSize /*poolSize*/, threadPoolSize * 2
/*maxPoolSize*/).threadName("threads_" + kafkaQueueRouteId)
// 
// This processor handles Kafka related processing.
// For example, determining the Kafka partitioning.
// 
.process(kafkaProcessor)
.id(kafkaProcessorId)
// ---
// Here we route to the final destination (e.g. Kafka)
// ---
.to(kafkaToURI);

// 
// SEDA_UDP_ACK_QUEUE
// 
from(ackQueueURI)
.setExchangePattern(ExchangePattern.InOut)
.routeId(ackQueueRouteId)
.startupOrder(ackQueueRouteStartupOrder)
.setProperty(Exchange.CHARSET_NAME,
ExpressionBuilder.constantExpression(charsetName))
.process(ackBackProcessor)
.id(ackBackProcessorId)

.to("netty:udp://dloco.m.mission.net:9998?clientPipelineFactory=#MY_CLIENT_PIPELINE_FACTORY&sendBufferSize=26214400&allowDefaultCodec=false");




--
View this message in context: 
http://camel

Re: Camel netty: How to send UDP reply based on NETTY_REMOTE_ADDRESS?

[SteveR]

Below is what I found that seems to work, with one issue that I can't yet
figure out:

After sending some test data into my application, and when I subsequently
use *VisualVM* to examine this Camel JMX MBean route, it always shows all
exchanges as *InflightExchanges* (i.e. the exchanges never show as
CompletedExchanges).

Any thoughts on how to debug this issue?

  Thanks, SteveR


// I extract the remote address in my processor, get the remote host
and remote port,
// and set them as input headers on the exchange.

Message inMsg = exchange.getIn();
String remoteAddress =
inMsg.getHeader(NettyConstants.NETTY_REMOTE_ADDRESS).toString();
String[] remoteAddressParts = remoteAddress.split(":");
String remoteHost = remoteAddressParts[0].replace("/", "");
String remotePort = remoteAddressParts[1];
inMsg.setHeader("REMOTE_HOST_IP", remoteHost);
inMsg.setHeader("REMOTE_PORT_NUMBER", remotePort);


// This in my route, I use recipientList EIP and the simple language
to dynamically add them to the netty URI.

from(ackQueueURI)
.setExchangePattern(ExchangePattern.InOnly)
.routeId(ackQueueRouteId)
.startupOrder(ackQueueRouteStartupOrder)
.setProperty(Exchange.CHARSET_NAME,
ExpressionBuilder.constantExpression(charsetName))
.threads(threadPoolSize /*poolSize*/, threadPoolSize * 2 
/*maxPoolSize*/)
.threadName("threads_" + sourceRouteId + "_ACK_QUEUE")
.process(cqmsAckBackProcessor)
.id(cqmsAckBackProcessorId)
.recipientList( // See
http://camel.apache.org/how-do-i-use-dynamic-uri-in-to.html
simple(
"netty:udp://${header.REMOTE_HOST_IP}:" +
"${header.REMOTE_PORT_NUMBER}?" +

"clientPipelineFactory=#CLIENT_PIPELINE_FACTORY_ROUTE_ID_RAW_CQMS_EVENTS&" +

"sendBufferSize=26214400&allowDefaultCodec=false"
)
)
.end();



--
View this message in context: 
http://camel.465427.n5.nabble.com/Camel-netty-How-to-send-UDP-reply-based-on-NETTY-REMOTE-ADDRESS-tp5773044p5773077.html
Sent from the Camel - Users mailing list archive at Nabble.com.

Re: NEED HELP: Exchanges remain inflight on route with recipientList() or toD()

[SteveR]

By the way, on the ack-back route in question that sends the UDP
acknowledgement via *netty:udp*, I've tried with and without the
*udpConnectionlessSending *option and neither way works.



--
View this message in context: 
http://camel.465427.n5.nabble.com/NEED-HELP-Exchanges-remain-inflight-on-route-with-recipientList-or-toD-tp5773079p5773082.html
Sent from the Camel - Users mailing list archive at Nabble.com.

Re: NEED HELP: Exchanges remain inflight on route with recipientList() or toD()

[SteveR]

This problem is severe enough to cause my Camel application to exhaust
resources (e.g. number of open files) which results in major application
failure.  Below are some of the errors and stack trace that I get when
sending in only 4000 messages using a Java UDP sender test program.

I've worked pretty hard, so far, trying to learn Camel and standing up this
UDP routing application. Everything else seems to be working just fine (e.g.
UDP receive, producing to kafka, creating backup files, throughput logging,
etc).  This route that deals with sending a UDP acknowledgement is the only
part that is causing issues. I would hate to give up now over an issue like
this, but I'm running out of ideas and my management is getting impatient
and a jaded view of Apache Camel.  It was my idea to use Camel, so I'm
trying real hard to make it work successfully.

   Thanks, SteveR

[2015-10-26 21:08:05,451 UTC] *ERROR *[Camel (MPLR_CQMS) thread #40 -
seda://SEDA_ACK_103] [FailureEventNotifier:notify:63]: Exchange failed:
exchangeId =
ID-talend-es-01-dloco-m-mission-net-32956-1445893522490-0-51974, fromRouteId
= ROUTE_ID_RAW_CQMS_EVENTS_SEDA_ACK, route =
EventDrivenConsumerRoute[Endpoint[seda://SEDA_ACK_103?concurrentConsumers=10&size=20]
-> Pipeline[[Channel[SetExchangePattern[InOnly]],
Channel[SetProperty(CamelCharsetName, iso-8859-1)],
Channel[DelegateSync[com.mission.mplr.multiprotocollistenerrouter.CqmsAckBackProcessor@3563739]],
Channel[sendTo(netty:udp://${header.REMOTE_HOST_IP}:${header.REMOTE_PORT_NUMBER}?clientPipelineFactory=#CQMS_CLIENT_PIPELINE_FACTORY_ROUTE_ID_RAW_CQMS_EVENTS&sendBufferSize=26214400&allowDefaultCodec=false)],
Channel[Stop, *exceptionMsg = Failed to open a DatagramChannel.*


[2015-10-26 21:09:37,663] *WARN 
*[qtp2066624561-39-acceptor-0@1b7d21a5-ServerConnector@5dcdfc98{HTTP/1.1}{talend-es-01.dloco.m.mission.net:8090}]
*java.io.IOException: Too many open files*
at sun.nio.ch.ServerSocketChannelImpl.accept0(Native 
Method)
~[?:1.7.0_75]
at
sun.nio.ch.ServerSocketChannelImpl.accept(ServerSocketChannelImpl.java:241)
~[?:1.7.0_75]
at
org.eclipse.jetty.server.ServerConnector.accept(ServerConnector.java:377)
~[jetty-server-9.2.11.v20150529.jar:9.2.11.v20150529]
at
org.eclipse.jetty.server.AbstractConnector$Acceptor.run(AbstractConnector.java:500)
[jetty-server-9.2.11.v20150529.jar:9.2.11.v20150529]
at
org.eclipse.jetty.util.thread.QueuedThreadPool.runJob(QueuedThreadPool.java:635)
[jetty-util-9.2.11.v20150529.jar:9.2.11.v20150529]
at
org.eclipse.jetty.util.thread.QueuedThreadPool$3.run(QueuedThreadPool.java:555)
[jetty-util-9.2.11.v20150529.jar:9.2.11.v20150529]
at java.lang.Thread.run(Thread.java:745) [?:1.7.0_75]



--
View this message in context: 
http://camel.465427.n5.nabble.com/NEED-HELP-Exchanges-remain-inflight-on-route-with-recipientList-or-toD-tp5773079p5773081.html
Sent from the Camel - Users mailing list archive at Nabble.com.

NEED HELP: Exchanges remain inflight on route with recipientList to netty:udp

[SteveR]

Hi:

*What I'm using:* Linux RHEL6, Camel 2.15.3, netty3, Java 1.7

I have to deliver my Camel application to production soon and I have a big
problem that I can't figure out.

I have a route that consumes from a *SEDA *queue, formulates a UDP
acknowledgement from the received exchange body, and then sends the ack back
to the associated UDP client specified in the
*NettyConstants.NETTY_REMOTE_ADDRESS* header on the exchange.

The UDP acks do get sent out, which I have verified via tcpdump, but when I
look at the Camel JMX route statistics (using VisualVM) it shows that all
exchanges flowing through this ack-back route are marked as
*ExchangesInflight* and remain that way.  If I run a long test, say sending
in a million messages, the application experiences
*java.net.SocketException: Too many open files.*

I'm including my route below.  By the way, I'm using the route-scoped
*onCompletion().onCompleteOnly()* approach within another route to send to
this ack-back route.

Any help on how to fix this is desperately needed and greatly appreciated!!!

I was also hoping that adding  .stop()
<http://camel.apache.org/maven/camel-2.15.0/camel-core/apidocs/org/apache/camel/model/ProcessorDefinition.html#stop%28%29>
  
to the route would force it to mark each exchange as completed, but it
doesn't appear to.

Also, the only reason I'm using the *recipientList* EIP is because of the
suggestion on the  How do I use dynamic URI
<http://camel.apache.org/how-do-i-use-dynamic-uri-in-to.html>   webpage. So
far, it's the only way I could figure out how to dynamically apply the
remote host and remote port to the netty URI.  I only need to send each ACK
to a single UDP client host, but I don't know of any other way to make it
work dynamically.  You would think that if there is a
*NettyConstants.NETTY_REMOTE_ADDRESS* header on the exchange that there
would be a way to tell *camel-netty* to just use that remote address.

  Thanks, SteveR


from(uri="seda:SEDA_ACK_QUEUE?size=20&concurrentConsumers=10&waitForTaskToComplete=Never&failIfNoConsumers=true&timeout=1000")
.setExchangePattern(ExchangePattern.InOnly)
.routeId(ackQueueRouteId)
.startupOrder(ackQueueRouteStartupOrder)
.setProperty(Exchange.CHARSET_NAME,
ExpressionBuilder.constantExpression(charsetName))
.threads(threadPoolSize /*poolSize*/, threadPoolSize * 2 
/*maxPoolSize*/)
.threadName("threads_" + sourceRouteId + "_ACK_QUEUE")
.process(cqmsAckBackProcessor).id(cqmsAckBackProcessorId)  // 
Formulate
ACK
.recipientList( // See
http://camel.apache.org/how-do-i-use-dynamic-uri-in-to.html
simple(
"netty:udp://${header.REMOTE_HOST_IP}:" +
"${header.REMOTE_PORT_NUMBER}?" +

"clientPipelineFactory=#MY_CLIENT_PIPELINE_FACTORY&" +

"sendBufferSize=26214400&allowDefaultCodec=false"
)
)
.setExchangePattern(ExchangePattern.InOnly)
.stop() // Stops subsequent routing of the current Exchange and 
marks it
as completed.
.end(); // Ends the current block.





--
View this message in context: 
http://camel.465427.n5.nabble.com/NEED-HELP-Exchanges-remain-inflight-on-route-with-recipientList-to-netty-udp-tp5773079.html
Sent from the Camel - Users mailing list archive at Nabble.com.

Re: NEED HELP: Exchanges remain inflight on route with recipientList to netty:udp

[SteveR]

Ok, I just tried upgrading to *Camel 2.16.0* in the hope that the new 
Dynamic To <http://camel.apache.org/message-endpoint.html>   would solve
this problem, but I still get the same result, that is all exchanges that
are sent to the route stay marked as *ExchangesInflight*.  This time I even
had a UDP listener running using the Linux  socat
<http://www.dest-unreach.org/socat/doc/socat.html>  command and it receives
the correct number of UDP acknowledgements.  So the route is sending all
exchanges that it received on its SEDA queue, but they never transition to
the completed state.

Below is my updated route that uses the new *toD()* method.  Below that is
the route which sends to this ack-back route.

I'm really puzzled by this behavior, but I need to figure it out soon
because my management is getting cranky ;-(

   Thanks for any help,  SteveR


from(uri="seda:SEDA_ACK_QUEUE?size=20&concurrentConsumers=10")
.setExchangePattern(ExchangePattern.InOnly)
.routeId(ackQueueRouteId)
.startupOrder(ackQueueRouteStartupOrder)
.setProperty(Exchange.CHARSET_NAME,
ExpressionBuilder.constantExpression(charsetName))
.process(cqmsAckBackProcessor).id(cqmsAckBackProcessorId)
*   .toD(
"netty:udp://${header.REMOTE_HOST_IP}:" +
"${header.REMOTE_PORT_NUMBER}?" +
"clientPipelineFactory=#MY_CLIENT_PIPELINE_FACTORY&" +
"sendBufferSize=26214400&allowDefaultCodec=false"
)*
.stop().end();


from(kafkaQueueURI)
.errorHandler(deadLetterChannelBuilder) // Add route-scoped
DeadLetterChannel error handler
*   .onCompletion()
.parallelProcessing() // Tells Camel to use a thread 
pool for
onCompletion route
.onCompleteOnly() // Synchronize only after 
Exchange completes
successfully with no errors
.to(ackQueueURI).id(sourceRouteId + 
"_ON_COMPLETION_ONLY_TO_ACK_QUEUE")
.end() // Must use end() to denote the end of the onCompletion 
route*
.setExchangePattern(ExchangePattern.InOnly)
.routeId(kafkaQueueRouteId)
.startupOrder(kafkaQueueRouteStartupOrder)
.setProperty(Exchange.CHARSET_NAME,
ExpressionBuilder.constantExpression(charsetName))
.threads(threadPoolSize /*poolSize*/, threadPoolSize * 2 
/*maxPoolSize*/)
.threadName("threads_" + kafkaQueueRouteId)
.callerRunsWhenRejected(true) // Hard-coded since we always 
want this
behavior!
//

// This processor handles Kafka related processing, e.g. 
determining the
Kafka partitioning
//

.process(kafkaProcessor).id(kafkaProcessorId)
// ---
// Here we route to the final destination (e.g. Kafka)
// ---
.to(kafkaToURI)
.setProperty(Exchange.CHARSET_NAME,
ExpressionBuilder.constantExpression(charsetName));


<http://camel.465427.n5.nabble.com/file/n5773080/IN_FLIGHT_1.png> 



--
View this message in context: 
http://camel.465427.n5.nabble.com/NEED-HELP-Exchanges-remain-inflight-on-route-with-recipientList-to-netty-udp-tp5773079p5773080.html
Sent from the Camel - Users mailing list archive at Nabble.com.

Camel 2.16.0 and camel-kafka using old verion of netty?

[SteveR]

Hi:

I've recently upgraded my Linux RHEL6 java application to use *Camel 2.16.0*
and *netty4*.  I see that *camel-kafka* is pulling in an old version of
netty3 (i.e. *netty-3.7.0.Final.jar*). Just wondering this is normal or if
it's an issue that can be addressed? 

My pom entry for camel-kafka:


org.apache.camel
camel-kafka
2.16.0
 

When running Maven's dependency:tree, i see this:

*[INFO] +- org.apache.camel:camel-kafka:jar:2.16.0:compile*
[INFO] |  +- (org.apache.camel:camel-core:jar:2.16.0:compile - omitted 
for
duplicate)
[INFO] |  +- org.apache.kafka:kafka_2.11:jar:0.8.2.1:compile
[INFO] |  |  +- org.scala-lang.modules:scala-xml_2.11:jar:1.0.2:compile
[INFO] |  |  +- com.yammer.metrics:metrics-core:jar:2.2.0:compile
[INFO] |  |  |  \- (org.slf4j:slf4j-api:jar:1.7.2:compile - omitted for
conflict with 1.7.10)
[INFO] |  |  +- net.sf.jopt-simple:jopt-simple:jar:3.2:compile
[INFO] |  |  +-
org.scala-lang.modules:scala-parser-combinators_2.11:jar:1.0.2:compile
[INFO] |  |  +- com.101tec:zkclient:jar:0.3:compile
*   [INFO] |  |  |  \- (org.apache.zookeeper:zookeeper:jar:3.3.1:compile -
omitted for conflict with 3.4.6)*
[INFO] |  |  +- org.apache.kafka:kafka-clients:jar:0.8.2.1:compile
[INFO] |  |  |  +- (org.slf4j:slf4j-api:jar:1.7.6:compile - omitted for
conflict with 1.7.10)
[INFO] |  |  |  +- net.jpountz.lz4:lz4:jar:1.2.0:compile
[INFO] |  |  |  \- (org.xerial.snappy:snappy-java:jar:1.1.1.6:compile -
omitted for conflict with 1.1.1.7)
[INFO] |  |  \- org.apache.zookeeper:zookeeper:jar:3.4.6:compile
[INFO] |  | +- (org.slf4j:slf4j-api:jar:1.6.1:compile - omitted for
conflict with 1.7.10)
[INFO] |  | +- jline:jline:jar:0.9.94:compile
[INFO] |  | |  \- (junit:junit:jar:3.8.1:compile - omitted for
duplicate)
*[INFO] |  | \- io.netty:netty:jar:3.7.0.Final:compile*
[INFO] |  +- (com.sun.xml.bind:jaxb-core:jar:2.2.11:compile - omitted 
for
duplicate)
[INFO] |  \- (com.sun.xml.bind:jaxb-impl:jar:2.2.11:compile - omitted 
for
duplicate)




--
View this message in context: 
http://camel.465427.n5.nabble.com/Camel-2-16-0-and-camel-kafka-using-old-verion-of-netty-tp5773381.html
Sent from the Camel - Users mailing list archive at Nabble.com.

Re: Integration Pattern

[SteveR]

Hi Ishada:

Maybe checkout the  Camel Mail Component <http://camel.apache.org/mail.html> 
.

SteveR



--
View this message in context: 
http://camel.465427.n5.nabble.com/Integration-Pattern-tp5773535p5773547.html
Sent from the Camel - Users mailing list archive at Nabble.com.

How to specify the UDP source port in dynamic To (toD)?

[SteveR]

I'm using *Camel 2.16.0*, *camel-netty4*, and the new dynamic *toD()* feature
to send UDP acknowledgements back to the associated UDP client.

In my Camel route that deals with acknowledgements, I have an
*AckBackProcessor* that composes the UDP response to be sent back to the UDP
client.  Right before returning from the *AckBackProcessor* I print out the
*NETTY_LOCAL_ADDRESS* and *NETTY_REMOTE_ADDRESS* headers in the Exchange's
inMsg and they look as expected.

I'm required to use an application-specific well-known UDP source port in
the acknowledgement (e.g. *61133*). However, somehow, after the
*AckBackProcessor* returns and the *toD()* executes, my tcpdump shows that
the UDP source port has changed from *61133* to what looks to be an
ephemeral source port.

So my question is, with *camel-netty4* and the new dynamic *toD()* feature,
how do I specify a UDP source port of my choosing?

Here's what my route looks like.  It's a SEDA queue:

from(ackQueueURI) 
.setExchangePattern(ExchangePattern.InOnly)
.id("setExchangePattern_" + ackQueueRouteId)
.routeId(ackQueueRouteId)
.startupOrder(ackQueueRouteStartupOrder)
.setProperty(Exchange.CHARSET_NAME,
ExpressionBuilder.constantExpression(charsetName))
.id("setProperty_" + ackQueueRouteId + "_" + 
charsetName)
 // AckBackProcessor formulates the UDP ack and does
setHeader("ACK_REQUIRED") to "true" or "false"
.process(ackBackProcessor).id(ackBackProcessorId)
.choice()
.when(header("ACK_REQUIRED").isEqualTo("false"))
.to("log:DEV_NULL_LOG?level=OFF") // Drop UDP 
packet since ack is not
required
.otherwise()
// Otherwise, this UDP request requires an 
acknowledgement.
// Dynamic To does remote address substitution 
from the in-flight
exchange header.
.toD(ackToURI).id("toD_" + ackQueueRouteId + 
"_TO_CLIENT")
.end();


Here's the method used to create the acknowledgement URI referenced within
the *toD(*):

/**
 * Builds and returns the string containing the acknowledgment to URI.
 *
 * @param dest
 * @param clientPipelineFactoryName
 * @return  The string containing the ackToURI
 */
private String getAckToURI(Destination dest, String
clientPipelineFactoryName) {
StringBuilder sb = new StringBuilder();

   
sb.append("netty4:udp").append(":/${header.CamelNettyRemoteAddress}")
 
.append("?clientPipelineFactory=#").append(clientPipelineFactoryName);

String ackOptions = dest.getOptions();
if(ackOptions != null && !ackOptions.isEmpty()) {
sb.append(ackOptions);
}

String ackExtraOptions = dest.getExtraOptions();
if(ackExtraOptions != null && !ackExtraOptions.isEmpty()) {
sb.append(ackExtraOptions);
}

logger.info("getAckToURI(): ackToURI = {}", sb.toString());
return sb.toString();
}



--
View this message in context: 
http://camel.465427.n5.nabble.com/How-to-specify-the-UDP-source-port-in-dynamic-To-toD-tp5774964.html
Sent from the Camel - Users mailing list archive at Nabble.com.

How to specify the UDP source port with camel-netty4 ?

[SteveR]

With *Camel 2.16.0* and the *camel-netty4* component, I need to send a UDP
response back to the associated client and I need control over the source IP
(i.e. my local host name) *AND *the UDP source port (i.e. a port number of
my choosing).

Currently, it appears that *camel-netty4* uses the local host IP for the
source IP, but allows the OS choose an ephemeral port number for the
sender's UDP source port.  The UDP client I send the response to expects the
UDP source port be a well-known value (e.g. 12345), for validation purposes.

For example, I have Camel route that consumes exchanges from
*MY_SEDA_QUEUE*, invokes a processor to formulate the UDP response, and
sends it via *netty4:udp* to the associated UDP client specified in the
*CamelNettyRemoteAddress* header of the exchange's input message:

   
from("seda:MY_SEDA_QUEUE?size=1000&concurrentConsumers=25&timeout=1")
.process(MyUdpResponseProcessor)
.toD("netty4:udp:/${header.CamelNettyRemoteAddress}?
clientPipelineFactory=#MY_CLIENT_PIPELINE_FACTORY&
   
sync=false&disconnect=false&sendBufferSize=26214400&allowDefaultCodec=false");

Any thoughts on how to achieve this is much appreciated.



--
View this message in context: 
http://camel.465427.n5.nabble.com/How-to-specify-the-UDP-source-port-with-camel-netty4-tp5775017.html
Sent from the Camel - Users mailing list archive at Nabble.com.

Re: How to specify the UDP source port with camel-netty4 ?

[SteveR]

Hi Greg:

Thanks for the reply. However, it's the UDP source port that I'd like
control over.  The *${header.CamelNettyRemoteAddress}* correctly resolves to
the in-flight destination *IP:port *of the client that I need to send the
UDP acknowledgement to.  My problem is that *camel-netty4* is not giving me
control over the UDP source port, instead it appears to be binding to a
local ephemeral port chosen by the Linux OS.

*
SOURCE  IP* Set correctly to the local IP address of the server
sending the UDP ack
*SOURCE  PORT*OS ephemeral port (I need to be able to specify this port
number)
*DEST   IP* Correctly taken from in-flight exchange
CamelNettyRemoteAddress header.
*DEST  PORT* Correctly taken from in-flight exchange
CamelNettyRemoteAddress header.

So my UDP acknowledgements get delivered to the intended client, but the
client rejects them because
it's examining the UDP source port and seeing that it's not the expected
value, but rather a different ephemeral source port in each acknowledgement.






--
View this message in context: 
http://camel.465427.n5.nabble.com/How-to-specify-the-UDP-source-port-with-camel-netty4-tp5775017p5775057.html
Sent from the Camel - Users mailing list archive at Nabble.com.

Re: How to specify the UDP source port with camel-netty4 ?

[SteveR]

I'm using *Camel-2.16* and I'm looking at the GitHub code for the 
NettyProducer.java

 
, and I think the below highlighted code is what causes netty to specify
using an ephemeral port for the UDP source port when the camel-netty4
*udpConnectionlessSending* option is set to *true*.  It would be nice if it
wasn't hard-coded and let you specify a UDP source port of your choosing!
I'm hoping someone with more knowledge will chime in.

// if udp connectionless sending is true we don't do a connect.
// we just send on the channel created with bind which means
// really fire and forget. You wont get an
PortUnreachableException
// if no one is listen on the port
if (!configuration.isUdpConnectionlessSending()) {
answer = connectionlessClientBootstrap.connect(new
InetSocketAddress(configuration.getHost(), configuration.getPort()));
} else {
*// bind and store channel so we can close it when stopping
answer = connectionlessClientBootstrap.bind(new
InetSocketAddress(0)).sync();*
Channel channel = answer.channel();
allChannels.add(channel);
}



--
View this message in context: 
http://camel.465427.n5.nabble.com/How-to-specify-the-UDP-source-port-with-camel-netty4-tp5775017p5775061.html
Sent from the Camel - Users mailing list archive at Nabble.com.

[ENHANCEMENT REQUEST]: camel-netty4 config support for specifying UDP source port

[SteveR]

I'm currently facing a show-stopping production issue: I'm unable to figure
out how to specify the UDP source port of my choosing for a Camel route that
uses the  camel-netty4 <http://camel.apache.org/netty4.html>   component to
send a UDP datagram.

If anyone has any thoughts/ideas about how I can do this (or hack this) it
would be much appreciated.

For details, see my recent Camel-Users post:  How to specify the UDP source
port with camel-netty4 ?
<http://camel.465427.n5.nabble.com/How-to-specify-the-UDP-source-port-with-camel-netty4-tc5775017.html>
  

It would be great if this could be taken under consideration for an upcoming
enhancement.  It looks like the main changes would be in the associated 
NettyProducer.java
<https://github.com/apache/camel/blob/camel-2.16.x/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/NettyProducer.java>
  
and possibly made configurable via an addition to the  camel-netty4
<http://camel.apache.org/netty4.html>   options.

For example:

.to("netty4:udp://remotehost:12345?*sourcePort=5*&
  clientPipelineFactory=#MY_CLIENT_PIPELINE_FACTORY&
 
sync=false&disconnect=false&sendBufferSize=26214400&allowDefaultCodec=false");

Thanks, SteveR




--
View this message in context: 
http://camel.465427.n5.nabble.com/ENHANCEMENT-REQUEST-camel-netty4-config-support-for-specifying-UDP-source-port-tp5775070.html
Sent from the Camel - Users mailing list archive at Nabble.com.

Easiest way to send a reply via camel-netty4:udp socket?

[SteveR]

- I have a Camel input route whereby camel-netty4 creates/binds a UDP socket
that listens on port 12345: 

*from("netty4:udp://host:12345? ...)
.to("SEDA_QUEUE1" ...);*

- The input route receives a UDP datagram, creates the exchange, and sends
it to *SEDA_QUEUE1*.

- Then the exchange from *SEDA_QUEUE1* is sent to some other queue (e.g.
*SEDA_QUEUE2*) which has a processor that formulates a UDP exchange message
that I want to send back as a reply message via the input route's UDP
socket.

This reply exchange has the *CamelNettyLocalAddress* and
*CamelNettyRemoteAddress *headers containing the associated local/remote IP
and port.

How, from a processor within the route consuming from *SEDA_QUEUE2*, do I
send the reply out the UDP socket on the input route?  There was to be an
easy way to do this, but I'm not seeing it.

  Thanks, SteveR


I've been trying with code similar to this within a processor on the
*SEDA_QUEUE2*, but no luck so far:

ProducerTemplate template = context.createProducerTemplate();
Endpoint endpoint = context.getEndpoint("netty4:udp ...");
Exchange replyExchange = endpoint.createExchange();
replyExchange.getIn().setBody("UDP REPLY MESSAGE");
replyExchange.getIn().setHeaders(exchange.getIn().getHeaders());
template.send(endpoint, replyExchange);












--
View this message in context: 
http://camel.465427.n5.nabble.com/Easiest-way-to-send-a-reply-via-camel-netty4-udp-socket-tp5775186.html
Sent from the Camel - Users mailing list archive at Nabble.com.