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.
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?