Simplify output stream API

7 files changed, 241 insertions, 107 deletions

diff --git a/src/block_dumper.c b/src/block_dumper.c
new file mode 100644
index 00000000..174f0215
--- /dev/null
+++ b/src/block_dumper.c
@@ -0,0 +1,54 @@
+// Copyright 2011-2021 David Robillard <d@drobilla.net>
+// SPDX-License-Identifier: ISC
+
+#include "block_dumper.h"
+#include "system.h"
+
+#include <stddef.h>
+
+SerdStatus
+serd_block_dumper_open(SerdBlockDumper* const  dumper,
+                       SerdOutputStream* const output,
+                       const size_t            block_size)
+{
+  if (!block_size) {
+    return SERD_BAD_ARG;
+  }
+
+  dumper->out        = output;
+  dumper->buf        = NULL;
+  dumper->size       = 0U;
+  dumper->block_size = block_size;
+
+  if (block_size == 1) {
+    return SERD_SUCCESS;
+  }
+
+  dumper->buf = (char*)serd_allocate_buffer(block_size);
+  return dumper->buf ? SERD_SUCCESS : SERD_BAD_ALLOC;
+}
+
+SerdStatus
+serd_block_dumper_flush(SerdBlockDumper* const dumper)
+{
+  SerdStatus st = SERD_SUCCESS;
+
+  if (dumper->out->stream && dumper->block_size > 1 && dumper->size > 0) {
+    const size_t written =
+      dumper->out->write(dumper->buf, 1, dumper->size, dumper->out->stream);
+
+    if (written != dumper->size) {
+      st = SERD_BAD_WRITE;
+    }
+
+    dumper->size = 0;
+  }
+
+  return st;
+}
+
+void
+serd_block_dumper_close(SerdBlockDumper* const dumper)
+{
+  serd_free_aligned(dumper->buf);
+}
diff --git a/src/block_dumper.h b/src/block_dumper.h
new file mode 100644
index 00000000..24fb977c
--- /dev/null
+++ b/src/block_dumper.h
@@ -0,0 +1,77 @@
+// Copyright 2011-2021 David Robillard <d@drobilla.net>
+// SPDX-License-Identifier: ISC
+
+#ifndef SERD_SRC_BLOCK_DUMPER_H
+#define SERD_SRC_BLOCK_DUMPER_H
+
+#include "serd/output_stream.h"
+#include "serd/status.h"
+#include "zix/attributes.h"
+
+#include <stddef.h>
+#include <string.h>
+
+typedef struct {
+  SerdOutputStream* ZIX_ALLOCATED out;        ///< Output stream to write to
+  char* ZIX_ALLOCATED             buf;        ///< Local buffer if needed
+  size_t                          size;       ///< Bytes pending for this block
+  size_t                          block_size; ///< Block size to write in bytes
+} SerdBlockDumper;
+
+/**
+   Set up a new output stream wrapper that writes in blocks.
+
+   This allocates a buffer internally, which must be eventually freed by
+   calling serd_block_dumper_close().
+*/
+SerdStatus
+serd_block_dumper_open(SerdBlockDumper* ZIX_NONNULL  dumper,
+                       SerdOutputStream* ZIX_NONNULL output,
+                       size_t                        block_size);
+
+SerdStatus
+serd_block_dumper_flush(SerdBlockDumper* ZIX_NONNULL dumper);
+
+void
+serd_block_dumper_close(SerdBlockDumper* ZIX_NONNULL dumper);
+
+/**
+   Write some bytes to the page writer.
+
+   This works like any other SerdWriteFunc, but will append to an internal
+   buffer and only actually write to the output when a whole block is ready.
+*/
+static inline size_t
+serd_block_dumper_write(const void* ZIX_NONNULL            buf,
+                        const size_t                       size,
+                        const size_t                       nmemb,
+                        SerdBlockDumper* ZIX_NONNULL const dumper)
+{
+  if (dumper->block_size == 1) {
+    return dumper->out->write(buf, size, nmemb, dumper->out->stream);
+  }
+
+  size_t       len      = size * nmemb;
+  const size_t orig_len = len;
+  while (len) {
+    const size_t space = dumper->block_size - dumper->size;
+    const size_t n     = space < len ? space : len;
+
+    // Write as much as possible into the remaining buffer space
+    memcpy(dumper->buf + dumper->size, buf, n);
+    dumper->size += n;
+    buf = (const char*)buf + n;
+    len -= n;
+
+    // Flush page if buffer is full
+    if (dumper->size == dumper->block_size) {
+      dumper->out->write(
+        dumper->buf, 1, dumper->block_size, dumper->out->stream);
+      dumper->size = 0;
+    }
+  }
+
+  return orig_len;
+}
+
+#endif // SERD_SRC_DUMPER_H
diff --git a/src/buffer.c b/src/buffer.c
index 12f343c2..befa3943 100644
--- a/src/buffer.c
+++ b/src/buffer.c
@@ -1,7 +1,7 @@
 // Copyright 2011-2021 David Robillard <d@drobilla.net>
 // SPDX-License-Identifier: ISC
 
