Streaming APIs (phase 2): stream()/start() terminals on the fluent WinRMClient

README.md

@@ -73,6 +73,44 @@ Per-operation options: `namespace(...)`, `timeout(...)`, and for WQL enumeration
 from the remote code set by default), and `upload(Path...)` to copy local script files and rewrite
 the command to reference the remote copies.
 
+### Streaming
+
+Both operations also have a streaming terminal for large result sets and long-running commands —
+everything upstream (authentication, TLS, namespace, options) is shared with the blocking
+terminals:
+
+```java
+// WQL rows are pulled from the server page by page as the stream advances:
+// memory stays bounded by one page (pageSize(int)), not by the whole result set.
+try (Stream<WqlRow> rows = client.wql("SELECT * FROM Win32_NTLogEvent").stream()) {
+    rows.filter(r -> "Error".equals(r.string("Type")))
+        .limit(100)
+        .forEach(System.out::println);
+}
+
+// Commands can be consumed while they run, java.lang.Process-style:
+try (RemoteProcess p = client.command("wevtutil qe System /f:text").start()) {
+    try (BufferedReader out = p.stdout()) {
+        out.lines().forEach(System.out::println);
+    }
+    int exitCode = p.waitFor();               // or waitFor(Duration) for a deadline
+}
+
+// Middle ground: tail the output live, keep the blocking terminal and its full result.
+client.command("longRunningThing.exe")
+    .onStdout(chunk -> log.info(chunk))
+    .onStderr(chunk -> log.warn(chunk))
+    .execute();
+```
+
+Streams and processes **must be closed** (try-with-resources): they hold the client's serial
+connection while open, and closing early releases the server-side enumeration (WS-Enumeration
+`Release`) or terminates the remote command (WinRM terminate `Signal`). For the streaming
+terminals the configured timeout is an **inactivity** timeout — the longest silence tolerated from
+the server between two responses — not an overall deadline, so long tails can stream indefinitely.
+Output is decoded incrementally: a multibyte character split across protocol chunks is decoded
+correctly.
+
 The pre-existing static helpers (`WinRMWqlExecutor.executeWql(...)`,
 `WinRMCommandExecutor.execute(...)`) keep working unchanged — see **Legacy API** below.
 
