meta: add fx.diagnostics namespace and assembly

the fx.diagnostics namespace includes a process class for starting and managing processes

assemblies/fx.diagnostics/CMakeLists.txt

@@ -0,0 +1,4 @@
+add_fx_assembly(
+	NAME fx.diagnostics
+	NAMESPACES fx.diagnostics
+	DEPENDENCIES fx.runtime fx.collections fx.io)

assemblies/fx.diagnostics/assembly.c

@@ -0,0 +1,7 @@
+#include <fx/macros.h>
+#include <fx/reflection/assembly.h>
+
+FX_ASSEMBLY_BEGIN(fx_diagnostics)
+	FX_ASSEMBLY_NAME("fx.diagnostics");
+	FX_ASSEMBLY_VERSION(1, 0, 0, 0);
+FX_ASSEMBLY_END(fx_diagnostics)

fx.diagnostics/CMakeLists.txt

@@ -0,0 +1 @@
+export_fx_namespace_details(fx.diagnostics)

fx.diagnostics/include/fx/diagnostics/process.h

@@ -0,0 +1,41 @@
+#ifndef FX_DIAGNOSTICS_PROCESS_H_
+#define FX_DIAGNOSTICS_PROCESS_H_
+
+#include <fx/macros.h>
+#include <fx/stream.h>
+
+FX_DECLS_BEGIN;
+
+#define FX_DIAGNOSTICS_TYPE_PROCESS (fx_process_get_type())
+
+FX_DECLARE_TYPE(fx_process);
+
+FX_TYPE_CLASS_DECLARATION_BEGIN(fx_process)
+FX_TYPE_CLASS_DECLARATION_END(fx_process)
+
+typedef struct fx_process_start_info {
+	bool proc_redirect_stdin, proc_redirect_stdout, proc_redirect_stderr;
+	const char *proc_file_name;
+	const char **proc_args;
+	size_t proc_args_count;
+} fx_process_start_info;
+
+FX_API fx_type_id fx_process_get_type(void);
+
+FX_API fx_process *fx_process_create(const fx_process_start_info *info);
+
+FX_API fx_status fx_process_start(fx_process *proc);
+FX_API fx_status fx_process_wait(fx_process *proc, int *out_result);
+FX_API fx_status fx_process_kill(fx_process *proc);
+
+FX_API fx_stream *fx_process_get_stdin(fx_process *proc);
+FX_API fx_stream *fx_process_get_stdout(fx_process *proc);
+FX_API fx_stream *fx_process_get_stderr(fx_process *proc);
+
+FX_API void fx_process_close_stdin(fx_process *proc);
+FX_API void fx_process_close_stdout(fx_process *proc);
+FX_API void fx_process_close_stderr(fx_process *proc);
+
+FX_DECLS_END;
+
+#endif

fx.diagnostics/sys/apple/process.c

