lib/zk/threaded_callback.rb in zk-1.2.0 vs lib/zk/threaded_callback.rb in zk-1.3.0

@@ -9,13 +9,15 @@
     attr_reader :callback
 
     def initialize(callback=nil, &blk)
       @callback = callback || blk
       @mutex = Monitor.new
-      @queue = Queue.new
-      @running = true
-      setup_dispatch_thread
+
+      @mutex.synchronize do
+        @running = true
+        reopen_after_fork!
+      end
     end
 
     def running?
       @mutex.synchronize { @running }
     end
@@ -34,12 +36,25 @@
 
     def call(*args)
       @queue.push(args)
     end
 
+    # called after a fork to replace a dead delivery thread
+    # special case, there should be ONLY ONE THREAD RUNNING, 
+    # (the one that survived the fork)
+    #
+    # @private
+    def reopen_after_fork!
+      return unless @running
+      return if @thread and @thread.alive?
+      @mutex = Monitor.new
+      @queue = Queue.new
+      @thread = spawn_dispatch_thread()
+    end
+
     protected
-      def setup_dispatch_thread
-        @thread ||= Thread.new do
+      def spawn_dispatch_thread
+        Thread.new do
           while running?
             args = @queue.pop
             break if args == KILL_TOKEN
             begin
               callback.call(*args)