Adamw has uploaded a new change for review.
https://gerrit.wikimedia.org/r/96820
Change subject: API for peeking at head messages in a queue
......................................................................
API for peeking at head messages in a queue
Change-Id: Id40d52d42a2348c15f0252016d55eeb256082f4a
---
M sites/all/modules/wmf_common/Queue.php
1 file changed, 26 insertions(+), 0 deletions(-)
git pull ssh://gerrit.wikimedia.org:29418/wikimedia/fundraising/crm
refs/changes/20/96820/1
diff --git a/sites/all/modules/wmf_common/Queue.php
b/sites/all/modules/wmf_common/Queue.php
index e694207..ea898f3 100644
--- a/sites/all/modules/wmf_common/Queue.php
+++ b/sites/all/modules/wmf_common/Queue.php
@@ -90,6 +90,32 @@
return $processed;
}
+ /**
+ * Preview several messages from the head of a queue, without ack'ing.
+ */
+ function peekMultiple( $queue, $count ) {
+ # TODO: prefetchPolicy, timeout as params?
+
+ $queue = $this->normalizeQueueName( $queue );
+ $con = $this->getFreshConnection();
+
+ $con->setReadTimeout( 1 );
+ # FIXME: probably the wrong value--does this buffer include msgs not
+ # specific to our subscription?
+ $con->prefetchSize = $count;
+
+ $con->subscribe( $queue, array( 'ack' => 'client' ) );
+ $messages = array();
+ while ( $con->hasFrameToRead() && $count-- >= 0 ) {
+ $msg = $con->readFrame();
+ if ( !$msg ) {
+ break;
+ }
+ $messages[] = $msg;
+ }
+ return $messages;
+ }
+
function getByCorrelationId( $queue, $correlationId ) {
$con = $this->getFreshConnection();
$properties = array(
--
To view, visit https://gerrit.wikimedia.org/r/96820
To unsubscribe, visit https://gerrit.wikimedia.org/r/settings
Gerrit-MessageType: newchange
Gerrit-Change-Id: Id40d52d42a2348c15f0252016d55eeb256082f4a
Gerrit-PatchSet: 1
Gerrit-Project: wikimedia/fundraising/crm
Gerrit-Branch: master
Gerrit-Owner: Adamw <[email protected]>
_______________________________________________
MediaWiki-commits mailing list
[email protected]
https://lists.wikimedia.org/mailman/listinfo/mediawiki-commits