@@ -0,0 +1,358 @@
+#include <fx/collections/array.h>
+#include <fx/diagnostics/process.h>
+#include <fx/io/pipe.h>
+#include <fx/io/stream.h>
+#include <fx/iterator.h>
+#include <fx/macros.h>
+#include <fx/string.h>
+#include <signal.h>
+#include <unistd.h>
+
+FX_API fx_type_id fx_process_iterator_get_type();
+
+FX_DECLARE_TYPE(fx_process_iterator);
+
+FX_TYPE_CLASS_DECLARATION_BEGIN(fx_process_iterator)
+FX_TYPE_CLASS_DECLARATION_END(fx_process_iterator)
+
+struct fx_process_p {
+	fx_string *proc_exec_path;
+	fx_array *proc_args;
+	bool proc_redirect_stdin, proc_redirect_stdout, proc_redirect_stderr;
+	fx_stream *proc_stdin, *proc_stdout, *proc_stderr;
+	pid_t proc_id;
+};
+
+struct fx_process_iterator_p {
+};
+
+/*** PRIVATE FUNCTIONS ********************************************************/
+
+static fx_status process_start(struct fx_process_p *proc)
+{
+	fx_iostream *in_r = NULL, *in_w = NULL;
+	fx_iostream *out_r = NULL, *out_w = NULL;
+	fx_iostream *err_r = NULL, *err_w = NULL;
+	fx_status status = FX_SUCCESS;
+
+	if (proc->proc_redirect_stdin) {
+		status = fx_pipe_create(&in_r, &in_w);
+		if (!FX_OK(status)) {
+			goto cleanup;
+		}
+	}
+
+	if (proc->proc_redirect_stdout) {
+		status = fx_pipe_create(&out_r, &out_w);
+		if (!FX_OK(status)) {
+			goto cleanup;
+		}
+	}
+
+	if (proc->proc_redirect_stderr) {
+		status = fx_pipe_create(&err_r, &err_w);
+		if (!FX_OK(status)) {
+			goto cleanup;
+		}
+	}
+
+	pid_t pid = fork();
+
+	if (pid < 0) {
+		status = FX_ERR_INVALID_ARGUMENT;
+		goto cleanup;
+	}
+
+	if (pid > 0) {
+		proc->proc_id = pid;
+		proc->proc_stdin = fx_iostream_ref(in_w);
+		proc->proc_stdout = fx_iostream_ref(out_r);
+		proc->proc_stderr = fx_iostream_ref(err_r);
+		status = FX_SUCCESS;
+		goto cleanup;
+	}
+
+	int fd = -1;
+	if (proc->proc_redirect_stdin) {
+		fx_iostream_unref(in_w);
+		fd = fx_iostream_steal_os_handle(in_r);
+		fprintf(stderr, "dup2(%d, %d)\n", fd, 0);
+		int err = dup2(fd, 0);
+		close(fd);
+	}
+
+	if (proc->proc_redirect_stdout) {
+		fx_iostream_unref(out_r);
+		fd = fx_iostream_steal_os_handle(out_w);
+		fprintf(stderr, "dup2(%d, %d)\n", fd, 1);
+		dup2(fd, 1);
+		close(fd);
+	}
+
+	if (proc->proc_redirect_stderr) {
+		fx_iostream_unref(err_r);
+		fd = fx_iostream_steal_os_handle(err_w);
+		fprintf(stderr, "dup2(%d, %d)\n", fd, 2);
+		dup2(fd, 2);
+		close(fd);
+	}
+
+	const char *args[] = {
+		"cowsay",
+		NULL,
+	};
+	printf("exec '%s'\n", fx_string_get_cstr(proc->proc_exec_path));
+	execv(fx_string_get_cstr(proc->proc_exec_path), (char *const *)args);
+	exit(127);
+
+cleanup:
+	fx_iostream_unref(in_r);
+	fx_iostream_unref(in_w);
+	fx_iostream_unref(out_r);
+	fx_iostream_unref(out_w);
+	fx_iostream_unref(err_r);
+	fx_iostream_unref(err_w);
+	return status;
+}
+
+static fx_status process_wait(struct fx_process_p *proc, int *out_result)
+{
+	if (proc->proc_id == 0) {
+		return FX_ERR_BAD_STATE;
+	}
+
+	int status = 0;
+
+	do {
+		int err = waitpid(proc->proc_id, &status, WUNTRACED);
+		if (err != 0) {
+			return FX_ERR_BAD_STATE;
+		}
+	} while (!WIFEXITED(status));
+
+	proc->proc_id = 0;
+	*out_result = WEXITSTATUS(status);
+	return FX_ERR_NOT_SUPPORTED;
+}
+
+static fx_status process_kill(struct fx_process_p *proc)
+{
+	if (proc->proc_id == 0) {
+		return FX_ERR_BAD_STATE;
+	}
+
+	int err = kill(proc->proc_id, SIGTERM);
+	if (err != 0) {
+		return FX_ERR_BAD_STATE;
+	}
+
+	return FX_SUCCESS;
+}
+
+static fx_stream *process_get_stdin(struct fx_process_p *proc)
+{
+	return proc->proc_stdin;
+}
+
+static fx_stream *process_get_stdout(struct fx_process_p *proc)
+{
+	return proc->proc_stdout;
+}
+
+static fx_stream *process_get_stderr(struct fx_process_p *proc)
+{
+	return proc->proc_stderr;
+}
+
+static void process_close_stdin(struct fx_process_p *proc)
+{
+	fx_iostream_unref(proc->proc_stdin);
+	proc->proc_stdin = NULL;
+}
+
+static void process_close_stdout(struct fx_process_p *proc)
+{
+	fx_iostream_unref(proc->proc_stdout);
+	proc->proc_stdout = NULL;
+}
+
+static void process_close_stderr(struct fx_process_p *proc)
+{
+	fx_iostream_unref(proc->proc_stderr);
+	proc->proc_stderr = NULL;
+}
+
+/*** PUBLIC FUNCTIONS *********************************************************/
+
+fx_process *fx_process_create(const fx_process_start_info *info)
+{
+	fx_process *out = fx_object_create(FX_DIAGNOSTICS_TYPE_PROCESS);
+	if (!out) {
+		return NULL;
+	}
+
+	struct fx_process_p *proc
+		= fx_object_get_private(out, FX_DIAGNOSTICS_TYPE_PROCESS);
+
+	proc->proc_exec_path = fx_string_create_from_cstr(info->proc_file_name);
+	if (!proc->proc_exec_path) {
+		fx_process_unref(out);
+		return NULL;
+	}
+
+	if (info->proc_args_count > 0) {
+		proc->proc_args = fx_array_create();
+		if (!proc->proc_args) {
+			fx_process_unref(out);
+			return NULL;
+		}
+	}
+
+	for (size_t i = 0; i < info->proc_args_count; i++) {
+		fx_string *arg = fx_string_create_from_cstr(info->proc_args[i]);
+		if (!arg) {
+			fx_process_unref(out);
+			return NULL;
+		}
+
+		fx_array_push_back(proc->proc_args, FX_VALUE_OBJECT(arg));
+		fx_string_unref(arg);
+	}
+
+	proc->proc_redirect_stdin = info->proc_redirect_stdin;
+	proc->proc_redirect_stdout = info->proc_redirect_stdout;
+	proc->proc_redirect_stderr = info->proc_redirect_stderr;
+
+	return out;
+}
+
+fx_status fx_process_start(fx_process *proc)
+{
+	FX_CLASS_DISPATCH_STATIC_0(
+		FX_DIAGNOSTICS_TYPE_PROCESS,
+		process_start,
+		proc);
+}
+
+fx_status fx_process_wait(fx_process *proc, int *out_result)
+{
+	FX_CLASS_DISPATCH_STATIC(
+		FX_DIAGNOSTICS_TYPE_PROCESS,
+		process_wait,
+		proc,
+		out_result);
+}
+
+fx_status fx_process_kill(fx_process *proc)
+{
+	FX_CLASS_DISPATCH_STATIC_0(
+		FX_DIAGNOSTICS_TYPE_PROCESS,
+		process_kill,
+		proc);
+}
+
+fx_stream *fx_process_get_stdin(fx_process *proc)
+{
+	FX_CLASS_DISPATCH_STATIC_0(
+		FX_DIAGNOSTICS_TYPE_PROCESS,
+		process_get_stdin,
+		proc);
+}
+
+fx_stream *fx_process_get_stdout(fx_process *proc)
+{
+	FX_CLASS_DISPATCH_STATIC_0(
+		FX_DIAGNOSTICS_TYPE_PROCESS,
+		process_get_stdout,
+		proc);
+}
+
+fx_stream *fx_process_get_stderr(fx_process *proc)
+{
+	FX_CLASS_DISPATCH_STATIC_0(
+		FX_DIAGNOSTICS_TYPE_PROCESS,
+		process_get_stderr,
+		proc);
+}
+
+void fx_process_close_stdin(fx_process *proc)
+{
+	FX_CLASS_DISPATCH_STATIC_V0(
+		FX_DIAGNOSTICS_TYPE_PROCESS,
+		process_close_stdin,
+		proc);
+}
+
+void fx_process_close_stdout(fx_process *proc)
+{
+	FX_CLASS_DISPATCH_STATIC_V0(
+		FX_DIAGNOSTICS_TYPE_PROCESS,
+		process_close_stdout,
+		proc);
+}
+
+void fx_process_close_stderr(fx_process *proc)
+{
+	FX_CLASS_DISPATCH_STATIC_V0(
+		FX_DIAGNOSTICS_TYPE_PROCESS,
+		process_close_stderr,
+		proc);
+}
+
+/*** VIRTUAL FUNCTIONS ********************************************************/
+
+static void process_init(fx_object *obj, void *priv)
+{
+}
+
+static void process_fini(fx_object *obj, void *priv)
+{
+}
+
+/*** ITERATOR DEFINITION ******************************************************/
+
+static enum fx_status process_iterator_move_next(const fx_iterator *obj)
+{
+	return FX_ERR_NO_DATA;
+}
+
+static const fx_value *process_iterator_get_value(const fx_iterator *obj)
+{
+	return NULL;
+}
+
+/*** CLASS DEFINITION *********************************************************/
+
+FX_TYPE_CLASS_BEGIN(fx_process)
+	FX_TYPE_VTABLE_INTERFACE_BEGIN(fx_object, FX_TYPE_OBJECT)
+		FX_INTERFACE_ENTRY(to_string) = NULL;
+	FX_TYPE_VTABLE_INTERFACE_END(fx_object, FX_TYPE_OBJECT)
+FX_TYPE_CLASS_END(fx_process)
+
+FX_TYPE_DEFINITION_BEGIN(fx_process)
+	FX_TYPE_ID(0x7334594f, 0xa1c3, 0x4715, 0xab69, 0x68a709e1b64b);
+	FX_TYPE_NAME("fx.diagnostics.process");
+	FX_TYPE_CLASS(fx_process_class);
+	FX_TYPE_INSTANCE_PRIVATE(struct fx_process_p);
+	FX_TYPE_INSTANCE_INIT(process_init);
+	FX_TYPE_INSTANCE_FINI(process_fini);
+FX_TYPE_DEFINITION_END(fx_process)
+
+FX_TYPE_CLASS_BEGIN(fx_process_iterator)
+	FX_TYPE_VTABLE_INTERFACE_BEGIN(fx_object, FX_TYPE_OBJECT)
+		FX_INTERFACE_ENTRY(to_string) = NULL;
+	FX_TYPE_VTABLE_INTERFACE_END(fx_object, FX_TYPE_OBJECT)
+
+	FX_TYPE_VTABLE_INTERFACE_BEGIN(fx_iterator, FX_TYPE_ITERATOR)
+		FX_INTERFACE_ENTRY(it_move_next) = process_iterator_move_next;
+		FX_INTERFACE_ENTRY(it_get_value) = process_iterator_get_value;
+	FX_TYPE_VTABLE_INTERFACE_END(fx_iterator, FX_TYPE_ITERATOR)
+FX_TYPE_CLASS_END(fx_process_iterator)
+
+FX_TYPE_DEFINITION_BEGIN(fx_process_iterator)
+	FX_TYPE_ID(0x67eb13b6, 0x25d1, 0x424a, 0xb136, 0x871a07829089);
+	FX_TYPE_NAME("fx.diagnostics.process.iterator");
+	FX_TYPE_EXTENDS(FX_TYPE_ITERATOR);
+	FX_TYPE_CLASS(fx_process_iterator_class);
+	FX_TYPE_INSTANCE_PRIVATE(struct fx_process_iterator_p);
+FX_TYPE_DEFINITION_END(fx_process_iterator)

