Simplify input stream API

More or less the same rationale as the previous commit, but for reading.  This
makes for nice symmetry with writing, at the cost of a slightly more annoying
reader interface since the source doesn't know its block size or name.

20 files changed, 422 insertions, 382 deletions

diff --git a/doc/conf.py.in b/doc/conf.py.in
index 374b6596..39d9568f 100644
--- a/doc/conf.py.in
+++ b/doc/conf.py.in
@@ -22,7 +22,6 @@ except ModuleNotFoundError:
 # Ignore everything opaque or external for nitpicky mode
 _opaque = [
     "FILE",
-    "SerdByteSourceImpl",
     "SerdCaretImpl",
     "SerdCursorImpl",
     "SerdEnvImpl",
diff --git a/include/serd/serd.h b/include/serd/serd.h
index db7520d9..6324c06e 100644
--- a/include/serd/serd.h
+++ b/include/serd/serd.h
@@ -2183,67 +2183,77 @@ serd_node_to_syntax(const SerdNode* SERD_NONNULL node,
 
 /**
    @}
-   @defgroup serd_byte_source Byte Source
+   @defgroup serd_input_stream Input Streams
+
+   An input stream is used for reading input as a raw stream of bytes.  It is
+   compatible with standard C `FILE` streams, but allows different functions to
+   be provided for things like reading from a buffer or a socket.
+
    @{
 */
 
-/// A source for bytes that provides text input
-typedef struct SerdByteSourceImpl SerdByteSource;
+/// An input stream that produces bytes
+typedef struct {
+  void* SERD_NULLABLE               stream; ///< Opaque parameter for functions
+  SerdReadFunc SERD_NULLABLE        read;   ///< Read bytes to input
+  SerdStreamErrorFunc SERD_NONNULL  error;  ///< Stream error accessor
+  SerdStreamCloseFunc SERD_NULLABLE close;  ///< Close input
+} SerdInputStream;
 
 /**
-   Create a new byte source that reads from a string.
+   Open a stream that reads from a provided function.
+
+   @param read_func Function to read input.
+   @param error_func Function used to detect errors.
+   @param close_func Function to close the stream after reading is done.
+   @param stream Opaque stream parameter for functions.
 
-   @param string Null-terminated UTF-8 string to read from.
-   @param name Optional name of stream for error messages (string or URI).
+   @return An opened input stream, or all zeros on error.
 */
-SERD_API
-SerdByteSource* SERD_ALLOCATED
-serd_byte_source_new_string(const char* SERD_NONNULL      string,
-                            const SerdNode* SERD_NULLABLE name);
+SERD_CONST_API
+SerdInputStream
+serd_open_input_stream(SerdReadFunc SERD_NONNULL         read_func,
+                       SerdStreamErrorFunc SERD_NONNULL  error_func,
+                       SerdStreamCloseFunc SERD_NULLABLE close_func,
+                       void* SERD_NULLABLE               stream);
 
 /**
-   Create a new byte source that reads from a file.
+   Open a stream that reads from a string.
 
-   An arbitrary `FILE*` can be used via serd_byte_source_new_function() as
-   well, this is just a convenience function that opens the file properly, sets
-   flags for optimized I/O if possible, and automatically sets the name of the
-   source to the file path.
+   The string pointer that position points to must remain valid until the
+   stream is closed.  This pointer serves as the internal stream state and will
+   be mutated as the stream is used.
 
-   @param path Path of file to open and read from.
-   @param page_size Number of bytes to read per call.
+   @param position Pointer to a valid string pointer for use as stream state.
+   @return An opened input stream, or all zeros on error.
 */
-SERD_API
-SerdByteSource* SERD_ALLOCATED
-serd_byte_source_new_filename(const char* SERD_NONNULL path, size_t page_size);
+SERD_CONST_API
+SerdInputStream
+serd_open_input_string(const char* SERD_NONNULL* SERD_NONNULL position);
 
 /**
-   Create a new byte source that reads from a user-specified function
+   Open a stream that reads from a file.
 
-   The `stream` will be passed to the `read_func`, which is compatible with the
-   standard C `fread` if `stream` is a `FILE*`.  Note that the reader only ever
-   reads individual bytes at a time, that is, the `size` parameter will always
-   be 1 (but `nmemb` may be higher).
+   An arbitrary `FILE*` can be used with serd_open_input_stream() as well, this
+   convenience function opens the file properly for reading with serd, and sets
+   flags for optimized I/O if possible.
 
-   @param read_func Stream read function, like `fread`.
-   @param error_func Stream error function, like `ferror`.
-   @param close_func Stream close function, like `fclose`.
-   @param stream Context parameter passed to `read_func` and `error_func`.
-   @param name Optional name of stream for error messages (string or URI).
-   @param page_size Number of bytes to read per call.
+   @param path Path of file to open and read from.
 */
 SERD_API
-SerdByteSource* SERD_ALLOCATED
-serd_byte_source_new_function(SerdReadFunc SERD_NONNULL         read_func,
-                              SerdStreamErrorFunc SERD_NONNULL  error_func,
-                              SerdStreamCloseFunc SERD_NULLABLE close_func,
-                              void* SERD_NULLABLE               stream,
-                              const SerdNode* SERD_NULLABLE     name,
-                              size_t                            page_size);
+SerdInputStream
+serd_open_input_file(const char* SERD_NONNULL path);
+
+/**
+   Close an input stream.
 
-/// Free `source`
+   This will call the close function, and reset the stream internally so that
+   no further reads can be made.  For convenience, this is safe to call on
+   NULL, and safe to call several times on the same input.
+*/
 SERD_API
-void
-serd_byte_source_free(SerdByteSource* SERD_NULLABLE source);
+SerdStatus
+serd_close_input(SerdInputStream* SERD_NULLABLE input);
 
 /**
    @}
@@ -2332,11 +2342,24 @@ serd_reader_new(SerdWorld* SERD_NONNULL      world,
                 const SerdSink* SERD_NONNULL sink,
                 size_t                       stack_size);
 
-/// Prepare to read from a byte source
+/**
+   Prepare to read some input.
+
+   This sets up the reader to read from the given input, but will not read any
+   bytes from it.  This should be followed by serd_reader_read_chunk() or
+   serd_reader_read_document() to actually read the input.
+
+   @param reader The reader.
+   @param input An opened input stream to read from.
+   @param input_name The name of the input stream for error messages.
+   @param block_size The number of bytes to read from the stream at once.
+*/
 SERD_API
 SerdStatus
-serd_reader_start(SerdReader* SERD_NONNULL     reader,
-                  SerdByteSource* SERD_NONNULL byte_source);
+serd_reader_start(SerdReader* SERD_NONNULL      reader,
+                  SerdInputStream* SERD_NONNULL input,
+                  const SerdNode* SERD_NULLABLE input_name,
+                  size_t                        block_size);
 
 /**
    Read a single "chunk" of data during an incremental read.
@@ -2473,11 +2496,11 @@ SerdOutputStream
 serd_open_output_buffer(SerdBuffer* SERD_NONNULL buffer);
 
 /**
-   Create a new byte sink that writes to a file.
+   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 readingn with serd,
-   and sets flags for optimized I/O if possible.
+   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.
 */
diff --git a/meson.build b/meson.build
index eebe55be..da14eb27 100644
--- a/meson.build
+++ b/meson.build
@@ -96,6 +96,7 @@ sources = [
   'src/describe.c',
   'src/env.c',
   'src/filter.c',
+  'src/input_stream.c',
   'src/inserter.c',
   'src/log.c',
   'src/model.c',
diff --git a/src/.clang-tidy b/src/.clang-tidy
index 6daee064..4245a751 100644
--- a/src/.clang-tidy
+++ b/src/.clang-tidy
@@ -3,7 +3,6 @@ Checks: >
   -*-magic-numbers,
   -*-uppercase-literal-suffix,
   -altera-*,
-  -android-cloexec-fopen,
   -bugprone-branch-clone,
   -bugprone-easily-swappable-parameters,
   -bugprone-reserved-identifier,
diff --git a/src/byte_source.c b/src/byte_source.c
index 567beee8..47753d83 100644
--- a/src/byte_source.c
+++ b/src/byte_source.c
@@ -17,21 +17,13 @@
 #include "byte_source.h"
 
 #include "caret.h"
-#include "serd_config.h"
 #include "system.h"
 
 #include "serd/serd.h"
 
-#include <sys/stat.h>
-
-#if USE_POSIX_FADVISE && USE_FILENO
-#  include <fcntl.h>
-#endif
-
 #include <assert.h>
 #include <stdbool.h>
 #include <stdint.h>
-#include <stdio.h>
 #include <stdlib.h>
 #include <string.h>
 
@@ -39,21 +31,21 @@ SerdStatus
 serd_byte_source_page(SerdByteSource* const source)
 {
   uint8_t* const buf =
-    (source->page_size > 1 ? source->file_buf : &source->read_byte);
+    (source->block_size > 1 ? source->block : &source->read_byte);
 
   const size_t n_read =
-    source->read_func(buf, 1, source->page_size, source->stream);
+    source->in->read(buf, 1, source->block_size, source->in->stream);
 
   source->buf_size  = n_read;
   source->read_head = 0;
   source->eof       = false;
 
-  if (n_read < source->page_size) {
+  if (n_read < source->block_size) {
     buf[n_read] = '\0';
     if (n_read == 0) {
       source->eof = true;
-      return (source->error_func(source->stream) ? SERD_ERR_UNKNOWN
-                                                 : SERD_FAILURE);
+      return (source->in->error(source->in->stream) ? SERD_ERR_UNKNOWN
+                                                    : SERD_FAILURE);
     }
   }
 
@@ -63,43 +55,34 @@ serd_byte_source_page(SerdByteSource* const source)
 static void
 serd_byte_source_init_buffer(SerdByteSource* const source)
 {
-  if (source->page_size > 1) {
-    source->file_buf = (uint8_t*)serd_allocate_buffer(source->page_size);
-    source->read_buf = source->file_buf;
-    memset(source->file_buf, '\0', source->page_size);
+  if (source->block_size > 1) {
+    source->block    = (uint8_t*)serd_allocate_buffer(source->block_size);
+    source->read_buf = source->block;
+    memset(source->block, '\0', source->block_size);
   } else {
     source->read_buf = &source->read_byte;
   }
 }
 
 SerdByteSource*
-serd_byte_source_new_function(const SerdReadFunc        read_func,
-                              const SerdStreamErrorFunc error_func,
-                              const SerdStreamCloseFunc close_func,
-                              void* const               stream,
-                              const SerdNode* const     name,
-                              const size_t              page_size)
+serd_byte_source_new_input(SerdInputStream* const input,
+                           const SerdNode* const  name,
+                           const size_t           block_size)
 {
-  assert(read_func);
-  assert(error_func);
+  assert(input);
 
-  if (!page_size) {
+  if (!block_size || !input->stream) {
     return NULL;
   }
 
   SerdByteSource* source = (SerdByteSource*)calloc(1, sizeof(SerdByteSource));
 
-  source->read_func  = read_func;
-  source->error_func = error_func;
-  source->close_func = close_func;
-  source->stream     = stream;
-  source->page_size  = page_size;
-  source->buf_size   = page_size;
-  source->type       = FROM_FUNCTION;
-
   source->name =
-    name ? serd_node_copy(name) : serd_new_string(SERD_STRING("func"));
+    name ? serd_node_copy(name) : serd_new_string(SERD_STRING("input"));
 
+  source->in         = input;
+  source->block_size = block_size;
+  source->buf_size   = block_size;
   source->caret.file = source->name;
   source->caret.line = 1u;
   source->caret.col  = 1u;
@@ -109,107 +92,27 @@ serd_byte_source_new_function(const SerdReadFunc        read_func,
   return source;
 }
 
-static bool
-is_directory(const char* const path)
-{
-#ifdef _MSC_VER
-  struct stat st;
-  return !stat(path, &st) && (st.st_mode & _S_IFDIR);
-#else
-  struct stat st;
-  return !stat(path, &st) && S_ISDIR(st.st_mode);
-#endif
-}
-
-SerdByteSource*
-serd_byte_source_new_filename(const char* const path, const size_t page_size)
+void
+serd_byte_source_free(SerdByteSource* const source)
 {
-  assert(path);
-
-  if (page_size == 0 || is_directory(path)) {
-    return NULL;
-  }
+  if (source) {
+    if (source->block_size > 1) {
+      serd_free_aligned(source->block);
+    }
 
-  FILE* const fd = fopen(path, "rb");
-  if (!fd) {
-    return NULL;
+    serd_node_free(source->name);
+    free(source);
   }
-
-  SerdByteSource* source = (SerdByteSource*)calloc(1, sizeof(SerdByteSource));
-
-  source->read_func  = (SerdReadFunc)fread;
-  source->error_func = (SerdStreamErrorFunc)ferror;
-  source->close_func = (SerdStreamCloseFunc)fclose;
-  source->stream     = fd;
-  source->page_size  = page_size;
-  source->buf_size   = page_size;
-
-  source->name = serd_new_file_uri(SERD_STRING(path), SERD_EMPTY_STRING());
-  source->type = FROM_FILENAME;
-
-  source->caret.file = source->name;
-  source->caret.line = 1u;
-  source->caret.col  = 1u;
-
-  serd_byte_source_init_buffer(source);
-
-#if USE_POSIX_FADVISE && USE_FILENO
-  posix_fadvise(fileno(fd), 0, 0, POSIX_FADV_SEQUENTIAL);
-#endif
-
-  return source;
-}
-
-SerdByteSource*
-serd_byte_source_new_string(const char* const     string,
-                            const SerdNode* const name)
-{
-  assert(string);
-
-  SerdByteSource* source = (SerdByteSource*)calloc(1, sizeof(SerdByteSource));
-
-  source->page_size = 1;
-  source->read_buf  = (const uint8_t*)string;
-  source->type      = FROM_STRING;
-
-  source->name =
-    name ? serd_node_copy(name) : serd_new_string(SERD_STRING("string"));
-
-  source->caret.file = source->name;
-  source->caret.line = 1u;
-  source->caret.col  = 1u;
-
-  return source;
 }
 
 SerdStatus
 serd_byte_source_prepare(SerdByteSource* const source)
 {
   source->prepared = true;
-  if (source->type != FROM_STRING) {
-    if (source->page_size > 1) {
-      return serd_byte_source_page(source);
-    }
 
-    return serd_byte_source_advance(source);
+  if (source->block_size > 1) {
+    return serd_byte_source_page(source);
   }
 
-  return SERD_SUCCESS;
-}
-
-void
-serd_byte_source_free(SerdByteSource* const source)
-{
-  if (source) {
-    if (source->close_func) {
-      source->close_func(source->stream);
-    }
-
-    if (source->page_size > 1) {
-      serd_free_aligned(source->file_buf);
-    }
-
-    serd_node_free(source->name);
-    free(source);
-  }
+  return serd_byte_source_advance(source);
 }
diff --git a/src/byte_source.h b/src/byte_source.h
index d054e156..2bd06bdf 100644
--- a/src/byte_source.h
+++ b/src/byte_source.h
@@ -1,5 +1,5 @@
 /*
-  Copyright 2011-2020 David Robillard <d@drobilla.net>
+  Copyright 2011-2021 David Robillard <d@drobilla.net>
 
   Permission to use, copy, modify, and/or distribute this software for any
   purpose with or without fee is hereby granted, provided that the above
@@ -26,29 +26,27 @@
 #include <stddef.h>
 #include <stdint.h>
 
-typedef enum {
-  FROM_STRING,   ///< Reading from a user-provided buffer
-  FROM_FILENAME, ///< Reading from a file we opened
-  FROM_FUNCTION, ///< Reading from a user-provided function
-} SerdByteSourceType;
-
-struct SerdByteSourceImpl {
-  SerdReadFunc        read_func;  ///< Read function (e.g. fread)
-  SerdStreamErrorFunc error_func; ///< Error function (e.g. ferror)
-  SerdStreamCloseFunc close_func; ///< Function for closing stream
-  void*               stream;     ///< Stream (e.g. FILE)
-  size_t              page_size;  ///< Number of bytes to read at a time
-  size_t              buf_size;   ///< Number of bytes in file_buf
-  SerdNode*           name;       ///< Name of stream (referenced by cur)
-  SerdCaret           caret;      ///< File position for error reporting
-  uint8_t*            file_buf;   ///< Buffer iff reading pages from a file
-  const uint8_t*      read_buf;   ///< Pointer to file_buf or read_byte
-  size_t              read_head;  ///< Offset into read_buf
-  SerdByteSourceType  type;       ///< Type of input
-  uint8_t             read_byte;  ///< 1-byte 'buffer' used when not paging
-  bool                prepared;   ///< True iff prepared for reading
-  bool                eof;        ///< True iff end of file reached
-};
+typedef struct {
+  SerdInputStream* in;         ///< Input stream to read from
+  size_t           block_size; ///< Number of bytes to read at a time
+  size_t           buf_size;   ///< Number of bytes in block
+  SerdNode*        name;       ///< Name of stream (for caret)
+  SerdCaret        caret;      ///< File position for error reporting
+  uint8_t*         block;      ///< Buffer if reading blocks
+  const uint8_t*   read_buf;   ///< Pointer to block or read_byte
+  size_t           read_head;  ///< Offset into read_buf
+  uint8_t          read_byte;  ///< 1-byte 'buffer' if reading bytes
+  bool             prepared;   ///< True iff prepared for reading
+  bool             eof;        ///< True iff end of file reached
+} SerdByteSource;
+
+SerdByteSource*
+serd_byte_source_new_input(SerdInputStream* input,
+                           const SerdNode*  name,
+                           size_t           block_size);
+
+void
+serd_byte_source_free(SerdByteSource* source);
 
 SerdStatus
 serd_byte_source_prepare(SerdByteSource* source);
@@ -71,6 +69,8 @@ serd_byte_source_advance(SerdByteSource* source)
   const bool was_eof = source->eof;
 
   switch (serd_byte_source_peek(source)) {
+  case '\0':
+    break;
   case '\n':
     ++source->caret.line;
     source->caret.col = 0;
@@ -79,12 +79,8 @@ serd_byte_source_advance(SerdByteSource* source)
     ++source->caret.col;
   }
 
-  if (source->type != FROM_STRING) {
-    if (++source->read_head >= source->buf_size) {
-      st = serd_byte_source_page(source);
-    }
-  } else if (!source->eof) {
-    source->eof = source->read_buf[++source->read_head] == '\0';
+  if (++source->read_head >= source->buf_size) {
+    st = serd_byte_source_page(source);
   }
 
   return (was_eof && source->eof) ? SERD_FAILURE : st;
diff --git a/src/input_stream.c b/src/input_stream.c
new file mode 100644
index 00000000..42dfc6f8
--- /dev/null
+++ b/src/input_stream.c
@@ -0,0 +1,150 @@
+/*
+  Copyright 2011-2021 David Robillard <d@drobilla.net>
+
+  Permission to use, copy, modify, and/or distribute this software for any
+  purpose with or without fee is hereby granted, provided that the above
+  copyright notice and this permission notice appear in all copies.
+
+  THIS SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
+  WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
+  MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
+  ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
+  WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
+  ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
+  OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
+*/
+
+#include "serd_config.h"
+
+#include "serd/serd.h"
+
+#include <sys/stat.h>
+
+#if USE_POSIX_FADVISE && USE_FILENO
+#  include <fcntl.h>
+#endif
+
+// IWYU pragma: no_include <features.h>
+
+#include <assert.h>
+#include <stdbool.h>
+#include <stdio.h>
+
+static size_t
+serd_string_read(void* const  buf,
+                 const size_t size,
+                 const size_t nmemb,
+                 void* const  stream)
+{
+  const char** position = (const char**)stream;
+
+  size_t       n_read = 0u;
+  const size_t len    = size * nmemb;
+  while (n_read < len && **position) {
+    ((char*)buf)[n_read++] = **position;
+
+    ++(*position);
+  }
+
+  return n_read;
+}
+
+static int
+serd_string_error(void* const stream)
+{
+  (void)stream;
+  return 0;
+}
+
+static int
+serd_string_close(void* const stream)
+{
+  (void)stream;
+  return 0;
+}
+
+SerdInputStream
+serd_open_input_stream(SerdReadFunc SERD_NONNULL         read_func,
+                       SerdStreamErrorFunc SERD_NONNULL  error_func,
+                       SerdStreamCloseFunc SERD_NULLABLE close_func,
+                       void* const                       stream)
+{
+  assert(read_func);
+  assert(error_func);
+
+  SerdInputStream input = {stream, read_func, error_func, close_func};
+  return input;
+}
+
+static bool
+is_directory(const char* const path)
+{
+#ifdef _MSC_VER
+  struct stat st;
+  return !stat(path, &st) && (st.st_mode & _S_IFDIR);
+#else
+  struct stat st;
+  return !stat(path, &st) && S_ISDIR(st.st_mode);
+#endif
+}
+
+SerdInputStream
+serd_open_input_string(const char** const position)
+{
+  assert(position);
+  assert(*position);
+
+  const SerdInputStream input = {
+    position, serd_string_read, serd_string_error, serd_string_close};
+
+  return input;
+}
+
+SerdInputStream
+serd_open_input_file(const char* const path)
+{
+  assert(path);
+
+  SerdInputStream input = {NULL, NULL, NULL, NULL};
+  if (is_directory(path)) {
+    return input;
+  }
+
+#ifdef __GLIBC__
+  FILE* const file = fopen(path, "rbe");
+#else
+  FILE* const file = fopen(path, "rb");
+#endif
+
+  if (!file) {
+    return input;
+  }
+
+#if USE_POSIX_FADVISE && USE_FILENO
+  posix_fadvise(fileno(file), 0, 0, POSIX_FADV_SEQUENTIAL);
+#endif
+
+  input.stream = file;
+  input.read   = (SerdReadFunc)fread;
+  input.error  = (SerdStreamErrorFunc)ferror;
+  input.close  = (SerdStreamCloseFunc)fclose;
+
+  return input;
+}
+
+SerdStatus
+serd_close_input(SerdInputStream* const input)
+{
+  int ret = 0;
+
+  if (input) {
+    if (input->close && input->stream) {
+      ret           = input->close(input->stream);
+      input->stream = NULL;
+    }
+
+    input->stream = NULL;
+  }
+
+  return ret ? SERD_ERR_UNKNOWN : SERD_SUCCESS;
+}
diff --git a/src/node_syntax.c b/src/node_syntax.c
index d2768c3e..5c579b84 100644
--- a/src/node_syntax.c
+++ b/src/node_syntax.c
@@ -54,8 +54,7 @@ serd_node_from_syntax_in(const char* const str,
   SerdWorld* const world  = serd_world_new();
   SerdSink* const  sink   = serd_sink_new(&object, on_node_string_event, NULL);
 
-  SerdByteSource* const source = serd_byte_source_new_string(doc, NULL);
-  SerdReader* const     reader =
+  SerdReader* const reader =
     serd_reader_new(world,
                     syntax,
                     SERD_READ_RELATIVE | SERD_READ_GLOBAL | SERD_READ_GENERATED,
@@ -63,11 +62,16 @@ serd_node_from_syntax_in(const char* const str,
                     sink,
                     1024 + doc_len);
 
-  serd_reader_start(reader, source);
+  const SerdNode* string_name =
+    serd_nodes_string(serd_world_nodes(world), SERD_STRING("string"));
+
+  const char*     position = doc;
+  SerdInputStream in       = serd_open_input_string(&position);
+  serd_reader_start(reader, &in, string_name, 1);
   serd_reader_read_document(reader);
   serd_reader_finish(reader);
+  serd_close_input(&in);
   serd_reader_free(reader);
-  serd_byte_source_free(source);
   serd_sink_free(sink);
   serd_world_free(world);
   free(doc);
diff --git a/src/reader.c b/src/reader.c
index c8a66c42..d051c34a 100644
--- a/src/reader.c
+++ b/src/reader.c
@@ -310,14 +310,17 @@ skip_bom(SerdReader* const me)
 }
 
 SerdStatus
-serd_reader_start(SerdReader* const reader, SerdByteSource* const byte_source)
+serd_reader_start(SerdReader* const      reader,
+                  SerdInputStream* const input,
+                  const SerdNode* const  input_name,
+                  const size_t           block_size)
 {
   assert(reader);
-  assert(byte_source);
+  assert(input);
 
   serd_reader_finish(reader);
 
-  reader->source = byte_source;
+  reader->source = serd_byte_source_new_input(input, input_name, block_size);
 
   return reader->source ? SERD_SUCCESS : SERD_ERR_BAD_ARG;
 }
@@ -358,6 +361,7 @@ serd_reader_finish(SerdReader* const reader)
 {
   assert(reader);
 
+  serd_byte_source_free(reader->source);
   reader->source = NULL;
   return SERD_SUCCESS;
 }
diff --git a/test/meson.build b/test/meson.build
index ab7666e3..37bae479 100644
--- a/test/meson.build
+++ b/test/meson.build
@@ -8,7 +8,6 @@ run_sort_suite = find_program('run_sort_suite.py')
 wrapper = meson.get_cross_property('exe_wrapper', '')
 
 unit_tests = [
-  'byte_source',
   'caret',
   'cursor',
   'env',
diff --git a/test/test_byte_source.c b/test/test_byte_source.c
deleted file mode 100644
index 14ef5819..00000000
--- a/test/test_byte_source.c
+++ /dev/null
@@ -1,40 +0,0 @@
-/*
-  Copyright 2021 David Robillard <d@drobilla.net>
-
-  Permission to use, copy, modify, and/or distribute this software for any
-  purpose with or without fee is hereby granted, provided that the above
-  copyright notice and this permission notice appear in all copies.
-
-  THIS SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
-  WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
-  MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
-  ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
-  WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
-  ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
-  OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
-*/
-
-#undef NDEBUG
-
-#include "serd/serd.h"
-
-#include <assert.h>
-#include <stddef.h>
-#include <stdio.h>
-
-static void
-test_bad_page_size(void)
-{
-  assert(!serd_byte_source_new_filename("file.ttl", 0));
-
-  assert(!serd_byte_source_new_function(
-    (SerdReadFunc)fread, (SerdStreamErrorFunc)ferror, NULL, NULL, NULL, 0));
-}
-
-int
-main(void)
-{
-  test_bad_page_size();
-
-  return 0;
-}
diff --git a/test/test_free_null.c b/test/test_free_null.c
index 51a3c2dc..85529a2b 100644
--- a/test/test_free_null.c
+++ b/test/test_free_null.c
@@ -24,7 +24,6 @@ int
 main(void)
 {
   serd_free(NULL);
-  serd_byte_source_free(NULL);
   serd_node_free(NULL);
   serd_world_free(NULL);
   serd_env_free(NULL);
diff --git a/test/test_overflow.c b/test/test_overflow.c
index bb2513d5..ca273b23 100644
--- a/test/test_overflow.c
+++ b/test/test_overflow.c
@@ -31,19 +31,24 @@ test_size(SerdWorld* const      world,
           const SerdReaderFlags flags,
           const size_t          stack_size)
 {
-  SerdSink*         sink        = serd_sink_new(NULL, NULL, NULL);
-  SerdByteSource*   byte_source = serd_byte_source_new_string(str, NULL);
-  SerdEnv* const    env         = serd_env_new(SERD_EMPTY_STRING());
+  SerdSink*         sink = serd_sink_new(NULL, NULL, NULL);
+  SerdEnv* const    env  = serd_env_new(SERD_EMPTY_STRING());
   SerdReader* const reader =
     serd_reader_new(world, syntax, flags, env, sink, stack_size);
 
   assert(reader);
 
-  serd_reader_start(reader, byte_source);
+  SerdNode*       string_name = serd_new_string(SERD_STRING("string"));
+  const char*     position    = str;
+  SerdInputStream in          = serd_open_input_string(&position);
+  serd_reader_start(reader, &in, string_name, 1);
+
   const SerdStatus st = serd_reader_read_document(reader);
+
+  serd_close_input(&in);
+  serd_node_free(string_name);
   serd_reader_free(reader);
   serd_env_free(env);
-  serd_byte_source_free(byte_source);
   serd_sink_free(sink);
 
   return st;
diff --git a/test/test_read_chunk.c b/test/test_read_chunk.c
index 52bb7804..a956a07d 100644
--- a/test/test_read_chunk.c
+++ b/test/test_read_chunk.c
@@ -94,23 +94,23 @@ main(void)
   SerdWorld* world = serd_world_new();
   SerdSink*  sink  = serd_sink_new(NULL, on_event, NULL);
 
-  SerdByteSource* byte_source =
-    serd_byte_source_new_string("@prefix eg: <http://example.org/> .\n"
-                                "@base <http://example.org/base> .\n"
-                                "eg:s1 eg:p1 eg:o1 ;\n"
-                                "      eg:p2 eg:o2 ,\n"
-                                "            eg:o3 .\n"
-                                "eg:s2 eg:p1 eg:o1 ;\n"
-                                "      eg:p2 eg:o2 .\n"
-                                "eg:s3 eg:p1 eg:o1 .\n"
-                                "eg:s4 eg:p1 [ eg:p3 eg:o1 ] .\n",
-                                NULL);
+  static const char* const string = "@prefix eg: <http://example.org/> .\n"
+                                    "@base <http://example.org/base> .\n"
+                                    "eg:s1 eg:p1 eg:o1 ;\n"
+                                    "      eg:p2 eg:o2 ,\n"
+                                    "            eg:o3 .\n"
+                                    "eg:s2 eg:p1 eg:o1 ;\n"
+                                    "      eg:p2 eg:o2 .\n"
+                                    "eg:s3 eg:p1 eg:o1 .\n"
+                                    "eg:s4 eg:p1 [ eg:p3 eg:o1 ] .\n";
 
   SerdEnv*    env    = serd_env_new(SERD_EMPTY_STRING());
   SerdReader* reader = serd_reader_new(world, SERD_TURTLE, 0, env, sink, 4096);
   assert(reader);
 
-  assert(!serd_reader_start(reader, byte_source));
+  const char*     position = string;
+  SerdInputStream in       = serd_open_input_string(&position);
+  assert(!serd_reader_start(reader, &in, NULL, 1));
 
   assert(!serd_reader_read_chunk(reader) && n_prefix == 1);
   assert(!serd_reader_read_chunk(reader) && n_base == 1);
@@ -123,9 +123,9 @@ main(void)
   assert(serd_reader_read_chunk(reader) == SERD_FAILURE);
   assert(!serd_reader_finish(reader));
 
+  serd_close_input(&in);
   serd_reader_free(reader);
   serd_env_free(env);
-  serd_byte_source_free(byte_source);
   serd_sink_free(sink);
   serd_world_free(world);
 
diff --git a/test/test_reader.c b/test/test_reader.c
index 32b475f9..1c28e3da 100644
--- a/test/test_reader.c
+++ b/test/test_reader.c
@@ -71,15 +71,17 @@ test_prepare_error(void)
 
   assert(reader);
 
-  SerdByteSource* byte_source = serd_byte_source_new_function(
-    prepare_test_read, prepare_test_error, NULL, f, NULL, 1);
+  SerdInputStream in =
+    serd_open_input_stream(prepare_test_read, prepare_test_error, NULL, f);
 
-  SerdStatus st = serd_reader_start(reader, byte_source);
+  assert(serd_reader_start(reader, &in, NULL, 0) == SERD_ERR_BAD_ARG);
+
+  SerdStatus st = serd_reader_start(reader, &in, NULL, 1);
   assert(!st);
 
   assert(serd_reader_read_document(reader) == SERD_ERR_UNKNOWN);
 
-  serd_byte_source_free(byte_source);
+  serd_close_input(&in);
   serd_reader_free(reader);
   serd_env_free(env);
   serd_sink_free(sink);
@@ -101,34 +103,38 @@ test_read_string(void)
 
   assert(reader);
 
-  SerdByteSource* byte_source =
-    serd_byte_source_new_string("<http://example.org/s> <http://example.org/p> "
-                                "<http://example.org/o> .",
-                                NULL);
+  static const char* const string1 =
+    "<http://example.org/s> <http://example.org/p> "
+    "<http://example.org/o> .";
+
+  const char*     position = string1;
+  SerdInputStream in       = serd_open_input_string(&position);
 
   // Test reading a string that ends exactly at the end of input (no newline)
-  assert(!serd_reader_start(reader, byte_source));
+  assert(!serd_reader_start(reader, &in, NULL, 1));
   assert(!serd_reader_read_document(reader));
   assert(n_statements == 1);
   assert(!serd_reader_finish(reader));
+  assert(!serd_close_input(&in));
+
+  static const char* const string2 =
+    "<http://example.org/s> <http://example.org/p> "
+    "<http://example.org/o> , _:blank .";
 
-  // Test reading the same but as a chunk
-  serd_byte_source_free(byte_source);
+  // Test reading a chunk
   n_statements = 0;
-  byte_source =
-    serd_byte_source_new_string("<http://example.org/s> <http://example.org/p> "
-                                "<http://example.org/o> , _:blank .",
-                                NULL);
+  position     = string2;
+  in           = serd_open_input_string(&position);
 
-  assert(!serd_reader_start(reader, byte_source));
+  assert(!serd_reader_start(reader, &in, NULL, 1));
   assert(!serd_reader_read_chunk(reader));
   assert(n_statements == 2);
   assert(serd_reader_read_chunk(reader) == SERD_FAILURE);
   assert(!serd_reader_finish(reader));
+  assert(!serd_close_input(&in));
 
   serd_reader_free(reader);
   serd_env_free(env);
-  serd_byte_source_free(byte_source);
   serd_sink_free(sink);
   serd_world_free(world);
 }
@@ -195,16 +201,16 @@ test_read_eof_by_page(void)
 
   SerdReader* reader = serd_reader_new(world, SERD_TURTLE, 0u, env, sink, 4096);
 
-  SerdByteSource* byte_source = serd_byte_source_new_function(
-    (SerdReadFunc)fread, (SerdStreamErrorFunc)ferror, NULL, temp, NULL, 4096);
+  SerdInputStream in = serd_open_input_stream(
+    (SerdReadFunc)fread, (SerdStreamErrorFunc)ferror, NULL, temp);
 
-  assert(serd_reader_start(reader, byte_source) == SERD_SUCCESS);
+  assert(serd_reader_start(reader, &in, NULL, 4096) == SERD_SUCCESS);
   assert(serd_reader_read_chunk(reader) == SERD_SUCCESS);
   assert(serd_reader_read_chunk(reader) == SERD_FAILURE);
   assert(serd_reader_read_chunk(reader) == SERD_FAILURE);
   assert(!serd_reader_finish(reader));
+  assert(!serd_close_input(&in));
 
-  serd_byte_source_free(byte_source);
   serd_reader_free(reader);
   serd_env_free(env);
   serd_sink_free(sink);
@@ -224,22 +230,20 @@ test_read_eof_by_byte(void)
   SerdReader* reader = serd_reader_new(world, SERD_TURTLE, 0u, env, sink, 4096);
 
   size_t          n_reads = 0u;
-  SerdByteSource* byte_source =
-    serd_byte_source_new_function((SerdReadFunc)eof_test_read,
-                                  (SerdStreamErrorFunc)eof_test_error,
-                                  NULL,
-                                  &n_reads,
-                                  NULL,
-                                  1);
-
-  assert(serd_reader_start(reader, byte_source) == SERD_SUCCESS);
+  SerdInputStream in =
+    serd_open_input_stream((SerdReadFunc)eof_test_read,
+                           (SerdStreamErrorFunc)eof_test_error,
+                           NULL,
+                           &n_reads);
+
+  assert(serd_reader_start(reader, &in, NULL, 1) == SERD_SUCCESS);
   assert(serd_reader_read_chunk(reader) == SERD_SUCCESS);
   assert(serd_reader_read_chunk(reader) == SERD_FAILURE);
   assert(serd_reader_read_chunk(reader) == SERD_SUCCESS);
   assert(serd_reader_read_chunk(reader) == SERD_FAILURE);
   assert(!serd_reader_finish(reader));
+  assert(!serd_close_input(&in));
 
-  serd_byte_source_free(byte_source);
   serd_reader_free(reader);
   serd_env_free(env);
   serd_sink_free(sink);
@@ -263,10 +267,10 @@ test_read_chunks(void)
 
   assert(reader);
 
-  SerdByteSource* byte_source = serd_byte_source_new_function(
-    (SerdReadFunc)fread, (SerdStreamErrorFunc)ferror, NULL, f, NULL, 1);
+  SerdInputStream in = serd_open_input_stream(
+    (SerdReadFunc)fread, (SerdStreamErrorFunc)ferror, NULL, f);
 
-  SerdStatus st = serd_reader_start(reader, byte_source);
+  SerdStatus st = serd_reader_start(reader, &in, NULL, 1);
   assert(st == SERD_SUCCESS);
 
   // Write two statement separated by null characters
@@ -307,7 +311,7 @@ test_read_chunks(void)
   assert(st == SERD_FAILURE);
   assert(n_statements == 2);
 
-  serd_byte_source_free(byte_source);
+  assert(!serd_close_input(&in));
   serd_reader_free(reader);
   serd_env_free(env);
   serd_sink_free(sink);
@@ -315,26 +319,6 @@ test_read_chunks(void)
   serd_world_free(world);
 }
 
-static size_t
-empty_test_read(void* buf, size_t size, size_t nmemb, void* stream)
-{
-  (void)buf;
-  (void)size;
-  (void)nmemb;
-  (void)stream;
-
-  assert(false);
-
-  return 0;
-}
-
-static int
-empty_test_error(void* stream)
-{
-  (void)stream;
-  return 0;
-}
-
 /// Test that reading SERD_SYNTAX_EMPTY "succeeds" without reading any input
 static void
 test_read_empty(void)
@@ -352,16 +336,17 @@ test_read_empty(void)
 
   assert(reader);
 
-  SerdByteSource* byte_source = serd_byte_source_new_function(
-    empty_test_read, empty_test_error, NULL, f, NULL, 1);
+  SerdInputStream in = serd_open_input_stream(
+    (SerdReadFunc)fread, (SerdStreamErrorFunc)ferror, NULL, f);
 
-  SerdStatus st = serd_reader_start(reader, byte_source);
+  SerdStatus st = serd_reader_start(reader, &in, NULL, 1);
   assert(!st);
 
   assert(serd_reader_read_document(reader) == SERD_SUCCESS);
   assert(n_statements == 0);
+  assert(!serd_reader_finish(reader));
+  assert(!serd_close_input(&in));
 
-  serd_byte_source_free(byte_source);
   serd_reader_free(reader);
   serd_env_free(env);
   serd_sink_free(sink);
@@ -400,17 +385,22 @@ test_error_cursor(void)
   assert(sink);
   assert(reader);
 
-  SerdByteSource* byte_source =
-    serd_byte_source_new_string("<http://example.org/s> <http://example.org/p> "
-                                "<http://example.org/o> .",
-                                NULL);
+  static const char* const string =
+    "<http://example.org/s> <http://example.org/p> "
+    "<http://example.org/o> .";
+
+  SerdNode* const string_name = serd_new_string(SERD_STRING("string"));
+  const char*     position    = string;
+  SerdInputStream in          = serd_open_input_string(&position);
 
-  SerdStatus st = serd_reader_start(reader, byte_source);
+  SerdStatus st = serd_reader_start(reader, &in, string_name, 1);
   assert(!st);
   assert(serd_reader_read_document(reader) == SERD_SUCCESS);
+  assert(!serd_reader_finish(reader));
   assert(called);
+  assert(!serd_close_input(&in));
 
-  serd_byte_source_free(byte_source);
+  serd_node_free(string_name);
   serd_reader_free(reader);
   serd_env_free(env);
   serd_sink_free(sink);
diff --git a/test/test_reader_writer.c b/test/test_reader_writer.c
index bd83b082..82af9f20 100644
--- a/test/test_reader_writer.c
+++ b/test/test_reader_writer.c
@@ -149,12 +149,12 @@ test_reader(const char* path)
   assert(serd_reader_read_document(reader) == SERD_ERR_BAD_CALL);
   assert(serd_reader_read_chunk(reader) == SERD_ERR_BAD_CALL);
 
-  SerdByteSource* byte_source = serd_byte_source_new_filename(path, 4096);
-  assert(!serd_reader_start(reader, byte_source));
+  SerdInputStream in = serd_open_input_file(path);
+  assert(!serd_reader_start(reader, &in, NULL, 4096));
   assert(!serd_reader_read_document(reader));
   assert(n_statements == 6);
   serd_reader_finish(reader);
-  serd_byte_source_free(byte_source);
+  serd_close_input(&in);
 
   serd_reader_free(reader);
   serd_env_free(env);
diff --git a/tools/console.c b/tools/console.c
index 5cfeb56c..30c57575 100644
--- a/tools/console.c
+++ b/tools/console.c
@@ -40,7 +40,7 @@ serd_tool_setup(SerdTool* const   tool,
   // Open the output first, since if that fails we have nothing to do
   const char* const out_path = options.out_filename;
 
-  if (!((tool->out = serd_open_output(out_path)).stream)) {
+  if (!((tool->out = serd_open_tool_output(out_path)).stream)) {
     fprintf(stderr,
             "%s: failed to open output file (%s)\n",
             program,
@@ -352,28 +352,22 @@ serd_file_read_byte(void* buf, size_t size, size_t nmemb, void* stream)
   return 1;
 }
 
-SerdByteSource*
-serd_open_input(const char* const filename, const size_t block_size)
+SerdInputStream
+serd_open_tool_input(const char* const filename)
 {
-  SerdByteSource* byte_source = NULL;
   if (!strcmp(filename, "-")) {
-    serd_set_stream_utf8_mode(stdin);
-
-    SerdNode* name = serd_new_string(SERD_STRING("stdin"));
-
-    byte_source = serd_byte_source_new_function(
-      serd_file_read_byte, (SerdStreamErrorFunc)ferror, NULL, stdin, name, 1);
+    const SerdInputStream in = serd_open_input_stream(
+      serd_file_read_byte, (SerdStreamErrorFunc)ferror, NULL, stdin);
 
-    serd_node_free(name);
-  } else {
-    byte_source = serd_byte_source_new_filename(filename, block_size);
+    serd_set_stream_utf8_mode(stdin);
+    return in;
   }
 
-  return byte_source;
+  return serd_open_input_file(filename);
 }
 
 SerdOutputStream
-serd_open_output(const char* const filename)
+serd_open_tool_output(const char* const filename)
 {
   if (!filename || !strcmp(filename, "-")) {
     serd_set_stream_utf8_mode(stdout);
@@ -422,13 +416,16 @@ serd_read_source(SerdWorld* const        world,
                  const SerdCommonOptions opts,
                  SerdEnv* const          env,
                  const SerdSyntax        syntax,
-                 SerdByteSource* const   in,
+                 SerdInputStream* const  in,
+                 const char* const       name,
                  const SerdSink* const   sink)
 {
   SerdReader* const reader = serd_reader_new(
     world, syntax, opts.input.flags, env, sink, opts.stack_size);
 
-  SerdStatus st = serd_reader_start(reader, in);
+  SerdNode* const name_node = serd_new_string(SERD_STRING(name));
+  SerdStatus st = serd_reader_start(reader, in, name_node, opts.block_size);
+  serd_node_free(name_node);
   if (!st) {
     st = serd_reader_read_document(reader);
   }
@@ -455,8 +452,8 @@ serd_read_inputs(SerdWorld* const        world,
     }
 
     // Open the input stream
-    SerdByteSource* const in = serd_open_input(in_path, opts.block_size);
-    if (!in) {
+    SerdInputStream in = serd_open_tool_input(in_path);
+    if (!in.stream) {
       return SERD_ERR_BAD_ARG;
     }
 
@@ -466,10 +463,11 @@ serd_read_inputs(SerdWorld* const        world,
       opts,
       env,
       serd_choose_syntax(world, opts.input, in_path, SERD_TRIG),
-      in,
+      &in,
+      !strcmp(in_path, "-") ? "stdin" : in_path,
       sink);
 
-    serd_byte_source_free(in);
+    serd_close_input(&in);
   }
 
   return st;
diff --git a/tools/console.h b/tools/console.h
index 97251f68..43bc7a12 100644
--- a/tools/console.h
+++ b/tools/console.h
@@ -119,11 +119,11 @@ serd_choose_syntax(SerdWorld*        world,
                    const char*       filename,
                    SerdSyntax        fallback);
 
-SerdByteSource*
-serd_open_input(const char* filename, size_t block_size);
+SerdInputStream
+serd_open_tool_input(const char* filename);
 
 SerdOutputStream
-serd_open_output(const char* filename);
+serd_open_tool_output(const char* filename);
 
 SerdStatus
 serd_set_base_uri_from_path(SerdEnv* env, const char* path);
@@ -133,7 +133,8 @@ serd_read_source(SerdWorld*        world,
                  SerdCommonOptions opts,
                  SerdEnv*          env,
                  SerdSyntax        syntax,
-                 SerdByteSource*   in,
+                 SerdInputStream*  in,
+                 const char*       name,
                  const SerdSink*   sink);
 
 SerdStatus
diff --git a/tools/serd-filter.c b/tools/serd-filter.c
index a6f9b484..147bf51d 100644
--- a/tools/serd-filter.c
+++ b/tools/serd-filter.c
@@ -66,10 +66,10 @@ on_pattern_event(void* const handle, const SerdEvent* const event)
 
 // Parse a pattern from some input and return a new filter for it
 static SerdSink*
-parse_pattern(SerdWorld* const      world,
-              const SerdSink* const sink,
-              SerdByteSource* const byte_source,
-              const bool            inclusive)
+parse_pattern(SerdWorld* const       world,
+              const SerdSink* const  sink,
+              SerdInputStream* const in,
+              const bool             inclusive)
 {
   SerdEnv* const env     = serd_env_new(SERD_EMPTY_STRING());
   FilterPattern  pat     = {NULL, NULL, NULL, NULL};
@@ -77,11 +77,15 @@ parse_pattern(SerdWorld* const      world,
   SerdReader*    reader  = serd_reader_new(
     world, SERD_NQUADS, SERD_READ_VARIABLES, env, in_sink, 4096);
 
-  SerdStatus st = serd_reader_start(reader, byte_source);
+  const SerdNode* pattern_name =
+    serd_nodes_string(serd_world_nodes(world), SERD_STRING("pattern"));
+
+  SerdStatus st = serd_reader_start(reader, in, pattern_name, 1);
   if (!st) {
     st = serd_reader_read_document(reader);
   }
 
+  serd_close_input(in);
   serd_reader_free(reader);
   serd_env_free(env);
   serd_sink_free(in_sink);
@@ -135,24 +139,29 @@ run(Options opts)
   const SerdSink* const target = serd_writer_sink(app.writer);
 
   // Open the pattern input (either a string or filename)
-  SerdByteSource* const pattern =
-    opts.pattern ? serd_byte_source_new_string(opts.pattern, NULL)
-    : opts.pattern_file
-      ? serd_byte_source_new_filename(opts.pattern_file, opts.common.block_size)
-      : NULL;
-  if (!pattern) {
+  SerdInputStream pattern  = {NULL, NULL, NULL, NULL};
+  const char*     position = opts.pattern;
+  if (opts.pattern) {
+    pattern = serd_open_input_string(&position);
+  } else if (opts.pattern_file) {
+    pattern = serd_open_input_file(opts.pattern_file);
+  }
+
+  if (!pattern.stream) {
     log_error(app.world, "failed to open pattern");
     return SERD_ERR_UNKNOWN;
   }
 
   // Set up the output pipeline: filter -> writer
   SerdSink* const filter =
-    parse_pattern(app.world, target, pattern, !opts.invert);
+    parse_pattern(app.world, target, &pattern, !opts.invert);
   if (!filter) {
     log_error(app.world, "failed to set up filter");
     return SERD_ERR_UNKNOWN;
   }
 
+  serd_close_input(&pattern);
+
   // Read all the inputs, which drives the writer to emit the output
   if (!(st = serd_read_inputs(app.world,
                               opts.common,
@@ -168,7 +177,6 @@ run(Options opts)
   }
 
   serd_sink_free(filter);
-  serd_byte_source_free(pattern);
 
   const SerdStatus cst = serd_tool_cleanup(app);
   return st ? st : cst;
diff --git a/tools/serd-pipe.c b/tools/serd-pipe.c
index 08479a74..25607d15 100644
--- a/tools/serd-pipe.c
+++ b/tools/serd-pipe.c
@@ -64,18 +64,19 @@ run(const Options opts)
   }
 
   if (opts.input_string) {
-    SerdByteSource* const in =
-      serd_byte_source_new_string(opts.input_string, NULL);
+    const char*     position = opts.input_string;
+    SerdInputStream in       = serd_open_input_string(&position);
 
     st = serd_read_source(
       app.world,
       opts.common,
       app.env,
       serd_choose_syntax(app.world, opts.common.input, NULL, SERD_TRIG),
-      in,
+      &in,
+      "string",
       sink);
 
-    serd_byte_source_free(in);
+    serd_close_input(&in);
   }
 
   // Read all the inputs, which drives the writer to emit the output