fx.io: import system-level streams and uni-directional pipes

2026-06-10 19:08:36 +01:00
parent 15784c3404
commit 2512611f8f
5 changed files with 240 additions and 0 deletions
@@ -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
@@ -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
@@ -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;
+}
@@ -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)
@@ -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;
+}