Refactor/fix the FrameReader to read plain bytes instead of lines

Author: gesellix

api-client/src/main/java/de/gesellix/docker/remote/api/core/Frame.java

@@ -21,6 +21,9 @@ public byte[] getPayload() {
   }
 
   public String getPayloadAsString() {
+    if (payload == null) {
+      return null;
+    }
     return new String(payload, StandardCharsets.UTF_8).trim();
   }
 

api-client/src/main/kotlin/de/gesellix/docker/remote/api/core/FrameReader.kt

@@ -1,31 +1,45 @@
 package de.gesellix.docker.remote.api.core
 
 import de.gesellix.docker.response.Reader
+import okio.Buffer
 import okio.BufferedSource
 import okio.Source
 import okio.buffer
 
 class FrameReader(source: Source, private val expectMultiplexedResponse: Boolean = false) : Reader<Frame> {
 
-  private val buffer: BufferedSource = source.buffer()
+  private val bufferedSource: BufferedSource = source.buffer()
+
+  private val buffer = Buffer()
 
   override fun readNext(type: Class<Frame>?): Frame {
     return if (expectMultiplexedResponse) {
       // See https://docs.docker.com/engine/api/v1.41/#operation/ContainerAttach for the stream format documentation.
       // header := [8]byte{STREAM_TYPE, 0, 0, 0, SIZE1, SIZE2, SIZE3, SIZE4}
 
-      val streamType = Frame.StreamType.valueOf(buffer.readByte())
-      buffer.skip(3)
-      val frameSize = buffer.readInt()
+      val streamType = Frame.StreamType.valueOf(bufferedSource.readByte())
+      bufferedSource.skip(3)
+      val frameSize = bufferedSource.readInt()
 
-      Frame(streamType, buffer.readByteArray(frameSize.toLong()))
+      Frame(streamType, bufferedSource.readByteArray(frameSize.toLong()))
     } else {
-      // TODO consider reading plain bytes, not line separated
-      Frame(Frame.StreamType.RAW, buffer.readUtf8Line()?.encodeToByteArray())
+      var byteCount: Long
+      bufferedSource.read(buffer, 8192L).also { byteCount = it }
+      if (byteCount < 0) {
+        Frame(Frame.StreamType.RAW, null)
+      } else {
+        Frame(Frame.StreamType.RAW, buffer.readByteArray(byteCount))
+      }
     }
   }
 
   override fun hasNext(): Boolean {
-    return !Thread.currentThread().isInterrupted && !buffer.exhausted()
+    return try {
+      !Thread.currentThread().isInterrupted
+//          && bufferedSource.isOpen
+          && !bufferedSource.peek().exhausted()
+    } catch (e: Exception) {
+      return false
+    }
   }
 }