Enable json-specified objects in queue_item frames

lib/dbqueuemanager.php

@@ -79,9 +79,14 @@ class DBQueueManager extends QueueManager
         }
 
         $queue = $qi->transport;
+        try {
             $item = $this->decode($qi->frame);
+        } catch (Exception $e) {
+            $this->_log(LOG_INFO, "[$queue] Discarding: ".$e->getMessage());
+            $this->_done($qi);
+            return true;
+        }
 
-        if ($item) {
         $rep = $this->logrep($item);
         $this->_log(LOG_DEBUG, "Got $rep for transport $queue");
         
@@ -98,10 +103,6 @@ class DBQueueManager extends QueueManager
             $this->_log(LOG_INFO, "[$queue:$rep] No handler for queue $queue; discarding.");
             $this->_done($qi);
         }
-        } else {
-            $this->_log(LOG_INFO, "[$queue] Got empty/deleted item, discarding");
-            $this->_done($qi);
-        }
         return true;
     }
 

lib/queuemanager.php

@@ -177,7 +177,39 @@ abstract class QueueManager extends IoManager
      */
     protected function decode($frame)
     {
-        return unserialize($frame);
+        $object = unserialize($frame);
+
+        // If it is a string, we really store a JSON object in there
+        if (is_string($object)) {
+            $json = json_decode($object);
+            if ($json === null) {
+                throw new Exception('Bad frame in queue item');
+            }
+
+            // The JSON object has a type parameter which contains the class
+            if (empty($json->type)) {
+                throw new Exception('Type not specified for queue item');
+            }
+            if (!is_a($json->type, 'Managed_DataObject', true)) {
+                throw new Exception('Managed_DataObject class does not exist for queue item');
+            }
+
+            // And each of these types should have a unique id (or uri)
+            if (isset($json->id) && !empty($json->id)) {
+                $object = call_user_func(array($json->type, 'getKV'), 'id', $json->id);
+            } elseif (isset($json->uri) && !empty($json->uri)) {
+                $object = call_user_func(array($json->type, 'getKV'), 'uri', $json->uri);
+            }
+
+            // But if no object was found, there's nothing we can handle
+            if (!$object instanceof Managed_DataObject) {
+                throw new Exception('Queue item frame referenced a non-existant object');
+            }
+        }
+
+        // If the frame was not a string, it's either an array or an object.
+
+        return $object;
     }
 
     /**

lib/stompqueuemanager.php

@@ -498,8 +498,9 @@ class StompQueueManager extends QueueManager
         // @fixme detect failing site switches
         $this->switchSite($site);
 
+        try {
             $item = $this->decode($message['payload']);
-        if (empty($item)) {
+        } catch (Exception $e) {
             $this->_log(LOG_ERR, "Skipping empty or deleted item in queue $queue from $host");
             $this->stats('baditem', $queue);
             return false;