Geek asks and answers

Content Based Throttling

I would like to know if it is possible with Camel to do throttling based on the contents of the exchange.

The situation is this: I have to call a web service through soap. Among the parameters sent to this web service is clientId. The problem is that the web service is sending back an error if there is more than 1 request per minute for the given customerId.

I am wondering if it is possible to implement throttling on customerId using Camel. Therefore, throttling should not be performed for all messages, but only for messages with the same client.

Let me know how I could implement this, or if I need to clarify my question.

+5
source share
3 answers

ActiveMQ . , JMS , JMSXGroupId customerId. -, .

. http://camel.apache.org/parallel-processing-and-ordering.html ...

+1

ActiveMQ , , Camel/ActiveMQ.

SLA. ( ) , , , .

, , , JIRA. . - :

<route>
  <from uri="activemq:pending?maxConcurrentConsumers=10"/>
  <throttle timePeriodMillis="60000">
    <constant>1</constant>
    <to uri="mock:endpoint"/>
  </throttle>
</route>

, , , , . , , . , , SLA , , JMSXGroupId.

+1

, , , .

:

  • ( )
  • , .

:

  • 1 customerID
  • Using Splitter to Break List Into Individual Messages
  • Send the first message from the splitter to the actual service
  • Re-move the rest of the list back to the aggregator.

The Java DSL version is a little understandable:

final AggregationStrategy aggregationStrategy = AggregationStrategies.flexible(Object.class)
        .accumulateInCollection(ArrayList.class);

from("direct:start")
    .log("Receiving ${body}")
    .aggregate(header("customerID"), aggregationStrategy).completionTimeout(60000)
        .log("Aggregate: releasing ${body}")
        .split(body())
        .choice()
            .when(header(Exchange.SPLIT_INDEX).isEqualTo(0))
                .log("*** Processing: ${body}")
                .to("mock:result")
            .otherwise()
              .to("seda:delay")
        .endChoice();

from("seda:delay")
    .delay(0)
    .to("direct:start");

Spring XML version is as follows:

 <!-- this is our aggregation strategy defined as a spring bean -->
 <!-- see http://stackoverflow.com/questions/27404726/how-does-one-set-the-pick-expression-for-apache-camels-flexibleaggregationstr -->
 <bean id="_flexible0" class="org.apache.camel.util.toolbox.FlexibleAggregationStrategy"/>
 <bean id="_flexible2" factory-bean="_flexible0" factory-method="accumulateInCollection">
     <constructor-arg value="java.util.ArrayList" />
 </bean>

<camelContext xmlns="http://camel.apache.org/schema/spring">
       <route>
           <from uri="direct:start"/>
           <log message="Receiving ${body}"/>
           <aggregate strategyRef="_flexible2" completionTimeout="60000" >
               <correlationExpression>
                   <xpath>/order/@customerID</xpath>
               </correlationExpression>
               <log message="Aggregate: releasing ${body}"/>
               <split>
                   <simple>${body}</simple>
                   <choice>
                       <when>
                           <simple>${header.CamelSplitIndex} == 0</simple>
                           <log message="*** Processing: ${body}"/>
                           <to uri="mock:result"/>
                       </when>
                       <otherwise>
                           <log message="--- Delaying: ${body}"/>
                           <to uri="seda:delay" />
                       </otherwise>
                   </choice>
               </split>
           </aggregate>
       </route>

       <route>
           <from uri="seda:delay"/>
           <to uri="direct:start"/>
       </route>
</camelContext>
0
source

More articles:


All Articles