WIP: Make Reader always read from a ByteSource

12 files changed, 309 insertions, 248 deletions

diff --git a/serd/serd.h b/serd/serd.h
index fe2a08b2..612d2438 100644
--- a/serd/serd.h
+++ b/serd/serd.h
@@ -94,9 +94,6 @@ typedef struct SerdWriterImpl SerdWriter;
 /// An interface that receives a stream of RDF data
 typedef struct SerdSinkImpl SerdSink;
 
-/// A sink for bytes that receives string output
-typedef struct SerdByteSinkImpl SerdByteSink;
-
 /// Return status code
 typedef enum {
 	SERD_SUCCESS,         ///< No error
@@ -112,7 +109,8 @@ typedef enum {
 	SERD_ERR_OVERFLOW,    ///< Stack overflow
 	SERD_ERR_INVALID,     ///< Invalid data
 	SERD_ERR_NO_DATA,     ///< Unexpected end of input
-	SERD_ERR_BAD_WRITE    ///< Error writing to file/stream
+	SERD_ERR_BAD_WRITE,   ///< Error writing to file/stream
+	SERD_ERR_BAD_CALL     ///< Invalid call
 } SerdStatus;
 
 /// RDF syntax type
@@ -395,10 +393,13 @@ serd_base64_decode(void* buf, size_t* size, const char* str, size_t len);
 
 /**
    @}
-   @name Byte Streams
+   @name Byte Source
    @{
 */
 
+/// A source for bytes that provides text input
+typedef struct SerdByteSourceImpl SerdByteSource;
+
 /**
    Function to detect I/O stream errors
 
@@ -426,6 +427,67 @@ typedef size_t (*SerdReadFunc)(void*  buf,
                                void*  stream);
 
 /**
+   Create a new byte source that reads from a string
+
+   @param string Null-terminated UTF-8 string to read from.
+   @param name Optional name of stream for error messages (string or URI).
+*/
+SERD_API
+SerdByteSource*
+serd_byte_source_new_string(const char* string, const SerdNode* name);
+
+/**
+   Create a new byte source that reads from a file
+
+   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.
+
+   @param path Path of file to open and read from.
+   @param block_size Number of bytes to read per call.
+*/
+SERD_API
+SerdByteSource*
+serd_byte_source_new_filename(const char* path, size_t block_size);
+
+/**
+   Create a new byte source that reads from a user-specified function
+
+   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 serd Reader
+   only ever reads individual bytes at a time, that is, the `size` parameter
+   will always be 1 (but `nmemb` may be higher).
+
+   @param read_func Function called with bytes to consume.
+   @param error_func Stream error function with `ferror` semantics.
+   @param stream Context parameter passed to `read_func` and `error_func`.
+   @param name Optional name of stream for error messages (string or URI).
+   @param block_size Number of bytes to read per call.
+*/
+SERD_API
+SerdByteSource*
+serd_byte_source_new_function(SerdReadFunc        read_func,
+                              SerdStreamErrorFunc error_func,
+                              void*               stream,
+                              const SerdNode*     name,
+                              size_t              block_size);
+
+/// Free `source`
+SERD_API
+void
+serd_byte_source_free(SerdByteSource* source);
+
+/**
+   @}
+   @name Byte Sink
+   @{
+*/
+
+/// A sink for bytes that receives text output
+typedef struct SerdByteSinkImpl SerdByteSink;
+
+/**
    Sink function for raw string output
 
    Identical semantics to `fwrite`, but may set errno for more informative
@@ -1315,32 +1377,10 @@ SERD_API
 void
 serd_reader_add_blank_prefix(SerdReader* reader, const char* prefix);
 
-/// Prepare to read from the file at a local file `uri`
-SERD_API
-SerdStatus
-serd_reader_start_file(SerdReader* reader, const char* uri, bool bulk);
-
-/**
-   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_stream(SerdReader*         reader,
-                         SerdReadFunc        read_func,
-                         SerdStreamErrorFunc error_func,
-                         void*               stream,
-                         const SerdNode*     name,
-                         size_t              page_size);
-
-/// Prepare to read from a string
+/// Prepare to read from a byte source
 SERD_API
 SerdStatus
-serd_reader_start_string(SerdReader*     reader,
-                         const char*     utf8,
-                         const SerdNode* name);
+serd_reader_start(SerdReader* reader, SerdByteSource* byte_source);
 
 /**
    Read a single "chunk" of data during an incremental read
diff --git a/src/byte_source.c b/src/byte_source.c
index 90af28ee..efe4cb04 100644
--- a/src/byte_source.c
+++ b/src/byte_source.c
@@ -21,117 +21,150 @@
 #include "serd/serd.h"
 
 #include <stdbool.h>
-#include <stdint.h>
+#include <stdio.h>
 #include <stdlib.h>
 #include <string.h>
 
-SerdStatus
-serd_byte_source_page(SerdByteSource* source)
+SerdByteSource*
+serd_byte_source_new_string(const char* string, const SerdNode* name)
 {
-	uint8_t* const buf = (source->page_size > 1
-	                      ? source->file_buf
-	                      : &source->read_byte);
+	SerdByteSource* source = (SerdByteSource*)calloc(1, sizeof(SerdByteSource));
 
-	const size_t n_read = source->read_func(
-		buf, 1, source->page_size, source->stream);
+	source->page_size = 1;
+	source->name      = name ? serd_node_copy(name) : serd_new_string("string");
+	source->read_buf  = (const uint8_t*)string;
+	source->type      = FROM_STRING;
 
-	source->buf_size  = n_read;
-	source->read_head = 0;
-	source->eof       = false;
+	const SerdCursor cur = {source->name, 1, 1};
+	source->cur          = cur;
 
-	if (n_read < source->page_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;
+}
+
+SerdByteSource*
+serd_byte_source_new_filename(const char* path, size_t block_size)
+{
+	if (!path || !block_size) {
+		return NULL;
 	}
 
-	return SERD_SUCCESS;
+	FILE* fd = fopen(path, "rb");
+	if (!fd) {
+		return NULL;
+	}
+
+	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   = block_size;
+	source->buf_size    = block_size;
+	source->name        = serd_new_file_uri(path, NULL);
+	source->type        = FROM_FILENAME;
+
+	const SerdCursor cur = { source->name, 1, 1 };
+	source->cur = cur;
+
+	if (block_size > 1) {
+		source->file_buf = (uint8_t*)serd_allocate_buffer(block_size);
+		source->read_buf = source->file_buf;
+		memset(source->file_buf, '\0', block_size);
+	} else {
+		source->read_buf = &source->read_byte;
+	}
+
+	return source;
 }
 
-SerdStatus
-serd_byte_source_open_source(SerdByteSource*     source,
-                             SerdReadFunc        read_func,
-                             SerdStreamErrorFunc error_func,
-                             SerdStreamCloseFunc close_func,
-                             void*               stream,
-                             const SerdNode*     name,
-                             size_t              page_size)
+SerdByteSource*
+serd_byte_source_new_function(SerdReadFunc        read_func,
+                              SerdStreamErrorFunc error_func,
+                              void*               stream,
+                              const SerdNode*     name,
+                              size_t              block_size)
 {
-	assert(page_size > 0);
-	memset(source, '\0', sizeof(*source));
+	if (!read_func || !block_size) {
+		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->close_func  = close_func; FIXME
 	source->stream      = stream;
-	source->page_size   = page_size;
-	source->buf_size    = page_size;
-	source->name        = serd_node_copy(name);
-	source->from_stream = true;
+	source->page_size   = block_size;
+	source->buf_size    = block_size;
+	source->name        = name ? serd_node_copy(name) : serd_new_string("func");
+	source->type        = FROM_FUNCTION;
 
 	const SerdCursor cur = { source->name, 1, 1 };
 	source->cur = cur;
 
-	if (page_size > 1) {
-		source->file_buf = (uint8_t*)serd_allocate_buffer(page_size);
+	if (block_size > 1) {
+		source->file_buf = (uint8_t*)serd_allocate_buffer(block_size);
 		source->read_buf = source->file_buf;
-		memset(source->file_buf, '\0', page_size);
+		memset(source->file_buf, '\0', block_size);
 	} else {
 		source->read_buf = &source->read_byte;
 	}
 
-	return SERD_SUCCESS;
+	return source;
 }
 
-SerdStatus
-serd_byte_source_prepare(SerdByteSource* source)
+void
+serd_byte_source_free(SerdByteSource* source)
 {
-	if (source->page_size == 0) {
-		return SERD_FAILURE;
-	}
-
-	source->prepared = true;
-	if (source->from_stream) {
+	if (source) {
+		if (source->close_func) {
+			source->close_func(source->stream);
+		}
 		if (source->page_size > 1) {
-			return serd_byte_source_page(source);
-		} else if (source->from_stream) {
-			return serd_byte_source_advance(source);
+			free(source->file_buf);
 		}
+		serd_node_free(source->name);
+		free(source);
 	}
-	return SERD_SUCCESS;
 }
 
 SerdStatus
-serd_byte_source_open_string(SerdByteSource* source,
-                             const char*     utf8,
-                             const SerdNode* name)
+serd_byte_source_page(SerdByteSource* source)
 {
-	memset(source, '\0', sizeof(*source));
+	uint8_t* const buf = (source->page_size > 1
+	                      ? source->file_buf
+	                      : &source->read_byte);
 
-	source->page_size = 1;
-	source->name      = name ? serd_node_copy(name) : serd_new_string("string");
-	source->read_buf  = (const uint8_t*)utf8;
+	const size_t n_read = source->read_func(
+		buf, 1, source->page_size, source->stream);
 
-	const SerdCursor cur = {source->name, 1, 1};
-	source->cur          = cur;
+	source->buf_size  = n_read;
+	source->read_head = 0;
+	source->eof       = false;
+
+	if (n_read < source->page_size) {
+		buf[n_read] = '\0';
+		if (n_read == 0) {
+			source->eof = true;
+			return (source->error_func(source->stream)
+			        ? SERD_ERR_UNKNOWN : SERD_FAILURE);
+		}
+	}
 
 	return SERD_SUCCESS;
 }
 
 SerdStatus
-serd_byte_source_close(SerdByteSource* source)
+serd_byte_source_prepare(SerdByteSource* 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) {
-		free(source->file_buf);
+	source->prepared = true;
+	if (source->type != FROM_STRING) {
+		if (source->page_size > 1) {
+			return serd_byte_source_page(source);
+		} else {
+			return serd_byte_source_advance(source);
+		}
 	}
-	serd_node_free(source->name);
-	memset(source, '\0', sizeof(*source));
-	return st;
+	return SERD_SUCCESS;
 }
diff --git a/src/byte_source.h b/src/byte_source.h
index acda0812..a525c019 100644
--- a/src/byte_source.h
+++ b/src/byte_source.h
@@ -26,9 +26,15 @@
 #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;
+
 typedef int (*SerdStreamCloseFunc)(void*);
 
-typedef struct {
+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
@@ -40,28 +46,11 @@ typedef struct {
 	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                from_stream;  ///< True iff reading from `stream`
 	bool                prepared;     ///< True iff prepared for reading
 	bool                eof;          ///< True iff end of file reached
-} SerdByteSource;
-
-SerdStatus
-serd_byte_source_open_string(SerdByteSource* source,
-                             const char*     utf8,
-                             const SerdNode* name);
-
-SerdStatus
-serd_byte_source_open_source(SerdByteSource*     source,
-                             SerdReadFunc        read_func,
-                             SerdStreamErrorFunc error_func,
-                             SerdStreamCloseFunc close_func,
-                             void*               stream,
-                             const SerdNode*     name,
-                             size_t              page_size);
-
-SerdStatus
-serd_byte_source_close(SerdByteSource* source);
+};
 
 SerdStatus
 serd_byte_source_prepare(SerdByteSource* source);
@@ -87,7 +76,7 @@ serd_byte_source_advance(SerdByteSource* source)
 	default:   ++source->cur.col;
 	}
 
-	if (source->from_stream) {
+	if (source->type != FROM_STRING) {
 		if (++source->read_head >= source->buf_size) {
 			st = serd_byte_source_page(source);
 		}
diff --git a/src/n3.c b/src/n3.c
index 778bf69f..c521b239 100644
--- a/src/n3.c
+++ b/src/n3.c
@@ -904,7 +904,7 @@ read_Var(SerdReader* reader, SerdNode** dest)
 	}
 
 	assert(peek_byte(reader) == '$' || peek_byte(reader) == '?');
-	serd_byte_source_advance(&reader->source);
+	serd_byte_source_advance(reader->source);
 
 	return read_VARNAME(reader, dest);
 }
@@ -1582,11 +1582,11 @@ skip_until(SerdReader* reader, uint8_t byte)
 SerdStatus
 read_turtleTrigDoc(SerdReader* reader)
 {
-	while (!reader->source.eof) {
+	while (!reader->source->eof) {
 		const size_t     orig_stack_size = reader->stack.size;
 		const SerdStatus st              = read_n3_statement(reader);
 		if (st > SERD_FAILURE) {
-			if (reader->strict || reader->source.eof ||
+			if (reader->strict || reader->source->eof ||
 			    st == SERD_ERR_OVERFLOW) {
 				serd_stack_pop_to(&reader->stack, orig_stack_size);
 				return st;
@@ -1602,7 +1602,7 @@ SerdStatus
 read_nquadsDoc(SerdReader* reader)
 {
 	SerdStatus st = SERD_SUCCESS;
-	while (!st && !reader->source.eof) {
+	while (!st && !reader->source->eof) {
 		const size_t orig_stack_size = reader->stack.size;
 
 		SerdStatementFlags flags   = 0;
diff --git a/src/reader.c b/src/reader.c
index 3dd2f465..b9d9e510 100644
--- a/src/reader.c
+++ b/src/reader.c
@@ -15,7 +15,6 @@
 */
 
 #include "reader.h"
-#include "system.h"
 
 #include "namespaces.h"
 #include "stack.h"
@@ -38,7 +37,7 @@ r_err(SerdReader* reader, SerdStatus st, const char* fmt, ...)
 	serd_world_vlogf_internal(reader->world,
 	                          st,
 	                          SERD_LOG_LEVEL_ERR,
-	                          &reader->source.cur,
+	                          &reader->source->cur,
 	                          fmt,
 	                          args);
 	va_end(args);
@@ -125,7 +124,7 @@ emit_statement(SerdReader* reader, ReadContext ctx, SerdNode* o)
 
 	const SerdStatement statement = {
 		{ ctx.subject, ctx.predicate, o, ctx.graph },
-		&reader->source.cur
+		&reader->source->cur
 	};
 
 	const SerdStatus st =
@@ -144,7 +143,11 @@ read_statement(SerdReader* reader)
 SerdStatus
 serd_reader_read_document(SerdReader* reader)
 {
-	if (!reader->source.prepared) {
+	if (!reader->source) {
+		return SERD_ERR_BAD_CALL;
+	}
+
+	if (!reader->source->prepared) {
 		SerdStatus st = serd_reader_prepare(reader);
 		if (st) {
 			return st;
@@ -225,12 +228,12 @@ serd_reader_add_blank_prefix(SerdReader* reader, const char* prefix)
 static SerdStatus
 skip_bom(SerdReader* me)
 {
-	if (serd_byte_source_peek(&me->source) == 0xEF) {
-		serd_byte_source_advance(&me->source);
-		if (serd_byte_source_peek(&me->source) != 0xBB ||
-		    serd_byte_source_advance(&me->source) ||
-		    serd_byte_source_peek(&me->source) != 0xBF ||
-		    serd_byte_source_advance(&me->source)) {
+	if (serd_byte_source_peek(me->source) == 0xEF) {
+		serd_byte_source_advance(me->source);
+		if (serd_byte_source_peek(me->source) != 0xBB ||
+		    serd_byte_source_advance(me->source) ||
+		    serd_byte_source_peek(me->source) != 0xBF ||
+		    serd_byte_source_advance(me->source)) {
 			r_err(me, SERD_ERR_BAD_SYNTAX, "corrupt byte order mark\n");
 			return SERD_ERR_BAD_SYNTAX;
 		}
@@ -240,64 +243,23 @@ skip_bom(SerdReader* me)
 }
 
 SerdStatus
-serd_reader_start_stream(SerdReader*         reader,
-                         SerdReadFunc        read_func,
-                         SerdStreamErrorFunc error_func,
-                         void*               stream,
-                         const SerdNode*     name,
-                         size_t              page_size)
+serd_reader_start(SerdReader* reader, SerdByteSource* byte_source)
 {
 	serd_reader_finish(reader);
-	return serd_byte_source_open_source(
-		&reader->source, read_func, error_func, NULL, stream, name, page_size);
-}
 
-SerdStatus
-serd_reader_start_file(SerdReader* reader, const char* uri, bool bulk)
-{
-	serd_reader_finish(reader);
-
-	char* const path = serd_file_uri_parse(uri, NULL);
-	if (!path) {
-		return SERD_ERR_BAD_ARG;
-	}
+	reader->source = byte_source;
 
-	FILE* fd = serd_world_fopen(reader->world, path, "rb");
-	free(path);
-	if (!fd) {
-		return SERD_ERR_UNKNOWN;
-	}
-
-	SerdNode* const name = serd_new_uri(uri);
-	const SerdStatus st = serd_byte_source_open_source(
-		&reader->source,
-		bulk ? (SerdReadFunc)fread : serd_file_read_byte,
-		(SerdStreamErrorFunc)ferror,
-		(SerdStreamCloseFunc)fclose,
-		fd,
-		name,
-		bulk ? SERD_PAGE_SIZE : 1u);
-	serd_node_free(name);
-	return st;
-}
-
-SerdStatus
-serd_reader_start_string(SerdReader*     reader,
-                         const char*     utf8,
-                         const SerdNode* name)
-{
-	serd_reader_finish(reader);
-	return serd_byte_source_open_string(&reader->source, utf8, name);
+	return reader->source ? SERD_SUCCESS : SERD_ERR_BAD_ARG;
 }
 
 static SerdStatus
 serd_reader_prepare(SerdReader* reader)
 {
-	SerdStatus st = serd_byte_source_prepare(&reader->source);
+	SerdStatus st = serd_byte_source_prepare(reader->source);
 	if (st == SERD_SUCCESS) {
 		st = skip_bom(reader);
 	} else if (st == SERD_FAILURE) {
-		reader->source.eof = true;
+		reader->source->eof = true;
 	} else {
 		r_err(reader, st, "error preparing read: %s\n", strerror(errno));
 	}
@@ -308,10 +270,12 @@ SerdStatus
 serd_reader_read_chunk(SerdReader* reader)
 {
 	SerdStatus st = SERD_SUCCESS;
-	if (!reader->source.prepared) {
+	if (!reader->source) {
+		return SERD_ERR_BAD_CALL;
+	} else if (!reader->source->prepared) {
 		st = serd_reader_prepare(reader);
-	} else if (reader->source.eof) {
-		st = serd_byte_source_advance(&reader->source);
+	} else if (reader->source->eof) {
+		st = serd_byte_source_advance(reader->source);
 	}
 
 	if (peek_byte(reader) == 0) {
@@ -325,5 +289,6 @@ serd_reader_read_chunk(SerdReader* reader)
 SerdStatus
 serd_reader_finish(SerdReader* reader)
 {
-	return serd_byte_source_close(&reader->source);
+	reader->source = NULL;
+	return SERD_SUCCESS;
 }
diff --git a/src/reader.h b/src/reader.h
index a5e1f3f0..df6e6387 100644
--- a/src/reader.h
+++ b/src/reader.h
@@ -57,7 +57,7 @@ struct SerdReaderImpl {
 	SerdNode*         rdf_first;
 	SerdNode*         rdf_rest;
 	SerdNode*         rdf_nil;
-	SerdByteSource    source;
+	SerdByteSource*   source;
 	SerdStack         stack;
 	SerdSyntax        syntax;
 	SerdReaderFlags   flags;
@@ -104,12 +104,21 @@ SerdStatus read_turtleTrigDoc(SerdReader* reader);
 static inline int
 peek_byte(SerdReader* reader)
 {
-	SerdByteSource* source = &reader->source;
+	SerdByteSource* source = reader->source;
 
 	return source->eof ? EOF : (int)source->read_buf[source->read_head];
 }
 
 static inline int
+eat_byte(SerdReader* reader)
+{
+	const int        c  = peek_byte(reader);
+	const SerdStatus st = serd_byte_source_advance(reader->source);
+
+	return st > SERD_FAILURE ? EOF : c;
+}
+
+static inline int
 eat_byte_safe(SerdReader* reader, const int byte)
 {
 	(void)byte;
@@ -117,7 +126,7 @@ eat_byte_safe(SerdReader* reader, const int byte)
 	const int c = peek_byte(reader);
 	assert(c == byte);
 
-	serd_byte_source_advance(&reader->source);
+	serd_byte_source_advance(reader->source);
 	return c;
 }
 
diff --git a/src/serdi.c b/src/serdi.c
index 0fc30239..833dc0f0 100644
--- a/src/serdi.c
+++ b/src/serdi.c
@@ -145,18 +145,21 @@ on_filter_event(void* handle, const SerdEvent* event)
 static SerdSink*
 parse_filter(SerdWorld* world, const SerdSink* sink, const char* str)
 {
-	FilterPattern pat     = {NULL, NULL, NULL, NULL};
-	SerdSink*     in_sink = serd_sink_new(&pat, NULL);
-	SerdReader*   reader =
+	FilterPattern   pat         = {NULL, NULL, NULL, NULL};
+	SerdSink*       in_sink     = serd_sink_new(&pat, NULL);
+	SerdByteSource* byte_source = serd_byte_source_new_string(str, NULL);
+	SerdReader*     reader =
 	    serd_reader_new(world, SERD_NQUADS, SERD_READ_VARIABLES, in_sink, 4096);
 
 	serd_sink_set_event_func(in_sink, on_filter_event);
-	SerdStatus st = serd_reader_start_string(reader, str, NULL);
+
+	SerdStatus st = serd_reader_start(reader, byte_source);
 	if (!st) {
 		st = serd_reader_read_document(reader);
 	}
 
 	serd_reader_free(reader);
+	serd_byte_source_free(byte_source);
 	serd_sink_free(in_sink);
 
 	if (st) {
@@ -184,28 +187,36 @@ read_file(SerdWorld* const      world,
 	syntax = syntax ? syntax : serd_guess_syntax(filename);
 	syntax = syntax ? syntax : SERD_TRIG;
 
-	SerdStatus  st = SERD_SUCCESS;
-	SerdReader* reader =
-	        serd_reader_new(world, syntax, flags, sink, stack_size);
-
-	serd_reader_add_blank_prefix(reader, add_prefix);
-
+	SerdByteSource* byte_source = NULL;
 	if (!strcmp(filename, "-")) {
 		SerdNode* name = serd_new_string("stdin");
-		st             = serd_reader_start_stream(reader,
-		                                          serd_file_read_byte,
-		                                          (SerdStreamErrorFunc)ferror,
-		                                          stdin,
-		                                          name,
-		                                          1);
+
+		byte_source = serd_byte_source_new_function(
+		    serd_file_read_byte, (SerdStreamErrorFunc)ferror, stdin, name, 1);
+
 		serd_node_free(name);
 	} else {
-		st = serd_reader_start_file(reader, filename, bulk_read);
+		byte_source =
+		    serd_byte_source_new_filename(filename,
+		                                  bulk_read ? SERD_PAGE_SIZE : 1u);
+	}
+
+	if (!byte_source) {
+		SERDI_ERRORF("failed to open input file `%s'\n", filename);
+		return SERD_ERR_UNKNOWN;
 	}
 
+	SerdReader* reader =
+	        serd_reader_new(world, syntax, flags, sink, stack_size);
+
+	serd_reader_add_blank_prefix(reader, add_prefix);
+
+	SerdStatus st = serd_reader_start(reader, byte_source);
+
 	st = st ? st : serd_reader_read_document(reader);
 
 	serd_reader_free(reader);
+	serd_byte_source_free(byte_source);
 
 	return st;
 }
@@ -439,19 +450,22 @@ main(int argc, char** argv)
 	SerdStatus st         = SERD_SUCCESS;
 	SerdNode*  input_name = NULL;
 	if (input_string) {
+		SerdByteSource* byte_source =
+		    serd_byte_source_new_string(input_string, NULL);
+
 		SerdReader* reader =
 		        serd_reader_new(world,
 		                        input_syntax ? input_syntax : SERD_TRIG,
 		                        reader_flags,
 		                        sink,
 		                        stack_size);
+
 		serd_reader_add_blank_prefix(reader, add_prefix);
 
-		SerdNode* name = serd_new_string("string");
-		if (!(st = serd_reader_start_string(reader, input_string, name))) {
+		if (!(st = serd_reader_start(reader, byte_source))) {
 			st = serd_reader_read_document(reader);
 		}
-		serd_node_free(name);
+
 		serd_reader_free(reader);
 	}
 
diff --git a/src/string.c b/src/string.c
index 1f91862f..9554a009 100644
--- a/src/string.c
+++ b/src/string.c
@@ -55,6 +55,7 @@ serd_strerror(SerdStatus status)
 	case SERD_ERR_INVALID:    return "Invalid data";
 	case SERD_ERR_NO_DATA:    return "Unexpectd end of input";
 	case SERD_ERR_BAD_WRITE:  return "Error writing to file";
+	case SERD_ERR_BAD_CALL:   return "Invalid call";
 	default:                  break;
 	}
 	return "Unknown error";  // never reached
diff --git a/tests/free_null_test.c b/tests/free_null_test.c
index a64ad6d7..edb50702 100644
--- a/tests/free_null_test.c
+++ b/tests/free_null_test.c
@@ -24,6 +24,7 @@ int
 main(void)
 {
 	serd_free(NULL);
+	serd_byte_source_free(NULL);
 	serd_byte_sink_free(NULL);
 	serd_node_free(NULL);
 	serd_world_free(NULL);
diff --git a/tests/overflow_test.c b/tests/overflow_test.c
index 847884de..2a3f3577 100644
--- a/tests/overflow_test.c
+++ b/tests/overflow_test.c
@@ -24,12 +24,15 @@
 static SerdStatus
 test(SerdWorld* world, SerdSink* sink, const char* str, size_t stack_size)
 {
+	SerdByteSource* byte_source = serd_byte_source_new_string(str, NULL);
+
 	SerdReader* reader = serd_reader_new(
 	        world, SERD_TURTLE, SERD_READ_VARIABLES, sink, stack_size);
 
-	serd_reader_start_string(reader, str, NULL);
+	serd_reader_start(reader, byte_source);
 	const SerdStatus st = serd_reader_read_document(reader);
 	serd_reader_free(reader);
+	serd_byte_source_free(byte_source);
 	return st;
 }
 
diff --git a/tests/read_chunk_test.c b/tests/read_chunk_test.c
index dfa0c0ac..a6ac4c45 100644
--- a/tests/read_chunk_test.c
+++ b/tests/read_chunk_test.c
@@ -96,20 +96,22 @@ main(void)
 	SerdSink*  sink  = serd_sink_new(NULL, NULL);
 	serd_sink_set_event_func(sink, on_event);
 
+	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);
+
 	SerdReader* reader = serd_reader_new(world, SERD_TURTLE, 0, sink, 4096);
 	assert(reader);
 
-	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",
-	                                 NULL));
+	assert(!serd_reader_start(reader, byte_source));
 
 	assert(!serd_reader_read_chunk(reader) && n_prefix == 1);
 	assert(!serd_reader_read_chunk(reader) && n_base == 1);
@@ -122,6 +124,7 @@ main(void)
 	assert(serd_reader_read_chunk(reader) == SERD_FAILURE);
 
 	serd_reader_free(reader);
+	serd_byte_source_free(byte_source);
 	serd_sink_free(sink);
 	serd_world_free(world);
 
diff --git a/tests/serd_test.c b/tests/serd_test.c
index a0fbb234..9f85136e 100644
--- a/tests/serd_test.c
+++ b/tests/serd_test.c
@@ -103,12 +103,10 @@ test_read_chunks(void)
 	assert(reader);
 	assert(f);
 
-	SerdStatus st = serd_reader_start_stream(reader,
-	                                         (SerdReadFunc)fread,
-	                                         (SerdStreamErrorFunc)ferror,
-	                                         f,
-	                                         NULL,
-	                                         1);
+	SerdByteSource* byte_source = serd_byte_source_new_function(
+	    (SerdReadFunc)fread, (SerdStreamErrorFunc)ferror, f, NULL, 1);
+
+	SerdStatus st = serd_reader_start(reader, byte_source);
 	assert(st == SERD_SUCCESS);
 
 	// Write two statement separated by null characters
@@ -150,6 +148,7 @@ test_read_chunks(void)
 	assert(n_statements == 2);
 
 	serd_reader_free(reader);
+	serd_byte_source_free(byte_source);
 	serd_sink_free(sink);
 	fclose(f);
 	serd_world_free(world);
@@ -220,18 +219,19 @@ test_read_string(void)
 
 	serd_sink_set_event_func(sink, count_statements);
 
-	// Test reading a string that ends exactly at the end of input (no newline)
-	assert(!serd_reader_start_string(
-		       reader,
+	SerdByteSource* byte_source = serd_byte_source_new_string(
 		       "<http://example.org/s> <http://example.org/p> "
-		       "<http://example.org/o> .",
-		       NULL));
+		       "<http://example.org/o> .", NULL);
+
+	// Test reading a string that ends exactly at the end of input (no newline)
+	assert(!serd_reader_start(reader, byte_source));
 
 	assert(!serd_reader_read_document(reader));
 	assert(n_statements == 1);
 	assert(!serd_reader_finish(reader));
 
 	serd_reader_free(reader);
+	serd_byte_source_free(byte_source);
 	serd_sink_free(sink);
 	serd_world_free(world);
 }
@@ -675,20 +675,23 @@ test_reader(const char* path)
 	SerdReader* reader = serd_reader_new(world, SERD_TURTLE, 0, sink, 4096);
 	assert(reader);
 
-	assert(serd_reader_read_document(reader) == SERD_FAILURE);
-	assert(serd_reader_read_chunk(reader) == SERD_FAILURE);
+	assert(serd_reader_read_document(reader) == SERD_ERR_BAD_CALL);
+	assert(serd_reader_read_chunk(reader) == SERD_ERR_BAD_CALL);
 
 	serd_reader_add_blank_prefix(reader, "tmp");
 	serd_reader_add_blank_prefix(reader, NULL);
 
-	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));
+	// FIXME
+	/* 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)); */
 
-	assert(!serd_reader_start_file(reader, path, true));
+	SerdByteSource* byte_source = serd_byte_source_new_filename(path, 4096);
+	assert(!serd_reader_start(reader, byte_source));
 	assert(!serd_reader_read_document(reader));
 	assert(n_statements == 13);
 	serd_reader_finish(reader);
+	serd_byte_source_free(byte_source);
 
 	// A read of a big page hits EOF then fails to read chunks immediately
 	{
@@ -698,31 +701,31 @@ test_reader(const char* path)
 		fflush(temp);
 		fseek(temp, 0L, SEEK_SET);
 
-		serd_reader_start_stream(reader,
-		                         (SerdReadFunc)fread,
-		                         (SerdStreamErrorFunc)ferror,
-		                         temp,
-		                         NULL,
-		                         4096);
+		byte_source = serd_byte_source_new_function(
+		    (SerdReadFunc)fread, (SerdStreamErrorFunc)ferror, temp, NULL, 4096);
 
+		assert(serd_reader_start(reader, byte_source) == 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);
 
 		serd_reader_finish(reader);
+		serd_byte_source_free(byte_source);
 		fclose(temp);
 	}
 
 	// A byte-wise reader that hits EOF once then continues (like a socket)
 	{
 		size_t n_reads = 0;
-		serd_reader_start_stream(reader,
-		                         (SerdReadFunc)eof_test_read,
-		                         (SerdStreamErrorFunc)eof_test_error,
-		                         &n_reads,
-		                         NULL,
-		                         1);
 
+		byte_source =
+		    serd_byte_source_new_function((SerdReadFunc)eof_test_read,
+		                                  (SerdStreamErrorFunc)eof_test_error,
+		                                  &n_reads,
+		                                  NULL,
+		                                  1);
+
+		assert(serd_reader_start(reader, byte_source) == 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);