Rely on Docker engine api v1.42 to return explicit content-types.

See https://docs.docker.com/reference/api/engine/version-history/#v142-api-changes

Author: gesellix

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

@@ -209,9 +209,8 @@ class ContainerApi(dockerClientConfig: DockerClientConfig = defaultClientConfig,
   ) {
     val localVariableConfig = containerAttachRequestConfig(id = id, detachKeys = detachKeys, logs = logs, stream = stream, stdin = stdin, stdout = stdout, stderr = stderr)
 
-    val expectMultiplexedResponse = !(containerInspect(id, false).config?.tty ?: false)
     val localVarResponse = requestFrames(
-      localVariableConfig, expectMultiplexedResponse
+      localVariableConfig
     )
 
     when (localVarResponse.responseType) {
@@ -851,9 +850,8 @@ class ContainerApi(dockerClientConfig: DockerClientConfig = defaultClientConfig,
   ) {
     val localVariableConfig = containerLogsRequestConfig(id = id, follow = follow, stdout = stdout, stderr = stderr, since = since, until = until, timestamps = timestamps, tail = tail)
 
-    val expectMultiplexedResponse = !(containerInspect(id, false).config?.tty ?: false)
     val localVarResponse = requestFrames(
-      localVariableConfig, expectMultiplexedResponse
+      localVariableConfig
     )
 
     when (localVarResponse.responseType) {

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

@@ -245,7 +245,7 @@ class ExecApi(dockerClientConfig: DockerClientConfig = defaultClientConfig, prox
       !(execInspect(id).processConfig?.tty ?: false)
     }
     val localVarResponse = requestFrames(
-      localVariableConfig, expectMultiplexedResponse
+      localVariableConfig
     )
 
     val timeout = if (timeoutMillis == null) {

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

@@ -315,7 +315,7 @@ class ServiceApi(dockerClientConfig: DockerClientConfig = defaultClientConfig, p
     val localVariableConfig = serviceLogsRequestConfig(id = id, details = details, follow = follow, stdout = stdout, stderr = stderr, since = since, timestamps = timestamps, tail = tail)
 
     val localVarResponse = requestFrames(
-      localVariableConfig, true /* do services/tasks always have container.tty == false? */
+      localVariableConfig
     )
 
     when (localVarResponse.responseType) {

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

@@ -185,7 +185,7 @@ class TaskApi(dockerClientConfig: DockerClientConfig = defaultClientConfig, prox
     val localVariableConfig = taskLogsRequestConfig(id = id, details = details, follow = follow, stdout = stdout, stderr = stderr, since = since, timestamps = timestamps, tail = tail)
 
     val localVarResponse = requestFrames(
-      localVariableConfig, true /* do services/tasks always have container.tty == false? */
+      localVariableConfig
     )
 
     when (localVarResponse.responseType) {

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

@@ -143,15 +143,15 @@ open class ApiClient(
     return requestStream(request, client)
   }
 
-  protected fun requestFrames(requestConfig: RequestConfig, expectMultiplexedResponse: Boolean = false): ApiInfrastructureResponse<Frame> {
+  protected fun requestFrames(requestConfig: RequestConfig): ApiInfrastructureResponse<Frame> {
     val engineRequest = EngineRequest(requestConfig.method, requestConfig.path).also {
       it.headers = requestConfig.headers
       it.query = requestConfig.query
       it.body = requestConfig.body
     }
-    val request = prepareRequest(engineRequest, DockerRawStreamMediaType)
+    val request = prepareRequest(engineRequest)
     val client = prepareClient(engineRequest)
-    return requestFrames(request, client, expectMultiplexedResponse)
+    return requestFrames(request, client)
   }
 
   protected fun prepareRequest(requestConfig: EngineRequest, fallbackContentType: String = ""): Request {
@@ -309,7 +309,7 @@ open class ApiClient(
     }
   }
 
-  protected fun requestFrames(request: Request, client: OkHttpClient, expectMultiplexedResponse: Boolean = false): ApiInfrastructureResponse<Frame> {
+  protected fun requestFrames(request: Request, client: OkHttpClient): ApiInfrastructureResponse<Frame> {
     val response = client.newCall(request).execute()
     val mediaType = response.header(ContentType)?.substringBefore(";")?.lowercase(Locale.getDefault())
 
@@ -325,7 +325,7 @@ open class ApiClient(
         response.headers.toMultimap()
       )
       response.isSuccessful -> return SuccessStream(
-        response.body.consumeFrames(mediaType, expectMultiplexedResponse),
+        response.body.consumeFrames(mediaType),
         response.code,
         response.headers.toMultimap()
       )

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

@@ -6,14 +6,16 @@ import okio.BufferedSource
 import okio.Source
 import okio.buffer
 
-class FrameReader(source: Source, private val expectMultiplexedResponse: Boolean = false) : Reader<Frame> {
+class FrameReader(source: Source, private val mediaType: String) : Reader<Frame> {
 
   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/reference/api/engine/version-history/#v142-api-changes
+    // see https://github.com/moby/moby/pull/39812
+    return if (mediaType == ApiClient.Companion.DockerMultiplexedStreamMediaType) {
       // 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}
 

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

@@ -57,16 +57,18 @@ inline fun <reified T : Any?> ResponseBody?.consumeStream(mediaType: String?): F
   }
 }
 
-fun ResponseBody?.consumeFrames(mediaType: String?, expectMultiplexedResponse: Boolean = false): Flow<Frame> {
+fun ResponseBody?.consumeFrames(mediaType: String?): Flow<Frame> {
   if (this == null) {
     return emptyFlow()
   }
   when (mediaType) {
+    // Requires api v1.42
     // multiplexed-stream: without attached Tty
     ApiClient.Companion.DockerMultiplexedStreamMediaType,
-      // raw-stream: with attached Tty
+    // Requires api v1.42
+    // raw-stream: with attached Tty
     ApiClient.Companion.DockerRawStreamMediaType -> {
-      val reader = FrameReader(source(), expectMultiplexedResponse)
+      val reader = FrameReader(source(), mediaType)
       val events = flow {
         while (reader.hasNext()) {
           val next = reader.readNext(Frame::class.java)

api-client/src/test/java/de/gesellix/docker/remote/api/client/ContainerApiIntegrationTest.java

@@ -808,7 +808,10 @@ public void run() {
     catch (InterruptedException e) {
       e.printStackTrace();
     }
-    assertSame(callback.frames.stream().findAny().get().getStreamType(), Frame.StreamType.STDOUT);
+
+    // we receive a RAW response because the connection is not upgraded - which is ok for non-interactive usage
+//    assertSame(Frame.StreamType.STDOUT, callback.frames.stream().findAny().get().getStreamType());
+    assertSame(Frame.StreamType.RAW, callback.frames.stream().findAny().get().getStreamType());
 
     removeContainer(engineApiClient, "container-attach-non-interactive-test");
   }