Feeding an Async Processor from an ActiveMQ topic

[Jörn Kottmann]

Hello,

in this route the messages are read from an ActiveMQ topic
and then send to an Async Processor. The Async Processor
drives the processing in an other systems which processes
100 messages parallel.
How can I configure camel to keep the Async Processor supplied
with at least 100 messages concurrently when there are messages
waiting to be processed on the topic ?

In my current configuration I think camel calls my processor synchronously,
which really slows the system down.

Here is my route:
getContext().addComponent("activemq", bean(ActiveMQComponent.class, 
"activemq"));
SpringTransactionPolicy required = bean(SpringTransactionPolicy.class,
   "PROPAGATION_REQUIRED");
      
errorHandler(transactionErrorHandler(required));

from("activemq:topic:TestTopic?clientId=test&durableSubscriptionName=test")
.policy(required)
.to("CustomAsyncProcessor:xxxxx");

Since the async processing can fail the route uses transactions to not 
loose messages.

And here is my camel-context.xml:

...

   <bean id="jmsConnectionFactory" 
class="org.apache.activemq.ActiveMQConnectionFactory">
       <property name="brokerURL"
           
value="failover:(tcp://10.45.2.100:61616)?jms.redeliveryPolicy.maximumRedeliveries=-1" 
/>
   </bean>

   <bean id="jmsTransactionManager"
       class="org.springframework.jms.connection.JmsTransactionManager">
       <property name="connectionFactory" ref="jmsConnectionFactory" />
   </bean>

   <bean id="jmsConfig" 
class="org.apache.camel.component.jms.JmsConfiguration">
       <property name="connectionFactory" ref="jmsConnectionFactory" />
       <property name="transactionManager" ref="jmsTransactionManager" />
       <property name="transacted" value="true" />
       <property name="concurrentConsumers" value="1" />
   </bean>

   <bean id="activemq" 
class="org.apache.activemq.camel.component.ActiveMQComponent">
       <property name="configuration" ref="jmsConfig" />
   </bean>

   <bean id="PROPAGATION_REQUIRED" 
class="org.apache.camel.spring.spi.SpringTransactionPolicy">
       <property name="transactionManager" ref="jmsTransactionManager" />
       <property name="propagationBehaviorName" 
value="PROPAGATION_REQUIRED"/>
   </bean>
...

Thanks,
Jörn