Standard stream reading and writing (#964)

mosaic-tty/api/mosaic-tty.api

@@ -2,9 +2,14 @@ public final class com/jakewharton/mosaic/tty/StandardStreams : java/lang/AutoCl
 	public static final field Companion Lcom/jakewharton/mosaic/tty/StandardStreams$Companion;
 	public static final fun bind ()Lcom/jakewharton/mosaic/tty/StandardStreams;
 	public fun close ()V
+	public final fun interruptInputRead ()V
 	public final fun isErrorTty ()Z
 	public final fun isInputTty ()Z
 	public final fun isOutputTty ()Z
+	public final fun readInput ([BII)I
+	public final fun readInputWithTimeout ([BIII)I
+	public final fun writeError ([BII)I
+	public final fun writeOutput ([BII)I
 }
 
 public final class com/jakewharton/mosaic/tty/StandardStreams$Companion {

mosaic-tty/api/mosaic-tty.klib.api

@@ -12,9 +12,14 @@ final class com.jakewharton.mosaic.tty/IOException : kotlin/Exception { // com.j
 
 final class com.jakewharton.mosaic.tty/StandardStreams : kotlin/AutoCloseable { // com.jakewharton.mosaic.tty/StandardStreams|null[0]
     final fun close() // com.jakewharton.mosaic.tty/StandardStreams.close|close(){}[0]
+    final fun interruptInputRead() // com.jakewharton.mosaic.tty/StandardStreams.interruptInputRead|interruptInputRead(){}[0]
     final fun isErrorTty(): kotlin/Boolean // com.jakewharton.mosaic.tty/StandardStreams.isErrorTty|isErrorTty(){}[0]
     final fun isInputTty(): kotlin/Boolean // com.jakewharton.mosaic.tty/StandardStreams.isInputTty|isInputTty(){}[0]
     final fun isOutputTty(): kotlin/Boolean // com.jakewharton.mosaic.tty/StandardStreams.isOutputTty|isOutputTty(){}[0]
+    final fun readInput(kotlin/ByteArray, kotlin/Int, kotlin/Int): kotlin/Int // com.jakewharton.mosaic.tty/StandardStreams.readInput|readInput(kotlin.ByteArray;kotlin.Int;kotlin.Int){}[0]
+    final fun readInputWithTimeout(kotlin/ByteArray, kotlin/Int, kotlin/Int, kotlin/Int): kotlin/Int // com.jakewharton.mosaic.tty/StandardStreams.readInputWithTimeout|readInputWithTimeout(kotlin.ByteArray;kotlin.Int;kotlin.Int;kotlin.Int){}[0]
+    final fun writeError(kotlin/ByteArray, kotlin/Int, kotlin/Int): kotlin/Int // com.jakewharton.mosaic.tty/StandardStreams.writeError|writeError(kotlin.ByteArray;kotlin.Int;kotlin.Int){}[0]
+    final fun writeOutput(kotlin/ByteArray, kotlin/Int, kotlin/Int): kotlin/Int // com.jakewharton.mosaic.tty/StandardStreams.writeOutput|writeOutput(kotlin.ByteArray;kotlin.Int;kotlin.Int){}[0]
 
     final object Companion { // com.jakewharton.mosaic.tty/StandardStreams.Companion|null[0]
         final fun bind(): com.jakewharton.mosaic.tty/StandardStreams // com.jakewharton.mosaic.tty/StandardStreams.Companion.bind|bind(){}[0]

mosaic-tty/src/commonMain/c/mosaic-streams-posix.c

@@ -1,6 +1,7 @@
 #if defined(__APPLE__) || defined(__linux__)
 
 #include "mosaic-streams-posix.h"
+#include "mosaic-tty-posix.h"
 
 #include <errno.h>
 #include <stdlib.h>
@@ -8,6 +9,8 @@
 
 typedef struct MosaicStreamsImpl {
 	int stdin;
+	int interrupt_stdin_reader;
+	int interrupt_stdin_writer;
 	int stdout;
 	int stderr;
 } MosaicStreamsImpl;
@@ -21,7 +24,15 @@ MosaicStreamsInitResult mosaic_streams_init_internal(int stdin, int stdout, int
 		goto ret;
 	}
 
+	int interruptPipe[2];
+	if (unlikely(pipe(interruptPipe)) != 0) {
+		result.error = errno;
+		goto err_free;
+	}
+
 	streams->stdin = stdin;
+	streams->interrupt_stdin_reader = interruptPipe[0];
+	streams->interrupt_stdin_writer = interruptPipe[1];
 	streams->stdout = stdout;
 	streams->stderr = stderr;
 
@@ -29,6 +40,10 @@ MosaicStreamsInitResult mosaic_streams_init_internal(int stdin, int stdout, int
 
 	ret:
 	return result;
+
+	err_free:
+	free(streams);
+	goto ret;
 }
 
 MosaicStreamsInitResult mosaic_streams_init() {
@@ -60,6 +75,32 @@ MosaicStreamsTtyResult mosaic_streams_is_stderr_tty(MosaicStreams *streams) {
 	return mosaic_streams_is_tty(streams->stderr);
 }
 
+MosaicIoResult mosaic_streams_read_input(MosaicStreams *streams, uint8_t *buffer, int count) {
+	return tty_readInternal(streams->stdin, streams->interrupt_stdin_reader, buffer, count, NULL);
+}
+
+MosaicIoResult mosaic_streams_read_input_with_timeout(MosaicStreams *streams, uint8_t *buffer, int count, int timeoutMillis) {
+	struct timeval timeout;
+	timeout.tv_sec = 0;
+	timeout.tv_usec = timeoutMillis * 1000;
+
+	return tty_readInternal(streams->stdin, streams->interrupt_stdin_reader, buffer, count, &timeout);
+}
+
+uint32_t mosaic_streams_interrupt_input_read(MosaicStreams *streams) {
+	uint8_t space = ' ';
+	MosaicIoResult result = tty_writeInternal(streams->interrupt_stdin_writer, &space, 1);
+	return result.error;
+}
+
+MosaicIoResult mosaic_streams_write_output(MosaicStreams *streams, uint8_t *buffer, int count) {
+	return tty_writeInternal(streams->stdout, buffer, count);
+}
+
+MosaicIoResult mosaic_streams_write_error(MosaicStreams *streams, uint8_t *buffer, int count) {
+	return tty_writeInternal(streams->stderr, buffer, count);
+}
+
 uint32_t mosaic_streams_free(MosaicStreams *streams) {
 	free(streams);
 	return 0;

mosaic-tty/src/commonMain/c/mosaic-streams-windows.c

@@ -1,9 +1,11 @@
 #if defined(_WIN32)
 
 #include "mosaic-streams-windows.h"
+#include "mosaic-tty-windows.h"
 
 typedef struct MosaicStreamsImpl {
 	HANDLE stdin;
+	HANDLE stdin_interrupt_event;
 	HANDLE stdout;
 	HANDLE stderr;
 } MosaicStreamsImpl;
@@ -17,7 +19,14 @@ MosaicStreamsInitResult mosaic_streams_init_internal(HANDLE stdin, HANDLE stdout
 		goto ret;
 	}
 
+	HANDLE stdinInterruptEvent = CreateEvent(NULL, FALSE, FALSE, NULL);
+	if (unlikely(stdinInterruptEvent == NULL)) {
+		result.error = GetLastError();
+		goto err_free;
+	}
+
 	streams->stdin = stdin;
+	streams->stdin_interrupt_event = stdinInterruptEvent;
 	streams->stdout = stdout;
 	streams->stderr = stderr;
 
@@ -25,6 +34,10 @@ MosaicStreamsInitResult mosaic_streams_init_internal(HANDLE stdin, HANDLE stdout
 
 	ret:
 	return result;
+
+	err_free:
+	free(streams);
+	goto ret;
 }
 
 MOSAIC_EXPORT MosaicStreamsInitResult mosaic_streams_init() {
@@ -76,9 +89,61 @@ MOSAIC_EXPORT MosaicStreamsTtyResult mosaic_streams_is_stderr_tty(MosaicStreams
 	return mosaic_streams_is_tty(streams->stderr);
 }
 
+MOSAIC_EXPORT MosaicIoResult mosaic_streams_read_input(MosaicStreams *streams, uint8_t *buffer, int count) {
+	return mosaic_streams_read_input_with_timeout(streams, buffer, count, INFINITE);
+}
+
+MOSAIC_EXPORT MosaicIoResult mosaic_streams_read_input_with_timeout(MosaicStreams *streams, uint8_t *buffer, int count, int timeoutMillis) {
+	MosaicIoResult result = {};
+
+	HANDLE waitHandles[2] = { streams->stdin, streams->stdin_interrupt_event };
+
+	DWORD waitResult = WaitForMultipleObjects(2, waitHandles, FALSE, timeoutMillis);
+	if (likely(waitResult == WAIT_OBJECT_0)) {
+		DWORD c;
+		if (!ReadFile(streams->stdin, buffer, count, &c, NULL)) {
+			goto err;
+		}
+		result.count = c;
+	} else if (unlikely(waitResult == WAIT_FAILED)) {
+		goto err;
+	}
+	// Else return a count of 0 because either:
+	// - The interrupt event was selected (which auto resets its state).
+	// - The user-supplied, non-infinite timeout ran out.
+
+	ret:
+	return result;
+
+	err:
+	result.error = GetLastError();
+	goto ret;
+}
+
+MOSAIC_EXPORT uint32_t mosaic_streams_interrupt_input_read(MosaicStreams *streams) {
+	return likely(SetEvent(streams->stdin_interrupt_event) != 0)
+		? 0
+		: GetLastError();
+}
+
+MOSAIC_EXPORT MosaicIoResult mosaic_streams_write_output(MosaicStreams *streams, uint8_t *buffer, int count) {
+	return tty_writeInternal(streams->stdout, buffer, count);
+}
+
+MOSAIC_EXPORT MosaicIoResult mosaic_streams_write_error(MosaicStreams *streams, uint8_t *buffer, int count) {
+	return tty_writeInternal(streams->stderr, buffer, count);
+}
+
 uint32_t mosaic_streams_free(MosaicStreams *streams) {
+	DWORD result = 0;
+
+	if (unlikely(CloseHandle(streams->stdin_interrupt_event) == 0)) {
+		result = GetLastError();
+	}
+
 	free(streams);
-	return 0;
+
+	return result;
 }
 
 #endif

mosaic-tty/src/commonMain/c/mosaic-streams.h

@@ -24,6 +24,13 @@ MOSAIC_EXPORT MosaicStreamsTtyResult mosaic_streams_is_stdin_tty(MosaicStreams *
 MOSAIC_EXPORT MosaicStreamsTtyResult mosaic_streams_is_stdout_tty(MosaicStreams *streams);
 MOSAIC_EXPORT MosaicStreamsTtyResult mosaic_streams_is_stderr_tty(MosaicStreams *streams);
 
+MOSAIC_EXPORT MosaicIoResult mosaic_streams_read_input(MosaicStreams *streams, uint8_t *buffer, int count);
+MOSAIC_EXPORT MosaicIoResult mosaic_streams_read_input_with_timeout(MosaicStreams *streams, uint8_t *buffer, int count, int timeoutMillis);
+MOSAIC_EXPORT uint32_t mosaic_streams_interrupt_input_read(MosaicStreams *streams);
+
+MOSAIC_EXPORT MosaicIoResult mosaic_streams_write_output(MosaicStreams *streams, uint8_t *buffer, int count);
+MOSAIC_EXPORT MosaicIoResult mosaic_streams_write_error(MosaicStreams *streams, uint8_t *buffer, int count);
+
 MOSAIC_EXPORT uint32_t mosaic_streams_free(MosaicStreams *streams);
 
 #endif // MOSAIC_STREAMS_H

mosaic-tty/src/commonMain/c/mosaic-tty-windows.c

@@ -176,11 +176,11 @@ MOSAIC_EXPORT uint32_t tty_interruptRead(MosaicTty *tty) {
 		: GetLastError();
 }
 
-MOSAIC_EXPORT MosaicIoResult tty_write(MosaicTty *tty, uint8_t *buffer, int count) {
+MosaicIoResult tty_writeInternal(HANDLE h, uint8_t *buffer, int count) {
 	MosaicIoResult result = {};
 
 	DWORD written;
-	if (WriteFile(tty->conout_for_write, buffer, count, &written, NULL)) {
+	if (WriteFile(h, buffer, count, &written, NULL)) {
 		result.count = written;
 	} else {
 		result.error = GetLastError();
@@ -189,6 +189,10 @@ MOSAIC_EXPORT MosaicIoResult tty_write(MosaicTty *tty, uint8_t *buffer, int coun
 	return result;
 }
 
+MOSAIC_EXPORT MosaicIoResult tty_write(MosaicTty *tty, uint8_t *buffer, int count) {
+	return tty_writeInternal(tty->conout_for_write, buffer, count);
+}
+
 MOSAIC_EXPORT uint32_t tty_enableRawMode(MosaicTty *tty) {
 	uint32_t result = 0;
 

mosaic-tty/src/commonMain/c/mosaic-tty-windows.h

@@ -28,4 +28,6 @@ MosaicTtyInitResult tty_initWithHandles(
 	bool conoutForWriteFake
 );
 
+MosaicIoResult tty_writeInternal(HANDLE h, uint8_t *buffer, int count);
+
 #endif // MOSAIC_TTY_WINDOWS_H

mosaic-tty/src/commonMain/kotlin/com/jakewharton/mosaic/tty/StandardStreams.kt

@@ -9,5 +9,12 @@ public expect class StandardStreams : AutoCloseable {
 	public fun isOutputTty(): Boolean
 	public fun isErrorTty(): Boolean
 
+	public fun readInput(buffer: ByteArray, offset: Int, count: Int): Int
+	public fun readInputWithTimeout(buffer: ByteArray, offset: Int, count: Int, timeoutMillis: Int): Int
+	public fun interruptInputRead()
+
+	public fun writeOutput(buffer: ByteArray, offset: Int, count: Int): Int
+	public fun writeError(buffer: ByteArray, offset: Int, count: Int): Int
+
 	override fun close()
 }

mosaic-tty/src/jvmJdk22/kotlin/com/jakewharton/mosaic/tty/StandardStreams.kt

@@ -2,11 +2,17 @@ package com.jakewharton.mosaic.tty
 
 import com.jakewharton.mosaic.tty.Libmosaic.mosaic_streams_free
 import com.jakewharton.mosaic.tty.Libmosaic.mosaic_streams_init
+import com.jakewharton.mosaic.tty.Libmosaic.mosaic_streams_interrupt_input_read
 import com.jakewharton.mosaic.tty.Libmosaic.mosaic_streams_is_stderr_tty
 import com.jakewharton.mosaic.tty.Libmosaic.mosaic_streams_is_stdin_tty
 import com.jakewharton.mosaic.tty.Libmosaic.mosaic_streams_is_stdout_tty
+import com.jakewharton.mosaic.tty.Libmosaic.mosaic_streams_read_input
+import com.jakewharton.mosaic.tty.Libmosaic.mosaic_streams_read_input_with_timeout
+import com.jakewharton.mosaic.tty.Libmosaic.mosaic_streams_write_error
+import com.jakewharton.mosaic.tty.Libmosaic.mosaic_streams_write_output
 import java.lang.foreign.Arena
 import java.lang.foreign.MemorySegment
+import java.lang.foreign.ValueLayout
 
 public class StandardStreams internal constructor(
 	private var ptr: MemorySegment,
@@ -57,6 +63,63 @@ public class StandardStreams internal constructor(
 		}
 	}
 
+	@Throws(IOException::class)
+	public fun readInput(buffer: ByteArray, offset: Int, count: Int): Int {
+		val segment = Libmosaic.LIBRARY_ARENA.allocate(count.toLong())
+		val result = mosaic_streams_read_input(Arena.global(), ptr, segment, count)
+		val error = MosaicIoResult.error(result)
+		if (error == 0) {
+			val read = MosaicIoResult.count(result)
+			MemorySegment.copy(segment, ValueLayout.JAVA_BYTE, 0L, buffer, offset, read)
+			return read
+		}
+		throwIoe(error)
+	}
+
+	@Throws(IOException::class)
+	public fun readInputWithTimeout(buffer: ByteArray, offset: Int, count: Int, timeoutMillis: Int): Int {
+		val segment = Libmosaic.LIBRARY_ARENA.allocate(count.toLong())
+		val result = mosaic_streams_read_input_with_timeout(Arena.global(), ptr, segment, count, timeoutMillis)
+		val error = MosaicIoResult.error(result)
+		if (error == 0) {
+			val read = MosaicIoResult.count(result)
+			MemorySegment.copy(segment, ValueLayout.JAVA_BYTE, 0L, buffer, offset, read)
+			return read
+		}
+		throwIoe(error)
+	}
+
+	@Throws(IOException::class)
+	public fun interruptInputRead() {
+		val error = mosaic_streams_interrupt_input_read(ptr)
+		if (error == 0) return
+		throwIoe(error)
+	}
+
+	@Throws(IOException::class)
+	public fun writeOutput(buffer: ByteArray, offset: Int, count: Int): Int {
+		val segment = Libmosaic.LIBRARY_ARENA.allocate(count.toLong())
+		MemorySegment.copy(buffer, offset, segment, ValueLayout.JAVA_BYTE, 0, count)
+		val result = mosaic_streams_write_output(Arena.global(), ptr, segment, count)
+		val error = MosaicIoResult.error(result)
+		if (error == 0) {
+			return MosaicIoResult.count(result)
+		}
+		throwIoe(error)
+	}
+
+	@Throws(IOException::class)
+	public fun writeError(buffer: ByteArray, offset: Int, count: Int): Int {
+		val segment = Libmosaic.LIBRARY_ARENA.allocate(count.toLong())
+		MemorySegment.copy(buffer, offset, segment, ValueLayout.JAVA_BYTE, 0, count)
+		val result = mosaic_streams_write_error(Arena.global(), ptr, segment, count)
+		val error = MosaicIoResult.error(result)
+		if (error == 0) {
+			return MosaicIoResult.count(result)
+		}
+		throwIoe(error)
+	}
+
 	override fun close() {
 		val ptr = ptr
 		if (ptr != MemorySegment.NULL) {

mosaic-tty/src/jvmMain/java/com/jakewharton/mosaic/tty/Jni.java

@@ -13,6 +13,37 @@ final class Jni {
 
 	static native boolean streamsErrorIsTty(long streamsPtr);
 
+	static native int streamsReadInput(
+		long streamsPtr,
+		byte[] buffer,
+		int offset,
+		int count
+	);
+
+	static native int streamsReadInputWithTimeout(
+		long streamsPtr,
+		byte[] buffer,
+		int offset,
+		int count,
+		int timeout
+	);
+
+	static native void streamsInterruptInputRead(long streamsPtr);
+
+	static native int streamsWriteOutput(
+		long streamsPtr,
+		byte[] buffer,
+		int offset,
+		int count
+	);
+
+	static native int streamsWriteError(
+		long streamsPtr,
+		byte[] buffer,
+		int offset,
+		int count
+	);
+
 	static native void streamsFree(long streamsPtr);
 
 	static native long ttyCallbackInit(Tty.Callback callback);

mosaic-tty/src/jvmMain/kotlin/com/jakewharton/mosaic/tty/StandardStreams.kt

@@ -25,6 +25,31 @@ public actual class StandardStreams internal constructor(
 		return Jni.streamsErrorIsTty(ptr)
 	}
 
+	@Throws(IOException::class)
+	public actual fun readInput(buffer: ByteArray, offset: Int, count: Int): Int {
+		return Jni.streamsReadInput(ptr, buffer, offset, count)
+	}
+
+	@Throws(IOException::class)
+	public actual fun readInputWithTimeout(buffer: ByteArray, offset: Int, count: Int, timeoutMillis: Int): Int {
+		return Jni.streamsReadInputWithTimeout(ptr, buffer, offset, count, timeoutMillis)
+	}
+
+	@Throws(IOException::class)
+	public actual fun interruptInputRead() {
+		return Jni.streamsInterruptInputRead(ptr)
+	}
+
+	@Throws(IOException::class)
+	public actual fun writeOutput(buffer: ByteArray, offset: Int, count: Int): Int {
+		return Jni.streamsWriteOutput(ptr, buffer, offset, count)
+	}
+
+	@Throws(IOException::class)
+	public actual fun writeError(buffer: ByteArray, offset: Int, count: Int): Int {
+		return Jni.streamsWriteError(ptr, buffer, offset, count)
+	}
+
 	@Throws(IOException::class)
 	actual override fun close() {
 		val ptr = ptr

mosaic-tty/src/nativeMain/kotlin/com/jakewharton/mosaic/tty/StandardStreams.kt

@@ -2,7 +2,9 @@ package com.jakewharton.mosaic.tty
 
 import kotlinx.cinterop.CValue
 import kotlinx.cinterop.CValuesRef
+import kotlinx.cinterop.addressOf
 import kotlinx.cinterop.useContents
+import kotlinx.cinterop.usePinned
 
 public actual class StandardStreams internal constructor(
 	ptr: CValuesRef<MosaicStreams>,
@@ -32,6 +34,48 @@ public actual class StandardStreams internal constructor(
 	public actual fun isOutputTty(): Boolean = mosaic_streams_is_stdout_tty(ptr).isTty
 	public actual fun isErrorTty(): Boolean = mosaic_streams_is_stderr_tty(ptr).isTty
 
+	public actual fun readInput(buffer: ByteArray, offset: Int, count: Int): Int {
+		buffer.asUByteArray().usePinned {
+			mosaic_streams_read_input(ptr, it.addressOf(offset), count).useContents {
+				if (error == 0U) return this.count
+				throwIoe(error)
+			}
+		}
+	}
+
+	public actual fun readInputWithTimeout(buffer: ByteArray, offset: Int, count: Int, timeoutMillis: Int): Int {
+		buffer.asUByteArray().usePinned {
+			mosaic_streams_read_input_with_timeout(ptr, it.addressOf(offset), count, timeoutMillis).useContents {
+				if (error == 0U) return this.count
+				throwIoe(error)
+			}
+		}
+	}
+
+	public actual fun interruptInputRead() {
+		val error = mosaic_streams_interrupt_input_read(ptr)
+		if (error == 0U) return
+		throwIoe(error)
+	}
+
+	public actual fun writeOutput(buffer: ByteArray, offset: Int, count: Int): Int {
+		buffer.asUByteArray().usePinned {
+			mosaic_streams_write_output(ptr, it.addressOf(offset), count).useContents {
+				if (error == 0U) return this.count
+				throwIoe(error)
+			}
+		}
+	}
+
+	public actual fun writeError(buffer: ByteArray, offset: Int, count: Int): Int {
+		buffer.asUByteArray().usePinned {
+			mosaic_streams_write_error(ptr, it.addressOf(offset), count).useContents {
+				if (error == 0U) return this.count
+				throwIoe(error)
+			}
+		}
+	}
+
 	actual override fun close() {
 		ptr?.let {
 			mosaic_streams_free(it)