[akka-user] [Camel, Akka, JMS] Deferred Message Acknowledgement: Can I Acknowledge A Previously Processed Message To The Broker?

Hi all,
I am posting my SO question 
here too, hopefully I will get some more focused responses here :)
I am using Akka (latest stable version), akka-camel and JMS (for the 
purposes of this conversation, let's say it's ActiveMQ, but ideally the 
solution should be generic).
The use-case
I have the following use-case. On a queue Q I receive messages like this:
time:   1     2    3    4    5    6
      | A1 | B1 | C1 | C2 | A2 | B2 | .... 
         ^                        ^
       first                     latest
My end goal is to pair them together in (A1,A2), (B1,B2), and so on; 
notwithstanding complications like duplicates and undelivered messages, the 
complication is that I have to ensure that, until a whole pair is matched 
and processes, *the broker will keep hold of all unacknowledged messages.*
Example
At 4 I received and processed 4 messages, and processed successfully pair (C1, 
C2), however I still cannot acknowledge anything back to the broker because 
A1 and B1 are still unmatched and pending, and in JMS *to 
acknowledge C2 means to acknowledge all messages up to C2*. As a matter of 
fact, the first acknowledgement I can send back is at time 5, when A2 is 
received: at this point I can acknowledge A1 (and only A1, as B1 is still 
pending).
The problem
Now, what I can't seem to figure out is how to do this kind of *delayed*
 and *async* acknowledgement via akka-camel. I have been reading online 
and, while I can find explanations on how to acknowledge messages manually (
docs 
 and 
an example 
), 
there is nothing showing how to acknowledge a previously processed message 
to the broker.
import akka.camel.{ CamelMessage, Consumer }import akka.camel.Ackimport akka.actor.Status.Failure
class Consumer3 extends Consumer {
  override def autoAck = false
  def endpointUri = "jms:queue:test"
  def receive = {
    case msg: CamelMessage =>
      sender() ! Ack
      // on success
      // ..
      val someException = new Exception("e1")
      // on failure
      sender() ! Failure(someException)
  }}
In this case, Ack is an object and its semantic is really just: *I 
acknowledge the current message* while I would need something like *Message 
X is now acknowledged*, where *X* is some previous message, but not 
necessarily the current one.
Is this use-case supported or supportable via akka-camel or should I just 
build it myself?
Thanks
You received this message because you are subscribed to the Google Groups "Akka User List" group.
To unsubscribe from this group and stop receiving emails from it, send an email to [email protected].
To post to this group, send email to [email protected].
Visit this group at https://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.

Hi all! Problem description: Sometimes (in my tests usually for 6-10 jms messages out of 60) when the camel jms consumer (configured with AUTO_ACKNOWLEDGE) consumes a message and an Exception is thrown in a later stage of the route (e.g. by a processor that uses this jms message body), then the ActiveMQ broker never receives the message consumption acknowledgement (or more likely, the jms
I have encoutered problem in a distributed environment with a network of two brokers. The ActiveMQ version is 5.1 The ActiveMQ is supposed to use store and forward network which ensure every message is delivered. But the real situation is I found message loss on one distributed queue. Senerio is like this: Queue name: TestQ Producer is on broker A and consumer is on broker B. Broker A and B is connected
[ https://issues.apache.org/jira/browse/CAMEL-5769?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=13490328#comment-13490328 ] Raul Kripalani commented on CAMEL-5769: --------------------------------------- I noticed that the issue doesn't necessarily happen with persistent messages only. It happens under any situation where the broker takes longer than the
Hi, in a network of two brokers with DemandFowardingBridges (conduitSubscriptions=false) with high latency, I'm getting consistently IllegalArgumentExceptions (The subscription does not exist). They happen, if a remote consumer disconnects before all locally produced messages have been consumed. In that case, the local broker is still sending a message, which on the remote broker is added to the
Hi guys, Is there a way to notify a message broker from a Camel consumer actor using Individual acknowledgement mode? I tried sending back an Ack message back to the sender() once the message is received but its never ack'ed by the broker (ActiveMQ). Any help would be greatly appreciated.
Hi, How can I use ActiveMQXA* classes in standalone JMS client? I tried to write this simple test: public class AMQXATest { public static void main(String[] args) throws Exception { // start broker BrokerService broker = new BrokerService(); broker.setPersistent(false); broker.addConnector("tcp://localhost:5000"); broker.start(); // start transaction manager TMService jotm
I need to provide JMS topics for external non-Akka clients from inside of my actor system. So far, I'm using the CamelExtension to do something like .. val factory = new ActiveMQConnectionFactory("vm:(broker:(tcp://localhost:61617)?persistent�lse)" ) camelContext.addComponent("ext-jms", JmsComponent.jmsComponentAutoAcknowledge(factory)) val exporter = system.actorOf(Props[JmsExporter
I would like to check the memory usage of activemq broker in the Linux environment. I know there is a UsageMemory.class, but I don't know how to call it and check the memory usage. -- View this message in context: http://www.nabble.com/How-can-I-check-the-memory-usage-of-activemq-broker-without-jconsole--tp17598144s2354p17598144.html Sent from the ActiveMQ - User mailing list archive at Nabble.com
hi buddies, suppose I have 4 AMQ-brokers name them : A,B,C,O there are all networkconnector wire to "O" AO BO CO now I want to produce message from A to B, B as the consumer. so it comes from A->O->B how ever, if I use TTL = 2 , then when I produce the message in A's Queue "QA", All the Queues(B,C,O) have the Queue(QA). so I even can consume the message in broker C. THE QUESTION IS ： HOW
Hi All, I am quite new to ActiveMQ and am trying to figure out a solution to the following. I have an embedded broker and a remote broker. From my code I do the following to create a session on the embedded broker and to create a queue there. factory = new ActiveMQConnectionFactory("tcp://localhost:61617"); connection = factory.createConnection();