Npipe: add integration test to verify async io

.github/actions/create-named-pipe/action.yml

@@ -0,0 +1,19 @@
+---
+name: "Creates a Named Pipe"
+description: "Creates a Named Pipe for integration tests on Windows"
+runs:
+  using: "composite"
+  steps:
+    - run: |
+        docker pull gesellix/npipe:windows
+        docker create --name npipe gesellix/npipe:windows
+        docker cp npipe:/npipe.exe ./npipe.exe
+        docker rm npipe
+        docker rmi gesellix/npipe:windows
+        ./npipe.exe \\\\.\\pipe\\hijack_test &
+        disown -h
+      shell: bash
+branding:
+  icon: "tag"
+  color: "blue"
+...

.github/workflows/cd.yml

@@ -18,6 +18,9 @@ jobs:
           distribution: 'zulu'
           java-version: 8
           cache: 'gradle'
+      - name: Prepare Named Pipe integration test on Windows
+        if: matrix.os == 'windows-latest'
+        uses: ./.github/actions/create-named-pipe
       - name: build publish
         run: ./gradlew clean build publish --info --stacktrace
         env:

.github/workflows/ci.yml

@@ -26,6 +26,9 @@ jobs:
       - name: Install Docker on macOS
         if: matrix.os == 'macos-latest'
         uses: ./.github/actions/docker-install-macos
+      - name: Prepare Named Pipe integration test on Windows
+        if: matrix.os == 'windows-latest'
+        uses: ./.github/actions/create-named-pipe
       - name: clean build
         run: ./gradlew clean build --info --stacktrace
       - name: Publish Test Report

.github/workflows/release.yml

@@ -20,6 +20,9 @@ jobs:
           distribution: 'zulu'
           java-version: 8
           cache: 'gradle'
+      - name: Prepare Named Pipe integration test on Windows
+        if: matrix.os == 'windows-latest'
+        uses: ./.github/actions/create-named-pipe
       - name: Set artifact version
         run: |
           echo "RELEASE_VERSION=$(echo '${{ github.event.release.tag_name }}' | sed -e s/^v//)" >> $GITHUB_ENV

src/test/java/de/gesellix/docker/client/filesocket/NamedPipeSocketTest.java

@@ -5,7 +5,14 @@
 import org.junit.jupiter.api.condition.OS;
 
 import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
 import java.net.InetSocketAddress;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 import static java.net.InetAddress.getByAddress;
 
@@ -17,4 +24,97 @@ void canConnect() throws IOException {
     NamedPipeSocket namedPipeSocket = new NamedPipeSocket();
     namedPipeSocket.connect(new InetSocketAddress(getByAddress(namedPipeSocket.encodeHostname("//./pipe/docker_engine"), new byte[] {0, 0, 0, 0}), 0));
   }
+
+  @Test
+  @EnabledOnOs(OS.WINDOWS)
+  void testPlainHijacked() throws Exception {
+    String namedPipeName = "//./pipe/hijack_test";
+    String containerId = "hijacking";
+
+    NamedPipeSocket namedPipeSocket = new NamedPipeSocket();
+    namedPipeSocket.connect(new InetSocketAddress(getByAddress(namedPipeSocket.encodeHostname(namedPipeName), new byte[] {0, 0, 0, 0}), 0));
+    namedPipeSocket.setKeepAlive(true);
+
+    OutputStream outputStream = namedPipeSocket.getOutputStream();
+    List<String> headers = new ArrayList<>();
+    headers.add("POST /containers/" + containerId + "/attach?logs=true&stream=true&stdin=true&stdout=true&stderr=true HTTP/1.1");
+    headers.add("Host: localhost");
+    headers.add("Content-Length: 0");
+    headers.add("Connection: Upgrade");
+    headers.add("Upgrade: tcp");
+    try {
+      for (String line : headers) {
+        System.out.println(">>|" + line);
+        outputStream.write(line.getBytes(StandardCharsets.UTF_8));
+        outputStream.write("\n".getBytes(StandardCharsets.UTF_8));
+      }
+      outputStream.write("\n".getBytes(StandardCharsets.UTF_8));
+    }
+    catch (IOException e) {
+      e.printStackTrace();
+    }
+
+    Object lock = new Object();
+
+    final AtomicBoolean readCanBeFinished = new AtomicBoolean(false);
+    Thread readerThread = new Thread(() -> {
+      InputStream inputStream = namedPipeSocket.getInputStream();
+
+      byte[] buf = new byte[1024];
+      while (!readCanBeFinished.get()) {
+        try {
+          Thread.sleep(100);
+        }
+        catch (InterruptedException ignored) {
+          // ignored
+        }
+        int read = 0;
+        try {
+          read = inputStream.read(buf, 0, 1024);
+        }
+        catch (IOException e) {
+          e.printStackTrace();
+        }
+        if (read > -1) {
+          System.out.println("<<|" + new String(buf, 0, read));
+        }
+      }
+      System.out.println("// reader done //");
+    });
+
+    Thread writerThread = new Thread(() -> {
+
+      String content = "yeah";
+      int n = 5;
+      while (n > 0) {
+        try {
+          System.out.println(">>: " + n);
+          outputStream.write((n + " " + content + " - " + new Date()).getBytes(StandardCharsets.UTF_8));
+
+          Thread.sleep(500);
+        }
+        catch (Exception e) {
+          e.printStackTrace();
+        }
+        finally {
+          n--;
+        }
+      }
+
+      System.out.println("// writer done //");
+      readCanBeFinished.set(true);
+      synchronized (lock) {
+        lock.notify();
+      }
+    });
+
+    synchronized (lock) {
+      readerThread.start();
+      Thread.sleep(1000);
+      writerThread.start();
+      lock.wait(10000);
+      System.out.println("// lock done //");
+    }
+    namedPipeSocket.close();
+  }
 }