Fix timing issues, add logging, other possible fixes

gesellix · gesellix · commit 19095efd6209 · 2025-08-10T18:26:07.000+02:00
diff --git a/build.gradle.kts b/build.gradle.kts
@@ -33,7 +33,7 @@ dependencies {
         }
       }
     }
-    api("com.squareup.okhttp3:okhttp") {
+    api(libs.okhttp) {
       version {
         strictly(libs.versions.okhttpVersionrange.get())
         prefer(libs.versions.okhttp.get())
@@ -83,6 +83,7 @@ dependencies {
   implementation(libs.junixsocketCommon)
   implementation(libs.bundles.jna)
 
+  testImplementation(libs.okhttpLoggingInterceptor)
   testImplementation("org.junit.jupiter:junit-jupiter-api:5.13.4")
   testRuntimeOnly("org.junit.jupiter:junit-jupiter-engine:5.13.4")
   testRuntimeOnly("org.junit.platform:junit-platform-launcher:1.13.4")
diff --git a/gradle/libs.versions.toml b/gradle/libs.versions.toml
@@ -25,6 +25,7 @@ kotlinJdk7 = { module = "org.jetbrains.kotlin:kotlin-stdlib-jdk7", version.ref =
 kotlinJdk8 = { module = "org.jetbrains.kotlin:kotlin-stdlib-jdk8", version.ref = "kotlin" }
 logback = { module = "ch.qos.logback:logback-classic", version.ref = "logback" }
 okhttp = { module = "com.squareup.okhttp3:okhttp", version.ref = "okhttp" }
+okhttpLoggingInterceptor = { module = "com.squareup.okhttp3:logging-interceptor", 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" }
@@ -33,4 +34,5 @@ slf4j = { module = "org.slf4j:slf4j-api", version.ref = "slf4j" }
 kotlin = ["kotlin", "kotlinCommon", "kotlinJdk7", "kotlinJdk8"]
 jna = ["jna", "jnaPlatform"]
 junixsocket = ["junixsocketCore", "junixsocketCommon"]
+okhttp = ["okhttp", "okhttpLoggingInterceptor"]
 okio = ["okio", "okioJvm"]
diff --git a/src/main/java/de/gesellix/docker/client/filesocket/NamedPipeSocket.java b/src/main/java/de/gesellix/docker/client/filesocket/NamedPipeSocket.java
@@ -6,7 +6,6 @@
 import static com.sun.jna.platform.win32.WinNT.GENERIC_WRITE;
 import static com.sun.jna.platform.win32.WinNT.OPEN_EXISTING;
 
-import java.io.FilterOutputStream;
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.OutputStream;
@@ -55,7 +54,13 @@ void connect(String socketPath) throws IOException {
     socketPath = socketPath.replace("/", "\\");
     log.debug("connect via '{}'...", socketPath);
 
-    Kernel32.INSTANCE.WaitNamedPipe(socketPath, 200);
+    boolean ok = Kernel32.INSTANCE.WaitNamedPipe(socketPath, 200);
+    if (!ok) {
+      int err = Kernel32.INSTANCE.GetLastError();
+      log.error("Failed to wait for Named Pipe '" + socketPath + "', WinError=" + err);
+      throw new IOException("Failed to wait for Named Pipe '" + socketPath + "', WinError=" + err);
+    }
+
     handle = Kernel32.INSTANCE.CreateFile(
         socketPath,
         GENERIC_READ | GENERIC_WRITE,
@@ -69,6 +74,7 @@ void connect(String socketPath) throws IOException {
 
     if (INVALID_HANDLE_VALUE.equals(handle)) {
       int err = Kernel32.INSTANCE.GetLastError();
+      log.error("Failed to open Named Pipe '" + socketPath + "', WinError=" + err);
       throw new IOException("Failed to open Named Pipe '" + socketPath + "', WinError=" + err);
     }
 
@@ -86,32 +92,27 @@ public InputStream getInputStream() throws IOException {
   @Override
   public OutputStream getOutputStream() throws IOException {
     ensureOpen();
-
-    // Wrap the Okio OutputStream so flush() also flushes the BufferedSink
-    return new FilterOutputStream(sink.outputStream()) {
-      @Override
-      public void flush() throws IOException {
-        super.flush();       // flush Java's wrapper
-        sink.flush();        // flush Okio buffer to NamedPipeSink
-      }
-    };
+    return sink.outputStream();
   }
 
   @Override
   public synchronized void close() throws IOException {
     if (closed) {
       return;
     }
+    log.debug("closing handle {}...", handle);
     try {
       if (handle != null && !INVALID_HANDLE_VALUE.equals(handle)) {
         // Cancel any pending read/write before closing to avoid CloseHandle() hang
         ExtendedKernel32.INSTANCE.CancelIoEx(handle, null);
       }
 
       if (source != null) {
+        log.debug("closing source {}...", source);
         source.close();
       }
       if (sink != null) {
+        log.debug("closing sink {}...", sink);
         sink.close();
       }
     } finally {
diff --git a/src/main/java/de/gesellix/docker/client/filesocket/NamedPipeSource.java b/src/main/java/de/gesellix/docker/client/filesocket/NamedPipeSource.java
@@ -44,10 +44,10 @@ public long read(@NotNull Buffer sink, long byteCount) {
         // Expected when CancelIoEx() is called during close()
         return -1;
       }
-      return -1; // Other read error
+      return err == 0 ? 0 : -1; // Other read error
     }
 
-    if (bytesRead.getValue() <= 0) {
+    if (bytesRead.getValue() < 0) {
       return -1; // EOF
     }
 
diff --git a/src/main/java/de/gesellix/docker/client/filesocket/NamedPipeUtils.java b/src/main/java/de/gesellix/docker/client/filesocket/NamedPipeUtils.java
@@ -2,6 +2,9 @@
 
 import static com.sun.jna.platform.win32.WinBase.INVALID_HANDLE_VALUE;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 import com.sun.jna.platform.win32.Kernel32;
 import com.sun.jna.platform.win32.WinBase;
 import com.sun.jna.platform.win32.WinError;
@@ -11,6 +14,8 @@
 
 public final class NamedPipeUtils {
 
+  private static final Logger log = LoggerFactory.getLogger(NamedPipeUtils.class);
+
   private NamedPipeUtils() {
   }
 
@@ -26,6 +31,12 @@ public static boolean readOverlapped(WinNT.HANDLE handle, byte[] buf, IntByRefer
     WinBase.OVERLAPPED overlapped = new WinBase.OVERLAPPED();
     overlapped.hEvent = Kernel32.INSTANCE.CreateEvent(null, true, false, null);
 
+    try {
+      Thread.sleep(10);
+    } catch (InterruptedException ignored) {
+      // ignored
+    }
+
     boolean ok = Kernel32.INSTANCE.ReadFile(handle, buf, buf.length, null, overlapped);
     if (!ok) {
       int err = Kernel32.INSTANCE.GetLastError();
@@ -44,13 +55,20 @@ public static boolean readOverlapped(WinNT.HANDLE handle, byte[] buf, IntByRefer
 
     ExtendedKernel32.INSTANCE.GetOverlappedResult(handle, overlapped, bytesRead, false);
     Kernel32.INSTANCE.CloseHandle(overlapped.hEvent);
-    return bytesRead.getValue() > 0;
+    log.debug("Bytes read: {}", bytesRead.getValue());
+    return bytesRead.getValue() >= 0;
   }
 
   public static boolean writeOverlapped(WinNT.HANDLE handle, byte[] buf, int len, IntByReference bytesWritten, int timeoutMillis) {
     WinBase.OVERLAPPED overlapped = new WinBase.OVERLAPPED();
     overlapped.hEvent = Kernel32.INSTANCE.CreateEvent(null, true, false, null);
 
+    try {
+      Thread.sleep(10);
+    } catch (InterruptedException ignored) {
+      // ignored
+    }
+
     boolean ok = Kernel32.INSTANCE.WriteFile(handle, buf, len, null, overlapped);
     if (!ok) {
       int err = Kernel32.INSTANCE.GetLastError();
@@ -69,7 +87,11 @@ public static boolean writeOverlapped(WinNT.HANDLE handle, byte[] buf, int len,
 
     ExtendedKernel32.INSTANCE.GetOverlappedResult(handle, overlapped, bytesWritten, false);
     Kernel32.INSTANCE.CloseHandle(overlapped.hEvent);
-    return bytesWritten.getValue() > 0;
+    log.debug("Bytes written: {}", bytesWritten.getValue());
+    if (bytesWritten.getValue() != len) {
+      log.warn("Incomplete write: {}/{}", bytesWritten.getValue(), len);
+    }
+    return bytesWritten.getValue() >= 0;
   }
 
   public static void closeHandle(HANDLE handle) {
diff --git a/src/test/java/de/gesellix/docker/client/filesocket/NamedPipeSocketTest.java b/src/test/java/de/gesellix/docker/client/filesocket/NamedPipeSocketTest.java
@@ -37,6 +37,14 @@
 import com.sun.jna.platform.win32.WinNT;
 import com.sun.jna.ptr.IntByReference;
 
+import okhttp3.HttpUrl;
+import okhttp3.OkHttpClient;
+import okhttp3.Request;
+import okhttp3.Response;
+import okhttp3.logging.HttpLoggingInterceptor;
+import okhttp3.logging.LoggingEventListener;
+import okio.Timeout;
+
 class NamedPipeSocketTest {
 
   @Test
@@ -49,6 +57,111 @@ void canConnect() throws IOException {
     }
   }
 
+  @Test
+  @EnabledOnOs(OS.WINDOWS)
+  void canPingViaSocket() throws IOException {
+    try (NamedPipeSocket namedPipeSocket = new NamedPipeSocket()) {
+      namedPipeSocket.connect(new InetSocketAddress(
+          getByAddress(namedPipeSocket.encodeHostname("//./pipe/docker_engine"), new byte[]{0, 0, 0, 0}),
+          0));
+
+      OutputStream os = namedPipeSocket.getOutputStream();
+
+      List<String> headers = new ArrayList<>();
+      headers.add("GET /_ping HTTP/1.1");
+      headers.add("Content-Type: application/json");
+      headers.add("Accept: application/json");
+//      headers.add("Host: localhost");
+      headers.add("Host: 2f2f2e2f706970652f646f636b65725f656e67696e65.socket");
+      headers.add("Connection: Keep-Alive");
+      headers.add("Accept-Encoding: gzip");
+      headers.add("User-Agent: NamedPipeSocketTest");
+
+      try {
+        for (String line : headers) {
+          System.out.println(">>|" + line);
+          os.write(line.getBytes(StandardCharsets.UTF_8));
+          os.write("\n".getBytes(StandardCharsets.UTF_8));
+        }
+        os.write("\n".getBytes(StandardCharsets.UTF_8));
+        os.flush();
+        System.out.println("[TEST] Sent request headers");
+        Thread.sleep(2000);
+      } catch (IOException e) {
+        e.printStackTrace();
+      } catch (InterruptedException e) {
+        e.printStackTrace();
+      }
+
+      InputStream is = namedPipeSocket.getInputStream();
+      Thread readerThread = new Thread(() -> {
+        try {
+          byte[] buffer = new byte[8024];
+          int bytesRead;
+          while ((bytesRead = is.read(buffer)) != -1) {
+            System.out.println("Read: " + new String(buffer, 0, bytesRead, StandardCharsets.UTF_8));
+          }
+          Thread.sleep(500);
+        } catch (Exception e) {
+          throw new RuntimeException(e);
+        }
+      });
+      readerThread.start();
+
+      try {
+        Thread.sleep(1000);
+      } catch (InterruptedException e) {
+        e.printStackTrace();
+      } finally {
+        os.close();
+      }
+      System.out.println("Closing reader");
+      if (is != null) {
+        try {
+          is.close();
+        } catch (IOException ignored) {
+        }
+      }
+    }
+  }
+
+  @Test
+  @EnabledOnOs(OS.WINDOWS)
+  void canPingViaOkHttp() throws IOException {
+    NamedPipeSocketFactory namedPipeSocketFactory = new NamedPipeSocketFactory();
+    OkHttpClient.Builder builder = new OkHttpClient.Builder()
+        .socketFactory(namedPipeSocketFactory)
+        .dns(namedPipeSocketFactory)
+        .connectTimeout(new Timeout().timeout(10, TimeUnit.SECONDS).timeoutNanos() / 1000, TimeUnit.MILLISECONDS)
+        .callTimeout(new Timeout().timeout(30, TimeUnit.SECONDS).timeoutNanos() / 1000, TimeUnit.MILLISECONDS)
+        .readTimeout(new Timeout().timeout(30, TimeUnit.SECONDS).timeoutNanos() / 1000, TimeUnit.MILLISECONDS)
+        .writeTimeout(new Timeout().timeout(30, TimeUnit.SECONDS).timeoutNanos() / 1000, TimeUnit.MILLISECONDS)
+        .addInterceptor(new HttpLoggingInterceptor().setLevel(HttpLoggingInterceptor.Level.BODY))
+        .eventListenerFactory(new LoggingEventListener.Factory());
+
+    HttpUrl url = new HttpUrl.Builder()
+        .scheme("http")
+        .host(new NamedPipeSocket().encodeHostname("//./pipe/docker_engine"))
+        .addPathSegments("_ping")
+        .build();
+
+    Request request = new Request.Builder()
+        .get()
+        .url(url)
+        .addHeader("Content-Type", "application/json")
+        .addHeader("Accept", "application/json")
+        .build();
+
+    OkHttpClient client = builder.build();
+    Response response = client.newCall(request).execute();
+    if (!response.isSuccessful()) {
+      response.close();
+    } else {
+      String content = response.toString();
+      System.out.println("Got: " + content);
+    }
+  }
+
   @Test
   @EnabledOnOs(OS.WINDOWS)
   public void testInProcessServerClient() throws Exception {
@@ -253,6 +366,7 @@ void testPlainHijacked() throws Exception {
 
     Thread readerThread = new Thread(() -> {
       try (InputStream inputStream = namedPipeSocket.getInputStream()) {
+        Thread.sleep(1000);
         byte[] buf = new byte[1024];
         int read;
         while ((read = inputStream.read(buf)) != -1) {

Original file line number	Diff line number	Diff line change
`@@ -33,7 +33,7 @@ dependencies {`
`33`	`33`	`}`
`34`	`34`	`}`
`35`	`35`	`}`
`36`		`- api("com.squareup.okhttp3:okhttp") {`
	`36`	`+ api(libs.okhttp) {`
`37`	`37`	`version {`
`38`	`38`	`strictly(libs.versions.okhttpVersionrange.get())`
`39`	`39`	`prefer(libs.versions.okhttp.get())`
`@@ -83,6 +83,7 @@ dependencies {`
`83`	`83`	`implementation(libs.junixsocketCommon)`
`84`	`84`	`implementation(libs.bundles.jna)`
`85`	`85`
	`86`	`+ testImplementation(libs.okhttpLoggingInterceptor)`
`86`	`87`	`testImplementation("org.junit.jupiter:junit-jupiter-api:5.13.4")`
`87`	`88`	`testRuntimeOnly("org.junit.jupiter:junit-jupiter-engine:5.13.4")`
`88`	`89`	`testRuntimeOnly("org.junit.platform:junit-platform-launcher:1.13.4")`
Original file line number	Diff line number	Diff line change
`@@ -44,10 +44,10 @@ public long read(@NotNull Buffer sink, long byteCount) {`
`44`	`44`	`// Expected when CancelIoEx() is called during close()`
`45`	`45`	`return -1;`
`46`	`46`	`}`
`47`		`- return -1; // Other read error`
	`47`	`+ return err == 0 ? 0 : -1; // Other read error`
`48`	`48`	`}`
`49`	`49`
`50`		`- if (bytesRead.getValue() <= 0) {`
	`50`	`+ if (bytesRead.getValue() < 0) {`
`51`	`51`	`return -1; // EOF`
`52`	`52`	`}`
`53`	`53`