@@ -129,56 +167,15 @@ java -jar target/winrm-java-<version>-standalone.jar \
   exec ipconfig /all
 ```
 
-Use `--help` for the complete option list and `--version` for the build version. HTTP is the
-default transport and uses port 5985; `--https` uses port 5986. `-P`/`--port` overrides either
-default, and `-t`/`--timeout` sets the operation timeout in milliseconds (60,000 by default).
-
-NTLM is used when neither authentication flag is supplied. `--ntlm` and `--kerberos` are mutually
-exclusive. Kerberos requires HTTPS. By default it uses the ambient JDK Kerberos configuration. The
-CLI can instead configure the JDK for the current invocation with `--kerberos-kdc <host>`. If no
-`--kerberos-realm <realm>` is supplied, the realm is inferred by removing the KDC hostname's first
-DNS label and uppercasing the remaining suffix. For example:
-
-```bash
-java -jar target/winrm-java-<version>-standalone.jar \
-  -h server.internal.sentrysoftware.net -u 'DOMAIN\user' -pf password.txt \
-  --https --kerberos --kerberos-kdc camus.internal.sentrysoftware.net \
-  command whoami
-```
+Use `--help` for the option list and `--version` for the build version. The CLI is built on the
+streaming API: WQL rows are written to stdout as UTF-8 [JSON Lines](https://jsonlines.org/) **as
+the enumeration pages arrive**, and remote command stdout and stderr are forwarded **live** to the
+corresponding local streams while the command runs. Diagnostics go only to stderr, and the exit
+codes are stable for scripting.
 
-This infers `INTERNAL.SENTRYSOFTWARE.NET`. The inference follows a common Active Directory DNS
-naming convention; it is not guaranteed by Kerberos. Specify `--kerberos-realm` when the realm does
-not match the KDC's DNS suffix or when the KDC is not a fully qualified DNS name. Both options are
-valid only with `--kerberos`, and `--kerberos-realm` requires `--kerberos-kdc`.
-
-HTTPS validates the certificate and hostname by default. `--https-permissive` trusts any
-certificate and hostname; it is intentionally insecure and should only be used for testing or
-isolated hosts.
-
-`-p`/`--password` is convenient for interactive use, but command-line arguments may be visible to
-other local processes. Prefer `-pf`/`--password-file` for automation. Password files are decoded as
-UTF-8. Exactly one final LF, CRLF, or CR is removed; all other bytes, including whitespace and
-earlier line endings, are part of the password. The two password options are mutually exclusive.
-If neither is supplied, the CLI securely requests the password from the interactive console without
-echoing it. Non-interactive runs must use `--password-file` (or, less securely, `--password`).
-
-WQL writes one compact UTF-8 JSON object per row to stdout
-([JSON Lines](https://jsonlines.org/)); property order follows the WinRM response. Diagnostics go
-only to stderr. Remote command stdout and stderr are forwarded to the corresponding local streams.
-The current backend buffers an operation's result; the CLI output boundary is ready to consume the
-streaming API when it becomes available.
-
-Exit behavior is stable:
-
-| Exit code | Meaning |
-| ---: | --- |
-| `0` | Successful WQL query or remote command |
-| `0`–`255` | Remote command exit code, when representable |
-| `64` | Invalid CLI usage |
-| `69` | Connection, DNS, socket, or TLS failure |
-| `70` | WinRM protocol or other remote failure |
-| `77` | Authentication failure |
-| `124` | Operation timeout |
+The full manual — options, password handling, Kerberos configuration, streaming and timeout
+semantics, exit codes — is the
+[Command-Line Client](https://metricshub.org/winrm-java/cli.html) page.
 
 ## Build instructions
 

src/main/java/org/metricshub/winrm/AuthScheme.java

@@ -4,7 +4,7 @@
  * ╱╲╱╲╱╲╱╲╱╲╱╲╱╲╱╲╱╲╱╲╱╲╱╲╱╲╱╲╱╲╱╲╱╲╱╲╱╲╱╲
  * WinRM Java Client
  * ჻჻჻჻჻჻
- * Copyright 2023 - 2026 MetricsHub
+ * Copyright (C) 2023 - 2026 MetricsHub
  * ჻჻჻჻჻჻
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.

src/main/java/org/metricshub/winrm/ChunkDecoder.java

@@ -0,0 +1,109 @@
+package org.metricshub.winrm;
+
+/*-
+ * ╱╲╱╲╱╲╱╲╱╲╱╲╱╲╱╲╱╲╱╲╱╲╱╲╱╲╱╲╱╲╱╲╱╲╱╲╱╲╱╲
+ * WinRM Java Client
+ * ჻჻჻჻჻჻
+ * Copyright (C) 2023 - 2026 MetricsHub
+ * ჻჻჻჻჻჻
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ╲╱╲╱╲╱╲╱╲╱╲╱╲╱╲╱╲╱╲╱╲╱╲╱╲╱╲╱╲╱╲╱╲╱╲╱╲╱╲╱
+ */
+
+import java.nio.ByteBuffer;
+import java.nio.CharBuffer;
+import java.nio.charset.Charset;
+import java.nio.charset.CharsetDecoder;
+import java.nio.charset.CoderResult;
+import java.nio.charset.CodingErrorAction;
+
+/**
+ * Incremental, stateful charset decoding for streamed command output. A multibyte character (e.g.
+ * UTF-8) can be split across WSMan Stream elements or Receive responses, so each chunk is decoded
+ * with a decoder that carries the partial-character bytes over to the next chunk — never a
+ * per-chunk {@code new String(bytes)}, which would corrupt the boundary bytes into replacement
+ * characters. Malformed and unmappable input is replaced, matching
+ * {@link String#String(byte[], Charset)}, so incrementally decoding a byte sequence yields the
+ * same text as decoding it in one piece.
+ * <p>
+ * This is a thin push-style convenience over the JDK's own {@link CharsetDecoder}, which does all
+ * the actual decoding. The JDK's ready-made incremental decoders ({@code InputStreamReader} and
+ * its underlying {@code StreamDecoder}) only fit <i>pull</i>-based streams — they block the caller
+ * until input arrives — whereas output chunks here are <i>pushed</i> by the protocol loop as each
+ * Receive response is processed; there is no public JDK type for that direction, only the
+ * {@link CharsetDecoder}/{@link ByteBuffer} primitives this class packages.
+ */
+final class ChunkDecoder {
+
+	private final CharsetDecoder decoder;
+
+	// Undecoded tail bytes of the previous chunk — an incomplete multibyte character — replayed in
+	// front of the next chunk.
+	private byte[] pending = new byte[0];
+
+	ChunkDecoder(final Charset charset) {
+		this.decoder = charset
+			.newDecoder()
+			.onMalformedInput(CodingErrorAction.REPLACE)
+			.onUnmappableCharacter(CodingErrorAction.REPLACE);
+	}
+
+	/**
+	 * Decode the next chunk, returning the characters that are complete so far. An incomplete
+	 * multibyte character at the end of the chunk is withheld until the next call completes it.
+	 */
+	String decode(final byte[] chunk) {
+		return decode(chunk, false);
+	}
+
+	/**
+	 * Flush the decoder at the end of the stream: a trailing incomplete character becomes a
+	 * replacement character, exactly as a whole-buffer {@code new String(bytes)} would render it.
+	 */
+	String finish() {
+		return decode(new byte[0], true);
+	}
+
+	private String decode(final byte[] chunk, final boolean endOfInput) {
+		final ByteBuffer in = ByteBuffer.allocate(pending.length + chunk.length);
+		in.put(pending);
+		in.put(chunk);
+		in.flip();
+
+		final StringBuilder text = new StringBuilder();
+		final CharBuffer out = CharBuffer.allocate(Math.max(16, in.remaining() * 2));
+		CoderResult result;
+		do {
+			result = decoder.decode(in, out, endOfInput);
+			out.flip();
+			text.append(out);
+			out.clear();
+			// With REPLACE in force the only non-underflow result is overflow: loop for more room.
+		} while (result.isOverflow());
+
+		if (endOfInput) {
+			do {
+				result = decoder.flush(out);
+				out.flip();
+				text.append(out);
+				out.clear();
+			} while (result.isOverflow());
+			pending = new byte[0];
+		} else {
+			// Whatever the decoder left in the input is an incomplete character: carry it over.
+			pending = new byte[in.remaining()];
+			in.get(pending);
+		}
+		return text.toString();
+	}
+}

src/main/java/org/metricshub/winrm/CommandCursor.java

@@ -0,0 +1,139 @@
+package org.metricshub.winrm;
+
+/*-
+ * ╱╲╱╲╱╲╱╲╱╲╱╲╱╲╱╲╱╲╱╲╱╲╱╲╱╲╱╲╱╲╱╲╱╲╱╲╱╲╱╲
+ * WinRM Java Client
+ * ჻჻჻჻჻჻
+ * Copyright (C) 2023 - 2026 MetricsHub
+ * ჻჻჻჻჻჻
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ╲╱╲╱╲╱╲╱╲╱╲╱╲╱╲╱╲╱╲╱╲╱╲╱╲╱╲╱╲╱╲╱╲╱╲╱╲╱╲╱
+ */
+
+import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
+import java.util.concurrent.TimeoutException;
+import org.metricshub.winrm.exceptions.WindowsRemoteException;
+
+/**
+ * A cursor over the raw output of a running remote command, returned by
+ * {@link WindowsRemoteExecutor#startCommand(String, String, long)}. Each {@link #next()} is one
+ * WSMan Receive round trip yielding the output bytes exactly as the server handed them out —
+ * undecoded, because a multibyte character can be split across chunks; decode with a stateful
+ * {@link java.nio.charset.CharsetDecoder} (or accumulate the bytes and decode once at the end).
+ * <p>
+ * The cursor owns the executor's serial connection until the command completes or the cursor is
+ * closed: no other operation can run on the same executor while the cursor is open. Completion
+ * (a {@code null} return from {@link #next()}) sends the protocol's terminate Signal and releases
+ * the connection on its own; closing earlier sends the same Signal, which actually stops the
+ * still-running remote command. Always close the cursor — use try-with-resources.
+ * <p>
+ * A cursor is not thread-safe: advance and close it from one thread at a time.
+ */
+public interface CommandCursor extends AutoCloseable {
+	/**
+	 * Block until the remote command produces output (or completes), for at most one
+	 * per-round-trip timeout.
+	 *
+	 * @return the next chunk of raw output — possibly empty — or {@code null} once the command has
+	 *         completed; the exit code is then available from {@link #exitCode()}
+	 * @throws TimeoutException when the command produces no output for a whole per-round-trip
+	 *         timeout (the inactivity timeout of the stream)
+	 * @throws WindowsRemoteException for any other failure while receiving
+	 */
+	Chunk next() throws TimeoutException, WindowsRemoteException;
+
+	/**
+	 * Bounded variant of {@link #next()}: block at most the given wait for output. When the
+	 * command produces nothing in that window, an <b>empty</b> chunk is returned — a bounded poll
+	 * expiring is not a failure, and the cursor remains fully usable — unlike {@link #next()},
+	 * whose whole per-round-trip timeout counts as the stream's inactivity limit. Deadline-bounded
+	 * waits (e.g. {@code RemoteProcess.waitFor(Duration)}) are built on this.
+	 * <p>
+	 * The default implementation does not bound the wait: it delegates to {@link #next()}.
+	 *
+	 * @param maxWaitMillis how long to block at most, capped by the cursor's per-round-trip timeout
+	 * @return the next chunk of raw output — empty when the wait elapsed first — or {@code null}
+	 *         once the command has completed
+	 * @throws TimeoutException when the server does not even answer the bounded request
+	 * @throws WindowsRemoteException for any other failure while receiving
+	 */
+	default Chunk poll(final long maxWaitMillis) throws TimeoutException, WindowsRemoteException {
+		return next();
+	}
+
+	/**
+	 * Get the command's exit code.
+	 *
+	 * @return the exit code
+	 * @throws IllegalStateException when the command has not completed yet — completion is
+	 *         observed as a {@code null} return from {@link #next()}
+	 */
+	int exitCode();
+
+	/**
+	 * Terminate the command (when it is still running) with the WinRM terminate Signal and release
+	 * the executor's connection. Idempotent; a no-op when the command already completed. After an
+	 * early close, {@link #next()} returns {@code null} without touching the connection again (and
+	 * no exit code is available, since the command never completed). May throw an unchecked
+	 * {@link org.metricshub.winrm.exceptions.WinRMClientException} when the Signal itself fails —
+	 * the remote command may then still be running.
+	 */
+	@Override
+	void close();
+
+	/** One Receive response's worth of raw output bytes, split by stream. */
+	final class Chunk {
+
+		private final byte[] stdout;
+		private final byte[] stderr;
+
+		/**
+		 * Create a chunk over the given stream bytes (not copied: a chunk is a transient carrier
+		 * between the protocol loop and the decoder, not a retained value).
+		 *
+		 * @param stdout the raw stdout bytes of this chunk (possibly empty, never null)
+		 * @param stderr the raw stderr bytes of this chunk (possibly empty, never null)
+		 */
+		@SuppressFBWarnings(value = "EI_EXPOSE_REP2", justification = "Chunks are transient carriers on the output hot path; defensive copies "
+			+
+			"would double the allocation for no benefit")
+		public Chunk(final byte[] stdout, final byte[] stderr) {
+			this.stdout = stdout;
+			this.stderr = stderr;
+		}
+
+		/**
+		 * Get the raw stdout bytes of this chunk.
+		 *
+		 * @return the stdout bytes, possibly empty
+		 */
+		@SuppressFBWarnings(value = "EI_EXPOSE_REP", justification = "Chunks are transient carriers on the output hot path; defensive copies "
+			+
+			"would double the allocation for no benefit")
+		public byte[] stdout() {
+			return stdout;
+		}
+
+		/**
+		 * Get the raw stderr bytes of this chunk.
+		 *
+		 * @return the stderr bytes, possibly empty
+		 */
+		@SuppressFBWarnings(value = "EI_EXPOSE_REP", justification = "Chunks are transient carriers on the output hot path; defensive copies "
+			+
+			"would double the allocation for no benefit")
+		public byte[] stderr() {
+			return stderr;
+		}
+	}
+}