Stop JMS Consumers in Camel-CXF on Exception using camel route policy

[SRIKANT MVS <sr...@gmail.com>]

Hi Team,

I am using a camel-cxf application where I have exposed cxfEndpoints for
SOAP over HTTP and SOAP over JMS. I am consuming events from IBM MQ and
publishing them to a WebService.



I want to implement the circuit break pattern when my WebService is not
reachable. Ideally it should not consume any new messages from IBM MQ when
the WebService is unavailable



Here is the current code setup



CXFEndpoint is Defined as below

<*cxf:cxfEndpoint **id*="cXFTestServiceHttp">

<*cxf:cxfEndpoint **id*="cXFTestServiceJMS">

*Camel Transport CXF Bus *

<*bean **class*="org.apache.camel.component.cxf.transport.CamelTransportFactory">
  <*property **name*="bus" *ref*="cxf" />
  <*property **name*="transportIds">
    <*list*>
      <*value*>http://cxf.apache.org/transports/camel</*value*>
    </*list*>
  </*property*>
  <*property **name*="checkException" *value*="true"></*property*>
</*bean*>

Routes are defined as below

  from("cxf:bean:cXFTestServiceHttp")
      .routeId("httpTestServiceRoute")
      .to(testProcessor)

  from("cxf:bean:cXFTestServiceJMS")
      .routeId("jmsTestServiceRoute")
      .to(testProcessor)
*      .routePolicy(getThrottlingExceptionRoutePolicy());*

*private *ThrottlingExceptionRoutePolicy getThrottlingExceptionRoutePolicy() {
  List<Class<?>> throttledExceptions = *new *ArrayList<>();
  throttledExceptions.add(WebServicesUnavailableException.*class*);

  ThrottlingExceptionRoutePolicy routePolicy = *new
*ThrottlingExceptionRoutePolicy(1, 30000,60000, throttledExceptions);
  routePolicy.setHalfOpenHandler(*this*::checkIfWebServiceAvailable);
  *return *routePolicy;
}

When I tested with the above setup, I see that the messages are actually
getting consumed from the MQ when the WebService is down.

Please advice what is missing here. Do I need configure something more as I
am using camel-cxf component.



Should I stop the actual cxfEndpoint as I feel that the route is still
active even after adding ThrottlingExceptionRoutePolicy.

I see from other references that we could actually stop that as mentioned
below



cxf:stop-endpoint <bus> <endpoint name>

cxf:start-endpoint <bus> <endpoint name>



But, Can I do this programmatically ?


I would like understand if the Route Policy is the right approach for this
setup, otherwise I need to think to a custom approach.


-Regards

Srikant Mantha