client.rb in blather-2.0.0

- old
+ new

@@ -30,10 +30,25 @@
   #     client.register_handler :message, :chat?, :body do |m|
   #       client.write Blather::Stanza::Message.new(m.from, "You sent: #{m.body}")
   #     end
   #
   class Client
+    class Job
+      include SuckerPunch::Job
+      class_attribute :client
+
+      def perform(stanza)
+        client.handle_data stanza
+      end
+
+      def self.shutdown
+        SuckerPunch::Queue.all.each do |queue|
+          queue.shutdown if queue.name == to_s
+        end
+      end
+    end
+
     attr_reader :jid,
                 :roster,
                 :caps,
                 :queue_size
 
@@ -60,10 +75,11 @@
       @state = :initializing
 
       @status = Stanza::Presence::Status.new
       @handlers = {}
       @tmp_handlers = {}
+      @tmp_handlers_mutex = Mutex.new
       @filters = {:before => [], :after => []}
       @roster = Roster.new self
       @caps = Stanza::Capabilities.new
       @queue_size = 5
 
@@ -122,11 +138,13 @@
     # the JID and live only until a stanza with said ID is received.
     #
     # @param [#to_s] id the ID of the stanza that should be handled
     # @yield [Blather::Stanza] stanza the incomming stanza
     def register_tmp_handler(id, &handler)
-      @tmp_handlers[id.to_s] = handler
+      @tmp_handlers_mutex.synchronize do
+        @tmp_handlers[id.to_s] = handler
+      end
     end
 
     # Clear handlers with given guards
     #
     # @param [Symbol, nil] type remove filters for a specific handler
@@ -192,11 +210,11 @@
     end
 
     # @private
     def receive_data(stanza)
       if handler_queue
-        handler_queue << stanza
+        handler_queue.perform_async stanza
       else
         handle_data stanza
       end
     end
 
@@ -227,13 +245,15 @@
     end
 
     # @private
     def handler_queue
       return if queue_size == 0
-      @handler_queue ||= GirlFriday::WorkQueue.new :handle_stanza, :size => queue_size do |stanza|
-        handle_data stanza
-      end
+      return @handler_queue if @handler_queue
+      @handler_queue = Class.new(Job)
+      @handler_queue.client = self
+      @handler_queue.workers queue_size
+      return @handler_queue
     end
 
     protected
 
     def stream
@@ -291,10 +311,14 @@
         catch(:pass) { call_handler filter, guards, stanza }
       end
     end
 
     def handle_stanza(stanza)
-      if handler = @tmp_handlers.delete(stanza.id)
+      handler = @tmp_handlers_mutex.synchronize do
+        @tmp_handlers.delete(stanza.id)
+      end
+
+      if handler
         handler.call stanza
       else
         stanza.handler_hierarchy.each do |type|
           break if call_handler_for(type, stanza)
         end