Merge with upstream

ruby-amqp · Apr 11, 2014 · ba0eee2 · ba0eee2
2 parents d3429ba + c9ba00b
commit ba0eee2
Showing 1 changed file with 17 additions and 15 deletions.
diff --git a/lib/bunny/channel.rb b/lib/bunny/channel.rb
@@ -539,8 +539,10 @@ def basic_publish(payload, exchange, routing_key, opts = {})
       opts[:priority]      ||= 0
 
       if @next_publish_seq_no > 0
-        @unconfirmed_set.add(@next_publish_seq_no)
-        @next_publish_seq_no += 1
+        @unconfirmed_set_mutex.synchronize do
+          @unconfirmed_set.add(@next_publish_seq_no)
+          @next_publish_seq_no += 1
+        end
       end
 
       frames = AMQ::Protocol::Basic::Publish.encode(@id,
@@ -1690,23 +1692,23 @@ def handle_basic_return(basic_return, properties, content)
 
     # @private
     def handle_ack_or_nack(delivery_tag, multiple, nack)
-      if nack
-        cloned_set = @unconfirmed_set.clone
+      @unconfirmed_set_mutex.synchronize do
+        if nack
+          cloned_set = @unconfirmed_set.clone
+          if multiple
+            cloned_set.keep_if { |i| i <= delivery_tag }
+            @nacked_set.merge(cloned_set)
+          else
+            @nacked_set.add(delivery_tag)
+          end
+        end
+
         if multiple
-          cloned_set.keep_if { |i| i <= delivery_tag }
-          @nacked_set.merge(cloned_set)
+          @unconfirmed_set.delete_if { |i| i <= delivery_tag }
         else
-          @nacked_set.add(delivery_tag)
+          @unconfirmed_set.delete(delivery_tag)
         end
-      end
 
-      if multiple
-        @unconfirmed_set.delete_if { |i| i <= delivery_tag }
-      else
-        @unconfirmed_set.delete(delivery_tag)
-      end
-
-      @unconfirmed_set_mutex.synchronize do
         @only_acks_received = (@only_acks_received && !nack)
 
         @confirms_continuations.push(true) if @unconfirmed_set.empty?