From c82fce8ec9f48e5997326fd02bd582c680288fe3 Mon Sep 17 00:00:00 2001
From: Michael van der Werve <mvdwerve@outlook.com>
Date: Wed, 19 Jun 2019 09:38:25 +0200
Subject: [PATCH 1/2] fix for incorrect handling of synchronous flag

---
 include/amqpcpp/channelimpl.h    | 17 +++++++++++++++--
 src/basiccancelokframe.h         |  2 +-
 src/basicconsumeokframe.h        |  2 +-
 src/basicgetemptyframe.h         |  2 +-
 src/basicqosokframe.h            |  2 +-
 src/channelcloseokframe.h        |  2 +-
 src/channelflowokframe.h         |  2 +-
 src/confirmselectokframe.h       |  2 +-
 src/exchangebindokframe.h        |  2 +-
 src/exchangedeclareokframe.h     |  2 +-
 src/exchangedeleteokframe.h      |  2 +-
 src/exchangeunbindokframe.h      |  2 +-
 src/queuebindokframe.h           |  2 +-
 src/queuedeclareokframe.h        |  2 +-
 src/queuedeleteokframe.h         |  2 +-
 src/queuepurgeokframe.h          |  2 +-
 src/queueunbindokframe.h         |  2 +-
 src/transactioncommitokframe.h   |  2 +-
 src/transactionrollbackokframe.h |  2 +-
 src/transactionselectokframe.h   |  2 +-
 20 files changed, 34 insertions(+), 21 deletions(-)

diff --git a/include/amqpcpp/channelimpl.h b/include/amqpcpp/channelimpl.h
index 71a08a33..89b93904 100644
--- a/include/amqpcpp/channelimpl.h
+++ b/include/amqpcpp/channelimpl.h
@@ -645,8 +645,17 @@ 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;
+        // remember whether or not we were synchronous
+        bool synchronous = _synchronous;
+
+        // is the queue empty at this moment?
+        bool empty = _queue.empty();
+        
+        // the last (possibly synchronous) operation was received, so we're no longer in synchronous mode. this
+        // is an optimization that makes sure that the first instruction _after_ a synchronous instruction
+        // that is installed during the success callback we make later does _not_ need to be buffered, but can be
+        // sent directly
+        if (synchronous && empty) _synchronous = false;
 
         // we are going to call callbacks that could destruct the channel
         Monitor monitor(this);
@@ -661,6 +670,10 @@ class ChannelImpl : public Watchable, public std::enable_shared_from_this<Channe
         // leap out if channel no longer exists
         if (!monitor.valid()) return false;
 
+        // if we were synchronous, but there were still messages in the queue, we process the queue now, because the synchronous
+        // operation was finished, and its callback was made, which means we're no longer in synchronous mode
+        if (synchronous && !empty) onSynchronized();
+
         // set the oldest callback
         _oldestCallback = next;
 
diff --git a/src/basiccancelokframe.h b/src/basiccancelokframe.h
index a894c5c3..d6b31ead 100644
--- a/src/basiccancelokframe.h
+++ b/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;
diff --git a/src/basicconsumeokframe.h b/src/basicconsumeokframe.h
index 9ec63976..129419d5 100644
--- a/src/basicconsumeokframe.h
+++ b/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;
diff --git a/src/basicgetemptyframe.h b/src/basicgetemptyframe.h
index 656c2df0..3efa9750 100644
--- a/src/basicgetemptyframe.h
+++ b/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;
diff --git a/src/basicqosokframe.h b/src/basicqosokframe.h
index 6698668a..72aa90a2 100644
--- a/src/basicqosokframe.h
+++ b/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;
diff --git a/src/channelcloseokframe.h b/src/channelcloseokframe.h
index 1c8bd1ed..9be59a93 100644
--- a/src/channelcloseokframe.h
+++ b/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;
diff --git a/src/channelflowokframe.h b/src/channelflowokframe.h
index 2111253c..d5c5aa03 100644
--- a/src/channelflowokframe.h
+++ b/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;
diff --git a/src/confirmselectokframe.h b/src/confirmselectokframe.h
index 4a20992d..c91c0184 100644
--- a/src/confirmselectokframe.h
+++ b/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;
diff --git a/src/exchangebindokframe.h b/src/exchangebindokframe.h
index 6116eca7..bbba17c1 100644
--- a/src/exchangebindokframe.h
+++ b/src/exchangebindokframe.h
@@ -67,7 +67,7 @@ class ExchangeBindOKFrame : public ExchangeFrame
         if(!channel) return false;
 
         // report to handler
