Replace multiple stream callbacks with SerdEvent

This makes plumbing easier since everything goes through the same "stream" and
only one callback is required to handling everything.  It's also more easily
extensible in case more event types need to be added in the future.

17 files changed, 353 insertions, 233 deletions

diff --git a/include/serd/env.h b/include/serd/env.h
index db2c1150..ca02523b 100644
--- a/include/serd/env.h
+++ b/include/serd/env.h
@@ -79,11 +79,10 @@ SERD_API SerdNode* SERD_ALLOCATED
 serd_env_expand_node(const SerdEnv* SERD_NULLABLE env,
                      const SerdNode* SERD_NONNULL node);
 
-/// Call `func` for each prefix defined in `env`
-SERD_API void
-serd_env_foreach(const SerdEnv* SERD_NONNULL env,
-                 SerdPrefixFunc SERD_NONNULL func,
-                 void* SERD_NULLABLE         handle);
+/// Write all prefixes in `env` to `sink`
+SERD_API SerdStatus
+serd_env_write_prefixes(const SerdEnv* SERD_NONNULL  env,
+                        const SerdSink* SERD_NONNULL sink);
 
 /**
    @}
diff --git a/include/serd/event.h b/include/serd/event.h
new file mode 100644
index 00000000..0c6437f6
--- /dev/null
+++ b/include/serd/event.h
@@ -0,0 +1,99 @@
+// Copyright 2011-2022 David Robillard <d@drobilla.net>
+// SPDX-License-Identifier: ISC
+
+#ifndef SERD_EVENT_H
+#define SERD_EVENT_H
+
+#include "serd/attributes.h"
+#include "serd/node.h"
+#include "serd/statement.h"
+#include "serd/status.h"
+
+SERD_BEGIN_DECLS
+
+/**
+   @defgroup serd_event Events
+   @ingroup serd_streaming
+   @{
+*/
+
+/// Type of a SerdEvent
+typedef enum {
+  SERD_BASE      = 1, ///< Base URI changed
+  SERD_PREFIX    = 2, ///< New URI prefix
+  SERD_STATEMENT = 3, ///< Statement
+  SERD_END       = 4, ///< End of anonymous node
+} SerdEventType;
+
+/**
+   Event for base URI changes.
+
+   Emitted whenever the base URI changes.
+*/
+typedef struct {
+  SerdEventType                type; ///< #SERD_BASE
+  const SerdNode* SERD_NONNULL uri;  ///< Base URI
+} SerdBaseEvent;
+
+/**
+   Event for namespace definitions.
+
+   Emitted whenever a prefix is defined.
+*/
+typedef struct {
+  SerdEventType                type; ///< #SERD_PREFIX
+  const SerdNode* SERD_NONNULL name; ///< Prefix name
+  const SerdNode* SERD_NONNULL uri;  ///< Namespace URI
+} SerdPrefixEvent;
+
+/**
+   Event for statements.
+
+   Emitted for every statement.
+*/
+typedef struct {
+  SerdEventType                     type;      ///< #SERD_STATEMENT
+  SerdStatementFlags                flags;     ///< Flags for pretty-printing
+  const SerdStatement* SERD_NONNULL statement; ///< Statement
+} SerdStatementEvent;
+
+/**
+   Event for the end of anonymous node descriptions.
+
+   This is emitted to indicate that the given anonymous node will no longer be
+   described.  This is used by the writer which may, for example, need to
+   write a delimiter.
+*/
+typedef struct {
+  SerdEventType                type; ///< #SERD_END
+  const SerdNode* SERD_NONNULL node; ///< Anonymous node that is finished
+} SerdEndEvent;
+
+/**
+   An event in a data stream.
+
+   Streams of data are represented as a series of events.  Events represent
+   everything that can occur in an RDF document, and are used to plumb together
+   different components.  For example, when parsing a document, a reader emits
+   a stream of events which can be sent to a writer to rewrite a document, or
+   to an inserter to build a model in memory.
+*/
+typedef union {
+  SerdEventType      type;      ///< Event type (always set)
+  SerdBaseEvent      base;      ///< Base URI changed
+  SerdPrefixEvent    prefix;    ///< New namespace prefix
+  SerdStatementEvent statement; ///< Statement
+  SerdEndEvent       end;       ///< End of anonymous node
+} SerdEvent;
+
+/// Function for handling events
+typedef SerdStatus (*SerdEventFunc)(void* SERD_NULLABLE           handle,
+                                    const SerdEvent* SERD_NONNULL event);
+
+/**
+   @}
+*/
+
+SERD_END_DECLS
+
+#endif // SERD_EVENT_H
diff --git a/include/serd/serd.h b/include/serd/serd.h
index 8de151c6..5b13f676 100644
--- a/include/serd/serd.h
+++ b/include/serd/serd.h
@@ -69,6 +69,7 @@
 */
 
 #include "serd/env.h"
+#include "serd/event.h"
 #include "serd/sink.h"
 
 /**
diff --git a/include/serd/sink.h b/include/serd/sink.h
index 95979b50..09fa7858 100644
--- a/include/serd/sink.h
+++ b/include/serd/sink.h
@@ -5,6 +5,7 @@
 #define SERD_SINK_H
 
 #include "serd/attributes.h"
+#include "serd/event.h"
 #include "serd/node.h"
 #include "serd/statement.h"
 #include "serd/status.h"
@@ -63,39 +64,23 @@ typedef void (*SerdFreeFunc)(void* SERD_NULLABLE ptr);
 /**
    Create a new sink.
 
-   Initially, the sink has no set functions and will do nothing.  Use the
-   serd_sink_set_*_func functions to set handlers for various events.
-
    @param handle Opaque handle that will be passed to sink functions.
+   @param event_func Function that will be called for every event.
    @param free_handle Free function to call on handle in serd_sink_free().
 */
 SERD_API SerdSink* SERD_ALLOCATED
