Simplify reader interface

9 files changed, 190 insertions, 284 deletions

diff --git a/include/serd/serd.h b/include/serd/serd.h
index 6b903b58..bfe53401 100644
--- a/include/serd/serd.h
+++ b/include/serd/serd.h
@@ -22,7 +22,6 @@
 #include <stdarg.h>
 #include <stdbool.h>
 #include <stdint.h>
-#include <stdio.h>
 #include <string.h> // IWYU pragma: keep
 
 #if defined(_WIN32) && !defined(SERD_STATIC) && defined(SERD_INTERNAL)
@@ -982,41 +981,33 @@ void
 serd_reader_set_default_graph(SerdReader* SERD_NONNULL      reader,
                               const SerdNode* SERD_NULLABLE graph);
 
-/// Read a file at a given `uri`
+/// Prepare to read from the file at a local file `uri`
 SERD_API
 SerdStatus
-serd_reader_read_file(SerdReader* SERD_NONNULL reader,
-                      const char* SERD_NONNULL uri);
+serd_reader_start_file(SerdReader* SERD_NONNULL reader,
+                       const char* SERD_NONNULL uri,
+                       bool                     bulk);
 
 /**
-   Start an incremental read from a file handle.
-
-   Iff `bulk` is true, `file` will be read a page at a time.  This is more
-   efficient, but uses a page of memory and means that an entire page of input
-   must be ready before any callbacks will fire.  To react as soon as input
-   arrives, set `bulk` to false.
-*/
-SERD_API
-SerdStatus
-serd_reader_start_stream(SerdReader* SERD_NONNULL  reader,
-                         FILE* SERD_NONNULL        file,
-                         const char* SERD_NULLABLE name,
-                         bool                      bulk);
-
-/**
-   Start an incremental read from a user-specified source.
+   Prepare to read from a stream.
 
    The `read_func` is guaranteed to only be called for `page_size` elements
    with size 1 (i.e. `page_size` bytes).
 */
 SERD_API
 SerdStatus
-serd_reader_start_source_stream(SerdReader* SERD_NONNULL         reader,
-                                SerdReadFunc SERD_NONNULL        read_func,
-                                SerdStreamErrorFunc SERD_NONNULL error_func,
-                                void* SERD_NONNULL               stream,
-                                const char* SERD_NULLABLE        name,
-                                size_t                           page_size);
+serd_reader_start_stream(SerdReader* SERD_NONNULL         reader,
+                         SerdReadFunc SERD_NONNULL        read_func,
+                         SerdStreamErrorFunc SERD_NONNULL error_func,
+                         void* SERD_NONNULL               stream,
+                         const char* SERD_NULLABLE        name,
+                         size_t                           page_size);
+
+/// Prepare to read from a string
+SERD_API
+SerdStatus
+serd_reader_start_string(SerdReader* SERD_NONNULL reader,
+                         const char* SERD_NONNULL utf8);
 
 /**
    Read a single "chunk" of data during an incremental read.
@@ -1030,35 +1021,31 @@ SERD_API
 SerdStatus
 serd_reader_read_chunk(SerdReader* SERD_NONNULL reader);
 
-/// Finish an incremental read from a file handle
-SERD_API
-SerdStatus
-serd_reader_end_stream(SerdReader* SERD_NONNULL reader);
+/**
+   Read a complete document from the source.
 
-/// Read `file`
+   This function will continue pulling from the source until a complete
+   document has been read.  Note that this may block when used with streams,
+   for incremental reading use serd_reader_read_chunk().
+*/
 SERD_API
 SerdStatus
-serd_reader_read_file_handle(SerdReader* SERD_NONNULL  reader,
-                             FILE* SERD_NONNULL        file,
-                             const char* SERD_NULLABLE name);
+serd_reader_read_document(SerdReader* SERD_NONNULL reader);
 
-/// Read a user-specified byte source
-SERD_API
-SerdStatus
-serd_reader_read_source(SerdReader* SERD_NONNULL         reader,
-                        SerdReadFunc SERD_NONNULL        source,
-                        SerdStreamErrorFunc SERD_NONNULL error,
-                        void* SERD_NONNULL               stream,
-                        const char* SERD_NULLABLE        name,
-                        size_t                           page_size);
-
-/// Read `utf8`
+/**
+   Finish reading from the source.
+
+   This should be called before starting to read from another source.
+*/
 SERD_API
 SerdStatus
