Harvest written byte count in WebSocket transport

Author: acogoluegnes

src/main/qpid/org/apache/qpid/protonj2/client/transport/netty4/TcpTransport.java

@@ -72,7 +72,7 @@ public class TcpTransport implements Transport {
     protected final SslOptions sslOptions;
     protected final Bootstrap bootstrap;
     private final IntConsumer readBytesConsumer;
-    private final IntConsumer writtenBytesConsumer;
+    protected final IntConsumer writtenBytesConsumer;
 
     protected Channel channel;
     protected volatile IOException failureCause;

src/main/qpid/org/apache/qpid/protonj2/client/transport/netty4/WebSocketTransport.java

@@ -22,6 +22,7 @@
 import java.nio.charset.StandardCharsets;
 import java.util.concurrent.TimeUnit;
 
+import io.netty.buffer.ByteBuf;
 import org.apache.qpid.protonj2.buffer.ProtonBuffer;
 import org.apache.qpid.protonj2.client.SslOptions;
 import org.apache.qpid.protonj2.client.TransportOptions;
@@ -87,10 +88,12 @@ public WebSocketTransport write(ProtonBuffer output, Runnable onComplete) throws
 
         LOG.trace("Attempted write of: {} bytes", length);
 
+        ByteBuf nettyBuf = toOutputBuffer(output);
+        this.writtenBytesConsumer.accept(nettyBuf.writableBytes());
         if (onComplete == null) {
-            channel.write(new BinaryWebSocketFrame(toOutputBuffer(output)), channel.voidPromise());
+            channel.write(new BinaryWebSocketFrame(nettyBuf), channel.voidPromise());
         } else {
-            channel.write(new BinaryWebSocketFrame(toOutputBuffer(output)), channel.newPromise().addListener(new GenericFutureListener<Future<? super Void>>() {
+            channel.write(new BinaryWebSocketFrame(nettyBuf), channel.newPromise().addListener(new GenericFutureListener<Future<? super Void>>() {
 
                 @Override
                 public void operationComplete(Future<? super Void> future) throws Exception {