-serd_sink_new(void* SERD_NULLABLE        handle,
-              SerdFreeFunc SERD_NULLABLE free_handle);
+serd_sink_new(void* SERD_NULLABLE         handle,
+              SerdEventFunc SERD_NULLABLE event_func,
+              SerdFreeFunc SERD_NULLABLE  free_handle);
 
 /// Free `sink`
 SERD_API void
 serd_sink_free(SerdSink* SERD_NULLABLE sink);
 
-/// Set a function to be called when the base URI changes
-SERD_API SerdStatus
-serd_sink_set_base_func(SerdSink* SERD_NONNULL     sink,
-                        SerdBaseFunc SERD_NULLABLE base_func);
-
-/// Set a function to be called when a namespace prefix is defined
-SERD_API SerdStatus
-serd_sink_set_prefix_func(SerdSink* SERD_NONNULL       sink,
-                          SerdPrefixFunc SERD_NULLABLE prefix_func);
-
-/// Set a function to be called when a statement is emitted
-SERD_API SerdStatus
-serd_sink_set_statement_func(SerdSink* SERD_NONNULL          sink,
-                             SerdStatementFunc SERD_NULLABLE statement_func);
-
-/// Set a function to be called when an anonymous node ends
+/// Send an event to the sink
 SERD_API SerdStatus
-serd_sink_set_end_func(SerdSink* SERD_NONNULL    sink,
-                       SerdEndFunc SERD_NULLABLE end_func);
+serd_sink_write_event(const SerdSink* SERD_NONNULL  sink,
+                      const SerdEvent* SERD_NONNULL event);
 
 /// Set the base URI
 SERD_API SerdStatus
