diff --git a/protonj2-client/src/main/java/org/apache/qpid/protonj2/client/impl/ClientConnection.java b/protonj2-client/src/main/java/org/apache/qpid/protonj2/client/impl/ClientConnection.java
index a1546eb..09f5955 100644
--- a/protonj2-client/src/main/java/org/apache/qpid/protonj2/client/impl/ClientConnection.java
+++ b/protonj2-client/src/main/java/org/apache/qpid/protonj2/client/impl/ClientConnection.java
@@ -615,8 +615,10 @@
         });
     }
 
-    void autoFlushOff() {
+    boolean autoFlushOff() {
+        boolean oldState = autoFlush;
         autoFlush = false;
+        return oldState;
     }
 
     void autoFlushOn() {
@@ -782,6 +784,10 @@
         } catch (Exception ignore) {}
 
         try {
+            transport.flush();
+        } catch (Exception ignore) {}
+
+        try {
             transport.close();
         } catch (Exception ignore) {}
 
@@ -976,7 +982,7 @@
             reconnectAttempts++;
             transport = ioContext.newTransport();
             LOG.trace("Connection {} Attempting connection to remote {}:{}", getId(), location.getHost(), location.getPort());
-            transport.connect(location.getHost(), location.getPort(), new ClientTransportListener(engine));
+            transport.connect(location.getHost(), location.getPort(), new ClientTransportListener(this, engine));
         } catch (Throwable error) {
             engine.engineFailed(ClientExceptionSupport.createOrPassthroughFatal(error));
         }
diff --git a/protonj2-client/src/main/java/org/apache/qpid/protonj2/client/impl/ClientSender.java b/protonj2-client/src/main/java/org/apache/qpid/protonj2/client/impl/ClientSender.java
index 7ff4849..3d73c99 100644
--- a/protonj2-client/src/main/java/org/apache/qpid/protonj2/client/impl/ClientSender.java
+++ b/protonj2-client/src/main/java/org/apache/qpid/protonj2/client/impl/ClientSender.java
@@ -337,14 +337,11 @@
             if (delivery == null) {
                 delivery = sender.protonLink().next();
                 delivery.setLinkedResource(sender.createTracker(delivery));
-            }
-
-            if (delivery.getTransferCount() == 0) {
                 delivery.setMessageFormat(messageFormat);
                 delivery.disposition(state, settled);
             }
 
-            sender.connection().autoFlushOff();
+            boolean wasAutoFlushOn = sender.connection().autoFlushOff();
             try {
                 delivery.streamBytes(payload, true);
                 if (payload != null && payload.isReadable()) {
@@ -352,9 +349,11 @@
                 } else {
                     succeeded();
                 }
-                sender.connection().flush();
             } finally {
-                sender.connection().autoFlushOn();
+                if (wasAutoFlushOn) {
+                    sender.connection().flush();
+                    sender.connection().autoFlushOn();
+                }
             }
         }
 
diff --git a/protonj2-client/src/main/java/org/apache/qpid/protonj2/client/impl/ClientStreamSender.java b/protonj2-client/src/main/java/org/apache/qpid/protonj2/client/impl/ClientStreamSender.java
index 511f902..2e0ef82 100644
--- a/protonj2-client/src/main/java/org/apache/qpid/protonj2/client/impl/ClientStreamSender.java
+++ b/protonj2-client/src/main/java/org/apache/qpid/protonj2/client/impl/ClientStreamSender.java
@@ -518,7 +518,7 @@
                 delivery.abort();
                 succeeded();
             } else {
-                sender.connection().autoFlushOff();
+                boolean wasAutoFlushOn = sender.connection().autoFlushOff();
                 try {
                     delivery.streamBytes(payload, complete);
                     if (payload != null && payload.isReadable()) {
@@ -526,9 +526,11 @@
                     } else {
                         succeeded();
                     }
-                    sender.connection().flush();
                 } finally {
-                    sender.connection().autoFlushOn();
+                    if (wasAutoFlushOn) {
+                        sender.connection().flush();
+                        sender.connection().autoFlushOn();
+                    }
                 }
             }
         }
diff --git a/protonj2-client/src/main/java/org/apache/qpid/protonj2/client/impl/ClientTransportListener.java b/protonj2-client/src/main/java/org/apache/qpid/protonj2/client/impl/ClientTransportListener.java
index 84049df..0345e5a 100644
--- a/protonj2-client/src/main/java/org/apache/qpid/protonj2/client/impl/ClientTransportListener.java
+++ b/protonj2-client/src/main/java/org/apache/qpid/protonj2/client/impl/ClientTransportListener.java
@@ -34,9 +34,11 @@
     private static final Logger LOG = LoggerFactory.getLogger(ClientTransportListener.class);
 
     private final Engine engine;
+    private final ClientConnection connection;
 
-    ClientTransportListener(Engine engine) {
+    ClientTransportListener(ClientConnection connection, Engine engine) {
         this.engine = engine;
+        this.connection = connection;
     }
 
     @Override
@@ -52,6 +54,7 @@
     @Override
     public void transportRead(ProtonBuffer incoming) {
         try {
+            connection.autoFlushOff();
             do {
                 engine.ingest(incoming);
             } while (incoming.isReadable() && engine.isWritable());
@@ -59,6 +62,8 @@
         } catch (EngineStateException e) {
             LOG.warn("Caught problem during incoming data processing: {}", e.getMessage(), e);
             engine.engineFailed(ClientExceptionSupport.createOrPassthroughFatal(e));
+        } finally {
+            connection.autoFlushOn();
         }
     }
 
diff --git a/protonj2-client/src/main/java/org/apache/qpid/protonj2/client/transport/TcpTransport.java b/protonj2-client/src/main/java/org/apache/qpid/protonj2/client/transport/TcpTransport.java
index a179075..119467c 100644
--- a/protonj2-client/src/main/java/org/apache/qpid/protonj2/client/transport/TcpTransport.java
+++ b/protonj2-client/src/main/java/org/apache/qpid/protonj2/client/transport/TcpTransport.java
@@ -451,6 +451,11 @@
         }
 
         @Override
+        public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
+            ctx.flush();
+        }
+
+        @Override
         public void channelInactive(ChannelHandlerContext context) throws Exception {
             handleTransportFailure(context.channel(), new IOException("Remote closed connection unexpectedly"));
         }