1

I need to find a specific message within an ActiveMQ queue using PHP and remove it.

AFAIK the only way to do so is to read all messages that are currently queued and ACK the one message I'm interested in. (The example in the PHP manual for Stomp::ack does more or less the same thing, they don't read all messages, but only ACK the one that matches).

So, I wrote this code (this is only the relevant part, obviously):

class StompController {

    private $con;

    public function __construct($stompSettings) {
        try {
            $this->con = new Stomp($stompSettings['scheme']."://".$stompSettings['host'].":".$stompSettings['port']);
            $this->con->connect();
            $this->con->setReadTimeout(5);
        } catch(StompException $e) {
            die('Connection failed:' .$e->getMessage());
        }
    }

    public function __destruct() {
        $this->con->disconnect();
    }

    public function ackMessageAsRead($recipient,$message) {
        if($this->con->isConnected()) {
            //Subscribe to the recipient user's message queue.
            $this->con->subscribe("/queue/".$recipient);
            //Read all messages currently in the queue (but only ACK the one we're interested in).
            while($this->con->hasFrameToRead()) {
                $msg = $this->con->readFrame();
                if($msg != null && $msg != false) {
                    //This is the message we are currently reading, ACK it to AMQ and be done with it.
                    if($msg->body == $message) {
                        $this->con->ack($msg);
                    }
                } 
            }
        } else {
            return false;
        }
    }
}

According to my logic, this should work. While running the code only one random message is being read though, despite checking for more frames.

The next frame seems only to be prepared when the frame we're currently reading has been ACK'ed. (When I manually ACK all messages the while loop works as intended, all messages are processed.

Does anyone know how to get the full set of messages from the queue, without ACK'ing all of them? I can ACK all of them and put the ones I wasn't interested in back into the queue afterwards, but this already inefficient way to find a single message gets a whole lot more inefficient that way.

3
  • 1
    Are you setting activemq.prefetchSize to any value? If activemq.prefetchSize is 1, broker will wait for the ack of the message before it sends the next message.
    – Buchi
    Commented Sep 10, 2012 at 15:50
  • @Buchi That sounds like exactly the thing I'm looking for! I'll check if changing that will do me any good when I'm back at the office tomorrow, but this pretty much sounds like the issue.
    – Rem.co
    Commented Sep 10, 2012 at 20:00
  • @Buchi That was the problem indeed, a higher activemq.prefetchSize header while subscribing to the queue did fetch the remaining messages. Unfortunately frames after the first one are incorrectly terminated (or incorrectly parsed by the PHP Stomp client, I'm guessing the latter) by a "�" char instead of \x00, which collates all remaining messages in the body field of the second message. I think I'm sticking with my original plan for now: feed messages back to the queue. Care to transform your comment into an answer so I can accept it as correct anyway?
    – Rem.co
    Commented Sep 11, 2012 at 8:40

1 Answer 1

2

I think this is an issue due to setting activemq.prefetchSize to 1. ActiveMQ uses a prefetch size on how many messages can be dispatched to a consumer at any point in time. Once the prefetch size is reached, no more messages are dispatched to the consumer until the consumer starts sending back acknowledgements. Increasing the prefetch size should fix your issue to my best knowledge.

Please read http://activemq.apache.org/what-is-the-prefetch-limit-for.html for more details on prefetch limit.

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.