Use upgraded connections for interactive requests

Author: gesellix

.github/workflows/ci.yml

@@ -62,6 +62,9 @@ jobs:
           go run .github/check.go
       - name: clean build
         run: ./gradlew clean build --info --stacktrace
+        env:
+          ORG_GRADLE_PROJECT_githubPackagesUsername: ${{ env.GITHUB_ACTOR }}
+          ORG_GRADLE_PROJECT_githubPackagesPassword: ${{ secrets.GITHUB_TOKEN }}
       - name: Upload Test Results
         # see publish-test-results.yml for workflow that publishes test results without security issues for forks
         # https://github.com/marketplace/actions/publish-test-results#support-fork-repositories-and-dependabot-branches

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

@@ -214,7 +214,8 @@ class ContainerApi(dockerClientConfig: DockerClientConfig = defaultClientConfig,
     )
 
     when (localVarResponse.responseType) {
-      ResponseType.Success -> {
+      ResponseType.Success,
+      ResponseType.Informational -> {
         runBlocking {
           launch {
             withTimeout(timeoutMillis) {
@@ -225,7 +226,6 @@ class ContainerApi(dockerClientConfig: DockerClientConfig = defaultClientConfig,
           }
         }
       }
-      ResponseType.Informational -> throw UnsupportedOperationException("Client does not support Informational responses.")
       ResponseType.Redirection -> throw UnsupportedOperationException("Client does not support Redirection responses.")
       ResponseType.ClientError -> {
         val localVarError = localVarResponse as ClientError<*>
@@ -282,6 +282,12 @@ class ContainerApi(dockerClientConfig: DockerClientConfig = defaultClientConfig,
         }
       }
     val localVariableHeaders: MutableMap<String, String> = mutableMapOf()
+    val requiresConnectionUpgrade = stdin != null
+    if (requiresConnectionUpgrade)
+      localVariableHeaders.apply {
+        put("Upgrade", "tcp")
+        put("Connection", "Upgrade")
+      }
 
     return RequestConfig(
       method = POST,

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

@@ -319,6 +319,11 @@ open class ApiClient(
         response.code,
         response.headers.toMultimap()
       )
+      response.code == 101 && request.isTcpUpgrade() && response.isTcpUpgrade() -> return SuccessStream(
+        response.socket.consumeFrames(mediaType),
+        response.code,
+        response.headers.toMultimap()
+      )
       response.isInformational -> return Informational(
         response.message,
         response.code,

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

@@ -6,15 +6,34 @@ import de.gesellix.docker.response.JsonChunksReader
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.emptyFlow
 import kotlinx.coroutines.flow.flow
+import okhttp3.Request
+import okhttp3.Response
 import okhttp3.ResponseBody
 import okio.Closeable
+import okio.Socket
 import okio.appendingSink
 import okio.buffer
 import java.io.File
 import java.io.InputStream
 import java.lang.reflect.Type
 import java.nio.file.Files
 
+fun Request?.isTcpUpgrade(): Boolean {
+  if (this == null) {
+    return false
+  }
+  return this.header("Connection")?.contains("Upgrade", ignoreCase = true) ?: false &&
+      this.header("Upgrade")?.contains("tcp", ignoreCase = true) ?: false
+}
+
+fun Response?.isTcpUpgrade(): Boolean {
+  if (this == null) {
+    return false
+  }
+  return this.header("Connection")?.contains("Upgrade", ignoreCase = true) ?: false &&
+      this.header("Upgrade")?.contains("tcp", ignoreCase = true) ?: false
+}
+
 fun ResponseBody?.consumeFile(): File? {
   if (this == null) {
     return null
@@ -57,6 +76,34 @@ inline fun <reified T : Any?> ResponseBody?.consumeStream(mediaType: String?): F
   }
 }
 
+fun Socket?.consumeFrames(mediaType: String?): Flow<Frame> {
+  if (this == null) {
+    return emptyFlow()
+  }
+  when (mediaType) {
+    // Requires api v1.42
+    // multiplexed-stream: without attached Tty
+    ApiClient.Companion.DockerMultiplexedStreamMediaType,
+    // Requires api v1.42
+    // raw-stream: with attached Tty
+    ApiClient.Companion.DockerRawStreamMediaType -> {
+      val reader = FrameReader(source, mediaType)
+      val events = flow {
+        while (reader.hasNext()) {
+          val next = reader.readNext(Frame::class.java)
+          emit(next)
+        }
+        source.closeQuietly()
+//        [email protected]()
+      }
+      return events
+    }
+    else -> {
+      throw UnsupportedOperationException("Can't handle media type $mediaType")
+    }
+  }
+}
+
 fun ResponseBody?.consumeFrames(mediaType: String?): Flow<Frame> {
   if (this == null) {
     return emptyFlow()

build.gradle.kts

@@ -40,15 +40,21 @@ val dependencyVersionsByGroup = mapOf(
 subprojects {
   repositories {
 //    mavenLocal()
-//    fun findProperty(s: String) = project.findProperty(s) as String?
-//    maven {
-//      name = "github"
-//      setUrl("https://maven.pkg.github.com/docker-client/*")
-//      credentials {
-//        username = System.getenv("PACKAGE_REGISTRY_USER") ?: findProperty("github.package-registry.username")
-//        password = System.getenv("PACKAGE_REGISTRY_TOKEN") ?: findProperty("github.package-registry.password")
-//      }
-//    }
+    listOf(
+      "gesellix/okhttp",
+//      "docker-client/*",
+    ).forEach { slug ->
+//      fun findProperty(s: String) = project.findProperty(s) as String?
+      maven {
+        name = "githubPackages"
+        url = uri("https://maven.pkg.github.com/${slug}")
+        credentials(PasswordCredentials::class)
+//        credentials {
+//          username = System.getenv("PACKAGE_REGISTRY_USER") ?: findProperty("github.package-registry.username")
+//          password = System.getenv("PACKAGE_REGISTRY_TOKEN") ?: findProperty("github.package-registry.password")
+//        }
+      }
+    }
     mavenCentral()
   }
 }
@@ -64,6 +70,10 @@ allprojects {
           useVersion(forcedVersion)
         }
       }
+      dependencySubstitution {
+        substitute(module("com.squareup.okhttp3:okhttp-jvm"))
+          .using(module("de.gesellix.okhttp3-forked:okhttp-jvm:${libs.versions.okhttp.get()}"))
+      }
     }
   }
 }

gradle/libs.versions.toml

@@ -7,7 +7,8 @@ logback = "1.3.15"
 logbackVersionrange = "[1.2,2)"
 moshi = "1.15.2"
 moshiVersionrange = "[1.12.0,2)"
-okhttp = "5.1.0"
+okhttp = "5.2.0-2025-08-02T07-52-48"
+#okhttp = "5.1.0"
 okhttpVersionrange = "[4,6)"
 okio = "3.16.0"
 okioVersionrange = "[3,4)"
@@ -30,8 +31,10 @@ kotlinTest = { module = "org.jetbrains.kotlin:kotlin-test", version.ref = "kotli
 logback = { module = "ch.qos.logback:logback-classic", version.ref = "logback" }
 moshi = { module = "com.squareup.moshi:moshi", version.ref = "moshi" }
 moshiKotlin = { module = "com.squareup.moshi:moshi-kotlin", version.ref = "moshi" }
-okhttp = { module = "com.squareup.okhttp3:okhttp", version.ref = "okhttp" }
-okhttpMockwebserver = { module = "com.squareup.okhttp3:mockwebserver", version.ref = "okhttp" }
+okhttp = { module = "de.gesellix.okhttp3-forked:okhttp", version.ref = "okhttp" }
+#okhttp = { module = "com.squareup.okhttp3:okhttp", version.ref = "okhttp" }
+okhttpMockwebserver = { module = "de.gesellix.okhttp3-forked:mockwebserver3-junit4", version.ref = "okhttp" }
+#okhttpMockwebserver = { module = "com.squareup.okhttp3:mockwebserver", version.ref = "okhttp" }
 okio = { module = "com.squareup.okio:okio", version.ref = "okio" }
 okioJvm = { module = "com.squareup.okio:okio-jvm", version.ref = "okio" }
 slf4j = { module = "org.slf4j:slf4j-api", version.ref = "slf4j" }