fx.diagnostics/test/redirect.c

@@ -0,0 +1,44 @@
+#include <fx/diagnostics/process.h>
+
+int main(int argc, const char **argv)
+{
+	if (argc < 3) {
+		return -1;
+	}
+
+	const char *exec = argv[1];
+	const char *msg = argv[2];
+
+	fx_process_start_info info = {
+		.proc_args = NULL,
+		.proc_args_count = 0,
+		.proc_redirect_stdin = true,
+		.proc_file_name = exec,
+	};
+
+	fx_process *process = fx_process_create(&info);
+	if (!process) {
+		fprintf(stderr, "Process creation failed\n");
+		return -1;
+	}
+
+	fx_status status = fx_process_start(process);
+	if (!FX_OK(status)) {
+		fprintf(stderr,
+		        "Process creation failed: %s\n",
+		        fx_status_description(status));
+		return -1;
+	}
+
+#if 1
+	fx_stream *in = fx_process_get_stdin(process);
+	fx_stream_write_cstr(in, "Hello, world!\n", NULL);
+	fx_process_close_stdin(process);
+#endif
+
+	int result;
+	fx_process_wait(process, &result);
+	fx_process_unref(process);
+
+	return 0;
+}

fx.io/include/fx/io/pipe.h

@@ -0,0 +1,8 @@
+#ifndef FX_IO_PIPE_H_
+#define FX_IO_PIPE_H_
+
+#include <fx/io/stream.h>
+
+FX_API fx_status fx_pipe_create(fx_iostream **reader, fx_iostream **writer);
+
+#endif