-#include "serd/serd.h"
+#include "serd/buffer.h"
 
 #include <assert.h>
 #include <stddef.h>
diff --git a/src/byte_sink.h b/src/byte_sink.h
deleted file mode 100644
index ada5e8a9..00000000
--- a/src/byte_sink.h
+++ /dev/null
@@ -1,89 +0,0 @@
-// Copyright 2011-2023 David Robillard <d@drobilla.net>
-// SPDX-License-Identifier: ISC
-
-#ifndef SERD_SRC_BYTE_SINK_H
-#define SERD_SRC_BYTE_SINK_H
-
-#include "macros.h"
-#include "system.h"
-
-#include "serd/sink.h"
-
-#include <stddef.h>
-#include <string.h>
-
-typedef struct SerdByteSinkImpl {
-  SerdWriteFunc sink;
-  void*         stream;
-  char*         buf;
-  size_t        size;
-  size_t        block_size;
-} SerdByteSink;
-
-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_allocate_buffer(block_size) : NULL);
-  return bsink;
-}
-
-static inline SerdStatus
-serd_byte_sink_flush(SerdByteSink* bsink)
-{
-  if (bsink->block_size > 1 && bsink->size > 0) {
-    const size_t size  = bsink->size;
-    const size_t n_out = bsink->sink(bsink->buf, 1, bsink->size, bsink->stream);
-    bsink->size        = 0;
-
-    return (n_out != size) ? SERD_BAD_WRITE : SERD_SUCCESS;
-  }
-
-  return SERD_SUCCESS;
-}
-
-static inline void
-serd_byte_sink_free(SerdByteSink* bsink)
-{
-  serd_byte_sink_flush(bsink);
-  serd_free_aligned(bsink->buf);
-  bsink->buf = NULL;
-}
-
-static inline size_t
-serd_byte_sink_write(const void* buf, size_t len, SerdByteSink* bsink)
-{
-  if (len == 0) {
-    return 0;
-  }
-
-  if (bsink->block_size == 1) {
-    return bsink->sink(buf, 1, len, bsink->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);
-
-    // 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;
-
-    // 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;
-    }
-  }
-  return orig_len;
-}
-
-#endif // SERD_SRC_BYTE_SINK_H
diff --git a/src/output_stream.c b/src/output_stream.c
new file mode 100644
index 00000000..9014a372
--- /dev/null
+++ b/src/output_stream.c
@@ -0,0 +1,80 @@
+// Copyright 2011-2021 David Robillard <d@drobilla.net>
+// SPDX-License-Identifier: ISC
+
+#include "serd_config.h"
+
+#include "serd/buffer.h"
+#include "serd/output_stream.h"
+#include "serd/status.h"
+#include "serd/stream.h"
+
+// IWYU pragma: no_include <features.h>
+
+#include <assert.h>
+#include <stdbool.h>
+#include <stddef.h>
+#include <stdio.h>
+
+#if USE_POSIX_FADVISE && USE_FILENO
+#  include <fcntl.h>
+#endif
+
+SerdOutputStream
+serd_open_output_stream(SerdWriteFunc const write_func,
+                        SerdErrorFunc const error_func,
+                        SerdCloseFunc const close_func,
+                        void* const         stream)
+{
+  assert(write_func);
+
+  SerdOutputStream output = {stream, write_func, error_func, close_func};
+  return output;
+}
+
+SerdOutputStream
+serd_open_output_buffer(SerdBuffer* const buffer)
+{
+  assert(buffer);
+
+  return serd_open_output_stream(
+    serd_buffer_write, NULL, serd_buffer_close, buffer);
+}
+
+SerdOutputStream
+serd_open_output_file(const char* const path)
+{
+  assert(path);
+
+#ifdef __GLIBC__
+  FILE* const file = fopen(path, "wbe");
+#else
+  FILE* const file = fopen(path, "wb");
+#endif
+
+  if (!file) {
+    const SerdOutputStream failure = {NULL, NULL, NULL, NULL};
+    return failure;
+  }
+
+#if USE_POSIX_FADVISE && USE_FILENO
+  (void)posix_fadvise(fileno(file), 0, 0, POSIX_FADV_SEQUENTIAL);
+#endif
+
+  return serd_open_output_stream(
+    (SerdWriteFunc)fwrite, (SerdErrorFunc)ferror, (SerdCloseFunc)fclose, file);
+}
+
+SerdStatus
+serd_close_output(SerdOutputStream* const output)
+{
+  if (!output || !output->stream) {
+    return SERD_FAILURE;
+  }
+
+  const bool had_error = output->error ? output->error(output->stream) : false;
+  int        close_st  = output->close ? output->close(output->stream) : 0;
+
+  output->stream = NULL;
+
+  return (had_error || close_st) ? SERD_BAD_STREAM : SERD_SUCCESS;
+}
diff --git a/src/serdi.c b/src/serdi.c
index 89d7ebd9..ac4d3178 100644
--- a/src/serdi.c
+++ b/src/serdi.c
@@ -5,6 +5,7 @@
 #include "serd/error.h"
 #include "serd/input_stream.h"
 #include "serd/node.h"
