eventually-postgres: add EventSubscriber using LISTEN feature (#96)

* feat(postgres): add EventSubscriber implementation using LISTEN

* feat: remove unnecessary Arc requirements for thread-safe objects

* docs(postgres): add documentation for EventSubscriber

* test(postgres): add integration tests

* fix: try to not limit threads for tarpaulin

Author: ar3s3ru

.github/workflows/rust.yml

@@ -40,8 +40,6 @@ jobs:
 
       - name: Run cargo-tarpaulin (main)
         uses: actions-rs/tarpaulin@v0.1
-        with:
-          args: '-- --test-threads 1'
 
       - name: Upload to codecov.io
         uses: codecov/codecov-action@v1.0.13

eventually-core/src/aggregate.rs

@@ -1,8 +1,6 @@
 //! Foundation traits for creating Domain abstractions
 //! using [the `Aggregate` pattern](https://martinfowler.com/bliki/DDD_Aggregate.html).
 
-use std::sync::Arc;
-
 use futures::future::BoxFuture;
 
 #[cfg(feature = "serde")]
@@ -102,23 +100,26 @@ impl<T> AggregateExt for T where T: Aggregate {}
 ///
 /// [`AggregateRoot`]: struct.AggregateRoot.html
 #[derive(Clone)]
-pub struct AggregateRootBuilder<T> {
-    aggregate: Arc<T>,
+pub struct AggregateRootBuilder<T>
+where
+    T: Aggregate + Clone,
+{
+    aggregate: T,
 }
 
-impl<T> From<Arc<T>> for AggregateRootBuilder<T>
+impl<T> From<T> for AggregateRootBuilder<T>
 where
-    T: Aggregate,
+    T: Aggregate + Clone,
 {
     #[inline]
-    fn from(aggregate: Arc<T>) -> Self {
+    fn from(aggregate: T) -> Self {
         Self { aggregate }
     }
 }
 
 impl<T> AggregateRootBuilder<T>
 where
-    T: Aggregate,
+    T: Aggregate + Clone,
 {
     /// Builds a new [`AggregateRoot`] instance for the specified Aggregate [`Id`].
     ///
@@ -172,7 +173,7 @@ where
 #[cfg_attr(feature = "serde", derive(Serialize))]
 pub struct AggregateRoot<T>
 where
-    T: Aggregate + 'static,
+    T: Aggregate + Clone + 'static,
 {
     id: T::Id,
     version: u32,
@@ -181,15 +182,15 @@ where
     state: T::State,
 
     #[cfg_attr(feature = "serde", serde(skip_serializing))]
-    aggregate: Arc<T>,
+    aggregate: T,
 
     #[cfg_attr(feature = "serde", serde(skip_serializing))]
     to_commit: Option<Vec<T::Event>>,
 }
 
 impl<T> PartialEq for AggregateRoot<T>
 where
-    T: Aggregate,
+    T: Aggregate + Clone,
 {
     #[inline]
     fn eq(&self, other: &Self) -> bool {
@@ -199,7 +200,7 @@ where
 
 impl<T> Versioned for AggregateRoot<T>
 where
-    T: Aggregate,
+    T: Aggregate + Clone,
 {
     #[inline]
     fn version(&self) -> u32 {
@@ -209,7 +210,7 @@ where
 
 impl<T> AggregateRoot<T>
 where
-    T: Aggregate,
+    T: Aggregate + Clone,
 {
     /// Returns a reference to the Aggregate [`Id`] that represents
     /// the entity wrapped by this [`AggregateRoot`] instance.
@@ -246,7 +247,7 @@ where
 
 impl<T> AggregateRoot<T>
 where
-    T: Aggregate,
+    T: Aggregate + Clone,
     T::Event: Clone,
     T::State: Clone,
 {

eventually-core/src/repository.rs

@@ -60,17 +60,19 @@ pub type Result<T, A, S> =
 /// [`State`]: ../aggregate/trait.Aggregate.html#associatedtype.State
 /// [`Event`]: ../aggregate/trait.Aggregate.html#associatedtype.Event
 /// [`EventStore`]: ../store/trait.EventStore.html
+#[derive(Clone)]
 pub struct Repository<T, Store>
 where
-    T: Aggregate + 'static,
+    T: Aggregate + Clone + 'static,
+    Store: EventStore<SourceId = T::Id, Event = T::Event>,
 {
     builder: AggregateRootBuilder<T>,
     store: Store,
 }
 
 impl<T, Store> Repository<T, Store>
 where
-    T: Aggregate,
+    T: Aggregate + Clone,
     Store: EventStore<SourceId = T::Id, Event = T::Event>,
 {
     /// Creates a new `Repository` instance, using the [`Aggregate`]

eventually-postgres/Cargo.toml

@@ -21,6 +21,8 @@ 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"
+tokio = { version = "0.2", features = ["sync"] }
 
 [dev-dependencies]
-tokio = { version = "0.2", features = ["sync"] }
+testcontainers = "0.9"
+tokio = { version = "0.2", features = ["sync", "macros"] }