Merge pull request #298 from CopernicaMarketingSoftware/synchronous-fix

fix for incorrect handling of synchronous flag

Author: EmielBruijntjes

include/amqpcpp/channelimpl.h

@@ -574,10 +574,10 @@ class ChannelImpl : public Watchable, public std::enable_shared_from_this<Channe
     }
 
     /**
-     *  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 @@ class ChannelImpl : public Watchable, public std::enable_shared_from_this<Channe
         // 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 @@ class ChannelImpl : public Watchable, public std::enable_shared_from_this<Channe
     {
         // 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 @@ class BasicCancelOKFrame : public BasicFrame
         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 @@ class BasicConsumeOKFrame : public BasicFrame
         if (!channel) return false;
 
         // report
-        if (channel->reportSuccess(consumerTag())) channel->onSynchronized();
+        channel->reportSuccess(consumerTag());
 
         // done
         return true;

src/basicgetemptyframe.h

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

src/basicqosokframe.h

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

src/channelcloseokframe.h

@@ -72,7 +72,7 @@ class ChannelCloseOKFrame : public ChannelFrame
         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 @@ class ChannelFlowOKFrame : public ChannelFrame
         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 @@ class ConfirmSelectOKFrame : public ConfirmFrame
         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();