+#include "serd/output_stream.h"
 #include "serd/reader.h"
 #include "serd/status.h"
 #include "serd/stream.h"
@@ -106,6 +107,7 @@ main(int argc, char** argv)
   bool            from_string   = false;
   bool            from_stdin    = false;
   bool            bulk_read     = true;
+  bool            bulk_write    = false;
   bool            osyntax_set   = false;
   bool            quiet         = false;
   size_t          stack_size    = 1048576U;
@@ -133,7 +135,7 @@ main(int argc, char** argv)
       if (opt == 'a') {
         writer_flags |= SERD_WRITE_ASCII;
       } else if (opt == 'b') {
-        writer_flags |= SERD_WRITE_BULK;
+        bulk_write = true;
       } else if (opt == 'e') {
         bulk_read = false;
       } else if (opt == 'f') {
@@ -247,8 +249,13 @@ main(int argc, char** argv)
   SerdEnv* const   env =
     serd_env_new(base ? serd_node_string_view(base) : serd_empty_string());
 
+  SerdOutputStream out = serd_open_output_stream((SerdWriteFunc)fwrite,
+                                                 (SerdErrorFunc)ferror,
+                                                 (SerdCloseFunc)fclose,
+                                                 out_fd);
+
   SerdWriter* const writer = serd_writer_new(
-    world, output_syntax, writer_flags, env, (SerdWriteFunc)fwrite, out_fd);
+    world, output_syntax, writer_flags, env, &out, bulk_write ? 4096U : 1U);
 
   const SerdLimits limits = {stack_size, MAX_DEPTH};
   serd_world_set_limits(world, limits);