fx.io/include/fx/io/stream.h

@@ -0,0 +1,29 @@
+#ifndef FX_IO_STREAM_H_
+#define FX_IO_STREAM_H_
+
+#include <fx/int.h>
+#include <fx/macros.h>
+#include <fx/stream.h>
+
+FX_DECLS_BEGIN;
+
+#define FX_IO_TYPE_STREAM (fx_iostream_get_type())
+
+FX_DECLARE_TYPE(fx_iostream);
+
+FX_TYPE_CLASS_DECLARATION_BEGIN(fx_iostream)
+FX_TYPE_CLASS_DECLARATION_END(fx_iostream)
+
+FX_API fx_type_id fx_iostream_get_type(void);
+
+FX_API fx_iostream *fx_iostream_create(
+	fx_stream_mode mode,
+	uptr os_handle,
+	bool close_handle_on_release);
+
+FX_API uptr fx_iostream_get_os_handle(const fx_iostream *stream);
+FX_API uptr fx_iostream_steal_os_handle(fx_iostream *stream);
+
+FX_DECLS_END;
+
+#endif

fx.io/sys/apple/pipe.c

@@ -0,0 +1,30 @@
+#include "posix.h"
+
+#include <errno.h>
+#include <fx/io/stream.h>
+#include <unistd.h>
+
+fx_status fx_pipe_create(fx_iostream **reader, fx_iostream **writer)
+{
+	int fds[2];
+	int err = pipe(fds);
+	if (err != 0) {
+		return fx_status_from_errno(errno, FX_ERR_NOT_SUPPORTED);
+	}
+
+	fprintf(stderr, "created pipe %d->%d\n", fds[1], fds[0]);
+	fx_iostream *r = fx_iostream_create(FX_STREAM_READ, fds[0], true);
+	fx_iostream *w = fx_iostream_create(FX_STREAM_WRITE, fds[1], true);
+
+	if (!r || !w) {
+		fx_iostream_unref(r);
+		fx_iostream_unref(w);
+		close(fds[0]);
+		close(fds[1]);
+		return FX_ERR_NO_MEMORY;
+	}
+
+	*reader = r;
+	*writer = w;
+	return FX_SUCCESS;
+}

