Retry using the waiting queue so as to preserve message ordering

3 files changed, 32 insertions, 11 deletions

diff --git a/plugins/Irc/IrcPlugin.php b/plugins/Irc/IrcPlugin.php
index 54b758521..e073d6f13 100644
--- a/plugins/Irc/IrcPlugin.php
+++ b/plugins/Irc/IrcPlugin.php
@@ -164,6 +164,7 @@ class IrcPlugin extends ImPlugin {
                                                  false, 'PRI', null, null, true),
                                    new ColumnDef('data', 'blob', null, false),
                                    new ColumnDef('prioritise', 'tinyint', 1, false),
+                                   new ColumnDef('attempts', 'integer', null, false),
                                    new ColumnDef('created', 'datetime', null, false),
                                    new ColumnDef('claimed', 'datetime')));
 
diff --git a/plugins/Irc/Irc_waiting_message.php b/plugins/Irc/Irc_waiting_message.php
index 05f72754c..59eec63d8 100644
--- a/plugins/Irc/Irc_waiting_message.php
+++ b/plugins/Irc/Irc_waiting_message.php
@@ -10,6 +10,7 @@ class Irc_waiting_message extends Memcached_DataObject {
     public $id;                              // int primary_key not_null auto_increment

     public $data;                            // blob not_null

     public $prioritise;                      // tinyint(1) not_null

+    public $attempts;                        // int not_null

     public $created;                         // datetime() not_null

     public $claimed;                         // datetime()

 

@@ -112,6 +113,22 @@ class Irc_waiting_message extends Memcached_DataObject {
     }

 

     /**

+    * Increment the attempts count

+    *

+    * @return void

+    * @throws Exception

+    */

+    public function incAttempts() {

+        $orig = clone($this);

+        $this->attempts++;

+        $result = $this->update($orig);

+

+        if (!$result) {

+            throw Exception(sprintf(_m("Could not increment attempts count for %d"), $this->id));

+        }

+    }

+

+    /**

      * Release a claimed item.

      */

     public function releaseClaim() {

diff --git a/plugins/Irc/ircmanager.php b/plugins/Irc/ircmanager.php
index aa0ff3539..5f55e6b34 100644
--- a/plugins/Irc/ircmanager.php
+++ b/plugins/Irc/ircmanager.php
@@ -101,19 +101,21 @@ class IrcManager extends ImManager {
                     $this->messageWaiting = false;
                     return;
                 }
-                $data = unserialize($wm->data);
 
-                if (!$this->send_raw_message($data)) {
-                    $this->plugin->enqueue_outgoing_raw(
-                        array(
-                            'type' => 'message',
-                            'prioritise' => $data['prioritise'],
-                            'data' => $data['data']
-                        )
-                    );
+                $data = unserialize($wm->data);
+                $wm->incAttempts();
+
+                if ($this->send_raw_message($data)) {
+                    $wm->delete();
+                } else {
+                    if ($wm->attempts <= common_config('queue', 'max_retries')) {
+                        // Try again next idle
+                        $wm->releaseClaim();
+                    } else {
+                        // Exceeded the maximum number of retries
+                        $wm->delete();
+                    }
                 }
-
-                $wm->delete();
             }
         }
     }
@@ -276,6 +278,7 @@ class IrcManager extends ImManager {
 
         $wm->data       = serialize($data);
         $wm->prioritise = $data['prioritise'];
+        $wm->attempts   = 0;
         $wm->created    = common_sql_now();
         $result         = $wm->insert();