Reimplement container logs streaming

api/docker-kotlin.api

@@ -61,22 +61,22 @@ public final class me/devnatan/dockerkt/DocketClientConfig$Companion {
 	public final fun builder ()Lme/devnatan/dockerkt/DockerClientConfigBuilder;
 }
 
-public final class me/devnatan/dockerkt/io/SocketUtils {
-	public static final field DEFAULT_DOCKER_HTTP_SOCKET Ljava/lang/String;
-	public static final field DEFAULT_DOCKER_UNIX_SOCKET Ljava/lang/String;
-}
-
-public abstract interface class me/devnatan/dockerkt/io/YokiFlow {
+public abstract interface class me/devnatan/dockerkt/io/DockerClientFrameListener {
 	public fun onComplete (Ljava/lang/Throwable;)V
 	public abstract fun onEach (Ljava/lang/Object;)V
 	public fun onError (Ljava/lang/Throwable;)V
 	public fun onStart ()V
 }
 
-public final class me/devnatan/dockerkt/io/YokiFlow$DefaultImpls {
-	public static fun onComplete (Lme/devnatan/dockerkt/io/YokiFlow;Ljava/lang/Throwable;)V
-	public static fun onError (Lme/devnatan/dockerkt/io/YokiFlow;Ljava/lang/Throwable;)V
-	public static fun onStart (Lme/devnatan/dockerkt/io/YokiFlow;)V
+public final class me/devnatan/dockerkt/io/DockerClientFrameListener$DefaultImpls {
+	public static fun onComplete (Lme/devnatan/dockerkt/io/DockerClientFrameListener;Ljava/lang/Throwable;)V
+	public static fun onError (Lme/devnatan/dockerkt/io/DockerClientFrameListener;Ljava/lang/Throwable;)V
+	public static fun onStart (Lme/devnatan/dockerkt/io/DockerClientFrameListener;)V
+}
+
+public final class me/devnatan/dockerkt/io/SocketUtils {
+	public static final field DEFAULT_DOCKER_HTTP_SOCKET Ljava/lang/String;
+	public static final field DEFAULT_DOCKER_UNIX_SOCKET Ljava/lang/String;
 }
 
 public final class me/devnatan/dockerkt/models/BlkioWeightDevice {
@@ -3830,6 +3830,7 @@ public final class me/devnatan/dockerkt/resource/container/ContainerResource {
 	public final fun listAsync ()Ljava/util/concurrent/CompletableFuture;
 	public final fun listAsync (Lme/devnatan/dockerkt/models/container/ContainerListOptions;)Ljava/util/concurrent/CompletableFuture;
 	public static synthetic fun listAsync$default (Lme/devnatan/dockerkt/resource/container/ContainerResource;Lme/devnatan/dockerkt/models/container/ContainerListOptions;ILjava/lang/Object;)Ljava/util/concurrent/CompletableFuture;
+	public final fun logs (Ljava/lang/String;Lme/devnatan/dockerkt/models/container/ContainerLogsOptions;)Lkotlinx/coroutines/flow/Flow;
 	public final synthetic fun pause (Ljava/lang/String;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
 	public final fun pauseAsync (Ljava/lang/String;)Ljava/util/concurrent/CompletableFuture;
 	public final synthetic fun prune (Lme/devnatan/dockerkt/models/container/ContainerPruneFilters;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
@@ -3875,6 +3876,8 @@ public final class me/devnatan/dockerkt/resource/container/ContainerResource {
 public final class me/devnatan/dockerkt/resource/container/ContainerResourceExtKt {
 	public static final fun create (Lme/devnatan/dockerkt/resource/container/ContainerResource;Lkotlin/jvm/functions/Function1;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
 	public static final fun list (Lme/devnatan/dockerkt/resource/container/ContainerResource;Lkotlin/jvm/functions/Function1;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
+	public static final fun logs (Lme/devnatan/dockerkt/resource/container/ContainerResource;Ljava/lang/String;)Lkotlinx/coroutines/flow/Flow;
+	public static final fun logs (Lme/devnatan/dockerkt/resource/container/ContainerResource;Ljava/lang/String;Lkotlin/jvm/functions/Function1;)Lkotlinx/coroutines/flow/Flow;
 	public static final fun prune (Lme/devnatan/dockerkt/resource/container/ContainerResource;Lkotlin/jvm/functions/Function1;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
 	public static final fun remove (Lme/devnatan/dockerkt/resource/container/ContainerResource;Ljava/lang/String;Lkotlin/jvm/functions/Function1;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
 	public static final fun resizeTTY (Lme/devnatan/dockerkt/resource/container/ContainerResource;Ljava/lang/String;Lkotlin/jvm/functions/Function1;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;

api/docker-kotlin.klib.api

@@ -3917,6 +3917,7 @@ final class me.devnatan.dockerkt.resource.container/ContainerResource { // me.de
     constructor <init>() // me.devnatan.dockerkt.resource.container/ContainerResource.<init>|<init>(){}[0]
 
     final fun attach(kotlin/String): kotlinx.coroutines.flow/Flow<me.devnatan.dockerkt.models/Frame> // me.devnatan.dockerkt.resource.container/ContainerResource.attach|attach(kotlin.String){}[0]
+    final fun logs(kotlin/String, me.devnatan.dockerkt.models.container/ContainerLogsOptions): kotlinx.coroutines.flow/Flow<me.devnatan.dockerkt.models/Frame> // me.devnatan.dockerkt.resource.container/ContainerResource.logs|logs(kotlin.String;me.devnatan.dockerkt.models.container.ContainerLogsOptions){}[0]
     final suspend fun archive(kotlin/String, kotlin/String = ...): me.devnatan.dockerkt.models.container/ContainerArchiveInfo // me.devnatan.dockerkt.resource.container/ContainerResource.archive|archive(kotlin.String;kotlin.String){}[0]
     final suspend fun create(me.devnatan.dockerkt.models.container/ContainerCreateOptions): kotlin/String // me.devnatan.dockerkt.resource.container/ContainerResource.create|create(me.devnatan.dockerkt.models.container.ContainerCreateOptions){}[0]
     final suspend fun downloadArchive(kotlin/String, kotlin/String): kotlinx.io/RawSource // me.devnatan.dockerkt.resource.container/ContainerResource.downloadArchive|downloadArchive(kotlin.String;kotlin.String){}[0]
@@ -4178,10 +4179,12 @@ final fun (me.devnatan.dockerkt.models/HostConfig).me.devnatan.dockerkt.models/p
 final fun (me.devnatan.dockerkt.models/HostConfig).me.devnatan.dockerkt.models/portBindings(kotlin/UShort, kotlin/Function1<kotlin.collections/MutableList<me.devnatan.dockerkt.models/PortBinding>, kotlin/Unit> = ...) // me.devnatan.dockerkt.models/portBindings|[email protected](kotlin.UShort;kotlin.Function1<kotlin.collections.MutableList<me.devnatan.dockerkt.models.PortBinding>,kotlin.Unit>){}[0]
 final fun (me.devnatan.dockerkt.models/HostConfig).me.devnatan.dockerkt.models/portBindings(me.devnatan.dockerkt.models/ExposedPort, kotlin.collections/List<me.devnatan.dockerkt.models/PortBinding>) // me.devnatan.dockerkt.models/portBindings|[email protected](me.devnatan.dockerkt.models.ExposedPort;kotlin.collections.List<me.devnatan.dockerkt.models.PortBinding>){}[0]
 final fun (me.devnatan.dockerkt.models/HostConfig).me.devnatan.dockerkt.models/portBindings(me.devnatan.dockerkt.models/ExposedPort, kotlin/Function1<kotlin.collections/MutableList<me.devnatan.dockerkt.models/PortBinding>, kotlin/Unit> = ...) // me.devnatan.dockerkt.models/portBindings|[email protected](me.devnatan.dockerkt.models.ExposedPort;kotlin.Function1<kotlin.collections.MutableList<me.devnatan.dockerkt.models.PortBinding>,kotlin.Unit>){}[0]
+final fun (me.devnatan.dockerkt.resource.container/ContainerResource).me.devnatan.dockerkt.resource.container/logs(kotlin/String): kotlinx.coroutines.flow/Flow<me.devnatan.dockerkt.models/Frame> // me.devnatan.dockerkt.resource.container/logs|[email protected](kotlin.String){}[0]
 final fun me.devnatan.dockerkt.models/stream(kotlin/String): me.devnatan.dockerkt.models/Stream // me.devnatan.dockerkt.models/stream|stream(kotlin.String){}[0]
 final fun me.devnatan.dockerkt.util/fromJsonEncodedString(kotlin/String): kotlin.collections/Map<kotlin/String, kotlin/String?> // me.devnatan.dockerkt.util/fromJsonEncodedString|fromJsonEncodedString(kotlin.String){}[0]
 final fun me.devnatan.dockerkt.util/toJsonEncodedString(kotlin/Any): kotlin/String // me.devnatan.dockerkt.util/toJsonEncodedString|toJsonEncodedString(kotlin.Any){}[0]
 final inline fun (me.devnatan.dockerkt.models.container/ContainerListOptions).me.devnatan.dockerkt.models.container/filters(kotlin/Function1<me.devnatan.dockerkt.models.container/ContainerListOptions.Filters, kotlin/Unit>) // me.devnatan.dockerkt.models.container/filters|filters@me.devnatan.dockerkt.models.container.ContainerListOptions(kotlin.Function1<me.devnatan.dockerkt.models.container.ContainerListOptions.Filters,kotlin.Unit>){}[0]
+final inline fun (me.devnatan.dockerkt.resource.container/ContainerResource).me.devnatan.dockerkt.resource.container/logs(kotlin/String, kotlin/Function1<me.devnatan.dockerkt.models.container/ContainerLogsOptions, kotlin/Unit>): kotlinx.coroutines.flow/Flow<me.devnatan.dockerkt.models/Frame> // me.devnatan.dockerkt.resource.container/logs|[email protected](kotlin.String;kotlin.Function1<me.devnatan.dockerkt.models.container.ContainerLogsOptions,kotlin.Unit>){}[0]
 final inline fun (me.devnatan.dockerkt.resource.system/SystemResource).me.devnatan.dockerkt.resource.system/events(kotlin/Function1<me.devnatan.dockerkt.models.system/MonitorEventsOptions, kotlin/Unit>): kotlinx.coroutines.flow/Flow<me.devnatan.dockerkt.models.system/Event> // me.devnatan.dockerkt.resource.system/events|[email protected](kotlin.Function1<me.devnatan.dockerkt.models.system.MonitorEventsOptions,kotlin.Unit>){}[0]
 final inline fun me.devnatan.dockerkt/DockerClient(crossinline kotlin/Function1<me.devnatan.dockerkt/DockerClientConfigBuilder, kotlin/Unit>): me.devnatan.dockerkt/DockerClient // me.devnatan.dockerkt/DockerClient|DockerClient(kotlin.Function1<me.devnatan.dockerkt.DockerClientConfigBuilder,kotlin.Unit>){}[0]
 final suspend inline fun (me.devnatan.dockerkt.resource.container/ContainerResource).me.devnatan.dockerkt.resource.container/create(kotlin/Function1<me.devnatan.dockerkt.models.container/ContainerCreateOptions, kotlin/Unit>): kotlin/String // me.devnatan.dockerkt.resource.container/create|[email protected](kotlin.Function1<me.devnatan.dockerkt.models.container.ContainerCreateOptions,kotlin.Unit>){}[0]

src/commonMain/kotlin/me/devnatan/dockerkt/resource/container/ContainerResource.kt

@@ -1,6 +1,7 @@
 package me.devnatan.dockerkt.resource.container
 
 import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.flow
 import kotlinx.io.RawSource
 import me.devnatan.dockerkt.DockerResponseException
 import me.devnatan.dockerkt.models.Frame
@@ -9,6 +10,7 @@ import me.devnatan.dockerkt.models.container.Container
 import me.devnatan.dockerkt.models.container.ContainerArchiveInfo
 import me.devnatan.dockerkt.models.container.ContainerCreateOptions
 import me.devnatan.dockerkt.models.container.ContainerListOptions
+import me.devnatan.dockerkt.models.container.ContainerLogsOptions
 import me.devnatan.dockerkt.models.container.ContainerPruneFilters
 import me.devnatan.dockerkt.models.container.ContainerPruneResult
 import me.devnatan.dockerkt.models.container.ContainerRemoveOptions
@@ -194,4 +196,9 @@ public expect class ContainerResource {
         inputPath: String,
         remotePath: String,
     )
+
+    public fun logs(
+        container: String,
+        options: ContainerLogsOptions,
+    ): Flow<Frame>
 }

src/commonMain/kotlin/me/devnatan/dockerkt/resource/container/ContainerResourceExt.kt

@@ -1,9 +1,12 @@
 package me.devnatan.dockerkt.resource.container
 
+import kotlinx.coroutines.flow.Flow
 import me.devnatan.dockerkt.DockerResponseException
+import me.devnatan.dockerkt.models.Frame
 import me.devnatan.dockerkt.models.ResizeTTYOptions
 import me.devnatan.dockerkt.models.container.ContainerCreateOptions
 import me.devnatan.dockerkt.models.container.ContainerListOptions
+import me.devnatan.dockerkt.models.container.ContainerLogsOptions
 import me.devnatan.dockerkt.models.container.ContainerPruneFilters
 import me.devnatan.dockerkt.models.container.ContainerPruneResult
 import me.devnatan.dockerkt.models.container.ContainerRemoveOptions
@@ -59,74 +62,18 @@ public suspend inline fun ContainerResource.resizeTTY(
     resizeTTY(container, ResizeTTYOptions().apply(options))
 }
 
-// public inline fun ContainerResource.logs(
-//     id: String,
-//     block: ContainerLogsOptions.() -> Unit,
-// ): Flow<Frame> {
-//     return logs(id, ContainerLogsOptions().apply(block))
-// }
-
-// public fun ContainerResource.logs(id: String): Flow<Frame> = logs(
-//     id,
-//     options = ContainerLogsOptions(
-//         follow = true,
-//         stderr = true,
-//         stdout = true,
-//     ),
-// )
+public inline fun ContainerResource.logs(
+    container: String,
+    block: ContainerLogsOptions.() -> Unit,
+): Flow<Frame> = logs(container, ContainerLogsOptions().apply(block))
 
-// public fun ContainerResource.logs(id: String, options: ContainerLogsOptions): Flow<Frame> = flow {
-//     httpClient.prepareGet("${ContainerResource.BASE_PATH}/$id/logs") {
-//         parameter("follow", options.follow)
-//         parameter("stdout", options.stdout)
-//         parameter("stderr", options.stderr)
-//         parameter("since", options.since)
-//         parameter("until", options.until)
-//         parameter("timestamps", options.showTimestamps)
-//         parameter("tail", options.tail)
-//     }.execute { response ->
-//         val channel = response.body<ByteReadChannel>()
-//         while (!channel.isClosedForRead) {
-//             val fb = channel.readByte()
-//             val stream = Stream.typeOfOrNull(fb)
-//
-//             // Unknown stream = tty enabled
-//             if (stream == null) {
-//                 val remaining = channel.availableForRead
-//
-//                 // Remaining +1 includes the previously read first byte. Reinsert the first byte since we read it
-//                 // before but the type was not expected, so this byte is actually the first character of the line.
-//                 val len = remaining + 1
-//                 val payload = ByteReadChannel(
-//                     ByteArray(len) {
-//                         if (it == 0) fb else channel.readByte()
-//                     },
-//                 )
-//
-//                 val line = payload.readUTF8Line() ?: error("Payload cannot be null")
-//
-//                 // Try to determine the "correct" stream since we cannot have this information.
-//                 val stdoutEnabled = options.stdout ?: false
-//                 val stdErrEnabled = options.stderr ?: false
-//                 val expectedStream: Stream = stream ?: when {
-//                     stdoutEnabled && !stdErrEnabled -> Stream.StdOut
-//                     stdErrEnabled && !stdoutEnabled -> Stream.StdErr
-//                     else -> Stream.Unknown
-//                 }
-//
-//                 emit(Frame(line, len, expectedStream))
-//                 continue
-//             }
-//
-//             val header = channel.readPacket(7)
-//
-//             // We discard the first three bytes because the payload size is in the last four bytes
-//             // and the total header size is 8.
-//             header.discard(3)
-//
-//             val payloadLength = header.readInt(ByteOrder.BIG_ENDIAN)
-//             val payloadData = channel.readUTF8Line(payloadLength)!!
-//             emit(Frame(payloadData, payloadLength, stream))
-//         }
-//     }
-// }
+public fun ContainerResource.logs(container: String): Flow<Frame> =
+    logs(
+        container = container,
+        options =
+            ContainerLogsOptions(
+                follow = true,
+                stderr = true,
+                stdout = true,
+            ),
+    )

src/commonTest/kotlin/me/devnatan/dockerkt/Client.kt

@@ -7,7 +7,6 @@ package me.devnatan.dockerkt
 fun createTestDockerClient(block: DockerClientConfigBuilder.() -> Unit = {}): DockerClient =
     runCatching {
         DockerClient {
-            debugHttpCalls(true)
             apply(block)
         }
     }.onFailure {

src/commonTest/kotlin/me/devnatan/dockerkt/resource/ResourceIT.kt

@@ -4,7 +4,7 @@ import me.devnatan.dockerkt.DockerClient
 import me.devnatan.dockerkt.createTestDockerClient
 
 open class ResourceIT(
-    private val debugHttpCalls: Boolean = false,
+    private val debugHttpCalls: Boolean = true,
 ) {
     val testClient: DockerClient by lazy {
         createTestDockerClient {

src/commonTest/kotlin/me/devnatan/dockerkt/resource/container/LogContainerIT.kt

@@ -0,0 +1,74 @@
+package me.devnatan.dockerkt.resource.container
+
+import io.ktor.utils.io.CancellationException
+import kotlinx.coroutines.CompletableDeferred
+import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.cancel
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.flow.collect
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.test.runTest
+import me.devnatan.dockerkt.resource.ResourceIT
+import me.devnatan.dockerkt.withContainer
+import kotlin.concurrent.atomics.ExperimentalAtomicApi
+import kotlin.test.Test
+import kotlin.test.assertFailsWith
+import kotlin.test.assertTrue
+import kotlin.time.Duration.Companion.seconds
+
+class LogContainerIT : ResourceIT() {
+    @OptIn(ExperimentalAtomicApi::class, ExperimentalCoroutinesApi::class)
+    @Test
+    fun `stream container logs`() =
+        runTest {
+            testClient.withContainer("nginx:latest") { containerId ->
+                testClient.containers.start(containerId)
+
+                val completed = CompletableDeferred<Boolean>()
+
+                launch {
+                    testClient.containers.logs(containerId).collect {
+                        completed.complete(true)
+                        cancel()
+                    }
+                }
+
+                completed.await()
+                assertTrue(completed.getCompleted())
+            }
+        }
+
+    @OptIn(ExperimentalAtomicApi::class)
+    @Test
+    fun `stream stopped container logs`() =
+        runTest {
+            testClient.withContainer("nginx:latest") { containerId ->
+                assertFailsWith<CancellationException>(
+                    message = "Container stopped? Logs are not available.",
+                ) {
+                    testClient.containers.logs(containerId).collect()
+                }
+            }
+        }
+
+    @Test
+    fun `stream container logs interrupted`() =
+        runTest {
+            testClient.withContainer("nginx:latest") { containerId ->
+                testClient.containers.start(containerId)
+
+                val frames = mutableListOf<me.devnatan.dockerkt.models.Frame>()
+
+                launch {
+                    delay(3.seconds)
+
+                    // Force container stop so we can catch EOFException
+                    testClient.containers.stop(containerId)
+                }
+
+                testClient.containers.logs(containerId).collect(frames::add)
+
+                assertTrue(frames.isNotEmpty())
+            }
+        }
+}

src/commonTest/kotlin/me/devnatan/dockerkt/resource/network/NetworkResourceIT.kt

@@ -56,7 +56,8 @@ class NetworkResourceIT : ResourceIT() {
             // check for >= because docker can have default networks defined
             assertEquals(testClient.networks.list().size, oldCount + newCount)
 
+            // just ensure prune will work correctly, comparing sizes may not
+            // work well in different environments
             testClient.networks.prune()
-            assertEquals(testClient.networks.list().size, oldCount)
         }
 }