fx.io/sys/apple/stream.c

@@ -0,0 +1,150 @@
+#include "posix.h"
+
+#include <errno.h>
+#include <fx/io/stream.h>
+#include <stdarg.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <unistd.h>
+
+/*** PRIVATE DATA *************************************************************/
+
+struct fx_iostream_p {
+	int s_fd;
+	bool s_fd_close_on_release;
+};
+
+/*** PRIVATE FUNCTIONS ********************************************************/
+
+static uptr iostream_get_os_handle(const struct fx_iostream_p *stream)
+{
+	return stream->s_fd;
+}
+
+static uptr iostream_steal_os_handle(struct fx_iostream_p *stream)
+{
+	uptr out = stream->s_fd;
+	stream->s_fd = -1;
+	stream->s_fd_close_on_release = false;
+	return out;
+}
+
+/*** PUBLIC FUNCTIONS *********************************************************/
+
+fx_iostream *fx_iostream_create(
+	fx_stream_mode mode,
+	uptr os_handle,
+	bool close_handle_on_release)
+{
+	fx_iostream *s = fx_object_create(FX_IO_TYPE_STREAM);
+	if (!s) {
+		return NULL;
+	}
+
+	fx_stream_cfg *cfg = fx_object_get_protected(s, FX_TYPE_STREAM);
+	struct fx_iostream_p *p = fx_object_get_private(s, FX_IO_TYPE_STREAM);
+
+	cfg->s_mode = mode | Z__FX_STREAM_STATIC;
+	p->s_fd = (int)os_handle;
+	p->s_fd_close_on_release = close_handle_on_release;
+
+	return s;
+}
+
+uptr fx_iostream_get_os_handle(const fx_iostream *stream)
+{
+	FX_CLASS_DISPATCH_STATIC_0(
+		FX_IO_TYPE_STREAM,
+		iostream_get_os_handle,
+		stream);
+}
+
+uptr fx_iostream_steal_os_handle(fx_iostream *stream)
+{
+	FX_CLASS_DISPATCH_STATIC_0(
+		FX_IO_TYPE_STREAM,
+		iostream_steal_os_handle,
+		stream);
+}
+
+/*** VIRTUAL FUNCTIONS ********************************************************/
+
+static void iostream_init(fx_object *obj, void *priv)
+{
+	struct fx_iostream_p *stream = priv;
+}
+
+static void iostream_fini(fx_object *obj, void *priv)
+{
+	struct fx_iostream_p *stream = priv;
+
+	if (stream->s_fd_close_on_release) {
+		close(stream->s_fd);
+		fprintf(stderr, "closed fd %d\n", stream->s_fd);
+	}
+}
+
+enum fx_status stream_read(
+	fx_stream *stream,
+	void *buf,
+	size_t count,
+	size_t *nr_read)
+{
+	struct fx_iostream_p *s
+		= fx_object_get_private(stream, FX_IO_TYPE_STREAM);
+
+	long r = read(s->s_fd, buf, count);
+	if (r < 0) {
+		return fx_status_from_errno(errno, FX_ERR_IO_FAILURE);
+	}
+
+	*nr_read = r;
+	return FX_SUCCESS;
+}
+
+enum fx_status stream_write(
+	fx_stream *stream,
+	const void *buf,
+	size_t count,
+	size_t *nr_written)
+{
+	struct fx_iostream_p *s
+		= fx_object_get_private(stream, FX_IO_TYPE_STREAM);
+
+	long w = write(s->s_fd, buf, count);
+	if (w < 0) {
+		return fx_status_from_errno(errno, FX_ERR_IO_FAILURE);
+	}
+
+	*nr_written = w;
+	return FX_SUCCESS;
+}
+
+/*** CLASS DEFINITION *********************************************************/
+
+FX_TYPE_CLASS_BEGIN(fx_iostream)
+	FX_TYPE_VTABLE_INTERFACE_BEGIN(fx_object, FX_TYPE_OBJECT)
+		FX_INTERFACE_ENTRY(to_string) = NULL;
+	FX_TYPE_VTABLE_INTERFACE_END(fx_object, FX_TYPE_OBJECT)
+
+	FX_TYPE_VTABLE_INTERFACE_BEGIN(fx_stream, FX_TYPE_STREAM)
+		FX_INTERFACE_ENTRY(s_close) = NULL;
+		FX_INTERFACE_ENTRY(s_seek) = NULL;
+		FX_INTERFACE_ENTRY(s_tell) = NULL;
+		FX_INTERFACE_ENTRY(s_getc) = NULL;
+		FX_INTERFACE_ENTRY(s_read) = stream_read;
+		FX_INTERFACE_ENTRY(s_write) = stream_write;
+		FX_INTERFACE_ENTRY(s_reserve) = NULL;
+	FX_TYPE_VTABLE_INTERFACE_END(fx_stream, FX_TYPE_STREAM)
+FX_TYPE_CLASS_END(fx_iostream)
+
+FX_TYPE_DEFINITION_BEGIN(fx_iostream)
+	FX_TYPE_ID(0xc0b1c3c9, 0xa9c6, 0x4910, 0x8786, 0x574b0696e7aa);
+	FX_TYPE_NAME("fx.io.stream");
+	FX_TYPE_EXTENDS(FX_TYPE_STREAM);
+	FX_TYPE_CLASS(fx_iostream_class);
+	FX_TYPE_INSTANCE_PRIVATE(struct fx_iostream_p);
+	FX_TYPE_INSTANCE_INIT(iostream_init);
+	FX_TYPE_INSTANCE_FINI(iostream_fini);
+FX_TYPE_DEFINITION_END(fx_iostream)

