Search code examples
javatransactionsapache-camelconsumer

Camel consume single message and stop, transacted

I am trying to use Camel to consume a single message from a JMS queue in a transacted manner. Specifically in a flow like this:

  1. Wait until message is published on JMS queue
  2. Try to consume and process the single message
  3. If processing fails (exception occurs), rollback the consumption
  4. If the processing passes, acknowledge and stop consuming anymore messages
  5. Later in the application lifecycle, another process triggers consumption to start again from (1)

At first I tried to do this using a polling consumer, using the ConsumerTemplate, but I can't figure out if its possible to do this transactionally - it seems like the transaction is internal to the ConsumerTemplate so regardless of what I do the message is already acknowledged as consumed by the time the ConsumerTemplate returns.

Can I do this using the ConsumerTemplate? Can I do this using Camel and if so what is the best approach (Simple examples would be appreciated)?

Solution

  • I ended up using the pollEnrich dsl to achieve this. For example my route builder looks like:

    from("direct:service-endpont").transacted("PROPOGATION_REQUIRED").setExchangePattern(ExchangePattern.InOut).pollEnrich("activemq:test-queue").bean(myHandler);

    I use the direct endpoint as a service, sending a "request" message to the direct endpoint polls the jms queue for a single message (blocking if required). The transaction started extends to the pollEnrich so if, for example, myHandler bean fails then the message taken during the pollEnrich is not consumed and left on the queue.