-serd_reader_read_string(SerdReader* SERD_NONNULL reader,
-                        const char* SERD_NONNULL utf8);
+serd_reader_finish(SerdReader* SERD_NONNULL reader);
 
-/// Free `reader`
+/**
+   Free `reader`.
+
+   The reader will be finished via `serd_reader_finish()` if necessary.
+*/
 SERD_API
 void
 serd_reader_free(SerdReader* SERD_NULLABLE reader);
diff --git a/src/byte_source.c b/src/byte_source.c
index 275133b9..727c655f 100644
--- a/src/byte_source.c
+++ b/src/byte_source.c
@@ -50,6 +50,7 @@ SerdStatus
 serd_byte_source_open_source(SerdByteSource* const     source,
                              const SerdReadFunc        read_func,
                              const SerdStreamErrorFunc error_func,
+                             const SerdStreamCloseFunc close_func,
                              void* const               stream,
                              const char* const         name,
                              const size_t              page_size)
@@ -57,13 +58,14 @@ serd_byte_source_open_source(SerdByteSource* const     source,
   const Cursor cur = {name, 1, 1};
 
   memset(source, '\0', sizeof(*source));
+  source->read_func   = read_func;
+  source->error_func  = error_func;
+  source->close_func  = close_func;
   source->stream      = stream;
-  source->from_stream = true;
   source->page_size   = page_size;
   source->buf_size    = page_size;
   source->cur         = cur;
-  source->error_func  = error_func;
-  source->read_func   = read_func;
+  source->from_stream = true;
 
   if (page_size > 1) {
     source->file_buf = (uint8_t*)serd_allocate_buffer(page_size);
@@ -104,10 +106,15 @@ serd_byte_source_open_string(SerdByteSource* const source,
 SerdStatus
 serd_byte_source_close(SerdByteSource* const source)
 {
+  SerdStatus st = SERD_SUCCESS;
+  if (source->close_func) {
+    st = source->close_func(source->stream) ? SERD_ERR_UNKNOWN : SERD_SUCCESS;
+  }
+
   if (source->page_size > 1) {
     serd_free_aligned(source->file_buf);
   }
 
   memset(source, '\0', sizeof(*source));
-  return SERD_SUCCESS;
+  return st;
 }
diff --git a/src/byte_source.h b/src/byte_source.h
index a5c98ef6..69f92295 100644
--- a/src/byte_source.h
+++ b/src/byte_source.h
@@ -23,7 +23,8 @@
 #include <stdbool.h>
 #include <stddef.h>
 #include <stdint.h>
-#include <stdio.h>
+
+typedef int (*SerdStreamCloseFunc)(void*);
 
 typedef struct {
   const char* filename;
@@ -34,6 +35,7 @@ typedef struct {
 typedef struct {
   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
@@ -48,15 +50,13 @@ typedef struct {
 } SerdByteSource;
 
 SerdStatus
-serd_byte_source_open_file(SerdByteSource* source, FILE* file, bool bulk);
-
-SerdStatus
 serd_byte_source_open_string(SerdByteSource* source, const char* utf8);
 
 SerdStatus
 serd_byte_source_open_source(SerdByteSource*     source,
                              SerdReadFunc        read_func,
                              SerdStreamErrorFunc error_func,
+                             SerdStreamCloseFunc close_func,
                              void*               stream,
                              const char*         name,
                              size_t              page_size);
diff --git a/src/n3.c b/src/n3.c
index 8e599458..c29908ae 100644
--- a/src/n3.c
+++ b/src/n3.c
@@ -1696,8 +1696,9 @@ read_nquadsDoc(SerdReader* const reader)
     }
 
     if (peek_byte(reader) == '@') {
-      return r_err(
+      r_err(
         reader, SERD_ERR_BAD_SYNTAX, "syntax does not support directives\n");
+      return SERD_ERR_BAD_SYNTAX;
     }
 
     // subject predicate object
diff --git a/src/reader.c b/src/reader.c
index 5989ab93..c8b0ca16 100644
--- a/src/reader.c
+++ b/src/reader.c
@@ -18,18 +18,19 @@
 
 #include "byte_source.h"
 #include "node.h"
+#include "serd_internal.h"
 #include "stack.h"
 #include "system.h"
 
-#include "serd_internal.h"
-
 #include <errno.h>
 #include <stdarg.h>
-#include <stdint.h>
 #include <stdio.h>
 #include <stdlib.h>
 #include <string.h>
 
+static SerdStatus
+serd_reader_prepare(SerdReader* reader);
+
 SerdStatus
 r_err(SerdReader* const reader, const SerdStatus st, const char* const fmt, ...)
 {
@@ -67,25 +68,6 @@ blank_id(SerdReader* const reader)
   return ref;
 }
 
-/** fread-like wrapper for getc (which is faster). */
-static size_t
-serd_file_read_byte(void* const  buf,
-                    const size_t size,
-                    const size_t nmemb,
-                    void* const  stream)
-{
-  (void)size;
-  (void)nmemb;
-
-  const int c = getc((FILE*)stream);
-  if (c == EOF) {
-    *((uint8_t*)buf) = 0;
-    return 0;
-  }
-  *((uint8_t*)buf) = (uint8_t)c;
-  return 1;
-}
-
 Ref
 push_node_padded(SerdReader* const  reader,
                  const size_t       maxlen,
@@ -170,9 +152,16 @@ read_statement(SerdReader* const reader)
   return read_n3_statement(reader);
 }
 
-static SerdStatus
-read_doc(SerdReader* const reader)
+SerdStatus
+serd_reader_read_document(SerdReader* const reader)
 {
+  if (!reader->source.prepared) {
+    SerdStatus st = serd_reader_prepare(reader);
+    if (st) {
+      return st;
+    }
+  }
+
   return ((reader->syntax == SERD_NQUADS) ? read_nquadsDoc(reader)
                                           : read_turtleTrigDoc(reader));
 }
@@ -221,6 +210,7 @@ serd_reader_free(SerdReader* const reader)
   pop_node(reader, reader->rdf_nil);
   pop_node(reader, reader->rdf_rest);
   pop_node(reader, reader->rdf_first);
+  serd_reader_finish(reader);
   serd_node_free(reader->default_graph);
 
 #ifdef SERD_STACK_CHECK
@@ -254,26 +244,6 @@ serd_reader_set_default_graph(SerdReader* const     reader,
   reader->default_graph = serd_node_copy(graph);
 }
 
-SerdStatus
-serd_reader_read_file(SerdReader* const reader, const char* const uri)
-{
-  char* const path = serd_parse_file_uri(uri, NULL);
-  if (!path) {
-    return SERD_ERR_BAD_ARG;
-  }
-
-  FILE* fd = serd_fopen(path, "rb");
-  if (!fd) {
-    serd_free(path);
-    return SERD_ERR_UNKNOWN;
-  }
-
-  SerdStatus ret = serd_reader_read_file_handle(reader, fd, path);
-  fclose(fd);
-  free(path);
-  return ret;
-}
-
 static SerdStatus
 skip_bom(SerdReader* const me)
 {
@@ -292,30 +262,45 @@ skip_bom(SerdReader* const me)
 }
 
 SerdStatus
-serd_reader_start_stream(SerdReader* const reader,
-                         FILE* const       file,
-                         const char* const name,
-                         const bool        bulk)
+serd_reader_start_stream(SerdReader* const         reader,
+                         const SerdReadFunc        read_func,
+                         const SerdStreamErrorFunc error_func,
+                         void* const               stream,
+                         const char* const         name,
+                         const size_t              page_size)
 {
-  return serd_reader_start_source_stream(reader,
-                                         bulk ? (SerdReadFunc)fread
-                                              : serd_file_read_byte,
-                                         (SerdStreamErrorFunc)ferror,
-                                         file,
-                                         name,
-                                         bulk ? SERD_PAGE_SIZE : 1);
+  return serd_byte_source_open_source(
+    &reader->source, read_func, error_func, NULL, stream, name, page_size);
 }
 
 SerdStatus
-serd_reader_start_source_stream(SerdReader* const         reader,
-                                const SerdReadFunc        read_func,
-                                const SerdStreamErrorFunc error_func,
-                                void* const               stream,
-                                const char* const         name,
-                                const size_t              page_size)
+serd_reader_start_file(SerdReader* reader, const char* uri, bool bulk)
 {
-  return serd_byte_source_open_source(
-    &reader->source, read_func, error_func, stream, name, page_size);
+  char* const path = serd_parse_file_uri(uri, NULL);
+  if (!path) {
+    return SERD_ERR_BAD_ARG;
+  }
+
+  FILE* fd = serd_fopen(path, "rb");
+  free(path);
+  if (!fd) {
+    return SERD_ERR_UNKNOWN;
+  }
+
+  return serd_byte_source_open_source(&reader->source,
+                                      bulk ? (SerdReadFunc)fread
+                                           : serd_file_read_byte,
+                                      (SerdStreamErrorFunc)ferror,
+                                      (SerdStreamCloseFunc)fclose,
+                                      fd,
+                                      uri,
+                                      bulk ? SERD_PAGE_SIZE : 1);
+}
+
+SerdStatus
+serd_reader_start_string(SerdReader* const reader, const char* const utf8)
+{
+  return serd_byte_source_open_string(&reader->source, utf8);
 }
 
 static SerdStatus
@@ -351,59 +336,7 @@ serd_reader_read_chunk(SerdReader* const reader)
 }
 
 SerdStatus
-serd_reader_end_stream(SerdReader* const reader)
+serd_reader_finish(SerdReader* const reader)
 {
   return serd_byte_source_close(&reader->source);
 }
-
-SerdStatus
-serd_reader_read_file_handle(SerdReader* const reader,
-                             FILE* const       file,
-                             const char* const name)
-{
-  return serd_reader_read_source(reader,
-                                 (SerdReadFunc)fread,
-                                 (SerdStreamErrorFunc)ferror,
-                                 file,
-                                 name,
-                                 SERD_PAGE_SIZE);
-}
-
-SerdStatus
-serd_reader_read_source(SerdReader* const         reader,
-                        const SerdReadFunc        source,
-                        const SerdStreamErrorFunc error,
-                        void* const               stream,
-                        const char* const         name,
-                        const size_t              page_size)
-{
-  SerdStatus st = serd_reader_start_source_stream(
-    reader, source, error, stream, name, page_size);
-
-  if (st || (st = serd_reader_prepare(reader))) {
-    serd_reader_end_stream(reader);
-    return st;
-  }
-
-  if ((st = read_doc(reader))) {
-    serd_reader_end_stream(reader);
-    return st;
-  }
-
-  return serd_reader_end_stream(reader);
-}
-
-SerdStatus
-serd_reader_read_string(SerdReader* const reader, const char* const utf8)
-{
-  serd_byte_source_open_string(&reader->source, utf8);
-
-  SerdStatus st = serd_reader_prepare(reader);
-  if (!st) {
-    st = read_doc(reader);
-  }
-
-  serd_byte_source_close(&reader->source);
-
-  return st;
-}
diff --git a/src/serdi.c b/src/serdi.c
index 0141b444..74ad358a 100644
--- a/src/serdi.c
+++ b/src/serdi.c
@@ -16,6 +16,7 @@
 
 #include "serd_config.h"
 #include "string_utils.h"
+#include "system.h"
 
 #include "serd/serd.h"
 
@@ -27,14 +28,8 @@
 #  include <io.h>
 #endif
 
-#if USE_POSIX_FADVISE && USE_FILENO
-#  include <fcntl.h>
-#endif
-
-#include <errno.h>
 #include <stdbool.h>
 #include <stdio.h>
-#include <stdlib.h>
 #include <string.h>
 
 #define SERDI_ERROR(msg) fprintf(stderr, "serdi: " msg)
@@ -132,22 +127,6 @@ quiet_error_sink(void* const handle, const SerdError* const e)
   return SERD_SUCCESS;
 }
 
-static FILE*
-serd_fopen(const char* const path, const char* const mode)
-{
-  FILE* fd = fopen(path, mode);
-  if (!fd) {
-    SERDI_ERRORF("failed to open file %s (%s)\n", path, strerror(errno));
-    return NULL;
-  }
-
-#if USE_POSIX_FADVISE && USE_FILENO
-  posix_fadvise(fileno(fd), 0, 0, POSIX_FADV_SEQUENTIAL | POSIX_FADV_NOREUSE);
-#endif
-
-  return fd;
-}
-
 static SerdWriterFlags
 choose_style(const SerdSyntax input_syntax,
              const SerdSyntax output_syntax,
@@ -186,25 +165,23 @@ main(int argc, char** argv)
     return print_usage(prog, true);
   }
 
-  FILE*       in_fd         = NULL;
   SerdSyntax  input_syntax  = (SerdSyntax)0;
   SerdSyntax  output_syntax = (SerdSyntax)0;
-  bool        from_file     = true;
+  bool        from_string   = false;
+  bool        from_stdin    = false;
   bool        ascii         = false;
   bool        bulk_read     = true;
   bool        bulk_write    = false;
   bool        full_uris     = false;
   bool        lax           = false;
   bool        quiet         = false;
-  const char* in_name       = NULL;
   const char* add_prefix    = NULL;
   const char* chop_prefix   = NULL;
   const char* root_uri      = NULL;
   int         a             = 1;
-  for (; a < argc && from_file && argv[a][0] == '-'; ++a) {
+  for (; a < argc && !from_string && argv[a][0] == '-'; ++a) {
     if (argv[a][1] == '\0') {
-      in_name = (const char*)"(stdin)";
-      in_fd   = stdin;
+      from_stdin = true;
       break;
     }
 
@@ -228,8 +205,7 @@ main(int argc, char** argv)
       } else if (opt == 'v') {
         return print_version();
       } else if (opt == 's') {
-        in_name   = "(string)";
-        from_file = false;
+        from_string = true;
         break;
       } else if (opt == 'c') {
         if (argv[a][o + 1] || ++a == argc) {
@@ -287,22 +263,9 @@ main(int argc, char** argv)
   _setmode(_fileno(stdout), _O_BINARY);
 #endif
 
-  char*       input_path = NULL;
-  const char* input      = (const char*)argv[a++];
-  if (from_file) {
-    in_name = in_name ? in_name : input;
-    if (!in_fd) {
-      if (!strncmp(input, "file:", 5)) {
-        input_path = serd_parse_file_uri(input, NULL);
-        input      = input_path;
-      }
-      if (!input || !(in_fd = serd_fopen(input, "rb"))) {
-        return 1;
-      }
-    }
-  }
+  const char* input = argv[a++];
 
-  if (!input_syntax && !(input_syntax = guess_syntax(in_name))) {
+  if (!input_syntax && !(input_syntax = guess_syntax(input))) {
     input_syntax = SERD_TRIG;
   }
 
@@ -319,7 +282,7 @@ main(int argc, char** argv)
   SerdNode* base = NULL;
   if (a < argc) { // Base URI given on command line
     base = serd_new_uri(SERD_STRING((const char*)argv[a]));
-  } else if (from_file && in_fd != stdin) { // Use input file URI
+  } else if (!from_string && !from_stdin) { // Use input file URI
     base = serd_new_file_uri(SERD_STRING(input), SERD_EMPTY_STRING());
   }
 
@@ -349,30 +312,31 @@ main(int argc, char** argv)
   serd_reader_add_blank_prefix(reader, add_prefix);
 
   SerdStatus st = SERD_SUCCESS;
-  if (!from_file) {
-    st = serd_reader_read_string(reader, input);
-  } else if (bulk_read) {
-    st = serd_reader_read_file_handle(reader, in_fd, in_name);
+  if (from_string) {
+    st = serd_reader_start_string(reader, input);
+  } else if (from_stdin) {
+    st = serd_reader_start_stream(reader,
+                                  serd_file_read_byte,
+                                  (SerdStreamErrorFunc)ferror,
+                                  stdin,
+                                  "(stdin)",
+                                  1);
   } else {
-    st = serd_reader_start_stream(reader, in_fd, in_name, false);
-    while (!st) {
-      st = serd_reader_read_chunk(reader);
-    }
-    serd_reader_end_stream(reader);
+    st = serd_reader_start_file(reader, input, bulk_read);
   }
 
+  if (!st) {
+    st = serd_reader_read_document(reader);
+  }
+
+  serd_reader_finish(reader);
   serd_reader_free(reader);
   serd_writer_finish(writer);
   serd_writer_free(writer);
   serd_env_free(env);
   serd_node_free(base);
-  free(input_path);
-
-  if (from_file) {
-    fclose(in_fd);
-  }
 
-  if (fclose(out_fd)) {
+  if (fclose(stdout)) {
     perror("serdi: write error");
     st = SERD_ERR_UNKNOWN;
   }
diff --git a/src/system.h b/src/system.h
index fae1cb84..7dae6a96 100644
--- a/src/system.h
+++ b/src/system.h
@@ -19,6 +19,7 @@
 
 #include "attributes.h"
 
+#include <stdint.h>
 #include <stdio.h>
 
 /// Open a file configured for fast sequential reading
@@ -42,4 +43,20 @@ serd_allocate_buffer(size_t size);
 void
 serd_free_aligned(void* ptr);
 
+/// Wrapper for getc that is compatible with SerdReadFunc
+static inline size_t
+serd_file_read_byte(void* buf, size_t size, size_t nmemb, void* stream)
+{
+  (void)size;
+  (void)nmemb;
+
+  const int c = getc((FILE*)stream);
+  if (c == EOF) {
+    *((uint8_t*)buf) = 0;
+    return 0;
+  }
+  *((uint8_t*)buf) = (uint8_t)c;
+  return 1;
+}
+
 #endif // SERD_SYSTEM_H
diff --git a/test/test_read_chunk.c b/test/test_read_chunk.c
index 8dd176e7..ce529b67 100644
--- a/test/test_read_chunk.c
+++ b/test/test_read_chunk.c
@@ -19,8 +19,7 @@
 #include "serd/serd.h"
 
 #include <assert.h>
-#include <stdbool.h>
-#include <stdio.h>
+#include <stddef.h>
 
 static size_t n_base      = 0;
 static size_t n_prefix    = 0;
@@ -80,21 +79,6 @@ on_end(void* handle, const SerdNode* node)
 int
 main(void)
 {
-  FILE* file = tmpfile();
-
-  fprintf(file,
-          "@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");
-
-  fseek(file, 0, SEEK_SET);
-
   SerdSink* sink = serd_sink_new(NULL, NULL);
   serd_sink_set_base_func(sink, on_base);
   serd_sink_set_prefix_func(sink, on_prefix);
@@ -104,22 +88,30 @@ main(void)
   SerdReader* reader = serd_reader_new(SERD_TURTLE, sink);
   assert(reader);
 
-  assert(reader);
-  assert(!serd_reader_start_stream(reader, file, NULL, true));
+  assert(!serd_reader_start_string(reader,
+                                   "@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"));
 
   assert(!serd_reader_read_chunk(reader) && n_prefix == 1);
   assert(!serd_reader_read_chunk(reader) && n_base == 1);
   assert(!serd_reader_read_chunk(reader) && n_statement == 3);
   assert(!serd_reader_read_chunk(reader) && n_statement == 5);
   assert(!serd_reader_read_chunk(reader) && n_statement == 6);
-  assert(!serd_reader_read_chunk(reader) && n_statement == 8 && n_end == 1);
+  assert(!serd_reader_read_chunk(reader) && n_statement == 8);
   assert(serd_reader_read_chunk(reader) == SERD_FAILURE);
+  assert(n_end == 1);
   assert(serd_reader_read_chunk(reader) == SERD_FAILURE);
+  assert(!serd_reader_finish(reader));
 
-  assert(!serd_reader_end_stream(reader));
   serd_reader_free(reader);
   serd_sink_free(sink);
-  fclose(file);
 
   return 0;
 }
diff --git a/test/test_reader_writer.c b/test/test_reader_writer.c
index ed1c87cb..d5669db0 100644
--- a/test/test_reader_writer.c
+++ b/test/test_reader_writer.c
@@ -107,7 +107,8 @@ test_read_chunks(void)
   assert(f);
   serd_sink_set_statement_func(sink, test_sink);
 
-  SerdStatus st = serd_reader_start_stream(reader, f, NULL, false);
+  SerdStatus st = serd_reader_start_stream(
+    reader, (SerdReadFunc)fread, (SerdStreamErrorFunc)ferror, f, NULL, 1);
   assert(st == SERD_SUCCESS);
 
   // Write two statement separated by null characters
@@ -167,13 +168,14 @@ test_read_string(void)
   serd_sink_set_statement_func(sink, test_sink);
 
   // Test reading a string that ends exactly at the end of input (no newline)
-  const SerdStatus st =
-    serd_reader_read_string(reader,
-                            "<http://example.org/s> <http://example.org/p> "
-                            "<http://example.org/o> .");
+  assert(
+    !serd_reader_start_string(reader,
+                              "<http://example.org/s> <http://example.org/p> "
+                              "<http://example.org/o> ."));
 
-  assert(!st);
+  assert(!serd_reader_read_document(reader));
   assert(rt->n_statements == 1);
+  assert(!serd_reader_finish(reader));
 
   serd_reader_free(reader);
   serd_sink_free(sink);
@@ -273,12 +275,12 @@ test_writer(const char* const path)
 static void
 test_reader(const char* path)
 {
-  ReaderTest*     rt     = (ReaderTest*)calloc(1, sizeof(ReaderTest));
-  SerdSink* const sink   = serd_sink_new(rt, NULL);
+  ReaderTest      rt     = {0, NULL};
+  SerdSink* const sink   = serd_sink_new(&rt, NULL);
   SerdReader*     reader = serd_reader_new(SERD_TURTLE, sink);
 
-  assert(sink);
   assert(reader);
+  assert(sink);
   serd_sink_set_statement_func(sink, test_sink);
 
   SerdNode* g = serd_new_uri(SERD_STRING("http://example.org/"));
@@ -296,17 +298,16 @@ test_reader(const char* path)
 
   serd_node_free(g);
 
-  assert(serd_reader_read_file(reader, "http://notafile"));
-  assert(serd_reader_read_file(reader, "file:///better/not/exist"));
-  assert(serd_reader_read_file(reader, "file://"));
+  assert(serd_reader_start_file(reader, "http://notafile", false));
+  assert(serd_reader_start_file(reader, "file://invalid", false));
+  assert(serd_reader_start_file(reader, "file:///nonexistant", false));
 
-  const SerdStatus st = serd_reader_read_file(reader, path);
-  assert(!st);
-  assert(rt->n_statements == 6);
-  assert(rt->graph && serd_node_string(rt->graph) &&
-         !strcmp(serd_node_string(rt->graph), "http://example.org/"));
-
-  assert(serd_reader_read_string(reader, "This isn't Turtle at all."));
+  assert(!serd_reader_start_file(reader, path, true));
+  assert(!serd_reader_read_document(reader));
+  assert(rt.n_statements == 6);
+  assert(rt.graph && serd_node_string(rt.graph) &&
+         !strcmp(serd_node_string(rt.graph), "http://example.org/"));
+  serd_reader_finish(reader);
 
   // A read of a big page hits EOF then fails to read chunks immediately
   {
@@ -316,25 +317,30 @@ test_reader(const char* path)
     fflush(temp);
     fseek(temp, 0L, SEEK_SET);
 
-    serd_reader_start_stream(reader, temp, NULL, true);
+    serd_reader_start_stream(reader,
+                             (SerdReadFunc)fread,
+                             (SerdStreamErrorFunc)ferror,
+                             temp,
+                             NULL,
+                             4096);
 
     assert(serd_reader_read_chunk(reader) == SERD_SUCCESS);
     assert(serd_reader_read_chunk(reader) == SERD_FAILURE);
     assert(serd_reader_read_chunk(reader) == SERD_FAILURE);
 
-    serd_reader_end_stream(reader);
+    serd_reader_finish(reader);
     fclose(temp);
   }
 
   // A byte-wise reader that hits EOF once then continues (like a socket)
   {
     size_t n_reads = 0;
-    serd_reader_start_source_stream(reader,
-                                    (SerdReadFunc)eof_test_read,
-                                    (SerdStreamErrorFunc)eof_test_error,
-                                    &n_reads,
-                                    NULL,
-                                    1);
+    serd_reader_start_stream(reader,
+                             (SerdReadFunc)eof_test_read,
+                             (SerdStreamErrorFunc)eof_test_error,
+                             &n_reads,
+                             NULL,
+                             1);
 
     assert(serd_reader_read_chunk(reader) == SERD_SUCCESS);
     assert(serd_reader_read_chunk(reader) == SERD_FAILURE);
@@ -344,7 +350,6 @@ test_reader(const char* path)
 
   serd_reader_free(reader);
   serd_sink_free(sink);
-  free(rt);
 }
 
 int