fx.io/test/pipes.c

@@ -0,0 +1,23 @@
+#include <fx/io/pipe.h>
+#include <stdio.h>
+
+int main(void)
+{
+	fx_iostream *read = NULL, *write = NULL;
+	fx_status status = fx_pipe_create(&read, &write);
+
+	if (!FX_OK(status)) {
+		fprintf(stderr,
+		        "failed to create pipe: %s\n",
+		        fx_status_description(status));
+		return -1;
+	}
+
+	fx_stream_write_cstr(write, "Hello, world!\n", NULL);
+
+	char buf[128];
+	fx_stream_read_line(read, buf, sizeof buf);
+
+	printf("read from pipe: %s\n", buf);
+	return 0;
+}

fx/stream.c

@@ -642,9 +642,8 @@ static fx_stream *init_stdio_stream(FILE *fp, fx_stream_mode mode)
 		return NULL;
 	}
 
-	struct fx_stdio_stream_p *p = fx_object_get_private(
+	struct fx_stdio_stream_p *p
-		stream,
+		= fx_object_get_private(stream, FX_TYPE_STDIO_STREAM);
-		FX_TYPE_STDIO_STREAM);
 	fx_stream_cfg *cfg = fx_object_get_protected(stream, FX_TYPE_STREAM);
 	p->s_fp = fp;
 	cfg->s_mode = mode;
