Camel dynamically build processor bean call

Question

So here’s a sample of my bean setup for one topic. Keep in mind there are multiple beans...one for each message type

I've been able to successfully receive and process all of my message types by having a route per message type, but I want to simplify the context file, so I'd to have just one route that receives all my topics.

Here is what my route setup looks like...

    <route id="Netty7001Route">
       <from uri="netty4:tcp://192.168.200.3:7001…"/>
       <to uri=”seda:ProcessRoute”/>
    </route>

    <route id=”ProcessRoute”/>
      <from uri=”seda:ProcessRoute”/>
           <setHeader headerName="LOGMESSAGE"><simple>A10::Entering endpoint for track message ${in.header.MSGNAME} via netty4</simple></setHeader>
           <to uri="log:messageLogger?level=ERROR"/>
           <setHeader headerName="LOGMESSAGE"><simple>A11::Leaving endpoint for track message ${in.header.MSGNAME}</simple></setHeader>
           <to uri="log:messageLogger?level=ERROR"/>
           <setHeader headerName="LOGMESSAGE"><constant>A14::Entering Mediation</constant></setHeader>
           <to uri="log:messageLogger?level=ERROR"/>
           **<process ref="DDS_C2_NewSystemReferencePointMediationBean"/>**
       <setHeader headerName="MSGNAME"><constant>C2_TM_NewSystemReferencePoint</constant></setHeader>
       <setHeader headerName="MSGTYPE"><constant>C2_NewSystemReferencePoint</constant></setHeader>
           <setHeader headerName="LOGMESSAGE"><constant>A15::Leaving Mediation</constant></setHeader>
           <to uri="log:messageLogger?level=ERROR"/>
           <setHeader headerName="LOGMESSAGE"><constant>A17::Sending message to DDS domain 4</constant></setHeader>   
           <to uri="log:messageLogger?level=ERROR"/>
       <to uri="dds://4/C2_TM_NewSystemReferencePoint?typeName=C2_NewSystemReferencePoint"/>
    </route>

So, process ref= line above is what I want to be dynamic since I won’t know which bean processor to call until I look at the MSGNAME in the header. I’ve tried using recipientList which fails because they are not endpoints but rather processor beans and I’ve tried using just ref:DDS_C2_{$in.header.MSGNAME}MediationBean, but camel won’t start and complains about the simple tag here. Is there a way to do this in a camel spring configuration?

I tried a workaround by using the tag which checks the MSGNAME and then calls the corresponding processor bean, but I need to have conditions for each of my message types. This works, but is extremely inefficient when compared to having a route per topic.

I’ve thought about just writing a processor which would call the appropriate bean processor in the java code, but I’m not sure if that's the correct way to do what I need and whether it would be more efficient over using the tag.

Thanks for the help.

Just to clarify, I'm speaking about the process ref= tag in the code. Also, writing a processor to determine the bean call for me will force code updates when the message types change or get added, hence my reason for needing a spring based solution. — gshepherd7
"I’ve thought about just writing a processor which would call the appropriate bean processor in the java code" I believe that would be a way (regardless security concerns providing a class to call in an http header..) the direct andpoint is evaluated when started, so you cannot provide the "ref" dynamically — gusto2
Why don't you use choice().when(header==''whatever") then call your process? Note I am saying use the choice/when functionality. There is the equivalent for the xml dsl. — Souciance Eqdam Rashti
I tried using the choice/when, but I have 96 different message types and during testing the performance is much slower than when I had a route per topic. — gshepherd7

mgyongyosi mgyongyosi · Accepted Answer · 2017-08-31T22:26:42

The best way could be to create a new Component with a ProcessorEndpoint as described here.

Working Example (with Processor)

Create your custom component:

package com.mgyongyosi.sample.component;

// imports

public class SpecifiedProcessor extends DefaultComponent {

@Override
protected Endpoint createEndpoint(String uri, String remaining, Map<String, Object> parameters) throws Exception {
    Object registryObj = getCamelContext().getRegistry().lookupByName(remaining);

    if(!(registryObj instanceof Processor))
        throw new IllegalArgumentException("The Processor with the specified name was not found in the Registry.");

    return new ProcessorEndpoint(uri, this, (Processor)registryObj);
}

Register your component by creating a file named specified-processor (this will be the component's uri scheme) in the META-INF/services/org/apache/camel/component directory with the following content (full documentation):
```
class=com.mgyongyosi.sample.component.SpecifiedProcessor
```

Example usage from Java DSL:

from("timer:helloworld?period=5000")
        .setHeader("MSGNAME", constant("NewSystemReferencePoint"))
        .toD("specified-processor://DDS_C2_${in.header.MSGNAME}MediationBean")
        .log("${body}");

In XML:

<toD uri="specified-processor://DDS_C2_${in.header.MSGNAME}MediationBean"/>

In case of Beans

Java DSL: .toD("bean:DDS_C2_${in.header.MSGNAME}MediationBean")

XML: <toD uri="bean:DDS_C2_${in.header.MSGNAME}MediationBean"/>

Camel dynamically build processor bean call

1 Answers

Working Example (with Processor)

In case of Beans