Switching Queueing to using JSON instead of YAML

This provides about a 75x speedup, so it's totally
worth it.  The downside is that queueing requires json,
but only on the server side.

4 files changed, 28 insertions, 16 deletions

diff --git a/lib/puppet/defaults.rb b/lib/puppet/defaults.rb
index 3d0ad4bb8..8743f859d 100644
--- a/lib/puppet/defaults.rb
+++ b/lib/puppet/defaults.rb
@@ -168,7 +168,7 @@ module Puppet
         :queue_source => ["stomp://localhost:61613/", "Which type of queue to use for asynchronous processing.  If your stomp server requires
             authentication, you can include it in the URI as long as your stomp client library is at least 1.1.1"],
         :async_storeconfigs => {:default => false, :desc => "Whether to use a queueing system to provide asynchronous database integration.
-            Requires that ``puppetqd`` be running.",
+            Requires that ``puppetqd`` be running and that 'JSON' support for ruby be installed.",
             :hook => proc do |value|
                 if value
                     # This reconfigures the terminii for Node, Facts, and Catalog
diff --git a/lib/puppet/indirector/queue.rb b/lib/puppet/indirector/queue.rb
index cd0e0c833..1fc72d6c1 100644
--- a/lib/puppet/indirector/queue.rb
+++ b/lib/puppet/indirector/queue.rb
@@ -1,6 +1,7 @@
 require 'puppet/indirector/terminus'
 require 'puppet/util/queue'
-require 'yaml'
+require 'puppet/util'
+require 'json'
 
 # Implements the <tt>:queue</tt> abstract indirector terminus type, for storing
 # model instances to a message queue, presumably for the purpose of out-of-process
@@ -20,6 +21,12 @@ require 'yaml'
 # creation is automatic and not a concern).
 class Puppet::Indirector::Queue < Puppet::Indirector::Terminus
     extend ::Puppet::Util::Queue
+    include Puppet::Util
+
+    def initialize(*args)
+        super
+        raise ArgumentError, "Queueing requires json support" unless Puppet.features.json?
+    end
 
     # Queue has no idiomatic "find"
     def find(request)
@@ -29,8 +36,11 @@ class Puppet::Indirector::Queue < Puppet::Indirector::Terminus
     # Place the request on the queue
     def save(request)
         begin
-            Puppet.info "Queueing catalog for %s" % request.key
-            client.send_message(queue, render(request.instance))
+            result = nil
+            benchmark :info, "Queued %s for %s" % [indirection.name, request.key] do
+                result = client.send_message(queue, request.instance.render(:json))
+            end
+            result
         rescue => detail
             raise Puppet::Error, "Could not write %s to queue: %s\nInstance::%s\n client : %s" % [request.key, detail,request.instance.to_s,client.to_s]
         end
@@ -49,15 +59,13 @@ class Puppet::Indirector::Queue < Puppet::Indirector::Terminus
         self.class.client
     end
 
-    # Formats the model instance associated with _request_ appropriately for message delivery.
-    # Uses YAML serialization.
-    def render(obj)
-        YAML::dump(obj)
-    end
-
     # converts the _message_ from deserialized format to an actual model instance.
     def self.intern(message)
-        YAML::load(message)
+        result = nil
+        benchmark :info, "Loaded queued %s" % [indirection.name] do
+            result = model.convert_from(:json, message)
+        end
+        result
     end
 
     # Provides queue subscription functionality; for a given indirection, use this method on the terminus
@@ -68,9 +76,8 @@ class Puppet::Indirector::Queue < Puppet::Indirector::Terminus
             begin
                 yield(self.intern(msg))
             rescue => detail
-                # really, this should log the exception rather than raise it all the way up the stack;
-                # we don't want exceptions resulting from a single message bringing down a listener
-                raise Puppet::Error, "Error occured with subscription to queue %s for indirection %s: %s" % [queue, indirection_name, detail]
+                puts detail.backtrace if Puppet[:trace]
+                Puppet.err "Error occured with subscription to queue %s for indirection %s: %s" % [queue, indirection_name, detail]
             end
         end
     end
diff --git a/lib/puppet/relationship.rb b/lib/puppet/relationship.rb
index 8efebf1e3..96a71c3f1 100644
--- a/lib/puppet/relationship.rb
+++ b/lib/puppet/relationship.rb
@@ -42,7 +42,7 @@ class Puppet::Relationship
     def initialize(source, target, options = {})
         @source, @target = source, target
 
-        options ||= {}
+        options = (options || {}).inject({}) { |h,a| h[a[0].to_sym] = a[1]; h }
         [:callback, :event].each do |option|
             if value = options[option]
                 send(option.to_s + "=", value)
diff --git a/lib/puppet/resource/catalog.rb b/lib/puppet/resource/catalog.rb
index 68e6d7de5..42e92f407 100644
--- a/lib/puppet/resource/catalog.rb
+++ b/lib/puppet/resource/catalog.rb
@@ -402,12 +402,14 @@ class Puppet::Resource::Catalog < Puppet::SimpleGraph
         end
 
         if resources = data['resources']
+            resources = JSON.parse(resources) if resources.is_a?(String)
             resources.each do |res|
                 resource_from_json(result, res)
             end
         end
 
         if edges = data['edges']
+            edges = JSON.parse(edges) if edges.is_a?(String)
             edges.each do |edge|
                 edge_from_json(result, edge)
             end
@@ -436,7 +438,10 @@ class Puppet::Resource::Catalog < Puppet::SimpleGraph
     def self.resource_from_json(result, res)
         # If no json_class information was presented, we manually find
         # the class.
-        res = Puppet::Resource.from_json(res) if res.is_a?(Hash)
+        if res.is_a?(Hash)
+            res = res['data'] if res['json_class']
+            res = Puppet::Resource.from_json(res)
+        end
         result.add_resource(res)
     end