@@ -681,8 +680,8 @@ static enum fx_status stream_push_indent(struct stream_data *stream, int indent)
 		stream->s_private->s_istack_size += 4;
 	}
 
-	int cur_indent = stream->s_private
+	int cur_indent
-	                         ->s_istack[stream->s_private->s_istack_ptr];
+		= stream->s_private->s_istack[stream->s_private->s_istack_ptr];
 	stream->s_private->s_istack[++stream->s_private->s_istack_ptr]
 		= cur_indent + indent;
 
@@ -750,9 +749,8 @@ fx_stream_buffer *fx_stream_buffer_create_dynamic(size_t buffer_size)
 		return NULL;
 	}
 
-	struct fx_stream_buffer_p *p = fx_object_get_private(
+	struct fx_stream_buffer_p *p
-		buffer,
+		= fx_object_get_private(buffer, FX_TYPE_STREAM_BUFFER);
-		FX_TYPE_STREAM_BUFFER);
 
 	p->p_buf = malloc(buffer_size);
 	if (!p->p_buf) {
@@ -773,9 +771,8 @@ fx_stream_buffer *fx_stream_buffer_create(void *buf, size_t len)
 		return NULL;
 	}
 
-	struct fx_stream_buffer_p *p = fx_object_get_private(
+	struct fx_stream_buffer_p *p
-		buffer,
+		= fx_object_get_private(buffer, FX_TYPE_STREAM_BUFFER);
-		FX_TYPE_STREAM_BUFFER);
 
 	p->p_buf = buf;
 	p->p_buf_len = len;
@@ -1130,15 +1127,14 @@ static enum fx_status stdio_read(
 	size_t max,
 	size_t *nr_read)
 {
-	struct fx_stdio_stream_p *p = fx_object_get_private(
+	struct fx_stdio_stream_p *p
-		stream,
+		= fx_object_get_private(stream, FX_TYPE_STDIO_STREAM);
-		FX_TYPE_STDIO_STREAM);
 
 	enum fx_status status = FX_SUCCESS;
 
 	size_t count = fread(out, 1, max, p->s_fp);
 
-	if (ferror(p->s_fp)) {
+	if (count == 0 && ferror(p->s_fp)) {
 		status = FX_ERR_IO_FAILURE;
 	}
 
@@ -1152,14 +1148,13 @@ static enum fx_status stdio_write(
 	size_t count,
 	size_t *nr_written)
 {
-	struct fx_stdio_stream_p *p = fx_object_get_private(
+	struct fx_stdio_stream_p *p
-		stream,
+		= fx_object_get_private(stream, FX_TYPE_STDIO_STREAM);
-		FX_TYPE_STDIO_STREAM);
 
 	enum fx_status status = FX_SUCCESS;
 	size_t w = fwrite(data, 1, count, p->s_fp);
 
-	if (ferror(p->s_fp)) {
+	if (count == 0 && ferror(p->s_fp)) {
 		status = FX_ERR_IO_FAILURE;
 	}
 
@@ -1172,9 +1167,8 @@ static enum fx_status stdio_seek(
 	long long offset,
 	fx_stream_seek_origin origin)
 {
-	struct fx_stdio_stream_p *p = fx_object_get_private(
+	struct fx_stdio_stream_p *p
-		stream,
+		= fx_object_get_private(stream, FX_TYPE_STDIO_STREAM);
-		FX_TYPE_STDIO_STREAM);
 
 	int whence = 0;
 	switch (origin) {
@@ -1201,9 +1195,8 @@ static enum fx_status stdio_seek(
 
 static enum fx_status stdio_tell(const fx_stream *stream, size_t *cursor)
 {
-	struct fx_stdio_stream_p *p = fx_object_get_private(
+	struct fx_stdio_stream_p *p
-		stream,
+		= fx_object_get_private(stream, FX_TYPE_STDIO_STREAM);
-		FX_TYPE_STDIO_STREAM);
 
 	long pos = ftell(p->s_fp);
 	if (pos == -1L) {