diff --git a/README.md b/README.md index ea7cb8e..27eebda 100644 --- a/README.md +++ b/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: diff --git a/lib/eventsourcingdb.ex b/lib/eventsourcingdb.ex index 011bd81..f6b32ca 100644 --- a/lib/eventsourcingdb.ex +++ b/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,7 +631,7 @@ defmodule EventSourcingDB do {[message], response} {:error, reason} -> - {:error, reason} + raise(reason) # handle heartbeat case nil -> @@ -617,7 +639,7 @@ defmodule EventSourcingDB do end {:error, reason} -> - {:error, reason} + raise(reason) # This is returned when the stream is done. {:ok, [:done]} -> diff --git a/lib/eventsourcingdb/events/event.ex b/lib/eventsourcingdb/events/event.ex index 03b0564..810ac85 100644 --- a/lib/eventsourcingdb/events/event.ex +++ b/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) diff --git a/lib/eventsourcingdb/events/management_event.ex b/lib/eventsourcingdb/events/management_event.ex index c94f620..037a731 100644 --- a/lib/eventsourcingdb/events/management_event.ex +++ b/lib/eventsourcingdb/events/management_event.ex @@ -5,12 +5,17 @@ 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() @@ -18,6 +23,13 @@ defmodule EventSourcingDB.ManagementEvent do @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 diff --git a/lib/eventsourcingdb/request_options/observe_events.ex b/lib/eventsourcingdb/request_options/observe_events.ex index 2a24c67..68318f9 100644 --- a/lib/eventsourcingdb/request_options/observe_events.ex +++ b/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 diff --git a/lib/eventsourcingdb/request_options/observe_from_latest_event.ex b/lib/eventsourcingdb/request_options/observe_from_latest_event.ex new file mode 100644 index 0000000..fe1cd4e --- /dev/null +++ b/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 diff --git a/lib/eventsourcingdb/request_options/read_events.ex b/lib/eventsourcingdb/request_options/read_events.ex index a7386b5..6af0e95 100644 --- a/lib/eventsourcingdb/request_options/read_events.ex +++ b/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 diff --git a/lib/eventsourcingdb/request_options/from_latest_event.ex b/lib/eventsourcingdb/request_options/read_from_latest_event.ex similarity index 76% rename from lib/eventsourcingdb/request_options/from_latest_event.ex rename to lib/eventsourcingdb/request_options/read_from_latest_event.ex index 0cd6083..9fcc8c6 100644 --- a/lib/eventsourcingdb/request_options/from_latest_event.ex +++ b/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 diff --git a/mix.exs b/mix.exs index 964d2d0..33071ef 100644 --- a/mix.exs +++ b/mix.exs @@ -70,7 +70,8 @@ defmodule EventSourcingDB.MixProject do ], "Request Options": [ EventSourcingDB.BoundOptions, - EventSourcingDB.FromLatestEventOptions, + EventSourcingDB.ObserveFromLatestEventOptions, + EventSourcingDB.ReadFromLatestEventOptions, EventSourcingDB.ObserveEventsOptions, EventSourcingDB.ReadEventsOptions ], diff --git a/test/observe_events_test.exs b/test/observe_events_test.exs index 38b074c..b863313 100644 --- a/test/observe_events_test.exs +++ b/test/observe_events_test.exs @@ -2,7 +2,7 @@ defmodule EventSourcingDBTest.ObserveEvents do alias EventSourcingDB.TestContainer alias EventSourcingDB.ObserveEventsOptions alias EventSourcingDB.BoundOptions - alias EventSourcingDB.FromLatestEventOptions + alias EventSourcingDB.ObserveFromLatestEventOptions import EventSourcingDBTest.Utils use ExUnit.Case, async: true @@ -70,7 +70,7 @@ defmodule EventSourcingDBTest.ObserveEvents do events = EventSourcingDB.observe_events!(client, "/", %ObserveEventsOptions{ recursive: true, - from_latest_event: %FromLatestEventOptions{ + from_latest_event: %ObserveFromLatestEventOptions{ subject: "/test", type: "io.eventsourcingdb.test.bar", if_event_is_missing: :read_everything diff --git a/test/read_events_test.exs b/test/read_events_test.exs index 968983f..270aa4b 100644 --- a/test/read_events_test.exs +++ b/test/read_events_test.exs @@ -1,7 +1,7 @@ defmodule EventSourcingDBTest.ReadEvents do alias EventSourcingDB.Errors.TransmissionError alias EventSourcingDB.Client - alias EventSourcingDB.FromLatestEventOptions + alias EventSourcingDB.ReadFromLatestEventOptions alias EventSourcingDB.BoundOptions alias EventSourcingDB.ReadEventsOptions alias EventSourcingDB.TestContainer @@ -181,7 +181,7 @@ defmodule EventSourcingDBTest.ReadEvents do events = EventSourcingDB.read_events!(client, "/test", %ReadEventsOptions{ - from_latest_event: %FromLatestEventOptions{ + from_latest_event: %ReadFromLatestEventOptions{ subject: "/", type: "io.eventsourcingdb.test.does-not-exist", if_event_is_missing: :read_nothing @@ -204,7 +204,7 @@ defmodule EventSourcingDBTest.ReadEvents do events = EventSourcingDB.read_events!(client, "/test", %ReadEventsOptions{ - from_latest_event: %FromLatestEventOptions{ + from_latest_event: %ReadFromLatestEventOptions{ subject: "/marker", type: "io.eventsourcingdb.test", if_event_is_missing: :read_nothing diff --git a/test/support/utils.ex b/test/support/utils.ex index d36f7cf..fb742da 100644 --- a/test/support/utils.ex +++ b/test/support/utils.ex @@ -40,7 +40,7 @@ defmodule EventSourcingDBTest.Utils do assert_event_match_eventcandidate(event, candidate) assert event.id == Integer.to_string(expected_id), "ID should match" - assert event.predecessorhash == previous_event_hash, "Previous hash should match" + assert event.predecessor_hash == previous_event_hash, "Previous hash should match" end @spec assert_event_match_eventcandidate(Event.t(), EventCandidate.t()) :: any() @@ -52,13 +52,13 @@ defmodule EventSourcingDBTest.Utils do assert event.type == candidate.type, "Type mismatch" # check metadata - assert event.datacontenttype == "application/json", + assert event.data_content_type == "application/json", "Data content type should be application/json" assert String.length(event.hash) == 64, "Hash should be present" assert not is_nil(event.id), "ID should be present" - assert not is_nil(event.predecessorhash), "Previous hash should be present" - assert event.specversion == "1.0", "Spec version should be 1.0" + assert not is_nil(event.predecessor_hash), "Previous hash should be present" + assert event.spec_version == "1.0", "Spec version should be 1.0" end @spec assert_event_match_eventcandidates([Event.t()], [EventCandidate.t()]) :: any()