diff --git a/include/serd/status.h b/include/serd/status.h
index 3ebdaf8f..e21297cd 100644
--- a/include/serd/status.h
+++ b/include/serd/status.h
@@ -34,6 +34,7 @@ typedef enum {
   SERD_BAD_STACK,   ///< Stack overflow
   SERD_BAD_TEXT,    ///< Invalid text encoding
   SERD_BAD_CALL,    ///< Invalid call
+  SERD_BAD_EVENT,   ///< Invalid event in stream
   SERD_BAD_URI,     ///< Invalid or unresolved URI
   SERD_BAD_DATA,    ///< Invalid data
   SERD_BAD_LITERAL, ///< Invalid literal
diff --git a/include/serd/writer.h b/include/serd/writer.h
index 96059932..db402fea 100644
--- a/include/serd/writer.h
+++ b/include/serd/writer.h
@@ -72,15 +72,6 @@ serd_writer_chop_blank_prefix(SerdWriter* SERD_NONNULL  writer,
                               const char* SERD_NULLABLE prefix);
 
 /**
-   Set the current output base URI, and emit a directive if applicable.
-
-   Note this function can be safely casted to #SerdBaseFunc.
-*/
-SERD_API SerdStatus
-serd_writer_set_base_uri(SerdWriter* SERD_NONNULL      writer,
-                         const SerdNode* SERD_NULLABLE uri);
-
-/**
    Set the current root URI.
 
    The root URI should be a prefix of the base URI.  The path of the root URI
diff --git a/meson.build b/meson.build
index 5cacf693..2afe226b 100644
--- a/meson.build
+++ b/meson.build
@@ -124,6 +124,7 @@ c_headers = files(
   'include/serd/caret.h',
   'include/serd/env.h',
   'include/serd/error.h',
+  'include/serd/event.h',
   'include/serd/memory.h',
   'include/serd/node.h',
   'include/serd/reader.h',
diff --git a/src/env.c b/src/env.c
index 8b3970b3..4ce0add4 100644
--- a/src/env.c
+++ b/src/env.c
@@ -238,12 +238,16 @@ serd_env_expand_node(const SerdEnv* const env, const SerdNode* const node)
   return NULL;
 }
 
-void
-serd_env_foreach(const SerdEnv* const env,
-                 const SerdPrefixFunc func,
-                 void* const          handle)
+SerdStatus
+serd_env_write_prefixes(const SerdEnv* const env, const SerdSink* const sink)
 {
-  for (size_t i = 0; i < env->n_prefixes; ++i) {
-    func(handle, env->prefixes[i].name, env->prefixes[i].uri);
+  SerdStatus st = SERD_SUCCESS;
+
+  for (size_t i = 0; !st && i < env->n_prefixes; ++i) {
+    const SerdPrefix* const prefix = &env->prefixes[i];
+
+    st = serd_sink_write_prefix(sink, prefix->name, prefix->uri);
   }
+
+  return st;
 }
diff --git a/src/sink.c b/src/sink.c
index 51b23ea5..8343a62a 100644
--- a/src/sink.c
+++ b/src/sink.c
@@ -14,11 +14,14 @@
 #include <stdlib.h>
 
 SerdSink*
-serd_sink_new(void* handle, SerdFreeFunc free_handle)
+serd_sink_new(void* const   handle,
+              SerdEventFunc event_func,
+              SerdFreeFunc  free_handle)
 {
   SerdSink* sink = (SerdSink*)calloc(1, sizeof(SerdSink));
 
   sink->handle      = handle;
+  sink->on_event    = event_func;
   sink->free_handle = free_handle;
 
   return sink;
@@ -37,37 +40,18 @@ serd_sink_free(SerdSink* sink)
 }
 
 SerdStatus
-serd_sink_set_base_func(SerdSink* sink, SerdBaseFunc base_func)
+serd_sink_write_event(const SerdSink* sink, const SerdEvent* event)
 {
-  sink->base = base_func;
-  return SERD_SUCCESS;
-}
-
-SerdStatus
-serd_sink_set_prefix_func(SerdSink* sink, SerdPrefixFunc prefix_func)
-{
-  sink->prefix = prefix_func;
-  return SERD_SUCCESS;
-}
-
-SerdStatus
-serd_sink_set_statement_func(SerdSink* sink, SerdStatementFunc statement_func)
-{
-  sink->statement = statement_func;
-  return SERD_SUCCESS;
-}
-
-SerdStatus
-serd_sink_set_end_func(SerdSink* sink, SerdEndFunc end_func)
-{
-  sink->end = end_func;
-  return SERD_SUCCESS;
+  return sink->on_event ? sink->on_event(sink->handle, event) : SERD_SUCCESS;
 }
 
 SerdStatus
 serd_sink_write_base(const SerdSink* sink, const SerdNode* uri)
 {
-  return sink->base ? sink->base(sink->handle, uri) : SERD_SUCCESS;
+  const SerdBaseEvent ev = {SERD_BASE, uri};
+
+  return sink->on_event ? sink->on_event(sink->handle, (const SerdEvent*)&ev)
+                        : SERD_SUCCESS;
 }
 
 SerdStatus
@@ -75,7 +59,10 @@ serd_sink_write_prefix(const SerdSink* sink,
                        const SerdNode* name,
                        const SerdNode* uri)
 {
-  return sink->prefix ? sink->prefix(sink->handle, name, uri) : SERD_SUCCESS;
+  const SerdPrefixEvent ev = {SERD_PREFIX, name, uri};
+
+  return sink->on_event ? sink->on_event(sink->handle, (const SerdEvent*)&ev)
+                        : SERD_SUCCESS;
 }
 
 SerdStatus
@@ -83,8 +70,11 @@ serd_sink_write_statement(const SerdSink*          sink,
                           const SerdStatementFlags flags,
                           const SerdStatement*     statement)
 {
-  return sink->statement ? sink->statement(sink->handle, flags, statement)
-                         : SERD_SUCCESS;
+  const SerdStatementEvent statement_ev = {SERD_STATEMENT, flags, statement};
+  SerdEvent                ev           = {SERD_STATEMENT};
+  ev.statement                          = statement_ev;
+
+  return sink->on_event ? sink->on_event(sink->handle, &ev) : SERD_SUCCESS;
 }
 
 SerdStatus
@@ -107,5 +97,8 @@ serd_sink_write(const SerdSink*          sink,
 SerdStatus
 serd_sink_write_end(const SerdSink* sink, const SerdNode* node)
 {
-  return sink->end ? sink->end(sink->handle, node) : SERD_SUCCESS;
+  const SerdEndEvent ev = {SERD_END, node};
+
+  return sink->on_event ? sink->on_event(sink->handle, (const SerdEvent*)&ev)
+                        : SERD_SUCCESS;
 }
diff --git a/src/sink.h b/src/sink.h
index 98884fe2..097e0d6d 100644
--- a/src/sink.h
+++ b/src/sink.h
@@ -4,18 +4,16 @@
 #ifndef SERD_SRC_SINK_H
 #define SERD_SRC_SINK_H
 
+#include "serd/event.h"
 #include "serd/sink.h"
 
 /**
    An interface that receives a stream of RDF data.
 */
 struct SerdSinkImpl {
-  void*             handle;
-  SerdFreeFunc      free_handle;
-  SerdBaseFunc      base;
-  SerdPrefixFunc    prefix;
-  SerdStatementFunc statement;
-  SerdEndFunc       end;
+  void*         handle;
+  SerdFreeFunc  free_handle;
+  SerdEventFunc on_event;
 };
 
 #endif // SERD_SRC_SINK_H
diff --git a/src/string.c b/src/string.c
index e13c0c01..8cc839bd 100644
--- a/src/string.c
+++ b/src/string.c
@@ -56,6 +56,8 @@ serd_strerror(const SerdStatus status)
     return "Invalid text encoding";
   case SERD_BAD_CALL:
     return "Invalid call";
+  case SERD_BAD_EVENT:
+    return "Invalid event in stream";
   case SERD_BAD_URI:
     return "Invalid or unresolved URI";
   case SERD_BAD_DATA:
diff --git a/src/writer.c b/src/writer.c
index 37263010..ca17bbd9 100644
--- a/src/writer.c
+++ b/src/writer.c
@@ -16,6 +16,7 @@
 #include "serd/attributes.h"
 #include "serd/buffer.h"
 #include "serd/env.h"
+#include "serd/event.h"
 #include "serd/node.h"
 #include "serd/sink.h"
 #include "serd/statement.h"
@@ -139,6 +140,9 @@ typedef enum { WRITE_STRING, WRITE_LONG_STRING } TextContext;
 typedef enum { RESET_GRAPH = 1U << 0U, RESET_INDENT = 1U << 1U } ResetFlag;
 
 SERD_NODISCARD static SerdStatus
+serd_writer_set_base_uri(SerdWriter* writer, const SerdNode* uri);
+
+SERD_NODISCARD static SerdStatus
 serd_writer_set_prefix(SerdWriter*     writer,
                        const SerdNode* name,
                        const SerdNode* uri);
@@ -945,7 +949,7 @@ write_list_next(SerdWriter* const        writer,
   return st;
 }
 
-static SerdStatus
+SERD_NODISCARD static SerdStatus
 terminate_context(SerdWriter* writer)
 {
   SerdStatus st = SERD_SUCCESS;
@@ -961,7 +965,7 @@ terminate_context(SerdWriter* writer)
   return st;
 }
 
-static SerdStatus
+SERD_NODISCARD static SerdStatus
 serd_writer_write_statement(SerdWriter* const          writer,
                             const SerdStatementFlags   flags,
                             const SerdStatement* const statement)
@@ -1103,7 +1107,7 @@ serd_writer_write_statement(SerdWriter* const          writer,
   return st;
 }
 
-static SerdStatus
+SERD_NODISCARD static SerdStatus
 serd_writer_end_anon(SerdWriter* writer, const SerdNode* node)
 {
   SerdStatus st = SERD_SUCCESS;
@@ -1113,7 +1117,7 @@ serd_writer_end_anon(SerdWriter* writer, const SerdNode* node)
   }
 
   if (serd_stack_is_empty(&writer->anon_stack)) {
-    return w_err(writer, SERD_BAD_CALL, "unexpected end of anonymous node\n");
+    return w_err(writer, SERD_BAD_EVENT, "unexpected end of anonymous node\n");
   }
 
   // Write the end separator ']' and pop the context
@@ -1129,6 +1133,25 @@ serd_writer_end_anon(SerdWriter* writer, const SerdNode* node)
   return st;
 }
 
