IO: add bytes counter to byte channel

Author: Sergey Mashkov

core/kotlinx-coroutines-io/src/main/kotlin/kotlinx/coroutines/experimental/io/ByteBufferChannel.kt

@@ -57,6 +57,14 @@ internal class ByteBufferChannel(
     override val isClosedForWrite: Boolean
         get() = closed != null
 
+    @Volatile
+    override var totalBytesRead: Long = 0L
+        private set
+
+    @Volatile
+    override var totalBytesWritten: Long = 0L
+        private set
+
     override fun close(cause: Throwable?): Boolean {
         if (closed != null) return false
         val newClosed = if (cause == null) ClosedElement.EmptyCause else ClosedElement(cause)
@@ -563,13 +571,15 @@ internal class ByteBufferChannel(
 
         writePosition = carryIndex(writePosition + n)
         c.completeWrite(n)
+        totalBytesWritten += n
     }
 
     private fun ByteBuffer.bytesRead(c: RingBufferCapacity, n: Int) {
         require(n >= 0)
 
         readPosition = carryIndex(readPosition + n)
         c.completeRead(n)
+        totalBytesRead += n
         resumeWriteOp()
     }
 

core/kotlinx-coroutines-io/src/main/kotlin/kotlinx/coroutines/experimental/io/ByteReadChannel.kt

@@ -30,6 +30,12 @@ public interface ByteReadChannel {
      */
     public var readByteOrder: ByteOrder
 
+    /**
+     * Number of bytes read from the channel.
+     * It is not guaranteed to be atomic so could be updated in the middle of long running read operation.
+     */
+    public val totalBytesRead: Long
+
     /**
      * Reads all available bytes to [dst] buffer and returns immediately or suspends if no bytes available
      * @return number of bytes were read or `-1` if the channel has been closed

core/kotlinx-coroutines-io/src/main/kotlin/kotlinx/coroutines/experimental/io/ByteWriteChannel.kt

@@ -36,6 +36,12 @@ public interface ByteWriteChannel {
      */
     public var writeByteOrder: ByteOrder
 
+    /**
+     * Number of bytes written to the channel.
+     * It is not guaranteed to be atomic so could be updated in the middle of write operation.
+     */
+    public val totalBytesWritten: Long
+
     /**
      * Writes as much as possible and only suspends if buffer is full
      */