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.

2 files changed, 130 insertions, 37 deletions

diff --git a/spec/integration/indirector/catalog/queue.rb b/spec/integration/indirector/catalog/queue.rb
new file mode 100755
index 000000000..22f29aac3
--- /dev/null
+++ b/spec/integration/indirector/catalog/queue.rb
@@ -0,0 +1,59 @@
+#!/usr/bin/env ruby
+
+Dir.chdir(File.dirname(__FILE__)) { (s = lambda { |f| File.exist?(f) ? require(f) : Dir.chdir("..") { s.call(f) } }).call("spec/spec_helper.rb") }
+
+require 'puppet/resource/catalog'
+
+Puppet::Resource::Catalog.indirection.terminus(:queue)
+
+describe Puppet::Resource::Catalog::Queue do
+    before do
+        @catalog = Puppet::Resource::Catalog.new
+
+        @one = Puppet::Resource.new(:file, "/one")
+        @two = Puppet::Resource.new(:file, "/two")
+        @catalog.add_resource(@one, @two)
+
+        @catalog.add_edge(@one, @two)
+
+        Puppet[:trace] = true
+    end
+
+    after { Puppet.settings.clear }
+
+    it "should render catalogs to json and send them via the queue client when catalogs are saved" do
+        terminus = Puppet::Resource::Catalog.indirection.terminus(:queue)
+
+        client = mock 'client'
+        terminus.stubs(:client).returns client
+
+        client.expects(:send_message).with(:catalog, @catalog.to_json)
+
+        request = Puppet::Indirector::Request.new(:catalog, :save, "foo", :instance => @catalog)
+
+        terminus.save(request)
+    end
+
+    it "should intern catalog messages when they are passed via a subscription" do
+        client = mock 'client'
+        Puppet::Resource::Catalog::Queue.stubs(:client).returns client
+
+        json = @catalog.to_json
+
+        client.expects(:subscribe).with(:catalog).yields(json)
+
+        Puppet.expects(:err).never
+
+        result = []
+        Puppet::Resource::Catalog::Queue.subscribe do |catalog|
+            result << catalog
+        end
+
+        catalog = result.shift
+        catalog.should be_instance_of(Puppet::Resource::Catalog)
+
+        catalog.resource(:file, "/one").should be_instance_of(Puppet::Resource)
+        catalog.resource(:file, "/two").should be_instance_of(Puppet::Resource)
+        catalog.should be_edge(catalog.resource(:file, "/one"), catalog.resource(:file, "/two"))
+    end
+end
diff --git a/spec/unit/indirector/queue.rb b/spec/unit/indirector/queue.rb
index 0e9074440..3bc066873 100755
--- a/spec/unit/indirector/queue.rb
+++ b/spec/unit/indirector/queue.rb
@@ -4,35 +4,32 @@ require File.dirname(__FILE__) + '/../../spec_helper'
 require 'puppet/indirector/queue'
 
 class Puppet::Indirector::Queue::TestClient
-    def self.reset
-        @queues = {}
-    end
+end
+
+class FooExampleData
+    attr_accessor :name
 
-    def self.queues
-        @queues ||= {}
+    def self.json_create(json)
+        new(json['data'].to_sym)
     end
 
-    def subscribe(queue)
-        stack = self.class.queues[queue] ||= []
-        while stack.length > 0 do
-            yield(stack.shift)
-        end
+    def initialize(name = nil)
+        @name = name if name
     end
 
-    def send_message(queue, message)
-        stack = self.class.queues[queue] ||= []
-        stack.push(message)
-        queue
+    def render(format = :json)
+        to_json
     end
-end
 
-class FooExampleData
-    attr_accessor :name
+    def to_json(*args)
+        {:json_class => self.class.to_s, :data => name}.to_json(*args)
+    end
 end
 
 describe Puppet::Indirector::Queue do
     before :each do
-        @indirection = stub 'indirection', :name => :my_queue, :register_terminus_type => nil
+        @model = mock 'model'
+        @indirection = stub 'indirection', :name => :my_queue, :register_terminus_type => nil, :model => @model
         Puppet::Indirector::Indirection.stubs(:instance).with(:my_queue).returns(@indirection)
         @store_class = Class.new(Puppet::Indirector::Queue) do
             def self.to_s
@@ -48,40 +45,77 @@ describe Puppet::Indirector::Queue do
         Puppet.settings.stubs(:value).returns("bogus setting data")
         Puppet.settings.stubs(:value).with(:queue_type).returns(:test_client)
         Puppet::Util::Queue.stubs(:queue_type_to_class).with(:test_client).returns(Puppet::Indirector::Queue::TestClient)
-        Puppet::Indirector::Queue::TestClient.reset
 
         @request = stub 'request', :key => :me, :instance => @subject
     end
 
+    it "should require JSON" do
+        Puppet.features.expects(:json?).returns false
+
+        lambda { @store_class.new }.should raise_error(ArgumentError)
+    end
+
     it 'should use the correct client type and queue' do
         @store.queue.should == :my_queue
         @store.client.should be_an_instance_of(Puppet::Indirector::Queue::TestClient)
     end
 
-    it 'should use render() to convert object to message' do
-        @store.expects(:render).with(@subject).once
-        @store.save(@request)
-    end
+    describe "when saving" do
+        it 'should render the instance using json' do
+            @subject.expects(:render).with(:json)
+            @store.client.stubs(:send_message)
+            @store.save(@request)
+        end
+
+        it "should send the rendered message to the appropriate queue on the client" do
+            @subject.expects(:render).returns "myjson"
 
-    it 'should save and restore with the appropriate queue, and handle subscribe block' do
-        @subject_two = @subject_class.new
-        @subject_two.name = :too
-        @store.save(@request)
-        @store.save(stub('request_two', :key => 'too', :instance => @subject_two))
+            @store.client.expects(:send_message).with(:my_queue, "myjson")
 
-        received = []
-        @store_class.subscribe do |obj|
-            received.push(obj)
+            @store.save(@request)
         end
 
-        received[0].name.should == @subject.name
-        received[1].name.should == @subject_two.name
+        it "should catch any exceptions raised" do
+            @store.client.expects(:send_message).raises ArgumentError
+
+            lambda { @store.save(@request) }.should raise_error(Puppet::Error)
+        end
     end
 
-    it 'should use intern() to convert message to object with subscribe()' do
-        @store.save(@request)
-        @store_class.expects(:intern).with(@store.render(@subject)).once
-        @store_class.subscribe {|o| o }
+    describe "when subscribing to the queue" do
+        before do
+            @store_class.stubs(:model).returns @model
+        end
+
+        it "should use the model's Format support to intern the message from json" do
+            @model.expects(:convert_from).with(:json, "mymessage")
+
+            @store_class.client.expects(:subscribe).yields("mymessage")
+            @store_class.subscribe {|o| o }
+        end
+
+        it "should yield each interned received message" do
+            @model.stubs(:convert_from).returns "something"
+
+            @subject_two = @subject_class.new
+            @subject_two.name = :too
+
+            @store_class.client.expects(:subscribe).with(:my_queue).multiple_yields(@subject, @subject_two)
+
+            received = []
+            @store_class.subscribe do |obj|
+                received.push(obj)
+            end
+
+            received.should == %w{something something}
+        end
+
+        it "should log but not propagate errors" do
+            @store_class.client.expects(:subscribe).yields("foo")
+            @store_class.expects(:intern).raises ArgumentError
+            Puppet.expects(:err)
+            @store_class.subscribe {|o| o }
+        end
     end
 end