+SERD_NODISCARD static SerdStatus
+serd_writer_on_event(SerdWriter* writer, const SerdEvent* event)
+{
+  switch (event->type) {
+  case SERD_BASE:
+    return serd_writer_set_base_uri(writer, event->base.uri);
+  case SERD_PREFIX:
+    return serd_writer_set_prefix(
+      writer, event->prefix.name, event->prefix.uri);
+  case SERD_STATEMENT:
+    return serd_writer_write_statement(
+      writer, event->statement.flags, event->statement.statement);
+  case SERD_END:
+    return serd_writer_end_anon(writer, event->end.node);
+  }
+
+  return SERD_BAD_ARG;
+}
+
 SerdStatus
 serd_writer_finish(SerdWriter* writer)
 {
@@ -1161,11 +1184,8 @@ serd_writer_new(SerdWorld*      world,
   writer->byte_sink  = serd_byte_sink_new(
     ssink, stream, (flags & SERD_WRITE_BULK) ? SERD_PAGE_SIZE : 1);
 
-  writer->iface.handle    = writer;
-  writer->iface.base      = (SerdBaseFunc)serd_writer_set_base_uri;
-  writer->iface.prefix    = (SerdPrefixFunc)serd_writer_set_prefix;
-  writer->iface.statement = (SerdStatementFunc)serd_writer_write_statement;
-  writer->iface.end       = (SerdEndFunc)serd_writer_end_anon;
+  writer->iface.handle   = writer;
+  writer->iface.on_event = (SerdEventFunc)serd_writer_on_event;
 
   return writer;
 }
@@ -1185,7 +1205,7 @@ serd_writer_chop_blank_prefix(SerdWriter* writer, const char* prefix)
   }
 }
 
