Implement interactive container attach

Author: gesellix

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

@@ -1,10 +1,21 @@
 package de.gesellix.docker.remote.api.core;
 
+import okio.Sink;
+
 public interface StreamCallback<T> {
 
   default void onStarting(Cancellable cancellable) {
   }
 
+  default void attachInput(Sink sink) {
+    try {
+      Thread.sleep(500);
+      sink.close();
+    } catch (Exception ignored) {
+    }
+    throw new IllegalStateException("Falling back to default implementation that closes the sink after 500ms. This is probably not what you want. Please provide a custom implementation of this method to handle the input stream.");
+  }
+
   void onNext(T element);
 
   default void onFailed(Exception e) {

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

@@ -41,11 +41,13 @@ import de.gesellix.docker.remote.api.core.ServerError
 import de.gesellix.docker.remote.api.core.ServerException
 import de.gesellix.docker.remote.api.core.StreamCallback
 import de.gesellix.docker.remote.api.core.Success
+import de.gesellix.docker.remote.api.core.SuccessBidirectionalStream
 import de.gesellix.docker.remote.api.core.SuccessStream
 import kotlinx.coroutines.cancel
 import kotlinx.coroutines.launch
 import kotlinx.coroutines.runBlocking
 import kotlinx.coroutines.withTimeout
+import okhttp3.RequestBody
 import okio.Source
 import okio.source
 import java.io.InputStream
@@ -216,14 +218,29 @@ class ContainerApi(dockerClientConfig: DockerClientConfig = defaultClientConfig,
     when (localVarResponse.responseType) {
       ResponseType.Success,
       ResponseType.Informational -> {
-        runBlocking {
-          launch {
-            withTimeout(timeoutMillis) {
-              callback.onStarting(this@launch::cancel)
-              (localVarResponse as SuccessStream<Frame>).data.collect { callback.onNext(it) }
-              callback.onFinished()
+        when (localVarResponse) {
+          is SuccessBidirectionalStream ->
+            runBlocking {
+              launch {
+                withTimeout(timeoutMillis) {
+                  callback.onStarting(this@launch::cancel)
+                  callback.attachInput(localVarResponse.socket.sink)
+                  localVarResponse.data.collect { callback.onNext(it) }
+                  callback.onFinished()
+                }
+              }
+            }
+
+          else ->
+            runBlocking {
+              launch {
+                withTimeout(timeoutMillis) {
+                  callback.onStarting(this@launch::cancel)
+                  (localVarResponse as SuccessStream<Frame>).data.collect { callback.onNext(it) }
+                  callback.onFinished()
+                }
+              }
             }
-          }
         }
       }
       ResponseType.Redirection -> throw UnsupportedOperationException("Client does not support Redirection responses.")
@@ -259,7 +276,7 @@ class ContainerApi(dockerClientConfig: DockerClientConfig = defaultClientConfig,
     stdout: Boolean?,
     stderr: Boolean?
   ): RequestConfig {
-    val localVariableBody = null
+    val localVariableBody = RequestBody.EMPTY
     val localVariableQuery: MultiValueMap = mutableMapOf<String, List<String>>()
       .apply {
         if (detachKeys != null) {
@@ -282,11 +299,11 @@ class ContainerApi(dockerClientConfig: DockerClientConfig = defaultClientConfig,
         }
       }
     val localVariableHeaders: MutableMap<String, String> = mutableMapOf()
-    val requiresConnectionUpgrade = stdin != null
+    val requiresConnectionUpgrade = stdin != null && stdin
     if (requiresConnectionUpgrade)
       localVariableHeaders.apply {
-        put("Upgrade", "tcp")
         put("Connection", "Upgrade")
+        put("Upgrade", "tcp")
       }
 
     return RequestConfig(

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

@@ -320,8 +320,9 @@ open class ApiClient(
         response.code,
         response.headers.toMultimap()
       )
-      response.code == 101 && request.isTcpUpgrade() && response.isTcpUpgrade() -> return SuccessStream(
+      response.code == 101 && request.isTcpUpgrade() && response.isTcpUpgrade() -> return SuccessBidirectionalStream(
         response.socket.consumeFrames(mediaType),
+        response.socket!!,
         response.code,
         response.headers.toMultimap()
       )

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

@@ -1,6 +1,7 @@
 package de.gesellix.docker.remote.api.core
 
 import kotlinx.coroutines.flow.Flow
+import okio.Socket
 
 enum class ResponseType {
   Success, Informational, Redirection, ClientError, ServerError
@@ -14,8 +15,15 @@ abstract class ApiInfrastructureResponse<T>(val responseType: ResponseType) : Re
   abstract val headers: Map<String, List<String>>
 }
 
-class SuccessStream<T>(
-  val data: Flow<T>,
+class SuccessBidirectionalStream<T>(
+  override val data: Flow<T>,
+  val socket: Socket,
+  override val statusCode: Int = -1,
+  override val headers: Map<String, List<String>> = mapOf()
+) : SuccessStream<T>(data, statusCode, headers)
+
+open class SuccessStream<T>(
+  open val data: Flow<T>,
   override val statusCode: Int = -1,
   override val headers: Map<String, List<String>> = mapOf()
 ) : ApiInfrastructureResponse<T>(ResponseType.Success)

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

@@ -81,11 +81,7 @@ fun Socket?.consumeFrames(mediaType: String?): Flow<Frame> {
     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 {
@@ -109,11 +105,7 @@ fun ResponseBody?.consumeFrames(mediaType: String?): Flow<Frame> {
     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 {

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

@@ -26,7 +26,10 @@
 import de.gesellix.docker.remote.api.testutil.InjectDockerClient;
 import de.gesellix.docker.remote.api.testutil.TarUtil;
 import de.gesellix.docker.remote.api.testutil.TestImage;
+import okio.BufferedSink;
 import okio.Okio;
+import okio.Sink;
+
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.slf4j.Logger;
@@ -759,6 +762,8 @@ public void containerStatsOnce() {
 
   @Test
   public void containerAttachNonInteractive() {
+    removeContainer(engineApiClient, "container-attach-non-interactive-test");
+
     imageApi.imageCreate(testImage.getImageName(), null, null, testImage.getImageTag(), null, null, null, null, null);
 
     ContainerCreateRequest containerCreateRequest = new ContainerCreateRequest(
@@ -816,21 +821,88 @@ public void run() {
     removeContainer(engineApiClient, "container-attach-non-interactive-test");
   }
 
-  static class LogFrameStreamCallback implements StreamCallback<Frame> {
+  @Test
+  public void containerAttachInteractive() {
+    removeContainer(engineApiClient, "container-attach-interactive-test");
 
-    List<Frame> frames = new ArrayList<>();
-    Cancellable job = null;
+    imageApi.imageCreate(testImage.getImageName(), null, null, testImage.getImageTag(), null, null, null, null, null);
 
-    @Override
-    public void onStarting(Cancellable cancellable) {
-      job = cancellable;
-    }
+    ContainerCreateRequest containerCreateRequest = new ContainerCreateRequest(
+        null, null, null,
+        true, true, true,
+        null,
+        true, true, null,
+        null,
+        null,
+        null,
+        null,
+        testImage.getImageWithTag(),
+        null, null, singletonList("/cat"),
+        null, null,
+        null,
+        singletonMap(LABEL_KEY, LABEL_VALUE),
+        null, null,
+        null,
+        null,
+        null
+    );
+    containerApi.containerCreate(containerCreateRequest, "container-attach-interactive-test");
+    containerApi.containerStart("container-attach-interactive-test", null);
 
-    @Override
-    public void onNext(Frame frame) {
-      frames.add(frame);
-      log.info("next: {}", frame);
+    Duration timeout = Duration.of(5, SECONDS);
+    LogFrameStreamCallback callback = new LogFrameStreamCallback() {
+      @Override
+      public void attachInput(Sink sink) {
+        System.out.println("attachInput, sending data...");
+        new Thread(() -> {
+          BufferedSink buffer = Okio.buffer(sink);
+          try {
+            buffer.writeUtf8("hello echo\n");
+            buffer.flush();
+            System.out.println("... data sent");
+          } catch (IOException e) {
+            e.printStackTrace();
+            System.err.println("Failed to write to stdin: " + e.getMessage());
+          } finally {
+            try {
+              Thread.sleep(100);
+              sink.close();
+            } catch (Exception ignored) {
+              // ignore
+            }
+          }
+        }).start();
+      }
+    };
+
+    new Thread(() -> containerApi.containerAttach(
+        "container-attach-interactive-test",
+        null, true, true, true, true, true,
+        callback, timeout.toMillis())).start();
+
+    CountDownLatch wait = new CountDownLatch(1);
+    new Timer().schedule(new TimerTask() {
+      @Override
+      public void run() {
+        if (callback.job != null) {
+          callback.job.cancel();
+        }
+        wait.countDown();
+      }
+    }, 5000);
+
+    try {
+      wait.await();
+    }
+    catch (InterruptedException e) {
+      e.printStackTrace();
     }
+    assertSame(Frame.StreamType.RAW, callback.frames.stream().findAny().get().getStreamType());
+    assertEquals(
+        "hello echo\nhello echo".replaceAll("[\\n\\r]", ""),
+        callback.frames.stream().map(Frame::getPayloadAsString).collect(Collectors.joining()).replaceAll("[\\n\\r]", ""));
+
+    removeContainer(engineApiClient, "container-attach-interactive-test");
   }
 
   static class LogObjectStreamCallback implements StreamCallback<Object> {

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

@@ -41,7 +41,9 @@ public void setup() {
   }
 
   @Test
-  public void containerExec() {
+  public void containerExecNonInteractive() {
+    removeContainer(engineApiClient, "container-exec-test");
+
     imageApi.imageCreate(testImage.getImageName(), null, null, testImage.getImageTag(), null, null, null, null, null);
 
     ContainerCreateRequest containerCreateRequest = new ContainerCreateRequest(

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

@@ -0,0 +1,30 @@
+package de.gesellix.docker.remote.api.client;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.slf4j.Logger;
+
+import de.gesellix.docker.remote.api.core.Cancellable;
+import de.gesellix.docker.remote.api.core.Frame;
+import de.gesellix.docker.remote.api.core.LoggingExtensionsKt;
+import de.gesellix.docker.remote.api.core.StreamCallback;
+
+class LogFrameStreamCallback implements StreamCallback<Frame> {
+
+  private static final Logger log = LoggingExtensionsKt.logger(LogFrameStreamCallback.class.getName()).getValue();
+
+  List<Frame> frames = new ArrayList<>();
+  Cancellable job = null;
+
+  @Override
+  public void onStarting(Cancellable cancellable) {
+    job = cancellable;
+  }
+
+  @Override
+  public void onNext(Frame frame) {
+    frames.add(frame);
+    log.info("next: {}", frame);
+  }
+}