Fix back-pressure in Netty frame handler

A race condition can cause to see the channel as non-writable but end up
with a new count down latch, not the one that was present when the
writability flag was true. So the code waits on this new latch which
will be never counted down. This can happen when the writability flips
from false to true very fast.

This commit makes sure to get the latch first, then check the
writability a second time. In case we get an "old" latch, it is counted
down automatically, so the enqueuing code will not be blocked.

(cherry picked from commit cdf43b8)

Conflicts:
	src/main/java/com/rabbitmq/client/impl/NettyFrameHandlerFactory.java

Author: acogoluegnes

src/main/java/com/rabbitmq/client/impl/NettyFrameHandlerFactory.java

@@ -316,16 +316,26 @@ public void writeFrame(Frame frame) throws IOException {
           // we do not wait in the event loop
           this.doWriteFrame(frame);
         } else {
-          try {
-            boolean canWriteNow =
-                this.handler.writableLatch().await(enqueuingTimeout.toMillis(), MILLISECONDS);
-            if (canWriteNow) {
-              this.doWriteFrame(frame);
-            } else {
-              throw new IOException("Frame enqueuing failed");
+          // we get the current latch
+          CountDownLatch latch = this.handler.writableLatch();
+          if (this.handler.isWritable()) {
+            // the channel became  writable
+            this.doWriteFrame(frame);
+          } else {
+            try {
+              // the channel is still non-writable
+              // in case its writability flipped, we have a reference to a latch that has been
+              // counted down
+              // so, worst case scenario, we'll enqueue only one frame right away
+              boolean canWriteNow = latch.await(enqueuingTimeout.toMillis(), MILLISECONDS);
+              if (canWriteNow) {
+                this.doWriteFrame(frame);
+              } else {
+                throw new IOException("Frame enqueuing failed");
+              }
+            } catch (InterruptedException e) {
+              Thread.currentThread().interrupt();
             }
-          } catch (InterruptedException e) {
-            Thread.currentThread().interrupt();
           }
         }
       }