-        if (channel->reportSuccess()) channel->onSynchronized();
+        channel->reportSuccess();
 
         // done
         return true;
diff --git a/src/exchangedeclareokframe.h b/src/exchangedeclareokframe.h
index 83347572..8d560305 100644
--- a/src/exchangedeclareokframe.h
+++ b/src/exchangedeclareokframe.h
@@ -70,7 +70,7 @@ class ExchangeDeclareOKFrame : public ExchangeFrame
         if(!channel) return false;
 
         // report exchange declare ok
-        if (channel->reportSuccess()) channel->onSynchronized();
+        channel->reportSuccess();
 
         // done
         return true;
diff --git a/src/exchangedeleteokframe.h b/src/exchangedeleteokframe.h
index 7f6dba07..624cc84a 100644
--- a/src/exchangedeleteokframe.h
+++ b/src/exchangedeleteokframe.h
@@ -71,7 +71,7 @@ class ExchangeDeleteOKFrame : public ExchangeFrame
         if(!channel) return false;
 
         // report to handler
-        if (channel->reportSuccess()) channel->onSynchronized();
+        channel->reportSuccess();
 
         // done
         return true;
diff --git a/src/exchangeunbindokframe.h b/src/exchangeunbindokframe.h
index 511b8fe2..3a300507 100644
--- a/src/exchangeunbindokframe.h
+++ b/src/exchangeunbindokframe.h
@@ -68,7 +68,7 @@ class ExchangeUnbindOKFrame : public ExchangeFrame
         if(!channel) return false;
 
         // report to handler
-        if (channel->reportSuccess()) channel->onSynchronized();
+        channel->reportSuccess();
 
         // done
         return true;
diff --git a/src/queuebindokframe.h b/src/queuebindokframe.h
index b80b11b9..e4e55e89 100644
--- a/src/queuebindokframe.h
+++ b/src/queuebindokframe.h
@@ -69,7 +69,7 @@ class QueueBindOKFrame : public QueueFrame
         if(!channel) return false;
 
         // report to handler
-        if (channel->reportSuccess()) channel->onSynchronized();
+        channel->reportSuccess();
 
         // done
         return true;
diff --git a/src/queuedeclareokframe.h b/src/queuedeclareokframe.h
index 99ad1e86..ca545e9b 100644
--- a/src/queuedeclareokframe.h
+++ b/src/queuedeclareokframe.h
@@ -133,7 +133,7 @@ class QueueDeclareOKFrame : public QueueFrame
         if (!channel) return false;
 
         // report success
-        if (channel->reportSuccess(name(), messageCount(), consumerCount())) channel->onSynchronized();
+        channel->reportSuccess(name(), messageCount(), consumerCount());
 
         // done
         return true;
diff --git a/src/queuedeleteokframe.h b/src/queuedeleteokframe.h
index e8bc7115..b7ece3fd 100644
--- a/src/queuedeleteokframe.h
+++ b/src/queuedeleteokframe.h
@@ -94,7 +94,7 @@ class QueueDeleteOKFrame : public QueueFrame
         if(!channel) return false;
 
         // report queue deletion success
-        if (channel->reportSuccess(this->messageCount())) channel->onSynchronized();
+        channel->reportSuccess(this->messageCount());
 
         // done
         return true;
diff --git a/src/queuepurgeokframe.h b/src/queuepurgeokframe.h
index 22cf1194..9098a769 100644
--- a/src/queuepurgeokframe.h
+++ b/src/queuepurgeokframe.h
@@ -94,7 +94,7 @@ class QueuePurgeOKFrame : public QueueFrame
         if(!channel) return false;
 
         // report queue purge success
-        if (channel->reportSuccess(this->messageCount())) channel->onSynchronized();
+        channel->reportSuccess(this->messageCount());
 
         // done
         return true;
diff --git a/src/queueunbindokframe.h b/src/queueunbindokframe.h
index 893908d0..d57cff05 100644
--- a/src/queueunbindokframe.h
+++ b/src/queueunbindokframe.h
@@ -73,7 +73,7 @@ class QueueUnbindOKFrame : public QueueFrame
         if(!channel) return false;
 
         // report queue unbind success
