feat: add EventSubscriber trait and inmemory impl (#94)

* feat: add EventSubscriber trait

* feat(inmemory): EventStore implements EventSubscriber interface

* fix(util): disable benchmarks temporarily

* fix(postgres): failing doc-tests

* chore: add documentation for new Subscriber

* chore(subscription): add missing documentation

Author: ar3s3ru

eventually-core/src/lib.rs

@@ -13,4 +13,5 @@
 pub mod aggregate;
 pub mod repository;
 pub mod store;
+pub mod subscription;
 pub mod versioning;

eventually-core/src/subscription.rs

@@ -0,0 +1,65 @@
+//! Module for creating and managing long-running Subscriptions
+//! to incoming events in the [`EventStore`].
+//!
+//! [`EventStore`]: ../store/trait.EventStore.html
+
+use futures::future::BoxFuture;
+use futures::stream::BoxStream;
+
+use crate::store::Persisted;
+
+/// Stream of events returned by the [`EventSubscriber::subscribe_all`] method.
+///
+/// [`EventSubscriber::subscribe_all`]: trait.EventSubscriber.html#method.subscribe_all
+pub type EventStream<'a, S> = BoxStream<
+    'a,
+    Result<
+        Persisted<<S as EventSubscriber>::SourceId, <S as EventSubscriber>::Event>,
+        <S as EventSubscriber>::Error,
+    >,
+>;
+
+/// Component to let users subscribe to newly-inserted events into the [`EventStore`].
+///
+/// Check out [`subscribe_all`] for more information.
+///
+/// Additional information can be found in the [_Volatile Subscription_] section
+/// of eventstore.com
+///
+/// [_Volatile Subscription_]: https://eventstore.com/docs/getting-started/reading-subscribing-events/index.html#volatile-subscriptions
+/// [`EventStore`]: ../store/trait.EventStore.html
+pub trait EventSubscriber {
+    /// Type of the Source id, typically an [`AggregateId`].
+    ///
+    /// [`AggregateId`]: ../aggregate/type.AggregateId.html
+    type SourceId: Eq;
+
+    /// Event type stored in the [`EventStore`], typically an [`Aggregate::Event`].
+    ///
+    /// [`Aggregate::Event`]: ../aggregate/trait.Aggregate.html#associatedtype.Event
+    /// [`EventStore`]: ../store/trait.EventStore.html
+    type Event;
+
+    /// Possible errors returned when receiving events from the notification channel.
+    type Error;
+
+    /// Subscribes to all new events persisted in the [`EventStore`], from
+    /// the moment of calling this function, in the future.
+    ///
+    /// Since this is a long-running stream, make sure not to *block*
+    /// or await the full computation of the stream.
+    ///
+    /// Prefer using a `while let` consumer for this [`EventStream`]:
+    ///
+    /// ```text
+    /// let stream = subscriber.subscribe_all().await?;
+    ///
+    /// while let Some(event) = stream.next().await {
+    ///     // Do stuff with the received event...
+    /// }
+    /// ```
+    ///
+    /// [`EventStore`]: ../store/trait.EventStore.html
+    /// [`EventStream`]: type.EventStream.html
+    fn subscribe_all(&self) -> BoxFuture<Result<EventStream<Self>, Self::Error>>;
+}

eventually-postgres/Cargo.toml

@@ -21,3 +21,6 @@ tokio-postgres = { version = "0.5", features = ["with-serde_json-1"] }
 thiserror = "1.0"
 refinery = { version = "0.3.0", features = ["tokio-postgres"] }
 anyhow = "1.0.32"
+
+[dev-dependencies]
+tokio = { version = "0.2", features = ["sync"] }

eventually-postgres/src/lib.rs

@@ -9,12 +9,11 @@
 //!
 //! ```no_run
 //! # use std::sync::Arc;
-//! # use tokio::sync::RwLock;
 //! # use eventually_postgres::EventStoreBuilder;
 //! #
 //! # async fn dox() -> Result<(), Box<dyn std::error::Error>> {
 //! // Open a connection with Postgres.
-//! let (client, connection) =
+//! let (mut client, connection) =
 //!     tokio_postgres::connect("postgres://user@pass:localhost:5432/db", tokio_postgres::NoTls)
 //!         .await
 //!         .map_err(|err| {
@@ -45,7 +44,7 @@
 //! // to distinguish between different aggregates.
 //! //
 //! // You can also use std::any::type_name for that.
-//! let store = builder::build::<String, SomeEvent>("aggregate-name").await?
+//! let store = builder.build::<String, SomeEvent>("aggregate-name").await?;
 //!
 //! # Ok(())
 //! # }

eventually-util/Cargo.toml

@@ -19,12 +19,8 @@ futures = { version = "0.3", features = ["async-await"] }
 parking_lot = "0.11.0"
 serde = { version = "1.0", features = ["derive"], optional = true }
 thiserror = "1.0"
+tokio = { version = "0.2", features = ["sync"] }
 
 [dev-dependencies]
 anyhow = "1.0"
-criterion = "0.3"
-tokio-test = "0.2"
-
-[[bench]]
-name = "event-store-inmemory-bench"
-harness = false
+tokio = { version = "0.2", features = ["macros"] }