WIP: Make Writer always write to a ByteSink

4 files changed, 137 insertions, 123 deletions

diff --git a/src/byte_sink.c b/src/byte_sink.c
index a4eb8098..03b46f03 100644
--- a/src/byte_sink.c
+++ b/src/byte_sink.c
@@ -14,32 +14,48 @@
   OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
 */
 
-#include "int_math.h"
+#define _POSIX_C_SOURCE 200809L /* for posix_fadvise and fileno */
+
+#include "byte_sink.h"
+
+#include "serd_config.h"
 #include "system.h"
 
 #include "serd/serd.h"
 
-#include <assert.h>
 #include <stddef.h>
+#include <stdio.h>
 #include <stdlib.h>
-#include <string.h>
 
-struct SerdByteSinkImpl {
-	SerdWriteFunc sink;
-	void*         stream;
-	char*         buf;
-	size_t        size;
-	size_t        block_size;
-};
+#if defined(HAVE_POSIX_FADVISE) || defined(HAVE_FILENO)
+#   include <fcntl.h>
+#endif
 
 SerdByteSink*
-serd_byte_sink_new(SerdWriteFunc write_func, void* stream, size_t block_size)
+serd_byte_sink_new_buffer(SerdBuffer* const buffer)
 {
 	SerdByteSink* sink = (SerdByteSink*)calloc(1, sizeof(SerdByteSink));
 
-	sink->sink       = write_func;
+	sink->write_func = serd_buffer_sink;
+	sink->stream     = buffer;
+	sink->block_size = 1;
+	sink->type       = TO_BUFFER;
+
+	return sink;
+}
+
+static SerdByteSink*
+serd_byte_sink_new_internal(const SerdWriteFunc    write_func,
+                            void* const            stream,
+                            const size_t           block_size,
+                            const SerdByteSinkType type)
+{
+	SerdByteSink* sink = (SerdByteSink*)calloc(1, sizeof(SerdByteSink));
+
+	sink->write_func = write_func;
 	sink->stream     = stream;
 	sink->block_size = block_size;
+	sink->type       = type;
 
 	if (block_size > 1) {
 		sink->buf = (char*)serd_allocate_buffer(block_size);
@@ -48,56 +64,63 @@ serd_byte_sink_new(SerdWriteFunc write_func, void* stream, size_t block_size)
 	return sink;
 }
 
-size_t
-serd_byte_sink_write(const void*   buf,
-                     size_t        size,
-                     size_t        nmemb,
-                     SerdByteSink* sink)
+SerdByteSink*
+serd_byte_sink_new_filename(const char* const path, const size_t block_size)
 {
-	assert(size == 1);
-	(void)size;
-
-	if (nmemb == 0) {
-		return 0;
-	} else if (sink->block_size == 1) {
-		return sink->sink(buf, 1, nmemb, sink->stream);
+	FILE* const file = fopen(path, "wb");
+	if (!file) {
+		return NULL;
 	}
 
-	const size_t orig_len = nmemb;
-	while (nmemb) {
-		const size_t space = sink->block_size - sink->size;
-		const size_t n     = MIN(space, nmemb);
-
-		// Write as much as possible into the remaining buffer space
-		memcpy(sink->buf + sink->size, buf, n);
-		sink->size += n;
-		buf = (const char*)buf + n;
-		nmemb -= n;
-
-		// Flush page if buffer is full
-		if (sink->size == sink->block_size) {
-			sink->sink(sink->buf, 1, sink->block_size, sink->stream);
-			sink->size = 0;
-		}
-	}
+#if defined(HAVE_POSIX_FADVISE) && defined(HAVE_FILENO)
+	posix_fadvise(fileno(file), 0, 0, POSIX_FADV_SEQUENTIAL);
+#endif
+
+	return serd_byte_sink_new_internal((SerdWriteFunc)fwrite,
+	                                   file,
+	                                   block_size,
+	                                   TO_FILENAME);
+}
 
-	return orig_len;
+SerdByteSink*
+serd_byte_sink_new_function(const SerdWriteFunc write_func,
+                            void* const         stream,
+                            const size_t        block_size)
+{
+	return serd_byte_sink_new_internal(write_func,
+	                                   stream,
+	                                   block_size,
+	                                   TO_FUNCTION);
 }
 
 void
 serd_byte_sink_flush(SerdByteSink* sink)
 {
 	if (sink->block_size > 1 && sink->size > 0) {
-		sink->sink(sink->buf, 1, sink->size, sink->stream);
+		sink->write_func(sink->buf, 1, sink->size, sink->stream);
 		sink->size = 0;
 	}
 }
 
+SerdStatus
+serd_byte_sink_close(SerdByteSink* sink)
+{
+	serd_byte_sink_flush(sink);
+
+	if (sink->type == TO_FILENAME && sink->stream) {
+		const int st = fclose((FILE*)sink->stream);
+		sink->stream = NULL;
+		return st ? SERD_ERR_UNKNOWN : SERD_SUCCESS;
+	}
+
+	return SERD_SUCCESS;
+}
+
 void
 serd_byte_sink_free(SerdByteSink* sink)
 {
 	if (sink) {
-		serd_byte_sink_flush(sink);
+		serd_byte_sink_close(sink);
 		free(sink->buf);
 		free(sink);
 	}
diff --git a/src/byte_sink.h b/src/byte_sink.h
index 1be84b48..b90cf4a5 100644
--- a/src/byte_sink.h
+++ b/src/byte_sink.h
@@ -17,79 +17,54 @@
 #ifndef SERD_BYTE_SINK_H
 #define SERD_BYTE_SINK_H
 
-#include "serd_internal.h"
-#include "system.h"
-
 #include "serd/serd.h"
 
 #include <stddef.h>
 #include <string.h>
 
-typedef struct SerdByteSinkImpl {
-	SerdWriteFunc sink;
-	void*         stream;
-	char*         buf;
-	size_t        size;
-	size_t        block_size;
-} SerdByteSink;
+typedef enum {
+	TO_BUFFER,   ///< Writing to a user-provided buffer
+	TO_FILENAME, ///< Writing to a file we opened
+	TO_FILE,     ///< Writing to a user-provided file
+	TO_FUNCTION, ///< Writing to a user-provided function
+} SerdByteSinkType;
 
-static inline SerdByteSink
-serd_byte_sink_new(SerdWriteFunc sink, void* stream, size_t block_size)
-{
-	SerdByteSink bsink;
-	bsink.sink       = sink;
-	bsink.stream     = stream;
-	bsink.size       = 0;
-	bsink.block_size = block_size;
-	bsink.buf        = ((block_size > 1)
-	                    ? (char*)serd_bufalloc(block_size)
-	                    : NULL);
-	return bsink;
-}
-
-static inline void
-serd_byte_sink_flush(SerdByteSink* bsink)
-{
-	if (bsink->block_size > 1 && bsink->size > 0) {
-		bsink->sink(bsink->buf, 1, bsink->size, bsink->stream);
-		bsink->size = 0;
-	}
-}
-
-static inline void
-serd_byte_sink_free(SerdByteSink* bsink)
-{
-	serd_byte_sink_flush(bsink);
-	free(bsink->buf);
-	bsink->buf = NULL;
-}
+struct SerdByteSinkImpl {
+	SerdWriteFunc    write_func; ///< User sink for TO_FUNCTION
+	void*            stream;     ///< Handle for TO_FILE* and TO_FUNCTION
+	char*            buf;        ///< Local buffer iff block_size > 1
+	size_t           size;       ///< Bytes written so far in this chunk
+	size_t           block_size; ///< Size of chunks to write
+	SerdByteSinkType type;       ///< Type of output
+};
 
 static inline size_t
-serd_byte_sink_write(const void* buf, size_t len, SerdByteSink* bsink)
+serd_byte_sink_write(const void* buf, size_t len, SerdByteSink* const sink)
 {
 	if (len == 0) {
 		return 0;
-	} else if (bsink->block_size == 1) {
-		return bsink->sink(buf, 1, len, bsink->stream);
+	} else if (sink->block_size == 1) {
+		return sink->write_func(buf, 1, len, sink->stream);
 	}
 
 	const size_t orig_len = len;
 	while (len) {
-		const size_t space = bsink->block_size - bsink->size;
-		const size_t n     = MIN(space, len);
+		const size_t space = sink->block_size - sink->size;
+		const size_t n     = space < len ? space : len;
 
 		// Write as much as possible into the remaining buffer space
-		memcpy(bsink->buf + bsink->size, buf, n);
-		bsink->size += n;
-		buf          = (const char*)buf + n;
-		len         -= n;
+		memcpy(sink->buf + sink->size, buf, n);
+		sink->size += n;
+		buf = (const char*)buf + n;
+		len -= n;
 
 		// Flush page if buffer is full
-		if (bsink->size == bsink->block_size) {
-			bsink->sink(bsink->buf, 1, bsink->block_size, bsink->stream);
-			bsink->size = 0;
+		if (sink->size == sink->block_size) {
+			sink->write_func(sink->buf, 1, sink->block_size, sink->stream);
+			sink->size = 0;
 		}
 	}
+
 	return orig_len;
 }
 
diff --git a/src/serdi.c b/src/serdi.c
index 1e9ceb88..0fc30239 100644
--- a/src/serdi.c
+++ b/src/serdi.c
@@ -236,6 +236,7 @@ main(int argc, char** argv)
 	const char*     add_prefix    = "";
 	const char*     chop_prefix   = NULL;
 	const char*     root_uri      = NULL;
+	const char*     out_filename  = NULL;
 	int             a             = 1;
 	for (; a < argc && argv[a][0] == '-'; ++a) {
 		if (argv[a][1] == '\0') {
@@ -323,6 +324,11 @@ main(int argc, char** argv)
 				return missing_arg(argv[0], 'r');
 			}
 			root_uri = argv[a];
+		} else if (argv[a][1] == 'w') {
+			if (++a == argc) {
+				return missing_arg(argv[0], 'w');
+			}
+			out_filename = argv[a];
 		} else if (argv[a][1] == 'x') {
 			reader_flags |= SERD_READ_VARIABLES;
 		} else {
@@ -336,11 +342,6 @@ main(int argc, char** argv)
 		return 1;
 	}
 
-#ifdef _WIN32
-	_setmode(_fileno(stdin), _O_BINARY);
-	_setmode(_fileno(stdout), _O_BINARY);
-#endif
-
 	char** inputs   = argv + a;
 	int    n_inputs = argc - a;
 
@@ -366,21 +367,35 @@ main(int argc, char** argv)
 		}
 	}
 
-	FILE*      out_fd = stdout;
 	SerdWorld* world  = serd_world_new();
 	SerdEnv*   env    = serd_env_new(base);
 
+#ifdef _WIN32
+	_setmode(_fileno(stdin), _O_BINARY);
+	if (!out_filename) {
+		_setmode(_fileno(stdout), _O_BINARY);
+	}
+#endif
+
 	const SerdSerialisationFlags serialisation_flags =
 		no_inline ? SERD_NO_INLINE_OBJECTS : 0u;
 
-	SerdByteSink* byte_sink = serd_byte_sink_new(
-		(SerdWriteFunc)fwrite, out_fd, bulk_write ? 4096u : 1u);
+	const size_t  block_size = bulk_write ? 4096u : 1u;
+	SerdByteSink* byte_sink =
+	    out_filename ? serd_byte_sink_new_filename(out_filename, block_size)
+	                 : serd_byte_sink_new_function((SerdWriteFunc)fwrite,
+	                                               stdout,
+	                                               block_size);
+
+	if (!byte_sink) {
+		perror("serdi: error opening output file");
+		return 1;
+	}
 
 	SerdWriter* writer = serd_writer_new(world,
 	                                     output_syntax,
 	                                     writer_flags,
 	                                     env,
-	                                     (SerdWriteFunc)serd_byte_sink_write,
 	                                     byte_sink);
 
 	SerdModel*      model    = NULL;
@@ -490,15 +505,16 @@ main(int argc, char** argv)
 	serd_sink_free(inserter);
 	serd_model_free(model);
 	serd_writer_free(writer);
-	serd_byte_sink_free(byte_sink);
 	serd_env_free(env);
 	serd_node_free(base);
 	serd_world_free(world);
 
-	if (fclose(stdout)) {
+	if (serd_byte_sink_close(byte_sink) || (!out_filename && fclose(stdout))) {
 		perror("serdi: write error");
 		st = SERD_ERR_UNKNOWN;
 	}
 
+	serd_byte_sink_free(byte_sink);
+
 	return (st > SERD_FAILURE) ? 1 : 0;
 }
diff --git a/src/writer.c b/src/writer.c
index 697d0e7c..ed3be4e7 100644
--- a/src/writer.c
+++ b/src/writer.c
@@ -14,6 +14,7 @@
   OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
 */
 
+#include "byte_sink.h"
 #include "env.h"
 #include "node.h"
 #include "sink.h"
@@ -131,8 +132,7 @@ struct SerdWriterImpl {
 	SerdURI         root_uri;
 	WriteContext*   anon_stack;
 	size_t          anon_stack_size;
-	SerdWriteFunc   write_func;
-	void*           stream;
+	SerdByteSink*   byte_sink;
 	SerdLogFunc     log_func;
 	void*           log_handle;
 	WriteContext    context;
@@ -236,7 +236,7 @@ ctx(SerdWriter* writer, const SerdField field)
 SERD_WARN_UNUSED_RESULT static inline size_t
 sink(const void* buf, size_t len, SerdWriter* writer)
 {
-	const size_t written = writer->write_func(buf, 1, len, writer->stream);
+	const size_t written = serd_byte_sink_write(buf, len, writer->byte_sink);
 	if (written != len) {
 		if (errno) {
 			SERD_LOG_ERRORF(writer->world, SERD_ERR_BAD_WRITE,
@@ -1077,22 +1077,22 @@ serd_writer_new(SerdWorld*      world,
                 SerdSyntax      syntax,
                 SerdWriterFlags flags,
                 SerdEnv*        env,
-                SerdWriteFunc   write_func,
-                void*           stream)
+                SerdByteSink*   byte_sink)
 {
 	const WriteContext context = WRITE_CONTEXT_NULL;
 	SerdWriter*        writer  = (SerdWriter*)calloc(1, sizeof(SerdWriter));
-	writer->world        = world;
-	writer->syntax       = syntax;
-	writer->flags        = flags;
-	writer->env          = env;
-	writer->root_node    = NULL;
-	writer->root_uri     = SERD_URI_NULL;
-	writer->anon_stack   = (WriteContext*)calloc(anon_stack_capacity, sizeof(WriteContext));
-	writer->write_func   = write_func;
-	writer->stream       = stream;
-	writer->context      = context;
-	writer->empty        = true;
+
+	writer->world     = world;
+	writer->syntax    = syntax;
+	writer->flags     = flags;
+	writer->env       = env;
+	writer->root_node = NULL;
+	writer->root_uri  = SERD_URI_NULL;
+	writer->anon_stack =
+	    (WriteContext*)calloc(anon_stack_capacity, sizeof(WriteContext));
+	writer->byte_sink = byte_sink;
+	writer->context   = context;
+	writer->empty     = true;
 
 	writer->iface.handle   = writer;
 	writer->iface.on_event = (SerdEventFunc)serd_writer_on_event;