-SerdStatus
+SERD_NODISCARD static SerdStatus
 serd_writer_set_base_uri(SerdWriter* writer, const SerdNode* uri)
 {
   if (uri && serd_node_type(uri) != SERD_URI) {
diff --git a/test/test_env.c b/test/test_env.c
index 8dbc24a1..31446f69 100644
--- a/test/test_env.c
+++ b/test/test_env.c
@@ -4,7 +4,9 @@
 #undef NDEBUG
 
 #include "serd/env.h"
+#include "serd/event.h"
 #include "serd/node.h"
+#include "serd/sink.h"
 #include "serd/status.h"
 #include "serd/string_view.h"
 
@@ -12,12 +14,12 @@
 #include <string.h>
 
 static SerdStatus
-count_prefixes(void* handle, const SerdNode* name, const SerdNode* uri)
+count_prefixes(void* handle, const SerdEvent* event)
 {
-  (void)name;
-  (void)uri;
+  if (event->type == SERD_PREFIX) {
+    ++*(int*)handle;
+  }
 
-  ++*(int*)handle;
   return SERD_SUCCESS;
 }
 
@@ -73,10 +75,13 @@ test_env(void)
   assert(!serd_env_expand_node(env, blank));
   serd_node_free(blank);
 
-  int n_prefixes = 0;
+  size_t          n_prefixes = 0;
+  SerdSink* const count_prefixes_sink =
+    serd_sink_new(&n_prefixes, count_prefixes, NULL);
+
   serd_env_set_prefix(
     env, serd_string("eg.2"), serd_string("http://example.org/"));
-  serd_env_foreach(env, count_prefixes, &n_prefixes);
+  serd_env_write_prefixes(env, count_prefixes_sink);
   assert(n_prefixes == 1);
 
   SerdNode* shorter_uri = serd_new_uri(serd_string("urn:foo"));
@@ -93,6 +98,7 @@ test_env(void)
   assert(!serd_env_set_base_uri(env, serd_empty_string()));
   assert(!serd_env_base_uri(env));
 
+  serd_sink_free(count_prefixes_sink);
   serd_node_free(shorter_uri);
   serd_node_free(badpre);
   serd_node_free(s);
diff --git a/test/test_overflow.c b/test/test_overflow.c
index f4c5a863..62154958 100644
--- a/test/test_overflow.c
+++ b/test/test_overflow.c
@@ -17,7 +17,7 @@ test_size(SerdWorld* const  world,
           const SerdSyntax  syntax,
           const size_t      stack_size)
 {
-  SerdSink*         sink   = serd_sink_new(NULL, NULL);
+  SerdSink*         sink   = serd_sink_new(NULL, NULL, NULL);
   SerdReader* const reader = serd_reader_new(world, syntax, sink, stack_size);
   if (!reader) {
     return SERD_BAD_STACK;
diff --git a/test/test_reader_writer.c b/test/test_reader_writer.c
index c4f8bb90..abea37b3 100644
--- a/test/test_reader_writer.c
+++ b/test/test_reader_writer.c
@@ -5,11 +5,11 @@
 
 #include "serd/buffer.h"
 #include "serd/env.h"
+#include "serd/event.h"
 #include "serd/memory.h"
 #include "serd/node.h"
 #include "serd/reader.h"
 #include "serd/sink.h"
-#include "serd/statement.h"
 #include "serd/status.h"
 #include "serd/stream.h"
 #include "serd/string_view.h"
@@ -65,46 +65,25 @@ static const char* const doc_string =
   "( eg:o ) eg:t eg:u .\n";
 
 static SerdStatus
-test_base_sink(void* handle, const SerdNode* uri)
+test_sink(void* handle, const SerdEvent* event)
 {
-  (void)uri;
-
-  ReaderTest* rt = (ReaderTest*)handle;
-  ++rt->n_base;
-  return SERD_SUCCESS;
-}
-
-static SerdStatus
-test_prefix_sink(void* handle, const SerdNode* name, const SerdNode* uri)
-{
-  (void)name;
-  (void)uri;
-
   ReaderTest* rt = (ReaderTest*)handle;
-  ++rt->n_prefix;
-  return SERD_SUCCESS;
-}
-
-static SerdStatus
-test_statement_sink(void*                    handle,
-                    const SerdStatementFlags flags,
-                    const SerdStatement*     statement)
-{
-  (void)flags;
-  (void)statement;
-
-  ReaderTest* rt = (ReaderTest*)handle;
-  ++rt->n_statement;
-  return SERD_SUCCESS;
-}
 
-static SerdStatus
-test_end_sink(void* handle, const SerdNode* node)
-{
-  (void)node;
+  switch (event->type) {
+  case SERD_BASE:
+    ++rt->n_base;
+    break;
+  case SERD_PREFIX:
+    ++rt->n_prefix;
+    break;
+  case SERD_STATEMENT:
+    ++rt->n_statement;
+    break;
+  case SERD_END:
+    ++rt->n_end;
+    break;
+  }
 
-  ReaderTest* rt = (ReaderTest*)handle;
-  ++rt->n_end;
   return SERD_SUCCESS;
 }
 
@@ -158,6 +137,7 @@ test_read_nquads_chunks(const char* const path)
   static const char null = 0;
 
   FILE* const f = fopen(path, "w+b");
+  assert(f);
 
   // Write two statements, a null separator, then another statement
 
@@ -177,18 +157,13 @@ test_read_nquads_chunks(const char* const path)
 
   fseek(f, 0, SEEK_SET);
 
-  SerdWorld* const  world  = serd_world_new();
-  ReaderTest* const rt     = (ReaderTest*)calloc(1, sizeof(ReaderTest));
-  SerdSink* const   sink   = serd_sink_new(rt, NULL);
-  SerdReader* const reader = serd_reader_new(world, SERD_NQUADS, sink, 4096);
+  SerdWorld* const world = serd_world_new();
+  ReaderTest       rt    = {0, 0, 0, 0};
+  SerdSink* const  sink  = serd_sink_new(&rt, test_sink, NULL);
+  assert(sink);
 
+  SerdReader* const reader = serd_reader_new(world, SERD_NQUADS, sink, 4096);
   assert(reader);
-  assert(sink);
-  assert(f);
-  serd_sink_set_base_func(sink, test_base_sink);
-  serd_sink_set_prefix_func(sink, test_prefix_sink);
-  serd_sink_set_statement_func(sink, test_statement_sink);
-  serd_sink_set_end_func(sink, test_end_sink);
 
   SerdStatus st = serd_reader_start_stream(
     reader, (SerdReadFunc)fread, (SerdStreamErrorFunc)ferror, f, NULL, 1);
@@ -197,48 +172,47 @@ test_read_nquads_chunks(const char* const path)
   // Read first statement
   st = serd_reader_read_chunk(reader);
   assert(st == SERD_SUCCESS);
-  assert(rt->n_base == 0);
-  assert(rt->n_prefix == 0);
-  assert(rt->n_statement == 1);
-  assert(rt->n_end == 0);
+  assert(rt.n_base == 0);
+  assert(rt.n_prefix == 0);
+  assert(rt.n_statement == 1);
+  assert(rt.n_end == 0);
 
   // Read second statement
   st = serd_reader_read_chunk(reader);
   assert(st == SERD_SUCCESS);
-  assert(rt->n_base == 0);
-  assert(rt->n_prefix == 0);
-  assert(rt->n_statement == 2);
-  assert(rt->n_end == 0);
+  assert(rt.n_base == 0);
+  assert(rt.n_prefix == 0);
+  assert(rt.n_statement == 2);
+  assert(rt.n_end == 0);
 
   // Read terminator
   st = serd_reader_read_chunk(reader);
   assert(st == SERD_FAILURE);
-  assert(rt->n_base == 0);
-  assert(rt->n_prefix == 0);
-  assert(rt->n_statement == 2);
-  assert(rt->n_end == 0);
+  assert(rt.n_base == 0);
+  assert(rt.n_prefix == 0);
+  assert(rt.n_statement == 2);
+  assert(rt.n_end == 0);
 
   // Read last statement
   st = serd_reader_read_chunk(reader);
   assert(st == SERD_SUCCESS);
-  assert(rt->n_base == 0);
-  assert(rt->n_prefix == 0);
-  assert(rt->n_statement == 3);
-  assert(rt->n_end == 0);
+  assert(rt.n_base == 0);
+  assert(rt.n_prefix == 0);
+  assert(rt.n_statement == 3);
+  assert(rt.n_end == 0);
 
   // EOF
   st = serd_reader_read_chunk(reader);
   assert(st == SERD_FAILURE);
-  assert(rt->n_base == 0);
-  assert(rt->n_prefix == 0);
-  assert(rt->n_statement == 3);
-  assert(rt->n_end == 0);
+  assert(rt.n_base == 0);
+  assert(rt.n_prefix == 0);
+  assert(rt.n_statement == 3);
+  assert(rt.n_end == 0);
 
   assert(serd_reader_read_chunk(reader) == SERD_FAILURE);
 
   serd_reader_free(reader);
   serd_sink_free(sink);
-  free(rt);
   serd_world_free(world);
   fclose(f);
   remove(path);
@@ -250,6 +224,7 @@ test_read_turtle_chunks(const char* const path)
   static const char null = 0;
 
   FILE* const f = fopen(path, "w+b");
+  assert(f);
 
   // Write two statements separated by null characters
   fprintf(f, "@base <http://example.org/base/> .\n");
@@ -261,18 +236,13 @@ test_read_turtle_chunks(const char* const path)
   fwrite(&null, sizeof(null), 1, f);
   fseek(f, 0, SEEK_SET);
 
-  SerdWorld*  world  = serd_world_new();
-  ReaderTest* rt     = (ReaderTest*)calloc(1, sizeof(ReaderTest));
-  SerdSink*   sink   = serd_sink_new(rt, NULL);
-  SerdReader* reader = serd_reader_new(world, SERD_TURTLE, sink, 4096);
+  SerdWorld* world = serd_world_new();
+  ReaderTest rt    = {0, 0, 0, 0};
+  SerdSink*  sink  = serd_sink_new(&rt, test_sink, NULL);
+  assert(sink);
 
+  SerdReader* reader = serd_reader_new(world, SERD_TURTLE, sink, 4096);
   assert(reader);
-  assert(sink);
-  assert(f);
-  serd_sink_set_base_func(sink, test_base_sink);
-  serd_sink_set_prefix_func(sink, test_prefix_sink);
-  serd_sink_set_statement_func(sink, test_statement_sink);
-  serd_sink_set_end_func(sink, test_end_sink);
 
   SerdStatus st = serd_reader_start_stream(
     reader, (SerdReadFunc)fread, (SerdStreamErrorFunc)ferror, f, NULL, 1);
@@ -281,64 +251,63 @@ test_read_turtle_chunks(const char* const path)
   // Read base
   st = serd_reader_read_chunk(reader);
   assert(st == SERD_SUCCESS);
-  assert(rt->n_base == 1);
-  assert(rt->n_prefix == 0);
-  assert(rt->n_statement == 0);
-  assert(rt->n_end == 0);
+  assert(rt.n_base == 1);
+  assert(rt.n_prefix == 0);
+  assert(rt.n_statement == 0);
+  assert(rt.n_end == 0);
 
   // Read prefix
   st = serd_reader_read_chunk(reader);
   assert(st == SERD_SUCCESS);
-  assert(rt->n_base == 1);
-  assert(rt->n_prefix == 1);
-  assert(rt->n_statement == 0);
-  assert(rt->n_end == 0);
+  assert(rt.n_base == 1);
+  assert(rt.n_prefix == 1);
+  assert(rt.n_statement == 0);
+  assert(rt.n_end == 0);
 
   // Read first two statements
   st = serd_reader_read_chunk(reader);
   assert(st == SERD_SUCCESS);
-  assert(rt->n_base == 1);
-  assert(rt->n_prefix == 1);
-  assert(rt->n_statement == 2);
-  assert(rt->n_end == 0);
+  assert(rt.n_base == 1);
+  assert(rt.n_prefix == 1);
+  assert(rt.n_statement == 2);
+  assert(rt.n_end == 0);
 
   // Read terminator
   st = serd_reader_read_chunk(reader);
   assert(st == SERD_FAILURE);
-  assert(rt->n_base == 1);
-  assert(rt->n_prefix == 1);
-  assert(rt->n_statement == 2);
-  assert(rt->n_end == 0);
+  assert(rt.n_base == 1);
+  assert(rt.n_prefix == 1);
+  assert(rt.n_statement == 2);
+  assert(rt.n_end == 0);
 
   // Read statements after null terminator
   st = serd_reader_read_chunk(reader);
   assert(st == SERD_SUCCESS);
-  assert(rt->n_base == 1);
-  assert(rt->n_prefix == 1);
-  assert(rt->n_statement == 4);
-  assert(rt->n_end == 1);
+  assert(rt.n_base == 1);
+  assert(rt.n_prefix == 1);
+  assert(rt.n_statement == 4);
+  assert(rt.n_end == 1);
 
   // Read terminator
   st = serd_reader_read_chunk(reader);
   assert(st == SERD_FAILURE);
-  assert(rt->n_base == 1);
-  assert(rt->n_prefix == 1);
-  assert(rt->n_statement == 4);
-  assert(rt->n_end == 1);
+  assert(rt.n_base == 1);
+  assert(rt.n_prefix == 1);
+  assert(rt.n_statement == 4);
+  assert(rt.n_end == 1);
 
   // EOF
   st = serd_reader_read_chunk(reader);
   assert(st == SERD_FAILURE);
-  assert(rt->n_base == 1);
-  assert(rt->n_prefix == 1);
-  assert(rt->n_statement == 4);
-  assert(rt->n_end == 1);
+  assert(rt.n_base == 1);
+  assert(rt.n_prefix == 1);
+  assert(rt.n_statement == 4);
+  assert(rt.n_end == 1);
 
   assert(serd_reader_read_chunk(reader) == SERD_FAILURE);
 
   serd_reader_free(reader);
   serd_sink_free(sink);
-  free(rt);
   serd_world_free(world);
   fclose(f);
   remove(path);
@@ -347,18 +316,13 @@ test_read_turtle_chunks(const char* const path)
 static void
 test_read_string(void)
 {
-  SerdWorld*  world  = serd_world_new();
-  ReaderTest* rt     = (ReaderTest*)calloc(1, sizeof(ReaderTest));
-  SerdSink*   sink   = serd_sink_new(rt, NULL);
-  SerdReader* reader = serd_reader_new(world, SERD_TURTLE, sink, 4096);
-
-  assert(reader);
+  SerdWorld* world = serd_world_new();
+  ReaderTest rt    = {0, 0, 0, 0};
+  SerdSink*  sink  = serd_sink_new(&rt, test_sink, NULL);
   assert(sink);
 
-  serd_sink_set_base_func(sink, test_base_sink);
-  serd_sink_set_prefix_func(sink, test_prefix_sink);
-  serd_sink_set_statement_func(sink, test_statement_sink);
-  serd_sink_set_end_func(sink, test_end_sink);
+  SerdReader* reader = serd_reader_new(world, SERD_TURTLE, sink, 4096);
+  assert(reader);
 
   // Test reading a string that ends exactly at the end of input (no newline)
   assert(
@@ -368,15 +332,14 @@ test_read_string(void)
                               NULL));
 
   assert(!serd_reader_read_document(reader));
-  assert(rt->n_base == 0);
-  assert(rt->n_prefix == 0);
-  assert(rt->n_statement == 1);
-  assert(rt->n_end == 0);
+  assert(rt.n_base == 0);
+  assert(rt.n_prefix == 0);
+  assert(rt.n_statement == 1);
+  assert(rt.n_end == 0);
   assert(!serd_reader_finish(reader));
 
   serd_reader_free(reader);
   serd_sink_free(sink);
-  free(rt);
   serd_world_free(world);
 }
 
@@ -525,7 +488,7 @@ test_writer(const char* const path)
 
   SerdNode* const base = serd_new_uri(serd_string("http://example.org/base"));
 
-  serd_writer_set_base_uri(writer, base);
+  serd_sink_write_base(serd_writer_sink(writer), base);
 
   serd_node_free(base);
   serd_writer_free(writer);
@@ -547,14 +510,9 @@ test_reader(const char* path)
 {
   SerdWorld*      world = serd_world_new();
   ReaderTest      rt    = {0, 0, 0, 0};
-  SerdSink* const sink  = serd_sink_new(&rt, NULL);
+  SerdSink* const sink  = serd_sink_new(&rt, test_sink, NULL);
   assert(sink);
 
-  serd_sink_set_base_func(sink, test_base_sink);
-  serd_sink_set_prefix_func(sink, test_prefix_sink);
-  serd_sink_set_statement_func(sink, test_statement_sink);
-  serd_sink_set_end_func(sink, test_end_sink);
-
   // Test that too little stack space fails gracefully
   assert(!serd_reader_new(world, SERD_TURTLE, sink, 32));
 
diff --git a/test/test_sink.c b/test/test_sink.c
index 4b9a2eb2..b0756fcc 100644
--- a/test/test_sink.c
+++ b/test/test_sink.c
@@ -4,6 +4,7 @@
 #undef NDEBUG
 
 #include "serd/env.h"
+#include "serd/event.h"
 #include "serd/node.h"
 #include "serd/sink.h"
 #include "serd/statement.h"
@@ -68,6 +69,24 @@ on_end(void* handle, const SerdNode* node)
   return state->return_status;
 }
 
+static SerdStatus
+on_event(void* const handle, const SerdEvent* const event)
+{
+  switch (event->type) {
+  case SERD_BASE:
+    return on_base(handle, event->base.uri);
+  case SERD_PREFIX:
+    return on_prefix(handle, event->prefix.name, event->prefix.uri);
+  case SERD_STATEMENT:
+    return on_statement(
+      handle, event->statement.flags, event->statement.statement);
+  case SERD_END:
+    return on_end(handle, event->end.node);
+  }
+
+  return SERD_BAD_ARG;
+}
+
 static void
 test_callbacks(void)
 {
@@ -81,23 +100,37 @@ test_callbacks(void)
   SerdStatement* const statement =
     serd_statement_new(base, uri, blank, NULL, NULL);
 
+  const SerdBaseEvent      base_event      = {SERD_BASE, uri};
+  const SerdPrefixEvent    prefix_event    = {SERD_PREFIX, name, uri};
+  const SerdStatementEvent statement_event = {SERD_STATEMENT, 0U, statement};
+  const SerdEndEvent       end_event       = {SERD_END, blank};
+
   // Call functions on a sink with no functions set
 
-  SerdSink* null_sink = serd_sink_new(&state, NULL);
+  SerdSink* null_sink = serd_sink_new(&state, NULL, NULL);
+
   assert(!serd_sink_write_base(null_sink, base));
   assert(!serd_sink_write_prefix(null_sink, name, uri));
   assert(!serd_sink_write_statement(null_sink, 0, statement));
   assert(!serd_sink_write(null_sink, 0, base, uri, blank, NULL));
   assert(!serd_sink_write_end(null_sink, blank));
+
+  SerdEvent event = {SERD_BASE};
+
+  event.base = base_event;
+  assert(!serd_sink_write_event(null_sink, &event));
+  event.prefix = prefix_event;
+  assert(!serd_sink_write_event(null_sink, &event));
+  event.statement = statement_event;
+  assert(!serd_sink_write_event(null_sink, &event));
+  event.end = end_event;
+  assert(!serd_sink_write_event(null_sink, &event));
+
   serd_sink_free(null_sink);
 
   // Try again with a sink that has the event handler set
 
-  SerdSink* sink = serd_sink_new(&state, NULL);
-  serd_sink_set_base_func(sink, on_base);
-  serd_sink_set_prefix_func(sink, on_prefix);
-  serd_sink_set_statement_func(sink, on_statement);
-  serd_sink_set_end_func(sink, on_end);
+  SerdSink* sink = serd_sink_new(&state, on_event, NULL);
 
   assert(!serd_sink_write_base(sink, base));
   assert(serd_node_equals(state.last_base, base));
@@ -112,6 +145,9 @@ test_callbacks(void)
   assert(!serd_sink_write_end(sink, blank));
   assert(serd_node_equals(state.last_end, blank));
 
+  const SerdEvent junk = {(SerdEventType)42};
+  assert(serd_sink_write_event(sink, &junk) == SERD_BAD_ARG);
+
   serd_sink_free(sink);
 
   serd_statement_free(statement);
@@ -130,7 +166,7 @@ test_free(void)
 
   // Set up a sink with dynamically allocated data and a free function
   uintptr_t* data = (uintptr_t*)calloc(1, sizeof(uintptr_t));
-  SerdSink*  sink = serd_sink_new(data, free);
+  SerdSink*  sink = serd_sink_new(data, NULL, free);
 
   // Free the sink, which should free the data (rely on valgrind or sanitizers)
   serd_sink_free(sink);
diff --git a/test/test_writer.c b/test/test_writer.c
index db68454f..d2f59af0 100644
--- a/test/test_writer.c
+++ b/test/test_writer.c
@@ -5,6 +5,7 @@
 
 #include "serd/buffer.h"
 #include "serd/env.h"
+#include "serd/event.h"
 #include "serd/memory.h"
 #include "serd/node.h"
 #include "serd/sink.h"
@@ -21,6 +22,31 @@
 #include <string.h>
 
 static void
+test_write_bad_event(void)
+{
+  SerdWorld*  world  = serd_world_new();
+  SerdEnv*    env    = serd_env_new(serd_empty_string());
+  SerdBuffer  buffer = {NULL, 0};
+  SerdWriter* writer =
+    serd_writer_new(world, SERD_TURTLE, 0U, env, serd_buffer_sink, &buffer);
+
+  assert(writer);
+
+  const SerdEvent event = {(SerdEventType)42};
+  assert(serd_sink_write_event(serd_writer_sink(writer), &event) ==
+         SERD_BAD_ARG);
+
+  char* const out = serd_buffer_sink_finish(&buffer);
+
+  assert(!strcmp(out, ""));
+  serd_free(out);
+
+  serd_writer_free(writer);
+  serd_env_free(env);
+  serd_world_free(world);
+}
+
+static void
 test_write_bad_prefix(void)
 {
   SerdWorld*  world  = serd_world_new();
@@ -129,7 +155,6 @@ test_writer_cleanup(void)
 
   // Finish writing without terminating nodes
   assert(!(st = serd_writer_finish(writer)));
-  assert(!(st = serd_writer_set_base_uri(writer, NULL)));
 
   // Set the base to an empty URI
   SerdNode* empty_uri = serd_new_uri(serd_string(""));
@@ -225,6 +250,7 @@ test_write_error(void)
 int
 main(void)
 {
+  test_write_bad_event();
   test_write_bad_prefix();
   test_write_long_literal();
   test_writer_cleanup();