Mwalker has uploaded a new change for review.
https://gerrit.wikimedia.org/r/73880
Change subject: Force STOMP Reconnections on Selector Change
......................................................................
Force STOMP Reconnections on Selector Change
There is a bug, suspected to be related to backend buffering in
the STOMP 1.0 protocol, that causes selectors to not correctly
pick up messages after the first time. E.g. changing a selector
on an open connection will not show new messages.
STOMP 1.1+ clients should not need this because they do per
subscription tracking of messages.
Change-Id: Ib57f8650551f6a5090ef60742ed6ef22275be641
---
M SmashPig/Core/DataStores/StompDataStore.php
A SmashPig/Maintenance/OrphanCollider.php
M SmashPig/config_defaults.php
3 files changed, 58 insertions(+), 12 deletions(-)
git pull
ssh://gerrit.wikimedia.org:29418/wikimedia/fundraising/PaymentsListeners
refs/changes/80/73880/1
diff --git a/SmashPig/Core/DataStores/StompDataStore.php
b/SmashPig/Core/DataStores/StompDataStore.php
index 79a8288..7e75995 100644
--- a/SmashPig/Core/DataStores/StompDataStore.php
+++ b/SmashPig/Core/DataStores/StompDataStore.php
@@ -17,6 +17,15 @@
/** @var string Object on STOMP server we're subscribing and pushing to
*/
protected $queue_id = null;
+ /** @var int Timeout to apply to the STOMP connection */
+ protected $timeout = 0;
+
+ /**
+ * @var bool If true will force a reconnection everytime the selectors
change. (Solves a bug
+ * where things in the STOMP RX buffer are old but not detected as old.)
+ */
+ protected $refreshConnection = false;
+
/** @var bool If true, we have successfully subscribed to @see
$queue_id on @see $stompObj */
protected $subscribed = false;
@@ -28,6 +37,7 @@
*
* Will connection to the server at data-store/stomp/uri
*
+ * @throws DataStoreException
* @param string $queueName The data-store/stomp/queue/* object to
connect to.
*/
public function __construct( $queueName ) {
@@ -46,8 +56,26 @@
}
$this->queue_id = $c->val(
"data-store/stomp/queues/{$queueName}" );
- // Start the connection
+ // Get some more configuration variables
$this->uri = $c->val( 'data-store/stomp/uri' );
+ $this->timeout = $c->val( 'data-store/stomp/timeout' );
+ $this->refreshConnection = $c->val(
'data-store/stomp/refresh-connection' );
+
+ // Start the connection
+ $this->createBackingObject();
+ }
+
+ /**
+ * Destroy the STOMP data store connection.
+ */
+ public function __destruct() {
+ $this->deleteSubscription();
+ }
+
+ /**
+ * Creates the STOMP store backing object.
+ */
+ protected function createBackingObject() {
Logger::debug( "Attempting connection to STOMP server
'{$this->uri}'" );
$this->stompObj = new \Stomp( $this->uri );
if ( method_exists( $this->stompObj, 'connect' ) ) {
@@ -56,15 +84,7 @@
Logger::debug( "STOMP server connection success." );
// Post connection configuration
- $timeout = $c->val( 'data-store/stomp/timeout' );
- $this->stompObj->setReadTimeout( $timeout );
- }
-
- /**
- * Destroy the STOMP data store connection.
- */
- public function __destruct() {
- $this->deleteSubscription();
+ $this->stompObj->setReadTimeout( $this->timeout );
}
/**
@@ -117,6 +137,7 @@
*
* @param KeyedOpaqueStorableObject $protoObj Prototype to remove.
*
+ * @throws DataStoreException
* @return int Count of messages removed.
*/
public function removeObjects( KeyedOpaqueStorableObject $protoObj ) {
@@ -190,10 +211,12 @@
*
* If there were no messages fitting the filter, null will be returned.
*
- * @param string|null $type The class of message to retrieve
(if null retrieves all)
+ * @param null|string $type The class of message to retrieve
(if null retrieves all)
* @param null|string $id The correlation ID of the message
(if null retrieves all)
*
* @throws DataStoreTransactionException
+ * @throws DataSerializationException
+ * @throws DataStoreException
* @return KeyedOpaqueStorableObject|null
*/
public function queueGetObject( $type = null, $id = null ) {
@@ -246,7 +269,8 @@
* new objects are being returned with the
current STOMP
* transaction ID in them.
*
- * @returns object STOMP message object
+ * @throws DataStoreTransactionException
+ * @return object STOMP message object
*/
protected function queueGetObjectRaw( $type = null, $id = null,
$checkTail = true ) {
if ( $this->queueMsg ) {
@@ -305,6 +329,12 @@
} elseif ( $this->subscribed ) {
// We need to create a new subscription; but we also
have to delete the old one
$this->deleteSubscription();
+
+ if ( $this->refreshConnection ) {
+ // Apparently the backend STOMP library has
some issues clearing
+ // out its buffer so we get old stuff :(
+ $this->createBackingObject();
+ }
}
$sType = $type;
diff --git a/SmashPig/Maintenance/OrphanCollider.php
b/SmashPig/Maintenance/OrphanCollider.php
new file mode 100644
index 0000000..3dd9691
--- /dev/null
+++ b/SmashPig/Maintenance/OrphanCollider.php
@@ -0,0 +1,15 @@
+<?php
+/**
+ * Created by JetBrains PhpStorm.
+ * User: mwalker
+ * Date: 6/27/13
+ * Time: 6:18 PM
+ * To change this template use File | Settings | File Templates.
+ */
+
+namespace SmashPig\Maintenance;
+
+
+class OrphanCollider {
+
+}
\ No newline at end of file
diff --git a/SmashPig/config_defaults.php b/SmashPig/config_defaults.php
index 826ba3c..2ff526a 100644
--- a/SmashPig/config_defaults.php
+++ b/SmashPig/config_defaults.php
@@ -35,6 +35,7 @@
'uri' => 'tcp://localhost:61613',
'timeout' => 1,
+ 'refresh-connection' => false,
'queues' => array(
'limbo' => '/queue/limbo',
--
To view, visit https://gerrit.wikimedia.org/r/73880
To unsubscribe, visit https://gerrit.wikimedia.org/r/settings
Gerrit-MessageType: newchange
Gerrit-Change-Id: Ib57f8650551f6a5090ef60742ed6ef22275be641
Gerrit-PatchSet: 1
Gerrit-Project: wikimedia/fundraising/PaymentsListeners
Gerrit-Branch: master
Gerrit-Owner: Mwalker <[email protected]>
_______________________________________________
MediaWiki-commits mailing list
[email protected]
https://lists.wikimedia.org/mailman/listinfo/mediawiki-commits