-        if (channel->reportSuccess()) channel->onSynchronized();
+        channel->reportSuccess();
 
         // done
         return true;
diff --git a/src/transactioncommitokframe.h b/src/transactioncommitokframe.h
index a02857d5..0b4bbfb4 100644
--- a/src/transactioncommitokframe.h
+++ b/src/transactioncommitokframe.h
@@ -74,7 +74,7 @@ class TransactionCommitOKFrame : public TransactionFrame
         if(!channel) return false;
 
         // report that the channel is open
-        if (channel->reportSuccess()) channel->onSynchronized();
+        channel->reportSuccess();
 
         // done
         return true;
diff --git a/src/transactionrollbackokframe.h b/src/transactionrollbackokframe.h
index d04532f0..ecbb98b4 100644
--- a/src/transactionrollbackokframe.h
+++ b/src/transactionrollbackokframe.h
@@ -74,7 +74,7 @@ class TransactionRollbackOKFrame : public TransactionFrame
         if(!channel) return false;
 
         // report that the channel is open
-        if (channel->reportSuccess()) channel->onSynchronized();
+        channel->reportSuccess();
 
         // done
         return true;
diff --git a/src/transactionselectokframe.h b/src/transactionselectokframe.h
index f83e8038..35737e48 100644
--- a/src/transactionselectokframe.h
+++ b/src/transactionselectokframe.h
@@ -74,7 +74,7 @@ class TransactionSelectOKFrame : public TransactionFrame
         if(!channel) return false;
 
         // report that the channel is open
-        if (channel->reportSuccess()) channel->onSynchronized();
+        channel->reportSuccess();
 
         // done
         return true;

From 7b79b7c6e2bfc415f5f09354327955783f4af949 Mon Sep 17 00:00:00 2001
From: Michael van der Werve <mvdwerve@outlook.com>
Date: Wed, 19 Jun 2019 10:54:50 +0200
Subject: [PATCH 2/2] now we always flush when a success is reported, and the
 funcction is renamed to flush

---
 include/amqpcpp/channelimpl.h | 27 +++++++--------------------
 src/channelimpl.cpp           |  2 +-
 src/deferredget.cpp           |  2 +-
 3 files changed, 9 insertions(+), 22 deletions(-)

diff --git a/include/amqpcpp/channelimpl.h b/include/amqpcpp/channelimpl.h
index 89b93904..a129fc38 100644
--- a/include/amqpcpp/channelimpl.h
+++ b/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,19 +644,10 @@ class ChannelImpl : public Watchable, public std::enable_shared_from_this<Channe
     {
         // skip if there is no oldest callback
         if (!_oldestCallback) return true;
-        
-        // remember whether or not we were synchronous
-        bool synchronous = _synchronous;
 
-        // is the queue empty at this moment?
-        bool empty = _queue.empty();
+        // flush the queue, which will send the next operation if the current operation was synchronous
+        flush();
         
-        // the last (possibly synchronous) operation was received, so we're no longer in synchronous mode. this
-        // is an optimization that makes sure that the first instruction _after_ a synchronous instruction
-        // that is installed during the success callback we make later does _not_ need to be buffered, but can be
-        // sent directly
-        if (synchronous && empty) _synchronous = false;
-
         // we are going to call callbacks that could destruct the channel
         Monitor monitor(this);
 
@@ -670,10 +661,6 @@ class ChannelImpl : public Watchable, public std::enable_shared_from_this<Channe
         // leap out if channel no longer exists
         if (!monitor.valid()) return false;
 
-        // if we were synchronous, but there were still messages in the queue, we process the queue now, because the synchronous
-        // operation was finished, and its callback was made, which means we're no longer in synchronous mode
-        if (synchronous && !empty) onSynchronized();
-
         // set the oldest callback
         _oldestCallback = next;
 
diff --git a/src/channelimpl.cpp b/src/channelimpl.cpp
index f3c57097..efefe24d 100644
--- a/src/channelimpl.cpp
+++ b/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;
diff --git a/src/deferredget.cpp b/src/deferredget.cpp
index e8985f19..743ce87b 100644
--- a/src/deferredget.cpp
+++ b/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();