fix: Improve client quality and align with Go client

README.md

@@ -212,7 +212,7 @@ EventSourcingDB.read_events(
   "/books/42",
   %EventSourcingDB.ReadEventsOptions{
     recursive: false,
-    from_latest_event: %EventSourcingDB.FromLatestEventOptions{
+    from_latest_event: %EventSourcingDB.ReadFromLatestEventOptions{
       subject: "/books/42",
       type: "io.eventsourcingdb.library.book-borrowed",
       if_event_is_missing: :read_everything
@@ -223,6 +223,18 @@ EventSourcingDB.read_events(
 
 *Note that `from_latest_event` and `lower_bound` can not be provided at the same time.*
 
+### Aborting Reading
+
+Since `read_events` returns a lazy stream, you can stop consuming it at any time. Use `Enum.take/2` to limit the number of events, or simply stop iterating:
+
+```elixir
+# Take only the first 10 events
+{:ok, events} = EventSourcingDB.read_events(client, "/books/42")
+first_ten = Enum.take(events, 10)
+```
+
+If you are consuming the stream in a separate process, you can abort by terminating that process.
+
 ## Running EventQL Queries
 
 To run an EventQL query, call the `run_eventql_query` function and provide the query as argument. The function returns a stream:
@@ -236,6 +248,15 @@ end
 
 *Note that each row returned by the stream matches the projection specified in your query.*
 
+### Aborting a Query
+
+Since `run_eventql_query` returns a lazy stream, you can stop consuming it at any time. Use `Enum.take/2` to limit the number of rows, or simply stop iterating:
+
+```elixir
+{:ok, rows} = EventSourcingDB.run_eventql_query(client, "FROM e IN events PROJECT INTO e")
+first_five = Enum.take(rows, 5)
+```
+
 ## Observing Events
 
 To observe all events of a subject, call the `observe_events` function with the subject.
@@ -297,7 +318,7 @@ EventSourcingDB.observe_events(
   "/books/42",
   %EventSourcingDB.ObserveEventsOptions{
     recursive: false,
-    from_latest_event: %EventSourcingDB.FromLatestEventOptions{
+    from_latest_event: %EventSourcingDB.ObserveFromLatestEventOptions{
       subject: "/books/42",
       type: "io.eventsourcingdb.library.book-borrowed",
       if_event_is_missing: :read_everything
@@ -308,6 +329,17 @@ EventSourcingDB.observe_events(
 
 *Note that `from_latest_event` and `lower_bound` can not be provided at the same time.*
 
+### Aborting Observing
+
+Since `observe_events` returns a lazy stream, you can stop consuming it at any time. Use `Enum.take/2` to limit the number of events, or simply stop iterating:
+
+```elixir
+{:ok, events} = EventSourcingDB.observe_events(client, "/books/42")
+first_event = Enum.take(events, 1)
+```
+
+If you are consuming the stream in a separate process, you can abort by terminating that process.
+
 ## Registering an Event Schema
 
 To register an event schema, call the `register_event_schema` function and hand over an event type and the desired schema:
@@ -350,6 +382,10 @@ If you only want to list subjects within a specific branch, provide the desired
 EventSourcingDB.read_subjects(client, "/books")
 ```
 
+### Aborting Listing
+
+Since `read_subjects` returns a lazy stream, you can stop consuming it at any time by simply stopping iteration or using `Enum.take/2`.
+
 ## Reading Event Types
 
 To list all event types, call the `read_event_types` function. The function returns a stream from which you can retrieve one event type at a time:
@@ -361,6 +397,10 @@ case EventSourcingDB.read_event_types(client) do
 end
 ```
 
+### Aborting Listing
+
+Since `read_event_types` returns a lazy stream, you can stop consuming it at any time by simply stopping iteration or using `Enum.take/2`.
+
 ## Reading a Specific Event Type
 
 To read a specific event type, call the `read_event_type` function with the event type as an argument. The function returns the detailed event type, which includes the schema:

lib/eventsourcingdb.ex

@@ -200,13 +200,13 @@ defmodule EventSourcingDB do
 
   """
   @spec write_events(Client.t(), nonempty_list(EventCandidate.t()), [precondition()]) ::
-          response(Event.t())
+          response([Event.t()])
   def write_events(client, events, preconditions \\ []) when is_list(events) do
     request_one_shot(client, WriteEvents.new(events, preconditions))
   end
 
   @spec write_events!(Client.t(), nonempty_list(EventCandidate.t()), [precondition()]) ::
-          response!(Event.t())
+          response!([Event.t()])
   def write_events!(client, events, preconditions \\ []) when is_list(events) do
     request_one_shot!(client, WriteEvents.new(events, preconditions))
   end
@@ -292,7 +292,7 @@ defmodule EventSourcingDB do
     "/books/42",
     %EventSourcingDB.ReadEventsOptions{
       recursive: false,
-      from_latest_event: %EventSourcingDB.FromLatestEventOptions{
+      from_latest_event: %EventSourcingDB.ReadFromLatestEventOptions{
         subject: "/books/42",
         type: "io.eventsourcingdb.library.book-borrowed",
         if_event_is_missing: :read_everything
@@ -303,6 +303,10 @@ defmodule EventSourcingDB do
 
   *Note that `from_latest_event` and `lower_bound` can not be provided at the same time.*
 
+  ### Aborting Reading
+
+  Since `read_events` returns a lazy stream, you can stop consuming it at any time. Use `Enum.take/2` to limit the number of events, or simply stop iterating. If you are consuming the stream in a separate process, you can abort by terminating that process.
+
   """
   @spec read_events(Client.t(), String.t(), ReadEventsOptions.t() | nil) ::
           stream_response(Event.t())
@@ -378,7 +382,7 @@ defmodule EventSourcingDB do
     "/books/42",
     %EventSourcingDB.ObserveEventsOptions{
       recursive: false,
-      from_latest_event: %EventSourcingDB.FromLatestEventOptions{
+      from_latest_event: %EventSourcingDB.ObserveFromLatestEventOptions{
         subject: "/books/42",
         type: "io.eventsourcingdb.library.book-borrowed",
         if_event_is_missing: :read_everything
@@ -389,6 +393,10 @@ defmodule EventSourcingDB do
 
   *Note that `from_latest_event` and `lower_bound` can not be provided at the same time.*
 
+  ### Aborting Observing
+
+  Since `observe_events` returns a lazy stream, you can stop consuming it at any time. Use `Enum.take/2` to limit the number of events, or simply stop iterating. If you are consuming the stream in a separate process, you can abort by terminating that process.
+
   """
   @spec observe_events(Client.t(), String.t(), ObserveEventsOptions.t() | nil) ::
           stream_response(Event.t())
@@ -416,6 +424,10 @@ defmodule EventSourcingDB do
 
   *Note that each row returned by the stream matches the projection specified in your query.*
 
+  ### Aborting a Query
+
+  Since `run_eventql_query` returns a lazy stream, you can stop consuming it at any time. Use `Enum.take/2` to limit the number of rows, or simply stop iterating.
+
   """
   @spec run_eventql_query(Client.t(), String.t()) :: stream_response(any())
   def run_eventql_query(client, query) do
@@ -479,6 +491,11 @@ defmodule EventSourcingDB do
   ```elixir
   EventSourcingDB.read_subjects(client, "/books")
   ```
+
+  ### Aborting Listing
+
+  Since `read_subjects` returns a lazy stream, you can stop consuming it at any time by simply stopping iteration or using `Enum.take/2`.
+
   """
   @spec read_subjects(Client.t(), String.t()) :: stream_response(String.t())
   def read_subjects(client, base_subject) do
@@ -523,6 +540,11 @@ defmodule EventSourcingDB do
     {:error, reason} -> # ...
   end
   ```
+
+  ### Aborting Listing
+
+  Since `read_event_types` returns a lazy stream, you can stop consuming it at any time by simply stopping iteration or using `Enum.take/2`.
+
   """
   @spec read_event_types(Client.t()) :: stream_response(EventType.t())
   def read_event_types(client) do
@@ -609,15 +631,15 @@ defmodule EventSourcingDB do
             {[message], response}
 
           {:error, reason} ->
-            {:error, reason}
+            raise(reason)
 
           # handle heartbeat case
           nil ->
             {[], response}
         end
 
       {:error, reason} ->
-        {:error, reason}
+        raise(reason)
 
       # This is returned when the stream is done.
       {:ok, [:done]} ->

lib/eventsourcingdb/events/event.ex

@@ -15,15 +15,21 @@ defmodule EventSourcingDB.Event do
 
   @signature_prefix "esdb:signature:v1:"
 
+  @key_mapping %{
+    "predecessorhash" => :predecessor_hash,
+    "datacontenttype" => :data_content_type,
+    "specversion" => :spec_version
+  }
+
   typedstruct do
     field :data, any(), enforce: true
-    field :datacontenttype, String.t(), enforce: true
+    field :data_content_type, String.t(), enforce: true
     field :hash, String.t(), enforce: true
     field :id, String.t(), enforce: true
-    field :predecessorhash, String.t(), enforce: true
+    field :predecessor_hash, String.t(), enforce: true
     field :signature, String.t()
     field :source, String.t(), enforce: true
-    field :specversion, String.t(), enforce: true
+    field :spec_version, String.t(), enforce: true
     field :subject, String.t(), enforce: true
     field :time, String.t(), enforce: true
     field :traceparent, String.t()
@@ -33,7 +39,14 @@ defmodule EventSourcingDB.Event do
 
   @spec new(map()) :: Event.t()
   def new(value \\ %{}) do
-    struct!(__MODULE__, value |> Map.new(fn {k, v} -> {String.to_existing_atom(k), v} end))
+    struct!(
+      __MODULE__,
+      value
+      |> Map.new(fn {k, v} ->
+        key = Map.get_lazy(@key_mapping, k, fn -> String.to_existing_atom(k) end)
+        {key, v}
+      end)
+    )
   end
 
   @doc """
@@ -55,7 +68,7 @@ defmodule EventSourcingDB.Event do
   @spec verify_hash(Event.t()) :: :ok | {:error, HashVerificationFailed.t()}
   def verify_hash(event) do
     metadata =
-      "#{event.specversion}|#{event.id}|#{event.predecessorhash}|#{event.time}|#{event.source}|#{event.subject}|#{event.type}|#{event.datacontenttype}"
+      "#{event.spec_version}|#{event.id}|#{event.predecessor_hash}|#{event.time}|#{event.source}|#{event.subject}|#{event.type}|#{event.data_content_type}"
 
     metadata_hash = :crypto.hash(:sha256, metadata) |> Base.encode16(case: :lower)
     data_hash = :crypto.hash(:sha256, Jason.encode!(event.data)) |> Base.encode16(case: :lower)

lib/eventsourcingdb/events/management_event.ex

@@ -5,19 +5,31 @@ defmodule EventSourcingDB.ManagementEvent do
   alias EventSourcingDB.ManagementEvent
   use TypedStruct
 
+  @key_mapping %{
+    "datacontenttype" => :data_content_type,
+    "specversion" => :spec_version
+  }
+
   typedstruct enforce: true do
     field :data, any()
-    field :datacontenttype, String.t()
+    field :data_content_type, String.t()
     field :id, String.t()
     field :source, String.t()
-    field :specversion, String.t()
+    field :spec_version, String.t()
     field :subject, String.t()
     field :time, String.t()
     field :type, String.t()
   end
 
   @spec new(map()) :: ManagementEvent.t()
   def new(value \\ %{}) do
-    struct!(__MODULE__, value |> Map.new(fn {k, v} -> {String.to_existing_atom(k), v} end))
+    struct!(
+      __MODULE__,
+      value
+      |> Map.new(fn {k, v} ->
+        key = Map.get_lazy(@key_mapping, k, fn -> String.to_existing_atom(k) end)
+        {key, v}
+      end)
+    )
   end
 end

lib/eventsourcingdb/request_options/observe_events.ex

@@ -5,8 +5,8 @@ defmodule EventSourcingDB.ObserveEventsOptions do
   use TypedStruct
 
   typedstruct do
-    field :recursive, boolean(), enforce: true
-    field :from_latest_event, EventSourcingDB.FromLatestEventOptions.t()
+    field :recursive, boolean(), default: false
+    field :from_latest_event, EventSourcingDB.ObserveFromLatestEventOptions.t()
     field :lower_bound, EventSourcingDB.BoundOptions.t()
   end
 

lib/eventsourcingdb/request_options/observe_from_latest_event.ex

@@ -0,0 +1,40 @@
+defmodule EventSourcingDB.ObserveFromLatestEventOptions do
+  @moduledoc """
+  Options for observing events starting from the latest event of a given type.
+  """
+  use TypedStruct
+
+  typedstruct enforce: true do
+    field :if_event_is_missing, :read_everything | :wait_for_event
+    field :subject, String.t()
+    field :type, String.t()
+  end
+
+  @spec new(keyword()) :: t()
+  def new(options) do
+    options =
+      options
+      |> Keyword.validate!([:if_event_is_missing, :subject, :type])
+
+    struct!(__MODULE__, options)
+  end
+
+  defimpl Jason.Encoder do
+    @spec encode(EventSourcingDB.ObserveFromLatestEventOptions.t(), Jason.Encode.opts()) ::
+            iodata()
+    def encode(value, opts) do
+      Jason.Encode.map(
+        %{
+          "ifEventIsMissing" =>
+            case value.if_event_is_missing do
+              :read_everything -> "read-everything"
+              :wait_for_event -> "wait-for-event"
+            end,
+          "subject" => value.subject,
+          "type" => value.type
+        },
+        opts
+      )
+    end
+  end
+end

lib/eventsourcingdb/request_options/read_events.ex

@@ -5,9 +5,9 @@ defmodule EventSourcingDB.ReadEventsOptions do
   use TypedStruct
 
   typedstruct do
-    field :recursive, boolean(), enforce: true
+    field :recursive, boolean(), default: false
     field :order, :chronological | :antichronological
-    field :from_latest_event, EventSourcingDB.FromLatestEventOptions.t()
+    field :from_latest_event, EventSourcingDB.ReadFromLatestEventOptions.t()
     field :lower_bound, EventSourcingDB.BoundOptions.t()
     field :upper_bound, EventSourcingDB.BoundOptions.t()
   end

lib/eventsourcingdb/request_options/from_latest_event.ex → lib/eventsourcingdb/request_options/read_from_latest_event.ex

@@ -1,11 +1,11 @@
-defmodule EventSourcingDB.FromLatestEventOptions do
+defmodule EventSourcingDB.ReadFromLatestEventOptions do
   @moduledoc """
-  Reading events from when a certain event occured.
+  Options for reading events starting from the latest event of a given type.
   """
   use TypedStruct
 
   typedstruct enforce: true do
-    field :if_event_is_missing, :read_everything | :read_nothing | :wait_for_event
+    field :if_event_is_missing, :read_everything | :read_nothing
     field :subject, String.t()
     field :type, String.t()
   end
@@ -20,7 +20,7 @@ defmodule EventSourcingDB.FromLatestEventOptions do
   end
 
   defimpl Jason.Encoder do
-    @spec encode(EventSourcingDB.FromLatestEventOptions.t(), Jason.Encode.opts()) ::
+    @spec encode(EventSourcingDB.ReadFromLatestEventOptions.t(), Jason.Encode.opts()) ::
             iodata()
     def encode(value, opts) do
       Jason.Encode.map(
@@ -29,7 +29,6 @@ defmodule EventSourcingDB.FromLatestEventOptions do
             case value.if_event_is_missing do
               :read_everything -> "read-everything"
               :read_nothing -> "read-nothing"
-              :wait_for_event -> "wait-for-event"
             end,
           "subject" => value.subject,
           "type" => value.type

mix.exs

@@ -70,7 +70,8 @@ defmodule EventSourcingDB.MixProject do
         ],
         "Request Options": [
           EventSourcingDB.BoundOptions,
-          EventSourcingDB.FromLatestEventOptions,
+          EventSourcingDB.ObserveFromLatestEventOptions,
+          EventSourcingDB.ReadFromLatestEventOptions,
           EventSourcingDB.ObserveEventsOptions,
           EventSourcingDB.ReadEventsOptions
         ],