lock around unconfirmed

lib/amqp/client/channel.rb

@@ -23,6 +23,7 @@ def initialize(connection, id)
           @on_return = nil
           @confirm = nil
           @unconfirmed = ::Queue.new
+          @unconfirmed_lock = Mutex.new
           @unconfirmed_empty = ::SizedQueue.new(1)
           @basic_gets = ::Queue.new
         end
@@ -407,7 +408,9 @@ def confirm_select(no_wait: false)
         # Block until all publishes messages are confirmed
         # @return [Boolean] True if all message where positivly acknowledged, false if not
         def wait_for_confirms
-          return true if @unconfirmed.empty?
+          @unconfirmed_lock.synchronize do
+            return true if @unconfirmed.empty?
+          end
 
           ok = @unconfirmed_empty.pop
           raise Error::Closed.new(@id, *@closed) if ok.nil?
@@ -419,24 +422,24 @@ def wait_for_confirms
         # @api private
         def confirm(args)
           ack_or_nack, delivery_tag, multiple = *args
-          loop do
-            tag = @unconfirmed.pop(true)
-            break if tag == delivery_tag
-            next if multiple && tag < delivery_tag
-
-            @unconfirmed << tag # requeue
-          rescue ThreadError
-            break
+          @unconfirmed_lock.synchronize do
+            loop do
+              tag = @unconfirmed.pop(true)
+              break if tag == delivery_tag
+              next if multiple && tag < delivery_tag
+
+              @unconfirmed.push(tag) # requeue
+            rescue ThreadError
+              break
+            end
+            return unless @unconfirmed.empty?
           end
-          return unless @unconfirmed.empty?
 
           ok = ack_or_nack == :ack
-          @unconfirmed_empty.push(ok) until @unconfirmed_empty.num_waiting.zero?
-          begin
-            @unconfirmed_empty.pop(true) # don't leave a residual ok in the queue
-          rescue ThreadError
-            nil
+          until @unconfirmed_empty.num_waiting.zero?
+            @unconfirmed_empty.push(ok, timeout: 0.001) || break
           end
+          @unconfirmed_empty.pop(timeout: 0) # don't leave a residual ok in the queue
         end
 
         # @!endgroup