EventStore behaviour (EventStore v1.2.1) View Source
EventStore allows you to define one or more event store modules to append, read, and subscribe to streams of events.
It uses PostgreSQL (v9.5 or later) as the underlying storage engine.
Defining an event store
An event store module is defined in your own application as follows:
defmodule MyApp.EventStore do
use EventStore, otp_app: :my_app
# Optional `init/1` function to modify config at runtime.
def init(config) do
{:ok, config}
end
end
Where the configuration for the event store must be in your application
environment, usually defined in config/config.exs
:
config :my_app, MyApp.EventStore,
serializer: EventStore.JsonSerializer,
username: "postgres",
password: "postgres",
database: "eventstore",
hostname: "localhost"
Or use a URL to connect instead:
config :my_app, MyApp.EventStore,
serializer: EventStore.JsonSerializer,
url: "postgres://postgres:postgres@localhost/eventstore"
Note: To use an EventStore with Commanded you should configure the event store to use Commanded's JSON serializer which provides additional support for JSON decoding:
config :my_app, MyApp.EventStore,
serializer: Commanded.Serialization.JsonSerializer
The event store module defines a start_link/1
function that needs to be
invoked before using the event store. In general, this function is not
called directly, but included as part of your application supervision tree.
If your application was generated with a supervisor (by passing --sup
to mix new
) you will have a lib/my_app/application.ex
file
containing the application start callback that defines and starts your
supervisor. You just need to edit the start/2
function to start the event
store in your application's supervisor:
def start(_type, _args) do
children = [
MyApp.EventStore
]
opts = [strategy: :one_for_one, name: MyApp.Supervisor]
Supervisor.start_link(children, opts)
end
Each event store module (e.g. MyApp.EventStore
) provides a public API to
read events from and write events to an event stream, and subscribe to event
notifications.
Postgres schema
By default the public
schema will be used for event store tables. An event
store can be configured to use an alternate Postgres schema:
defmodule MyApp.EventStore do
use EventStore, otp_app: :my_app, schema: "schema_name"
end
Or provide the schema as an option in the init/1
callback function:
defmodule MyApp.EventStore do
use EventStore, otp_app: :my_app
def init(config) do
{:ok, Keyword.put(config, :schema, "schema_name")}
end
end
Or define it in environment config when configuring the database connection settings:
# config/config.exs
config :my_app, MyApp.EventStore, schema: "schema_name"
This feature allows you to define and start multiple event stores sharing a single Postgres database, but with their data isolated and segregated by schema.
Note the mix event_store.<task>
tasks to create, initialize, and drop an
event store database will also handle creating and/or dropping the schema.
Dynamic named event store
An event store can be started multiple times by providing a name when starting. The name must be provided as an option to all event store operations to identify the correct instance.
Example
Define an event store:
defmodule MyApp.EventStore do
use EventStore, otp_app: :my_app
end
Start multiple instances of the event store, each with a unique name:
{:ok, _pid} = EventStore.start_link(name: :eventstore1)
{:ok, _pid} = EventStore.start_link(name: :eventstore2)
{:ok, _pid} = EventStore.start_link(name: :eventstore3)
Use a dynamic event store by providing its name as an option to each function:
:ok = EventStore.append_to_stream(stream_uuid, expected_version, events, name: :eventstore1)
{:ok, events} = EventStore.read_stream_forward(stream_uuid, 0, 1_000, name: :eventstore1)
Dynamic schemas
This feature also allows you to start each event store instance using a different schema:
{:ok, _pid} = EventStore.start_link(name: :tenant1, schema: "tenant1")
{:ok, _pid} = EventStore.start_link(name: :tenant2, schema: "tenant2")
Or start supervised:
children =
for tenant <- [:tenant1, :tenant2, :tenant3] do
{MyApp.EventStore, name: tenant, schema: "#{tenant}"}
end
opts = [strategy: :one_for_one, name: MyApp.Supervisor]
Supervisor.start_link(children, opts)
The above can be used for multi-tenancy where the data for each tenant is stored in a separate, isolated schema.
Shared database connection pools
By default each event store will start its own Postgrex
database connection
pool. The size of the pool is configured with the pool_size
config option.
When you have multiple event stores running you will also end up with multiple
connection pools. If they are all connecting to the same physical Postgres
database then it can be useful to share a single pool amongst all event
stores. Use the shared_connection_pool
config option to specify a name for
the shared connection pool. Then configure the event stores you'd like to
share the pool with the same name.
This can be done in config:
# config/config.exs
config :my_app, MyApp.EventStore, shared_connection_pool: :shared_pool
Or when starting the event stores, such as via a Supervisor
:
Supervisor.start_link(
[
{MyApp.EventStore, name: :eventstore1, shared_connection_pool: :shared_pool},
{MyApp.EventStore, name: :eventstore2, shared_connection_pool: :shared_pool},
{MyApp.EventStore, name: :eventstore3, shared_connection_pool: :shared_pool}
], opts)
Using an existing database connection or transaction
In some situations you might want to execute the event store operations using an existing Postgres database connection or transaction. For instance, if you want to persist changes to one or more other tables, such as a read-model projection.
To do this you can provide a Postgrex connection process or transaction as a
:conn
option to any of the supported EventStore
functions.
{:ok, pid} = Postgrex.start_link(config)
Postgrex.transaction(pid, fn conn ->
:ok = EventStore.append_to_stream(stream_uuid, expected_version, events, conn: conn)
end)
This can also be used with an Ecto Repo
which is configured to use the
Postgres SQL adapter. The connection process may be looked up as follows:
Repo.transaction(fn ->
%{pid: pool} = Ecto.Adapter.lookup_meta(Repo)
conn = Process.get({Ecto.Adapters.SQL, pool})
:ok = EventStore.append_to_stream(stream_uuid, expected_version, events, conn: conn)
end)
Guides
Please refer to the following guides to learn more:
- Getting started
- Usage
- Subscriptions
- Running on a cluster of nodes
- Event serialization
- Upgrading an existing EventStore database
Link to this section Summary
Callbacks
Acknowledge receipt of the given events received from a subscription.
Append one or more events to a stream atomically.
Returns the event store configuration stored in the :otp_app
environment.
Delete an existing persistent subscription to all streams.
Delete a previously recorded snapshop for a given source.
Delete an existing stream.
Delete an existing persistent subscription.
A callback executed when the event store starts or when configuration is read.
Link one or more existing events to another stream.
Reads the requested number of events from all streams, in the order in which they were originally written.
Read a snapshot, if available, for a given source.
Reads the requested number of events from the given stream, in the order in which they were originally written.
Record a snapshot of the data and metadata for a given source.
Starts any connection pooling or supervision and return {:ok, pid}
or just :ok
if nothing needs to be done.
Shuts down the event store.
Streams events from all streams, in the order in which they were originally written.
Streams events from the given stream, in the order in which they were originally written.
Create a transient subscription to a given stream.
Create a subscription to all streams. By default the subscription is persistent.
Create a subscription to a single stream. By default the subscription is persistent.
Unsubscribe an existing subscriber from all event notifications.
Unsubscribe an existing subscriber from event notifications.
Link to this section Types
Specs
expected_version() :: :any_version | :no_stream | :stream_exists | non_neg_integer()
Specs
option() :: {:name, atom()} | {:conn, Postgrex.conn() | DBConnection.t()} | {:timeout, timeout()}
Specs
options() :: [option()]
Specs
persistent_subscription_option() :: transient_subscribe_option() | {:concurrency_limit, pos_integer()} | {:buffer_size, pos_integer()} | {:start_from, :origin | :current | non_neg_integer()} | {:partition_by, (EventStore.RecordedEvent.t() -> any())}
Specs
persistent_subscription_options() :: [persistent_subscription_option()]
Specs
start_from() :: :origin | :current | non_neg_integer()
Specs
t() :: module()
Specs
transient_subscribe_option() :: {:name, atom()} | {:selector, (EventStore.RecordedEvent.t() -> any())} | {:mapper, (EventStore.RecordedEvent.t() -> any())}
Specs
transient_subscribe_options() :: [transient_subscribe_option()]
Link to this section Callbacks
Specs
ack( subscription :: pid(), EventStore.RecordedEvent.t() | [EventStore.RecordedEvent.t()] | non_neg_integer() ) :: :ok | {:error, reason :: term()}
Acknowledge receipt of the given events received from a subscription.
Accepts a single EventStore.RecordedEvent
struct, a list of
EventStore.RecordedEvent
s, or the event number of the recorded event to
acknowledge.
Specs
append_to_stream( stream_uuid :: String.t(), expected_version(), events :: [EventStore.EventData.t()], opts :: options() ) :: :ok | {:error, :cannot_append_to_all_stream} | {:error, :stream_exists} | {:error, :stream_not_found} | {:error, :wrong_expected_version} | {:error, :stream_deleted} | {:error, reason :: term()}
Append one or more events to a stream atomically.
stream_uuid
is used to uniquely identify a stream.expected_version
is used for optimistic concurrency checks. You can provide a non-negative integer to specify the expected stream version. This is used to ensure you can only append to the stream if it is at exactly that version.You can also provide one of the following values to alter the concurrency check behaviour:
:any_version
- No concurrency checking and allow any stream version (including no stream).:no_stream
- Ensure the stream does not exist.:stream_exists
- Ensure the stream exists.
events
is a list of%EventStore.EventData{}
structs.opts
an optional keyword list containing:name
the name of the event store if provided tostart_link/1
.timeout
an optional timeout for the database transaction, in milliseconds. Defaults to 15,000ms.
Returns :ok
on success, or an {:error, reason}
tagged tuple. The returned
error may be due to one of the following reasons:
{:error, :wrong_expected_version}
when the actual stream version differs from the provided expected version.{:error, :stream_exists}
when the stream exists, but expected version was:no_stream
.{:error, :stream_not_found}
when the stream does not exist, but expected version was:stream_exists
.
Specs
config() :: Keyword.t()
Returns the event store configuration stored in the :otp_app
environment.
Specs
delete_all_streams_subscription( subscription_name :: String.t(), opts :: options() ) :: :ok | {:error, term()}
Delete an existing persistent subscription to all streams.
subscription_name
is used to identify the existing subscription to remove.
Returns :ok
on success.
Specs
Delete a previously recorded snapshop for a given source.
Returns :ok
on success, or when the snapshot does not exist.
Specs
delete_stream( stream_uuid :: String.t(), expected_version :: :any_version | :stream_exists | non_neg_integer(), type :: :soft | :hard, opts :: Keyword.t() ) :: :ok | {:error, :stream_not_found} | {:error, :stream_deleted} | {:error, term()}
Delete an existing stream.
stream_uuid
identity of the stream to be deleted.expected_version
is used for optimistic concurrency checking. You can provide a non-negative integer to specify the expected stream version. This is used to ensure you can only delete a stream if it is at exactly that version.You can also provide one of the following values to alter the concurrency checking behaviour:
:any_version
- No concurrency check, allow any stream version.:stream_exists
- Ensure the stream exists, at any version.
type
- used to indicate how the stream is deleted::soft
- the stream is marked as deleted, but no events are removed.:hard
- the stream and its events are permanently deleted from the database. Soft deletion is the default if the type is not provided.
Returns :ok
on success or an error tagged tuple on failure.
Soft delete
Will mark the stream as deleted, but will not delete its events. Events from
soft deleted streams will still appear in the globally ordered all events
($all
) stream and in any linked streams.
A soft deleted stream cannot be read nor appended to. Subscriptions to the deleted stream will not receive any events but subscriptions containing linked events from the deleted stream, such as the global all events stream, will still receive events from the deleted stream.
Hard delete
Will permanently delete the stream and its events. This is irreversible and will remove data. Events will be removed from the globally ordered all events stream and any linked streams.
After being hard deleted, a stream can later be appended to and read as if it had never existed.
Examples
Soft delete a stream
Delete a stream at any version:
:ok = MyApp.EventStore.delete_stream("stream1", :any_version, :soft)
Delete a stream at an expected version:
:ok = MyApp.EventStore.delete_stream("stream2", 3, :soft)
Delete stream will use soft delete by default so you can omit the type:
:ok = MyApp.EventStore.delete_stream("stream1", :any_version)
Hard delete a stream
Since hard deletes are destructive and irreversible they are disabled by default. To use hard deletes you must first enable them for the event store:
defmodule MyApp.EventStore do
use EventStore, otp_app: :my_app, enable_hard_deletes: true
end
Or via config:
# config/config.exs
config :my_app, MyApp.EventStore, enable_hard_deletes: true
Hard delete a stream at any version:
:ok = MyApp.EventStore.delete_stream("stream1", :any_version, :hard)
Hard delete a stream that should exist:
:ok = MyApp.EventStore.delete_stream("stream2", :stream_exists, :hard)
Specs
delete_subscription( stream_uuid :: String.t(), subscription_name :: String.t(), opts :: options() ) :: :ok | {:error, term()}
Delete an existing persistent subscription.
stream_uuid
is the stream the subscription is subscribed to.subscription_name
is used to identify the existing subscription to remove.
Returns :ok
on success.
Specs
A callback executed when the event store starts or when configuration is read.
It must return {:ok, keyword}
with the updated list of configuration.
Specs
link_to_stream( stream_uuid :: String.t(), expected_version(), events :: [EventStore.RecordedEvent.t()] | [non_neg_integer()], opts :: options() ) :: :ok | {:error, :cannot_append_to_all_stream} | {:error, :stream_exists} | {:error, :stream_not_found} | {:error, :wrong_expected_version} | {:error, :stream_deleted} | {:error, reason :: term()}
Link one or more existing events to another stream.
Allows you to construct streams containing events already appended to any other stream. This is more efficient than copying events between streams since only a reference to the existing event is created.
stream_uuid
is used to uniquely identify the target stream.expected_version
is used for optimistic concurrency checks. You can provide a non-negative integer to specify the expected stream version. This is used to ensure you can only append to the stream if it is at exactly that version.You can also provide one of the following values to affect the concurrency check behaviour:
:any_version
- No concurrency checking; allow any stream version (including no stream).:no_stream
- Ensure the stream does not exist.:stream_exists
- Ensure the stream exists.
events_or_event_ids
is a list of%EventStore.EventData{}
structs or event ids.opts
an optional keyword list containing:name
the name of the event store if provided tostart_link/1
.timeout
an optional timeout for the database transaction, in milliseconds. Defaults to 15,000ms.
Returns :ok
on success, or an {:error, reason}
tagged tuple. The returned
error may be due to one of the following reasons:
{:error, :wrong_expected_version}
when the actual stream version differs from the provided expected version.{:error, :stream_exists}
when the stream exists, but expected version was:no_stream
.{:error, :stream_not_found}
when the stream does not exist, but expected version was:stream_exists
.
Specs
read_all_streams_forward( start_version :: non_neg_integer(), count :: non_neg_integer(), opts :: options() ) :: {:ok, [EventStore.RecordedEvent.t()]} | {:error, reason :: term()}
Reads the requested number of events from all streams, in the order in which they were originally written.
start_version
optionally, the number of the first event to read. Defaults to the beginning of the stream if not set.count
optionally, the maximum number of events to read. If not set it will be limited to returning 1,000 events from all streams.opts
an optional keyword list containing:name
the name of the event store if provided tostart_link/1
.timeout
an optional timeout for the database transaction, in milliseconds. Defaults to 15,000ms.
Specs
read_snapshot(source_uuid :: String.t(), opts :: options()) :: {:ok, EventStore.Snapshots.SnapshotData.t()} | {:error, :snapshot_not_found}
Read a snapshot, if available, for a given source.
Returns {:ok, %EventStore.Snapshots.SnapshotData{}}
on success, or
{:error, :snapshot_not_found}
when unavailable.
Specs
read_stream_forward( stream_uuid :: String.t(), start_version :: non_neg_integer(), count :: non_neg_integer(), opts :: options() ) :: {:ok, [EventStore.RecordedEvent.t()]} | {:error, :stream_deleted} | {:error, reason :: term()}
Reads the requested number of events from the given stream, in the order in which they were originally written.
stream_uuid
is used to uniquely identify a stream.start_version
optionally, the version number of the first event to read. Defaults to the beginning of the stream if not set.count
optionally, the maximum number of events to read. If not set it will be limited to returning 1,000 events from the stream.opts
an optional keyword list containing:name
the name of the event store if provided tostart_link/1
.timeout
an optional timeout for the database transaction, in milliseconds. Defaults to 15,000ms.
Specs
record_snapshot( snapshot :: EventStore.Snapshots.SnapshotData.t(), opts :: options() ) :: :ok | {:error, reason :: term()}
Record a snapshot of the data and metadata for a given source.
Returns :ok
on success.
Specs
start_link(opts :: Keyword.t()) :: {:ok, pid()} | {:error, {:already_started, pid()}} | {:error, term()}
Starts any connection pooling or supervision and return {:ok, pid}
or just :ok
if nothing needs to be done.
Returns {:error, {:already_started, pid}}
if the event store is already
started or {:error, term}
in case anything else goes wrong.
Specs
stop(Supervisor.supervisor(), timeout()) :: :ok
Shuts down the event store.
Specs
stream_all_forward( start_version :: non_neg_integer(), opts :: [options() | {:read_batch_size, non_neg_integer()}] ) :: Enumerable.t() | {:error, :stream_deleted} | {:error, reason :: term()}
Streams events from all streams, in the order in which they were originally written.
start_version
optionally, the number of the first event to read. Defaults to the beginning of the stream if not set.opts
an optional keyword list containing:name
the name of the event store if provided tostart_link/1
.timeout
an optional timeout for the database transaction, in milliseconds. Defaults to 15,000ms.read_batch_size
optionally, the number of events to read at a time from storage. Defaults to reading 1,000 events per batch.
Specs
stream_forward( stream_uuid :: String.t(), start_version :: non_neg_integer(), opts :: [options() | {:read_batch_size, non_neg_integer()}] ) :: Enumerable.t() | {:error, :stream_deleted} | {:error, reason :: term()}
Streams events from the given stream, in the order in which they were originally written.
start_version
optionally, the version number of the first event to read. Defaults to the beginning of the stream if not set.opts
an optional keyword list containing:name
the name of the event store if provided tostart_link/1
.timeout
an optional timeout for the database transaction, in milliseconds. Defaults to 15,000ms.read_batch_size
optionally, the number of events to read at a time from storage. Defaults to reading 1,000 events per batch.
Specs
subscribe(stream_uuid :: String.t(), opts :: transient_subscribe_options()) :: :ok | {:error, term()}
Create a transient subscription to a given stream.
stream_uuid
is the stream to subscribe to. Use the$all
identifier to subscribe to events from all streams.opts
is an optional map providing additional subscription configuration:name
the name of the event store if provided tostart_link/1
.selector
to define a function to filter each event, i.e. returns only those elements for which fun returns a truthy valuemapper
to define a function to map each recorded event before sending to the subscriber.
The calling process will be notified whenever new events are appended to
the given stream_uuid
.
As the subscription is transient you do not need to acknowledge receipt of
each event. The subscriber process will miss any events if it is restarted
and resubscribes. If you need a persistent subscription with guaranteed
at-least-once event delivery and back-pressure you should use
EventStore.subscribe_to_stream/4
.
Notification message
Events will be sent to the subscriber, in batches, as {:events, events}
where events is a collection of EventStore.RecordedEvent
structs.
Example
{:ok, subscription} = EventStore.subscribe(stream_uuid)
# receive first batch of events
receive do
{:events, events} ->
IO.puts "Received events: " <> inspect(events)
end
Specs
subscribe_to_all_streams( subscription_name :: String.t(), subscriber :: pid(), opts :: persistent_subscription_options() ) :: {:ok, subscription :: pid()} | {:error, :already_subscribed} | {:error, :subscription_already_exists} | {:error, :too_many_subscribers} | {:error, reason :: term()}
Create a subscription to all streams. By default the subscription is persistent.
The subscriber
process will be notified of each batch of events appended to
any stream.
subscription_name
is used to uniquely identify the subscription.subscriber
is a process that will be sent{:events, events}
notification messages.opts
is an optional map providing additional subscription configuration:name
the name of the event store if provided tostart_link/1
.start_from
is a pointer to the first event to receive. It must be one of::origin
for all events from the start of the stream (default).:current
for any new events appended to the stream after the subscription has been created.- any positive integer for an event id to receive events after that exact event.
selector
to define a function to filter each event, i.e. returns only those elements for which fun returns a truthy valuemapper
to define a function to map each recorded event before sending to the subscriber.concurrency_limit
defines the maximum number of concurrent subscribers allowed to connect to the subscription. By default only one subscriber may connect. If too many subscribers attempt to connect to the subscription an{:error, :too_many_subscribers}
is returned.transient
is an optional boolean flag to create a transient subscription. Seesubscribe_to_stream
for the full information.
The subscription will resume from the last acknowledged event if it already
exists. It will ignore the start_from
argument in this case.
Returns {:ok, subscription}
when subscription succeeds.
Example
{:ok, subscription} = EventStore.subscribe_to_all_streams("all_subscription", self())
# wait for the subscription confirmation
receive do
{:subscribed, ^subscription} ->
IO.puts "Successfully subscribed to all streams"
end
receive do
{:events, events} ->
IO.puts "Received events: " <> inspect(events)
# acknowledge receipt
EventStore.ack(subscription, events)
end
subscribe_to_stream(stream_uuid, subscription_name, subscriber, opts)
View SourceSpecs
subscribe_to_stream( stream_uuid :: String.t(), subscription_name :: String.t(), subscriber :: pid(), opts :: persistent_subscription_options() ) :: {:ok, subscription :: pid()} | {:error, :already_subscribed} | {:error, :subscription_already_exists} | {:error, :too_many_subscribers} | {:error, reason :: term()}
Create a subscription to a single stream. By default the subscription is persistent.
The subscriber
process will be notified of each batch of events appended to
the single stream identified by stream_uuid
.
stream_uuid
is the stream to subscribe to. Use the$all
identifier to subscribe to events from all streams.subscription_name
is used to uniquely identify the subscription.subscriber
is a process that will be sent{:events, events}
notification messages.opts
is an optional map providing additional subscription configuration:name
the name of the event store if provided tostart_link/1
.start_from
is a pointer to the first event to receive. It must be one of::origin
for all events from the start of the stream (default).:current
for any new events appended to the stream after the subscription has been created.- any positive integer for a stream version to receive events after.
selector
to define a function to filter each event, i.e. returns only those elements for which fun returns a truthy value.mapper
to define a function to map each recorded event before sending to the subscriber.concurrency_limit
defines the maximum number of concurrent subscribers allowed to connect to the subscription. By default only one subscriber may connect. If too many subscribers attempt to connect to the subscription an{:error, :too_many_subscribers}
is returned.buffer_size
limits how many in-flight events will be sent to the subscriber process before acknowledgement of successful processing. This limits the number of messages sent to the subscriber and stops their message queue from getting filled with events. Defaults to one in-flight event.partition_by
is an optional function used to partition events to subscribers. It can be used to guarantee processing order when multiple subscribers have subscribed to a single subscription. The function is passed a single argument (anEventStore.RecordedEvent
struct) and must return the partition key. As an example to guarantee events for a single stream are processed serially, but different streams are processed concurrently, you could use thestream_uuid
as the partition key.by_stream = fn %EventStore.RecordedEvent{stream_uuid: stream_uuid} -> stream_uuid end {:ok, _subscription} = EventStore.subscribe_to_stream(stream_uuid, "example", self(), concurrency_limit: 10, partition_by: by_stream )
transient
is an optional boolean flag to create a transient subscription. By default this is set tofalse
. If you want to create a transient subscription set this flag to true. Your subscription will not be persisted, so if the subscription is restarted, you will receive the events again starting fromstart_from
. An example usage are short lived event handlers that keep their state in memory but still want to have the guarantee to have received all events. It's possible to create a persistent subscription with some name, stop it and later create a transient subscription with the same name. The transient subscription will now receive all events starting fromstart_from
. If you later stop thistransient
subscription and start a persistent subscription again with the same name, you will receive the events again as if the transient subscription never existed.
The subscription will resume from the last acknowledged event if it already
exists. It will ignore the start_from
argument in this case.
Returns {:ok, subscription}
when subscription succeeds.
Notification messages
Subscribers will initially receive a {:subscribed, subscription}
message
once the subscription has successfully subscribed.
After this message events will be sent to the subscriber, in batches, as
{:events, events}
where events is a collection of EventStore.RecordedEvent
structs.
Example
{:ok, subscription} = EventStore.subscribe_to_stream(stream_uuid, "example", self())
# wait for the subscription confirmation
receive do
{:subscribed, ^subscription} ->
IO.puts "Successfully subscribed to stream: " <> inspect(stream_uuid)
end
receive do
{:events, events} ->
IO.puts "Received events: " <> inspect(events)
# acknowledge receipt
EventStore.ack(subscription, events)
end
Specs
Unsubscribe an existing subscriber from all event notifications.
subscription_name
is used to identify the existing subscription process to stop.
Returns :ok
on success.
Specs
unsubscribe_from_stream( stream_uuid :: String.t(), subscription_name :: String.t(), opts :: options() ) :: :ok
Unsubscribe an existing subscriber from event notifications.
stream_uuid
is the stream to unsubscribe from.subscription_name
is used to identify the existing subscription process to stop.
Returns :ok
on success.