Merge pull request #298 from CopernicaMarketingSoftware/synchronous-fix

fix for incorrect handling of synchronous flag

include/amqpcpp/channelimpl.h

@@ -574,10 +574,10 @@ public:
     }
 
     /**
-     *  Signal the channel that a synchronous operation was completed.
-     *  After this operation, waiting frames can be sent out.
+     *  Signal the channel that a synchronous operation was completed, and that any
+     *  queued frames can be sent out.
      */
-    void onSynchronized();
+    void flush();
 
     /**
      *  Report to the handler that the channel is opened
@@ -596,8 +596,8 @@ public:
         // inform handler
         if (_readyCallback) _readyCallback();
 
-        // if the monitor is still valid, we exit synchronous mode now
-        if (monitor.valid()) onSynchronized();
+        // if the monitor is still valid, we flush any waiting operations 
+        if (monitor.valid()) flush();
     }
 
     /**
@@ -644,10 +644,10 @@ public:
     {
         // skip if there is no oldest callback
         if (!_oldestCallback) return true;
-        
-        // the last (possibly synchronous) operation was received, so we're no longer in synchronous mode
-        if (_synchronous && _queue.empty()) _synchronous = false;
 
+        // flush the queue, which will send the next operation if the current operation was synchronous
+        flush();
+        
         // we are going to call callbacks that could destruct the channel
         Monitor monitor(this);
 

src/basiccancelokframe.h

@@ -94,7 +94,7 @@ public:
         if (!channel) return false;
 
         // report
-        if (channel->reportSuccess<const std::string&>(consumerTag())) channel->onSynchronized();
+        channel->reportSuccess<const std::string&>(consumerTag());
 
         // done
         return true;

src/basicconsumeokframe.h

@@ -94,7 +94,7 @@ public:
         if (!channel) return false;
 
         // report
-        if (channel->reportSuccess(consumerTag())) channel->onSynchronized();
+        channel->reportSuccess(consumerTag());
 
         // done
         return true;

src/basicgetemptyframe.h

@@ -84,7 +84,7 @@ public:
         if (!channel) return false;
 
         // report
-        if (channel->reportSuccess()) channel->onSynchronized();
+        channel->reportSuccess();
 
         // done
         return true;

src/basicqosokframe.h

@@ -67,7 +67,7 @@ public:
         if (!channel) return false;
 
         // report
-        if (channel->reportSuccess()) channel->onSynchronized();
+        channel->reportSuccess();
 
         // done
         return true;

src/channelcloseokframe.h

@@ -72,7 +72,7 @@ public:
         if (!channel) return false;
 
         // report that the channel is closed
-        if (channel->reportClosed()) channel->onSynchronized();
+        channel->reportClosed();
 
         // done
         return true;

src/channelflowokframe.h

@@ -93,7 +93,7 @@ public:
         if (!channel) return false;
 
         // report success for the call
-        if (channel->reportSuccess()) channel->onSynchronized();
+        channel->reportSuccess();
 
         // done
         return true;

src/channelimpl.cpp

@@ -754,7 +754,7 @@ bool ChannelImpl::send(const Frame &frame)
  *  Signal the channel that a synchronous operation was completed. After 
  *  this operation, waiting frames can be sent out.
  */
-void ChannelImpl::onSynchronized()
+void ChannelImpl::flush()
 {
     // we are no longer waiting for synchronous operations
     _synchronous = false;

src/confirmselectokframe.h

@@ -75,7 +75,7 @@ public:
         if(!channel) return false;
 
         // report that the channel is open
-        if (channel->reportSuccess()) channel->onSynchronized();
+        channel->reportSuccess();
 
         // done
         return true;

src/deferredget.cpp

@@ -62,7 +62,7 @@ const std::shared_ptr<Deferred> &DeferredGet::reportSuccess() const
 void DeferredGet::complete()
 {
     // the channel is now synchronized, delayed frames may now be sent
-    _channel->onSynchronized();
+    _channel->flush();
     
     // pass on to normal implementation
     DeferredExtReceiver::complete();

src/exchangebindokframe.h

@@ -67,7 +67,7 @@ public:
         if(!channel) return false;
 
         // report to handler
-        if (channel->reportSuccess()) channel->onSynchronized();
+        channel->reportSuccess();
 
         // done
         return true;

src/exchangedeclareokframe.h

@@ -70,7 +70,7 @@ public:
         if(!channel) return false;
 
         // report exchange declare ok
-        if (channel->reportSuccess()) channel->onSynchronized();
+        channel->reportSuccess();
 
         // done
         return true;

src/exchangedeleteokframe.h

@@ -71,7 +71,7 @@ public:
         if(!channel) return false;
 
         // report to handler
-        if (channel->reportSuccess()) channel->onSynchronized();
+        channel->reportSuccess();
 
         // done
         return true;

src/exchangeunbindokframe.h

@@ -68,7 +68,7 @@ public:
         if(!channel) return false;
 
         // report to handler
-        if (channel->reportSuccess()) channel->onSynchronized();
+        channel->reportSuccess();
 
         // done
         return true;

src/queuebindokframe.h

@@ -69,7 +69,7 @@ public:
         if(!channel) return false;
 
         // report to handler
-        if (channel->reportSuccess()) channel->onSynchronized();
+        channel->reportSuccess();
 
         // done
         return true;

src/queuedeclareokframe.h

@@ -133,7 +133,7 @@ public:
         if (!channel) return false;
 
         // report success
-        if (channel->reportSuccess(name(), messageCount(), consumerCount())) channel->onSynchronized();
+        channel->reportSuccess(name(), messageCount(), consumerCount());
 
         // done
         return true;

src/queuedeleteokframe.h

@@ -94,7 +94,7 @@ public:
         if(!channel) return false;
 
         // report queue deletion success
-        if (channel->reportSuccess(this->messageCount())) channel->onSynchronized();
+        channel->reportSuccess(this->messageCount());
 
         // done
         return true;

src/queuepurgeokframe.h

@@ -94,7 +94,7 @@ public:
         if(!channel) return false;
 
         // report queue purge success
-        if (channel->reportSuccess(this->messageCount())) channel->onSynchronized();
+        channel->reportSuccess(this->messageCount());
 
         // done
         return true;

src/queueunbindokframe.h

@@ -73,7 +73,7 @@ public:
         if(!channel) return false;
 
         // report queue unbind success
-        if (channel->reportSuccess()) channel->onSynchronized();
+        channel->reportSuccess();
 
         // done
         return true;

src/transactioncommitokframe.h

@@ -74,7 +74,7 @@ public:
         if(!channel) return false;
 
         // report that the channel is open
-        if (channel->reportSuccess()) channel->onSynchronized();
+        channel->reportSuccess();
 
         // done
         return true;

src/transactionrollbackokframe.h

@@ -74,7 +74,7 @@ public:
         if(!channel) return false;
 
         // report that the channel is open
-        if (channel->reportSuccess()) channel->onSynchronized();
+        channel->reportSuccess();
 
         // done
         return true;

src/transactionselectokframe.h

@@ -74,7 +74,7 @@ public:
         if(!channel) return false;
 
         // report that the channel is open
-        if (channel->reportSuccess()) channel->onSynchronized();
+        channel->reportSuccess();
 
         // done
         return true;