diff --git a/src/writer.c b/src/writer.c
index 58700f6f..60a387c0 100644
--- a/src/writer.c
+++ b/src/writer.c
@@ -1,7 +1,7 @@
 // Copyright 2011-2023 David Robillard <d@drobilla.net>
 // SPDX-License-Identifier: ISC
 
-#include "byte_sink.h"
+#include "block_dumper.h"
 #include "env.h"
 #include "namespaces.h"
 #include "node.h"
@@ -17,10 +17,10 @@
 #include "serd/env.h"
 #include "serd/event.h"
 #include "serd/node.h"
+#include "serd/output_stream.h"
 #include "serd/sink.h"
 #include "serd/statement.h"
 #include "serd/status.h"
-#include "serd/stream.h"
 #include "serd/string_view.h"
 #include "serd/syntax.h"
 #include "serd/uri.h"
@@ -136,7 +136,7 @@ struct SerdWriterImpl {
   WriteContext*   anon_stack;
   size_t          max_depth;
   size_t          anon_stack_size;
-  SerdByteSink    byte_sink;
+  SerdBlockDumper output;
   WriteContext    context;
   char*           bprefix;
   size_t          bprefix_len;
@@ -257,7 +257,8 @@ pop_context(SerdWriter* writer)
 SERD_NODISCARD static size_t
 sink(const void* buf, size_t len, SerdWriter* writer)
 {
-  const size_t written = serd_byte_sink_write(buf, len, &writer->byte_sink);
+  const size_t written = serd_block_dumper_write(buf, 1, len, &writer->output);
+
   if (written != len) {
     if (errno) {
       char message[1024] = {0};
@@ -1331,7 +1332,7 @@ serd_writer_finish(SerdWriter* writer)
   assert(writer);
 
   const SerdStatus st0 = terminate_context(writer);
-  const SerdStatus st1 = serd_byte_sink_flush(&writer->byte_sink);
+  const SerdStatus st1 = serd_block_dumper_flush(&writer->output);
 
   free_anon_stack(writer);
   reset_context(writer, RESET_GRAPH | RESET_INDENT);
@@ -1340,16 +1341,21 @@ serd_writer_finish(SerdWriter* writer)
 }
 
 SerdWriter*
-serd_writer_new(SerdWorld*      world,
-                SerdSyntax      syntax,
-                SerdWriterFlags flags,
-                SerdEnv*        env,
-                SerdWriteFunc   ssink,
-                void*           stream)
+serd_writer_new(SerdWorld*        world,
+                SerdSyntax        syntax,
+                SerdWriterFlags   flags,
+                SerdEnv*          env,
+                SerdOutputStream* output,
+                size_t            block_size)
 {
   assert(world);
   assert(env);
-  assert(ssink);
+  assert(output);
+
+  SerdBlockDumper dumper = {NULL, NULL, 0U, 0U};
+  if (serd_block_dumper_open(&dumper, output, block_size)) {
+    return NULL;
+  }
 
   const size_t       max_depth = world->limits.writer_max_depth;
   const WriteContext context   = WRITE_CONTEXT_NULL;
@@ -1361,9 +1367,8 @@ serd_writer_new(SerdWorld*      world,
   writer->env       = env;
   writer->root_node = NULL;
   writer->root_uri  = SERD_URI_NULL;
+  writer->output    = dumper;
   writer->context   = context;
-  writer->byte_sink = serd_byte_sink_new(
-    ssink, stream, (flags & SERD_WRITE_BULK) ? SERD_PAGE_SIZE : 1);
 
   if (max_depth) {
     writer->max_depth  = max_depth;
@@ -1476,9 +1481,9 @@ serd_writer_free(SerdWriter* writer)
   serd_writer_finish(writer);
   free_context(&writer->context);
   free_anon_stack(writer);
+  serd_block_dumper_close(&writer->output);
   free(writer->anon_stack);
   free(writer->bprefix);
-  serd_byte_sink_free(&writer->byte_sink);
   serd_node_free(writer->root_node);
   free(writer);
 }