Use upgraded connections for interactive requests

# Conflicts:
#	gradle/libs.versions.toml

Author: gesellix

.github/workflows/ci.yml

@@ -61,6 +61,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

@@ -77,6 +77,7 @@ open class ApiClient(
 
   protected inline fun <reified T> requestBody(content: T, mediaType: String = JsonMediaType): RequestBody =
     when {
+      content is RequestBody -> content
       content is File -> content.asRequestBody(
         mediaType.toMediaTypeOrNull()
       )
@@ -319,6 +320,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()
+//        this@consumeFrames.cancel()
+      }
+      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,13 +40,19 @@ 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<String>(
+////      "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

@@ -31,15 +31,15 @@ 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" }
+okhttpMockwebserverJunit5 = { module = "com.squareup.okhttp3:mockwebserver3-junit5", 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" }
 
 [bundles]
 kotlin = ["kotlin", "kotlinCommon", "kotlinJdk7", "kotlinJdk8", "kotlinReflect", "kotlinScriptingJvm", "kotlinStdlib", "kotlinTest"]
 moshi = ["moshi", "moshiKotlin"]
-okhttp = ["okhttp", "okhttpMockwebserver"]
+okhttp = ["okhttp", "okhttpMockwebserverJunit5"]
 okio = ["okio", "okioJvm"]
 
 [plugins]