Simplify output stream API

16 files changed, 449 insertions, 194 deletions

diff --git a/include/serd/buffer.h b/include/serd/buffer.h
index 556e7e1f..20abfb29 100644
--- a/include/serd/buffer.h
+++ b/include/serd/buffer.h
@@ -17,8 +17,8 @@ SERD_BEGIN_DECLS
 
    The #SerdBuffer type represents a writable area of memory with a known size.
 
-   A #SerdWriteFunc function is provided which enable writing output to a
-   memory buffer (as `fwrite` does for files).
+   #SerdWriteFunc and #SerdCloseFunc functions are provided which enable
+   writing output to a memory buffer (as `fwrite` and `fclose` do for files).
 
    @{
 */
diff --git a/include/serd/output_stream.h b/include/serd/output_stream.h
new file mode 100644
index 00000000..91c04e6b
--- /dev/null
+++ b/include/serd/output_stream.h
@@ -0,0 +1,94 @@
+// Copyright 2011-2022 David Robillard <d@drobilla.net>
+// SPDX-License-Identifier: ISC
+
+#ifndef SERD_OUTPUT_STREAM_H
+#define SERD_OUTPUT_STREAM_H
+
+#include "serd/attributes.h"
+#include "serd/buffer.h"
+#include "serd/status.h"
+#include "serd/stream.h"
+#include "zix/attributes.h"
+
+SERD_BEGIN_DECLS
+
+/**
+   @defgroup serd_output_stream Output Streams
+   @ingroup serd_reading_writing
+   @{
+*/
+
+/**
+   An output stream that receives bytes.
+
+   An output stream is used for writing output as a raw stream of bytes.  It is
+   compatible with standard C `FILE` streams, but allows different functions to
+   be provided for things like writing to a buffer or a socket.
+
+   Output from serd is UTF-8 encoded text.
+*/
+typedef struct {
+  void* ZIX_NULLABLE         stream; ///< Opaque parameter for functions
+  SerdWriteFunc ZIX_NONNULL  write;  ///< Write bytes to output
+  SerdErrorFunc ZIX_NULLABLE error;  ///< Stream error accessor
+  SerdCloseFunc ZIX_NULLABLE close;  ///< Close output
+} SerdOutputStream;
+
+/**
+   Open a stream that writes to a provided function.
+
+   @param write_func Function to write bytes to the stream.
+   @param error_func Function to detect errors in the stream.
+   @param close_func Function to close the stream.
+   @param stream Stream parameter passed to `write_func` and `close_func`.
+   @return An opened output stream, or all zeros on error.
+*/
+SERD_CONST_API SerdOutputStream
+serd_open_output_stream(SerdWriteFunc ZIX_NONNULL  write_func,
+                        SerdErrorFunc ZIX_NULLABLE error_func,
+                        SerdCloseFunc ZIX_NULLABLE close_func,
+                        void* ZIX_NULLABLE         stream);
+
+/**
+   Open a stream that writes to a buffer.
+
+   The `buffer` is owned by the caller, but will be expanded using `realloc` as
+   necessary.  Note that the string in the buffer will not be null terminated
+   until the stream is closed.
+
+   @param buffer Buffer to write output to.
+   @return An opened output stream, or all zeros on error.
+*/
+SERD_CONST_API SerdOutputStream
+serd_open_output_buffer(SerdBuffer* ZIX_NONNULL buffer);
+
+/**
+   Open a stream that writes to a file.
+
+   An arbitrary `FILE*` can be used with serd_open_output_stream() as well,
+   this convenience function opens the file properly for writing with serd, and
+   sets flags for optimized I/O if possible.
+
+   @param path Path of file to open and write to.
+*/
+SERD_API SerdOutputStream
+serd_open_output_file(const char* ZIX_NONNULL path);
+
+/**
+   Close an output stream.
+
+   This will call the close function, and reset the stream internally so that
+   no further writes can be made.  For convenience, this is safe to call on
+   NULL, and safe to call several times on the same output.  Failure is
+   returned in both of those cases.
+*/
+SERD_API SerdStatus
+serd_close_output(SerdOutputStream* ZIX_NULLABLE output);
+
+/**
+   @}
+*/
+
+SERD_END_DECLS
+
+#endif // SERD_OUTPUT_STREAM_H
diff --git a/include/serd/serd.h b/include/serd/serd.h
index 58c8e7ec..2b09eff2 100644
--- a/include/serd/serd.h
+++ b/include/serd/serd.h
@@ -79,6 +79,7 @@
 */
 
 #include "serd/input_stream.h"
+#include "serd/output_stream.h"
 #include "serd/reader.h"
 #include "serd/stream.h"
 #include "serd/writer.h"
diff --git a/include/serd/stream.h b/include/serd/stream.h
index f59e2c9d..a6fabb69 100644
--- a/include/serd/stream.h
+++ b/include/serd/stream.h
@@ -24,24 +24,6 @@ SERD_BEGIN_DECLS
 */
 
 /**
-   Function to detect I/O stream errors.
-
-   Identical semantics to `ferror`.
-
-   @return Non-zero if `stream` has encountered an error.
-*/
-typedef int (*SerdStreamErrorFunc)(void* ZIX_NONNULL stream);
-
-/**
-   Function to close an I/O stream.
-
-   Identical semantics to `fclose`.
-
-   @return Non-zero if `stream` has encountered an error.
-*/
-typedef int (*SerdStreamCloseFunc)(void* ZIX_NONNULL stream);
-
-/**
    Function for reading input bytes from a stream.
 
    This has identical semantics to `fread`, but may set `errno` for more
@@ -62,7 +44,7 @@ typedef size_t (*SerdReadFunc)(void* ZIX_NONNULL buf,
    Function for writing output bytes to a stream.
 
    This has identical semantics to `fwrite`, but may set `errno` for more
-   informative error reporting than supported by #SerdStreamErrorFunc.
+   informative error reporting than supported by #SerdErrorFunc.
 
    @param buf Input buffer.
    @param size Size of a single element of data in bytes (always 1).
diff --git a/include/serd/writer.h b/include/serd/writer.h
index 3cd7e2f4..41ffb8cd 100644
--- a/include/serd/writer.h
+++ b/include/serd/writer.h
@@ -6,14 +6,15 @@
 
 #include "serd/attributes.h"
 #include "serd/env.h"
+#include "serd/output_stream.h"
 #include "serd/sink.h"
 #include "serd/status.h"
-#include "serd/stream.h"
 #include "serd/string_view.h"
 #include "serd/syntax.h"
 #include "serd/world.h"
 #include "zix/attributes.h"
 
+#include <stddef.h>
 #include <stdint.h>
 
 SERD_BEGIN_DECLS
@@ -38,9 +39,8 @@ typedef enum {
   SERD_WRITE_ASCII       = 1U << 0U, ///< Escape all non-ASCII characters
   SERD_WRITE_UNQUALIFIED = 1U << 1U, ///< Do not shorten URIs into CURIEs
   SERD_WRITE_UNRESOLVED  = 1U << 2U, ///< Do not make URIs relative
-  SERD_WRITE_BULK        = 1U << 3U, ///< Write output in pages
-  SERD_WRITE_LAX         = 1U << 4U, ///< Tolerate lossy output
-  SERD_WRITE_TERSE       = 1U << 5U, ///< Write terser output without newlines
+  SERD_WRITE_LAX         = 1U << 3U, ///< Tolerate lossy output
+  SERD_WRITE_TERSE       = 1U << 4U, ///< Write terser output without newlines
 } SerdWriterFlag;
 
 /// Bitwise OR of #SerdWriterFlag values
@@ -48,12 +48,12 @@ typedef uint32_t SerdWriterFlags;
 
 /// Create a new RDF writer
 SERD_API SerdWriter* ZIX_ALLOCATED
-serd_writer_new(SerdWorld* ZIX_NONNULL    world,
-                SerdSyntax                syntax,
-                SerdWriterFlags           flags,
-                SerdEnv* ZIX_NONNULL      env,
-                SerdWriteFunc ZIX_NONNULL ssink,
-                void* ZIX_NULLABLE        stream);
+serd_writer_new(SerdWorld* ZIX_NONNULL        world,
+                SerdSyntax                    syntax,
+                SerdWriterFlags               flags,
+                SerdEnv* ZIX_NONNULL          env,
+                SerdOutputStream* ZIX_NONNULL output,
+                size_t                        block_size);
 
 /// Free `writer`
 SERD_API void
diff --git a/meson.build b/meson.build
index cc173f44..71f2dd3a 100644
--- a/meson.build
+++ b/meson.build
@@ -136,6 +136,7 @@ c_headers = files(
   'include/serd/input_stream.h',
   'include/serd/memory.h',
   'include/serd/node.h',
+  'include/serd/output_stream.h',
   'include/serd/reader.h',
   'include/serd/serd.h',
   'include/serd/sink.h',
@@ -153,12 +154,14 @@ c_headers = files(
 )
 
 sources = files(
+  'src/block_dumper.c',
   'src/buffer.c',
   'src/byte_source.c',
   'src/caret.c',
   'src/env.c',
   'src/input_stream.c',
   'src/node.c',
+  'src/output_stream.c',
   'src/read_nquads.c',
   'src/read_ntriples.c',
   'src/read_trig.c',
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);
 }
diff --git a/test/test_reader_writer.c b/test/test_reader_writer.c
index 1c4a415b..aa820fa6 100644
--- a/test/test_reader_writer.c
+++ b/test/test_reader_writer.c
@@ -9,10 +9,10 @@
 #include "serd/input_stream.h"
 #include "serd/memory.h"
 #include "serd/node.h"
+#include "serd/output_stream.h"
 #include "serd/reader.h"
 #include "serd/sink.h"
 #include "serd/status.h"
-#include "serd/stream.h"
 #include "serd/string_view.h"
 #include "serd/syntax.h"
 #include "serd/world.h"
@@ -109,9 +109,12 @@ test_write_errors(void)
       ctx.n_written    = 0;
       ctx.error_offset = o;
 
-      SerdEnv* const    env = serd_env_new(serd_empty_string());
+      SerdEnv* const   env = serd_env_new(serd_empty_string());
+      SerdOutputStream out =
+        serd_open_output_stream(faulty_sink, NULL, NULL, &ctx);
+
       SerdWriter* const writer =
-        serd_writer_new(world, syntax, 0U, env, faulty_sink, &ctx);
+        serd_writer_new(world, syntax, 0U, env, &out, 1U);
 
       const SerdSink* const sink = serd_writer_sink(writer);
       SerdReader* const reader   = serd_reader_new(world, SERD_TRIG, 0U, sink);
@@ -137,14 +140,14 @@ test_write_errors(void)
 static void
 test_writer(const char* const path)
 {
-  FILE*    fd  = fopen(path, "wb");
-  SerdEnv* env = serd_env_new(serd_empty_string());
-  assert(fd);
-
   SerdWorld* world = serd_world_new();
+  SerdEnv*   env   = serd_env_new(serd_empty_string());
+
+  SerdOutputStream output = serd_open_output_file(path);
+
+  SerdWriter* writer =
+    serd_writer_new(world, SERD_TURTLE, SERD_WRITE_LAX, env, &output, 1U);
 
-  SerdWriter* writer = serd_writer_new(
-    world, SERD_TURTLE, SERD_WRITE_LAX, env, (SerdWriteFunc)fwrite, fd);
   assert(writer);
 
   serd_writer_chop_blank_prefix(writer, "tmp");
@@ -205,6 +208,7 @@ test_writer(const char* const path)
   serd_node_free(hello);
 
   serd_writer_free(writer);
+  serd_close_output(&output);
 
   serd_node_free(lit);
   serd_node_free(o);
@@ -212,19 +216,20 @@ test_writer(const char* const path)
   serd_node_free(l);
 
   // Test buffer sink
-  SerdBuffer buffer = {NULL, 0};
-  writer =
-    serd_writer_new(world, SERD_TURTLE, 0, env, serd_buffer_write, &buffer);
+  SerdBuffer      buffer = {NULL, 0};
+  SerdNode* const base   = serd_new_uri(serd_string("http://example.org/base"));
 
-  SerdNode* const base = serd_new_uri(serd_string("http://example.org/base"));
+  output = serd_open_output_buffer(&buffer);
+  writer = serd_writer_new(world, SERD_TURTLE, 0, env, &output, 1U);
 
   serd_sink_write_base(serd_writer_sink(writer), base);
 
   serd_node_free(base);
   serd_writer_free(writer);
-  serd_buffer_close(&buffer);
+  serd_close_output(&output);
 
   char* const out = (char*)buffer.buf;
+  assert(out);
   assert(!strcmp(out, "@base <http://example.org/base> .\n"));
   serd_free(out);
 
@@ -233,7 +238,6 @@ test_writer(const char* const path)
 
   serd_env_free(env);
   serd_world_free(world);
-  fclose(fd);
 }
 
 static void
diff --git a/test/test_terse_write.c b/test/test_terse_write.c
index 17bacc13..39e3767a 100644
--- a/test/test_terse_write.c
+++ b/test/test_terse_write.c
@@ -6,6 +6,7 @@
 #include "serd/buffer.h"
 #include "serd/env.h"
 #include "serd/node.h"
+#include "serd/output_stream.h"
 #include "serd/sink.h"
 #include "serd/statement.h"
 #include "serd/string_view.h"
@@ -28,6 +29,8 @@ check_output(SerdWriter* writer, SerdBuffer* buffer, const char* expected)
 
   const char* output = (const char*)buffer->buf;
 
+  fprintf(stderr, "output: %s\n", output);
+  fprintf(stderr, "expected: %s\n", expected);
   assert(!strcmp(output, expected));
 
   buffer->len = 0;
@@ -53,8 +56,9 @@ test(void)
 
   serd_env_set_prefix(env, serd_string("rdf"), serd_string(NS_RDF));
 
-  SerdWriter* writer =
-    serd_writer_new(world, SERD_TURTLE, 0, env, serd_buffer_write, &buffer);
+  SerdOutputStream  output = serd_open_output_buffer(&buffer);
+  SerdWriter* const writer =
+    serd_writer_new(world, SERD_TURTLE, 0, env, &output, 1U);
 
   const SerdSink* sink = serd_writer_sink(writer);
 
@@ -86,7 +90,6 @@ test(void)
   serd_sink_write(sink, 0, l2, rdf_rest, rdf_nil, NULL);
   check_output(writer, &buffer, "[] rdf:value ( \"s1\" \"s2\" ) .\n");
 
-  serd_buffer_close(&buffer);
   serd_writer_free(writer);
   serd_node_free(rdf_nil);
   serd_node_free(rdf_rest);
@@ -97,6 +100,7 @@ test(void)
   serd_node_free(l2);
   serd_node_free(l1);
   serd_node_free(b1);
+  serd_close_output(&output);
   serd_env_free(env);
   serd_world_free(world);
   free(buffer.buf);
diff --git a/test/test_writer.c b/test/test_writer.c
index fe685437..2a852562 100644
--- a/test/test_writer.c
+++ b/test/test_writer.c
@@ -8,6 +8,7 @@
 #include "serd/event.h"
 #include "serd/memory.h"
 #include "serd/node.h"
+#include "serd/output_stream.h"
 #include "serd/sink.h"
 #include "serd/statement.h"
 #include "serd/status.h"
@@ -23,23 +24,39 @@
 #include <string.h>
 
 static void
+test_writer_new(void)
+{
+  SerdWorld*       world  = serd_world_new();
+  SerdEnv*         env    = serd_env_new(serd_empty_string());
+  SerdBuffer       buffer = {NULL, 0};
+  SerdOutputStream output = serd_open_output_buffer(&buffer);
+
+  assert(!serd_writer_new(world, SERD_TURTLE, 0U, env, &output, 0U));
+
+  serd_world_free(world);
+  serd_env_free(env);
+}
+
+static void
 test_write_bad_event(void)
 {
-  SerdWorld*  world  = serd_world_new();
-  SerdEnv*    env    = serd_env_new(serd_empty_string());
-  SerdBuffer  buffer = {NULL, 0};
-  SerdWriter* writer =
-    serd_writer_new(world, SERD_TURTLE, 0U, env, serd_buffer_write, &buffer);
+  SerdWorld*       world  = serd_world_new();
+  SerdEnv*         env    = serd_env_new(serd_empty_string());
+  SerdBuffer       buffer = {NULL, 0};
+  SerdOutputStream output = serd_open_output_buffer(&buffer);
 
+  SerdWriter* writer =
+    serd_writer_new(world, SERD_TURTLE, 0U, env, &output, 1U);
   assert(writer);
 
   const SerdEvent event = {(SerdEventType)42};
   assert(serd_sink_write_event(serd_writer_sink(writer), &event) ==
          SERD_BAD_ARG);
 
-  serd_buffer_close(&buffer);
+  assert(!serd_close_output(&output));
 
   char* const out = (char*)buffer.buf;
+  assert(out);
   assert(!strcmp(out, ""));
   serd_free(out);
 
@@ -51,11 +68,12 @@ test_write_bad_event(void)
 static void
 test_write_bad_prefix(void)
 {
-  SerdWorld*  world  = serd_world_new();
-  SerdEnv*    env    = serd_env_new(serd_empty_string());
-  SerdBuffer  buffer = {NULL, 0};
-  SerdWriter* writer =
-    serd_writer_new(world, SERD_TURTLE, 0U, env, serd_buffer_write, &buffer);
+  SerdWorld*       world  = serd_world_new();
+  SerdEnv*         env    = serd_env_new(serd_empty_string());
+  SerdBuffer       buffer = {NULL, 0};
+  SerdOutputStream output = serd_open_output_buffer(&buffer);
+  SerdWriter*      writer =
+    serd_writer_new(world, SERD_TURTLE, 0U, env, &output, 1U);
 
   assert(writer);
 
@@ -81,12 +99,13 @@ test_write_bad_prefix(void)
 static void
 test_write_long_literal(void)
 {
-  SerdWorld*  world  = serd_world_new();
-  SerdEnv*    env    = serd_env_new(serd_empty_string());
-  SerdBuffer  buffer = {NULL, 0};
-  SerdWriter* writer =
-    serd_writer_new(world, SERD_TURTLE, 0U, env, serd_buffer_write, &buffer);
+  SerdWorld*       world  = serd_world_new();
+  SerdEnv*         env    = serd_env_new(serd_empty_string());
+  SerdBuffer       buffer = {NULL, 0};
+  SerdOutputStream output = serd_open_output_buffer(&buffer);
 
+  SerdWriter* writer =
+    serd_writer_new(world, SERD_TURTLE, 0U, env, &output, 1U);
   assert(writer);
 
   SerdNode* s = serd_new_uri(serd_string("http://example.org/s"));
@@ -99,6 +118,7 @@ test_write_long_literal(void)
   serd_node_free(p);
   serd_node_free(s);
   serd_writer_free(writer);
+  serd_close_output(&output);
   serd_env_free(env);
   serd_buffer_close(&buffer);
 
@@ -108,7 +128,7 @@ test_write_long_literal(void)
     "<http://example.org/s>\n"
     "\t<http://example.org/p> \"\"\"hello \"\"\\\"world\"\"\\\"!\"\"\" .\n";
 
-  assert(!strcmp((char*)out, expected));
+  assert(!strcmp(out, expected));
   serd_free(out);
 
   serd_world_free(world);
@@ -129,11 +149,14 @@ null_sink(const void* const buf,
 static void
 test_writer_cleanup(void)
 {
-  SerdStatus  st    = SERD_SUCCESS;
-  SerdWorld*  world = serd_world_new();
-  SerdEnv*    env   = serd_env_new(serd_empty_string());
+  SerdStatus       st    = SERD_SUCCESS;
+  SerdWorld*       world = serd_world_new();
+  SerdEnv*         env   = serd_env_new(serd_empty_string());
+  SerdOutputStream output =
+    serd_open_output_stream(null_sink, NULL, NULL, NULL);
+
   SerdWriter* writer =
-    serd_writer_new(world, SERD_TURTLE, 0U, env, null_sink, NULL);
+    serd_writer_new(world, SERD_TURTLE, 0U, env, &output, 1U);
 
   const SerdSink* sink = serd_writer_sink(writer);
 
@@ -183,8 +206,9 @@ test_strict_write(void)
 
   SerdWorld*        world = serd_world_new();
   SerdEnv* const    env   = serd_env_new(serd_empty_string());
+  SerdOutputStream  out   = serd_open_output_stream(null_sink, NULL, NULL, fd);
   SerdWriter* const writer =
-    serd_writer_new(world, SERD_TURTLE, 0U, env, null_sink, fd);
+    serd_writer_new(world, SERD_TURTLE, 0U, env, &out, 1U);
 
   assert(writer);
 
@@ -229,15 +253,16 @@ error_sink(const void* const buf,
 static void
 test_write_error(void)
 {
-  SerdWorld* const world  = serd_world_new();
-  SerdEnv* const   env    = serd_env_new(serd_empty_string());
+  SerdWorld* const world = serd_world_new();
+  SerdEnv* const   env   = serd_env_new(serd_empty_string());
+  SerdOutputStream out = serd_open_output_stream(error_sink, NULL, NULL, NULL);
   SerdWriter*      writer = NULL;
   SerdStatus       st     = SERD_SUCCESS;
 
   SerdNode* u = serd_new_uri(serd_string("http://example.com/u"));
 
-  writer = serd_writer_new(
-    world, SERD_TURTLE, (SerdWriterFlags)0, env, error_sink, NULL);
+  writer =
+    serd_writer_new(world, SERD_TURTLE, (SerdWriterFlags)0, env, &out, 1U);
   assert(writer);
 
   const SerdSink* const sink = serd_writer_sink(writer);
@@ -257,8 +282,11 @@ test_writer_stack_overflow(void)
   SerdWorld* world = serd_world_new();
   SerdEnv*   env   = serd_env_new(serd_empty_string());
 
+  SerdOutputStream output =
+    serd_open_output_stream(null_sink, NULL, NULL, NULL);
+
   SerdWriter* writer =
-    serd_writer_new(world, SERD_TURTLE, 0U, env, null_sink, NULL);
+    serd_writer_new(world, SERD_TURTLE, 0U, env, &output, 1U);
 
   const SerdSink* sink = serd_writer_sink(writer);
 
@@ -293,6 +321,7 @@ test_writer_stack_overflow(void)
   serd_node_free(p);
   serd_node_free(s);
   serd_writer_free(writer);
+  serd_close_output(&output);
   serd_env_free(env);
   serd_world_free(world);
 }
@@ -307,18 +336,19 @@ test_write_empty_syntax(void)
   SerdNode* p = serd_new_uri(serd_string("http://example.org/p"));
   SerdNode* o = serd_new_curie(serd_string("eg:o"));
 
-  SerdBuffer buffer = {NULL, 0};
+  SerdBuffer       buffer = {NULL, 0};
+  SerdOutputStream output = serd_open_output_buffer(&buffer);
 
-  SerdWriter* writer = serd_writer_new(
-    world, SERD_SYNTAX_EMPTY, 0U, env, serd_buffer_write, &buffer);
+  SerdWriter* writer =
+    serd_writer_new(world, SERD_SYNTAX_EMPTY, 0U, env, &output, 1U);
 
   assert(writer);
 
   assert(!serd_sink_write(serd_writer_sink(writer), 0U, s, p, o, NULL));
-
-  serd_buffer_close(&buffer);
+  assert(!serd_close_output(&output));
 
   char* const out = (char*)buffer.buf;
+  assert(out);
   assert(strlen(out) == 0);
   serd_free(out);
 
@@ -326,6 +356,7 @@ test_write_empty_syntax(void)
   serd_node_free(o);
   serd_node_free(p);
   serd_node_free(s);
+  serd_close_output(&output);
   serd_env_free(env);
   serd_world_free(world);
 }
@@ -333,13 +364,13 @@ test_write_empty_syntax(void)
 static void
 check_pname_escape(const char* const lname, const char* const expected)
 {
-  SerdWorld* world  = serd_world_new();
-  SerdEnv*   env    = serd_env_new(serd_empty_string());
-  SerdBuffer buffer = {NULL, 0};
+  SerdWorld*       world  = serd_world_new();
+  SerdEnv*         env    = serd_env_new(serd_empty_string());
+  SerdBuffer       buffer = {NULL, 0};
+  SerdOutputStream output = serd_open_output_buffer(&buffer);
 
   SerdWriter* writer =
-    serd_writer_new(world, SERD_TURTLE, 0U, env, serd_buffer_write, &buffer);
-
+    serd_writer_new(world, SERD_TURTLE, 0U, env, &output, 1U);
   assert(writer);
 
   static const char* const prefix     = "http://example.org/";
@@ -362,11 +393,12 @@ check_pname_escape(const char* const lname, const char* const expected)
   serd_node_free(p);
   serd_node_free(s);
   serd_writer_free(writer);
+  serd_close_output(&output);
   serd_env_free(env);
   serd_buffer_close(&buffer);
 
   char* const out = (char*)buffer.buf;
-  assert(!strcmp((char*)out, expected));
+  assert(!strcmp(out, expected));
   serd_free(out);
 
   serd_world_free(world);
@@ -406,6 +438,7 @@ test_write_pname_escapes(void)
 int
 main(void)
 {
+  test_writer_new();
   test_write_bad_event();
   test_write_bad_prefix();
   test_write_long_literal();