feat: Projection trait is now async and might fail (#97)

* feat: Projection trait is now async and might fail

* fix: solve some clippy linter warnings

Author: ar3s3ru

eventually-core/src/projection.rs

@@ -7,6 +7,8 @@
 //! [`Projection`]: trait.Projection.html
 //! [`Aggregate`]: ../aggregate/trait.Aggregate.html
 
+use futures::future::BoxFuture;
+
 use crate::store::Persisted;
 
 /// A `Projection` is an optimized read model (or materialized view)
@@ -18,7 +20,7 @@ use crate::store::Persisted;
 ///
 /// [`Aggregate`]: ../aggregate/trait.Aggregate.html
 /// [`EventStore`]: ../store/trait.EventStore.html
-pub trait Projection: Default {
+pub trait Projection {
     /// Type of the Source id, typically an [`AggregateId`].
     ///
     /// [`AggregateId`]: ../aggregate/type.AggregateId.html
@@ -29,7 +31,13 @@ pub trait Projection: Default {
     /// [`Aggregate::Event`]: ../aggregate/trait.Aggregate.html#associatedtype.Event
     type Event;
 
-    /// Updates the next value of the `Projection` using the provided
-    /// event value.
-    fn project(self, event: Persisted<Self::SourceId, Self::Event>) -> Self;
+    /// Type of the possible error that might occur when projecting
+    /// the next state.
+    type Error;
+
+    /// Updates the next value of the `Projection` using the provided event value.
+    fn project<'a>(
+        &'a mut self,
+        event: Persisted<Self::SourceId, Self::Event>,
+    ) -> BoxFuture<'a, Result<(), Self::Error>>;
 }

eventually-test/src/api.rs

@@ -69,7 +69,7 @@ pub(crate) async fn history(req: Request<AppState>) -> Result<Response, Error> {
         .await
         .map_err(Error::from)?
         .try_filter(|event| {
-            futures::future::ready(match from {
+            ready(match from {
                 None => true,
                 Some(from) => event.happened_at() >= &from,
             })

eventually-test/src/lib.rs

@@ -9,8 +9,6 @@ use eventually::aggregate::Optional;
 use eventually::inmemory::{EventStoreBuilder, ProjectorBuilder};
 use eventually::{AggregateRootBuilder, Repository};
 
-use futures::stream::StreamExt;
-
 use tokio::sync::RwLock;
 
 use crate::config::Config;
@@ -38,18 +36,13 @@ pub async fn run(config: Config) -> anyhow::Result<()> {
     // Put the store behind an Arc to allow for clone-ness of a single instance.
     let store = Arc::new(store);
 
+    // Create a new in-memory projection to keep the total orders computed by
+    // the application.
+    let total_orders_projection = Arc::new(RwLock::new(TotalOrdersProjection::default()));
+
     // Create a new Projector for the desired projection.
     let mut total_orders_projector =
-        ProjectorBuilder::new(store.clone(), store.clone()).build::<TotalOrdersProjection>();
-
-    // Get a watch channel from the Projector: updates to the projector values
-    // will be sent here.
-    let mut total_orders_projector_rx = total_orders_projector.watch();
-
-    // Keep the projection value in memory.
-    // We can use it to access it from the context of an endpoint and serialize the read model.
-    let total_orders_projection = Arc::new(RwLock::new(TotalOrdersProjection::default()));
-    let total_orders_projection_state = total_orders_projection.clone();
+        ProjectorBuilder::new(store.clone(), store.clone()).build(total_orders_projection.clone());
 
     // Spawn a dedicated coroutine to run the projector.
     //
@@ -58,19 +51,6 @@ pub async fn run(config: Config) -> anyhow::Result<()> {
     // and it will progressively update the projection as events arrive.
     tokio::spawn(async move { total_orders_projector.run().await.expect("should not fail") });
 
-    // Spawn a dedicated coroutine to listen to changes to the projection.
-    //
-    // In this case we're logging the latest version, but in more advanced
-    // scenario you might want to do something more with it.
-    //
-    // In some cases you might not need to watch the projection changes.
-    tokio::spawn(async move {
-        while let Some(total_orders) = total_orders_projector_rx.next().await {
-            log::info!("Total orders: {:?}", total_orders);
-            *total_orders_projection_state.write().await = total_orders;
-        }
-    });
-
     // Set up the HTTP router.
     let mut app = tide::new();
 

eventually-test/src/order.rs

@@ -2,7 +2,8 @@ use std::fmt::{Display, Formatter, Result as FmtResult};
 
 use chrono::{DateTime, Utc};
 
-use futures::{future, future::BoxFuture};
+use futures::future;
+use futures::future::{BoxFuture, FutureExt};
 
 use serde::{Deserialize, Serialize};
 
@@ -20,16 +21,20 @@ pub struct TotalOrdersProjection {
 impl Projection for TotalOrdersProjection {
     type SourceId = String;
     type Event = OrderEvent;
+    type Error = std::convert::Infallible;
 
-    fn project(mut self, event: Persisted<Self::SourceId, Self::Event>) -> Self {
+    fn project<'a>(
+        &'a mut self,
+        event: Persisted<Self::SourceId, Self::Event>,
+    ) -> BoxFuture<'a, Result<(), Self::Error>> {
         match event.take() {
             OrderEvent::Created { .. } => self.created += 1,
             OrderEvent::Completed { .. } => self.completed += 1,
             OrderEvent::Cancelled { .. } => self.cancelled += 1,
             _ => (),
         };
 
-        self
+        future::ok(()).boxed()
     }
 }
 

eventually-util/src/inmemory/projector.rs

@@ -1,15 +1,14 @@
 use std::error::Error as StdError;
-use std::fmt::Debug;
 use std::sync::atomic::{AtomicU32, Ordering};
 use std::sync::Arc;
 
 use eventually_core::projection::Projection;
 use eventually_core::store::{EventStore, Select};
 use eventually_core::subscription::EventSubscriber;
 
-use futures::stream::{Stream, StreamExt, TryStreamExt};
+use futures::stream::{StreamExt, TryStreamExt};
 
-use tokio::sync::watch::{channel, Receiver, Sender};
+use tokio::sync::RwLock;
 
 /// Reusable builder for multiple [`Projector`] instances.
 ///
@@ -29,21 +28,21 @@ impl<Store, Subscriber> ProjectorBuilder<Store, Subscriber> {
         Self { store, subscriber }
     }
 
-    /// Builds a new [`Projector`] for the [`Projection`]
-    /// specified in the function type.
+    /// Builds a new [`Projector`] for the [`Projection`] specified in the function type.
     ///
     /// [`Projector`]: struct.Projector.html
     /// [`Projection`]: ../../../eventually-core/projection/trait.Projection.html
-    pub fn build<P>(&self) -> Projector<P, Store, Subscriber>
+    pub fn build<P>(&self, projection: Arc<RwLock<P>>) -> Projector<P, Store, Subscriber>
     where
         // NOTE: these bounds are required for Projector::run.
-        P: Projection + Debug + Clone,
+        P: Projection,
         Store: EventStore<SourceId = P::SourceId, Event = P::Event>,
         Subscriber: EventSubscriber<SourceId = P::SourceId, Event = P::Event>,
+        <P as Projection>::Error: StdError + Send + Sync + 'static,
         <Store as EventStore>::Error: StdError + Send + Sync + 'static,
         <Subscriber as EventSubscriber>::Error: StdError + Send + Sync + 'static,
     {
-        Projector::new(self.store.clone(), self.subscriber.clone())
+        Projector::new(projection, self.store.clone(), self.subscriber.clone())
     }
 }
 
@@ -68,44 +67,31 @@ pub struct Projector<P, Store, Subscriber>
 where
     P: Projection,
 {
-    tx: Sender<P>,
-    rx: Receiver<P>, // Keep the receiver to be able to clone it in watch().
+    projection: Arc<RwLock<P>>,
     store: Arc<Store>,
     subscriber: Arc<Subscriber>,
-    state: P,
     last_sequence_number: AtomicU32,
-    projection: std::marker::PhantomData<P>,
 }
 
 impl<P, Store, Subscriber> Projector<P, Store, Subscriber>
 where
-    P: Projection + Debug + Clone,
+    P: Projection,
     Store: EventStore<SourceId = P::SourceId, Event = P::Event>,
     Subscriber: EventSubscriber<SourceId = P::SourceId, Event = P::Event>,
     // NOTE: these bounds are needed for anyhow::Error conversion.
+    <P as Projection>::Error: StdError + Send + Sync + 'static,
     <Store as EventStore>::Error: StdError + Send + Sync + 'static,
     <Subscriber as EventSubscriber>::Error: StdError + Send + Sync + 'static,
 {
-    fn new(store: Arc<Store>, subscriber: Arc<Subscriber>) -> Self {
-        let state: P = Default::default();
-        let (tx, rx) = channel(state.clone());
-
+    fn new(projection: Arc<RwLock<P>>, store: Arc<Store>, subscriber: Arc<Subscriber>) -> Self {
         Self {
-            tx,
-            rx,
             store,
             subscriber,
-            state,
+            projection,
             last_sequence_number: Default::default(),
-            projection: std::marker::PhantomData,
         }
     }
 
-    /// Provides a `Stream` that receives the latest copy of the `Projection` state.
-    pub fn watch(&self) -> impl Stream<Item = P> {
-        self.rx.clone()
-    }
-
     /// Starts the update of the `Projection` by processing all the events
     /// coming from the [`EventStore`].
     ///
@@ -124,32 +110,30 @@ where
         let subscription = self.subscriber.subscribe_all().await?;
         let one_off_stream = self.store.stream_all(Select::All).await?;
 
-        let mut stream = one_off_stream
+        let stream = one_off_stream
             .map_err(anyhow::Error::from)
             .chain(subscription.map_err(anyhow::Error::from));
 
-        while let Some(event) = stream.next().await {
-            let event = event?;
-            let expected_sequence_number = self.last_sequence_number.load(Ordering::SeqCst);
-            let event_sequence_number = event.sequence_number();
+        stream
+            .try_for_each(|event| async {
+                let expected_sequence_number = self.last_sequence_number.load(Ordering::SeqCst);
+                let event_sequence_number = event.sequence_number();
 
-            if event_sequence_number < expected_sequence_number {
-                continue; // Duplicated event detected, let's skip it.
-            }
+                if event_sequence_number < expected_sequence_number {
+                    return Ok(()); // Duplicated event detected, let's skip it.
+                }
 
-            self.state = P::project(self.state.clone(), event);
+                self.projection.write().await.project(event).await?;
 
-            self.last_sequence_number.compare_and_swap(
-                expected_sequence_number,
-                event_sequence_number,
-                Ordering::SeqCst,
-            );
+                self.last_sequence_number.compare_and_swap(
+                    expected_sequence_number,
+                    event_sequence_number,
+                    Ordering::SeqCst,
+                );
 
-            // Notify watchers of the latest projection state.
-            self.tx.broadcast(self.state.clone()).expect(
-                "since this struct holds the original receiver, failures should not happen",
-            );
-        }
+                Ok(())
+            })
+            .await?;
 
         Ok(())
     }

eventually-util/src/inmemory/store.rs

@@ -23,29 +23,30 @@ const SUBSCRIBE_CHANNEL_DEFAULT_CAP: usize = 128;
 /// Error returned by the [`EventStore::append`] when a conflict has been detected.
 ///
 /// [`EventStore::append`]: trait.EventStore.html#method.append
-#[derive(Debug, thiserror::Error, PartialEq)]
-pub enum Error {
-    /// Version conflict registered.
-    #[error(
-        "inmemory::EventStore: conflicting versions, expected {expected}, got instead {actual}"
-    )]
-    Conflict {
-        /// The last version value found the Store.
-        expected: u32,
-        /// The actual version passed by the caller to the Store.
-        actual: u32,
-    },
-
-    #[error("inmemory::EventStore: failed to read event from subscription: {0}")]
-    ReceiveEvent(#[source] RecvError),
+#[derive(Debug, thiserror::Error, PartialEq, Eq)]
+#[error("conflicting versions, expected {expected}, got instead {actual}")]
+pub struct ConflictError {
+    /// The last version value found the Store.
+    pub expected: u32,
+    /// The actual version passed by the caller to the Store.
+    pub actual: u32,
 }
 
-impl AppendError for Error {
+impl AppendError for ConflictError {
     fn is_conflict_error(&self) -> bool {
         true
     }
 }
 
+/// Error returned by the [`EventStore::append`] when a conflict has been detected.
+///
+/// [`EventStore::append`]: trait.EventStore.html#method.append
+#[derive(Debug, thiserror::Error)]
+pub enum SubscriberError {
+    #[error("failed to read event from subscription watch channel: {0}")]
+    ReceiveEvent(#[source] RecvError),
+}
+
 /// Builder for [`EventStore`] instances.
 ///
 /// [`EventStore`]: struct.EventStore.html
@@ -119,7 +120,7 @@ where
 {
     type SourceId = Id;
     type Event = Event;
-    type Error = Error;
+    type Error = SubscriberError;
 
     fn subscribe_all(
         &self,
@@ -130,7 +131,12 @@ where
         // with the definition of the EventStream.
         let rx = self.tx.subscribe();
 
-        Box::pin(async move { Ok(rx.into_stream().map_err(Error::ReceiveEvent).boxed()) })
+        Box::pin(async move {
+            Ok(rx
+                .into_stream()
+                .map_err(SubscriberError::ReceiveEvent)
+                .boxed())
+        })
     }
 }
 
@@ -141,7 +147,7 @@ where
 {
     type SourceId = Id;
     type Event = Event;
-    type Error = Error;
+    type Error = ConflictError;
 
     fn append(
         &mut self,
@@ -160,7 +166,7 @@ where
 
             if let Expected::Exact(actual) = version {
                 if expected != actual {
-                    return Err(Error::Conflict { expected, actual });
+                    return Err(ConflictError { expected, actual });
                 }
             }
 
@@ -274,7 +280,7 @@ where
 
 #[cfg(test)]
 mod tests {
-    use super::{Error, EventStore as InMemoryStore};
+    use super::{ConflictError, EventStore as InMemoryStore};
 
     use std::cell::RefCell;
     use std::sync::Arc;
@@ -480,7 +486,7 @@ mod tests {
             .await;
 
         assert_eq!(
-            Err(Error::Conflict {
+            Err(ConflictError {
                 expected: last_version,
                 actual: poisoned_last_version
             }),