Merge branch 'main' into renovate/all

Cargo.lock

@@ -596,6 +596,7 @@ dependencies = [
  "anyhow",
  "async-trait",
  "futures",
+ "prost 0.12.1",
  "thiserror",
  "tokio",
  "uuid",

README.md

@@ -9,22 +9,18 @@ The value of crunch is that you can separate your own business domain from other
 See [examples](crates/crunch/examples/) for a more holistic listing of features
 
 ```rust
-impl Event for SomeEvent {
-    fn event_info(&self) -> EventInfo {
-        EventInfo {
-            domain: "some-domain",
-            entity_type: "some-entity",
-        }
-    }
-}
-
 #[tokio::main]
 async fn main() -> anyhow::Result<()> {
-    let in_memory = Persistence::in_memory();
+    let crunch = crunch::Builder::default().build()?;
-    OutboxHandler::new(in_memory.clone()).spawn();
-    let publisher = Publisher::new(in_memory);
 
-    publisher
+    crunch.subscribe(|event| async move {
+        println!("received event: {:?}", event);
+
+        Ok(())
+    })
+    .await?;
+
+    crunch
         .publish(SomeEvent {
             name: "some-name".into(),
         })
@@ -66,8 +62,12 @@ See [docs](docs/index.md) for more information (TBA)
 
 When crunch is used in services it needs some supportive tooling, it isn't a requirement, but it helps ease development when using them.
 
-- [ ] [Cli](crates/crunch-cli) Used to generate code, add subscriptions, publish event schema, bump versions and more
+- [x] [Cli](crates/crunch-cli) Used to generate code, add subscriptions, publish event schema, bump versions and more
-- [ ] [Codegen](crates/crunch-codegen) Can be used to automatically generate rust code depending on your crunch.toml file
+  - [x] Codegen done (at least for an alpha)
+  - [ ] Rest
+- [x] [Codegen](crates/crunch-codegen) Can be used to automatically generate rust code depending on your crunch.toml file
+  - [x] Main serialization and protobuf -> rust
+  - [ ] Domain information
 
 ## Extensions
 

crates/crunch-codegen/src/lib.rs

@@ -64,28 +64,28 @@ impl Node {
         let padding = " ".repeat(indent * 4);
 
         let mut message_tokens = Vec::new();
-
         if let Some(file) = &self.file {
             if let Some(messages) = &self.messages {
                 for message in messages.iter() {
                     let tokens: genco::lang::rust::Tokens = quote! {
-                    $['\r']$(&padding)impl ::crunch::Serializer for $(message) {
+                    $['\r']$(&padding)impl ::crunch::traits::Serializer for $(message) {
                     $['\r']$(&padding)    fn serialize(&self) -> Result<Vec<u8>, ::crunch::errors::SerializeError> {
-                    $['\r']$(&padding)        todo!()
+                    $['\r']$(&padding)        Ok(self.encode_to_vec())
                     $['\r']$(&padding)    }
                     $['\r']$(&padding)}
-                    $['\r']$(&padding)impl ::crunch::Deserializer for $(message) {
+                    $['\r']$(&padding)impl ::crunch::traits::Deserializer for $(message) {
-                    $['\r']$(&padding)    fn deserialize(_raw: Vec<u8>) -> Result<Self, ::crunch::errors::DeserializeError>
+                    $['\r']$(&padding)    fn deserialize(raw: Vec<u8>) -> Result<Self, ::crunch::errors::DeserializeError>
                     $['\r']$(&padding)    where
                     $['\r']$(&padding)        Self: Sized,
                     $['\r']$(&padding)    {
-                    $['\r']$(&padding)        todo!()
+                    $['\r']$(&padding)        let output  = Self::decode(raw.as_slice()).map_err(|e| ::crunch::errors::DeserializeError::ProtoErr(e))?;
+                    $['\r']$(&padding)        Ok(output)
                     $['\r']$(&padding)    }
                     $['\r']$(&padding)}
                     $['\r']$(&padding)
-                    $['\r']$(&padding)impl Event for $(message) {
+                    $['\r']$(&padding)impl crunch::traits::Event for $(message) {
                     $['\r']$(&padding)    fn event_info() -> ::crunch::traits::EventInfo {
-                    $['\r']$(&padding)        EventInfo {
+                    $['\r']$(&padding)        ::crunch::traits::EventInfo {
                     $['\r']$(&padding)            domain: "my-domain",
                     $['\r']$(&padding)            entity_type: "my-entity-type",
                     $['\r']$(&padding)            event_name: "my-event-name",
@@ -100,6 +100,7 @@ impl Node {
 
             quote! {
                 $['\r']$(&padding)pub mod $(&self.segment) {
+                    use prost::Message;
                     $['\r']$(&padding)include!($(quoted(file)));
                     $['\r']$(&padding)$(for tokens in message_tokens join ($['\r']) => $tokens)
                 $['\r']$(&padding)}

crates/crunch-traits/Cargo.toml

@@ -12,3 +12,4 @@ thiserror.workspace = true
 async-trait.workspace = true
 uuid.workspace = true
 futures.workspace = true
+prost.workspace = true

crates/crunch-traits/src/errors.rs

@@ -8,8 +8,10 @@ pub enum SerializeError {
 
 #[derive(Error, Debug)]
 pub enum DeserializeError {
-    #[error("failed to serialize {0}")]
+    #[error("failed to deserialize {0}")]
     FailedToDeserialize(anyhow::Error),
+    #[error("failed to deserialize {0}")]
+    ProtoErr(prost::DecodeError),
 }
 
 #[derive(Error, Debug)]

crates/crunch-traits/src/lib.rs

@@ -1,4 +1,4 @@
-use std::{fmt::Display};
+use std::fmt::Display;
 
 use async_trait::async_trait;
 use errors::{DeserializeError, PersistenceError, SerializeError};

crates/crunch/src/impls.rs

@@ -56,13 +56,17 @@ impl crunch_traits::Persistence for InMemoryPersistence {
     }
 
     async fn get(&self, event_id: &str) -> Result<Option<(EventInfo, Vec<u8>)>, PersistenceError> {
-        Ok(self
+        let store = self.store.read().await;
-            .store
+
-            .read()
+        let event = match store.get(event_id).filter(|m| m.state == MsgState::Pending) {
-            .await
+            Some(event) => event,
-            .get(event_id)
+            None => return Ok(None),
-            .filter(|m| m.state == MsgState::Pending).cloned()
+        };
-            .map(|m| (m.info, m.msg)))
+
+        let (content, _) = crunch_envelope::proto::unwrap(event.msg.as_slice())
+            .map_err(|e| PersistenceError::GetErr(anyhow::anyhow!(e)))?;
+
+        Ok(Some((event.info, content)))
     }
 
     async fn update_published(&self, event_id: &str) -> Result<(), PersistenceError> {

crates/crunch/src/lib.rs

@@ -56,7 +56,8 @@ impl std::ops::Deref for Crunch {
     }
 }
 
-pub mod builder {
+pub use builder::*;
+mod builder {
     use crate::{errors, Crunch, OutboxHandler, Persistence, Publisher, Subscriber, Transport};
 
     #[derive(Clone)]

crates/crunch/src/subscriber.rs

@@ -1,4 +1,4 @@
-use crunch_traits::{Event};
+use crunch_traits::Event;
 use futures::StreamExt;
 
 use crate::{errors, Transport};

examples/basic-setup/crunch.toml

@@ -5,4 +5,4 @@ codegen = ["rust"]
 
 [[publish]]	
 schema-path = "schemas/crunch"
-output-path = "src/crunch"
+output-path = "src/gencrunch"

examples/basic-setup/src/crunch/mod.rs

@@ -1,58 +0,0 @@
-pub mod basic {
-    pub mod my_event {
-    include!("basic.my_event.rs");
-    
-    impl ::crunch::Serializer for MyEvent {
-     fn serialize(&self) -> Result<Vec<u8>, ::crunch::errors::SerializeError> {
-     todo!()
-     }
-    }
-    impl ::crunch::Deserializer for MyEvent {
-     fn deserialize(_raw: Vec<u8>) -> Result<Self, ::crunch::errors::DeserializeError>
-     where
-     Self: Sized,
-     {
-     todo!()
-     }
-    }
-    
-    impl Event for MyEvent {
-     fn event_info() -> ::crunch::traits::EventInfo {
-     EventInfo {
-     domain: "my-domain",
-     entity_type: "my-entity-type",
-     event_name: "my-event-name",
-     }
-     }
-    }
-    }
-    pub mod includes {     
-        pub mod my_include {
-        include!("basic.includes.my_include.rs");
-        
-        impl ::crunch::Serializer for MyInclude {
-         fn serialize(&self) -> Result<Vec<u8>, ::crunch::errors::SerializeError> {
-         todo!()
-         }
-        }
-        impl ::crunch::Deserializer for MyInclude {
-         fn deserialize(_raw: Vec<u8>) -> Result<Self, ::crunch::errors::DeserializeError>
-         where
-         Self: Sized,
-         {
-         todo!()
-         }
-        }
-        
-        impl Event for MyInclude {
-         fn event_info() -> ::crunch::traits::EventInfo {
-         EventInfo {
-         domain: "my-domain",
-         entity_type: "my-entity-type",
-         event_name: "my-event-name",
-         }
-         }
-        }
-        }
-    }
-}

examples/basic-setup/src/gencrunch/mod.rs

@@ -0,0 +1,60 @@
+pub mod basic {
+    pub mod includes {     
+        pub mod my_include { use prost::Message;
+        include!("basic.includes.my_include.rs");
+        
+        impl ::crunch::traits::Serializer for MyInclude {
+         fn serialize(&self) -> Result<Vec<u8>, ::crunch::errors::SerializeError> {
+         Ok(self.encode_to_vec())
+         }
+        }
+        impl ::crunch::traits::Deserializer for MyInclude {
+         fn deserialize(raw: Vec<u8>) -> Result<Self, ::crunch::errors::DeserializeError>
+         where
+         Self: Sized,
+         {
+         let output = Self::decode(raw.as_slice()).map_err(|e| ::crunch::errors::DeserializeError::ProtoErr(e))?;
+         Ok(output)
+         }
+        }
+        
+        impl crunch::traits::Event for MyInclude {
+         fn event_info() -> ::crunch::traits::EventInfo {
+         ::crunch::traits::EventInfo {
+         domain: "my-domain",
+         entity_type: "my-entity-type",
+         event_name: "my-event-name",
+         }
+         }
+        }
+        }
+    }
+    pub mod my_event { use prost::Message;
+    include!("basic.my_event.rs");
+    
+    impl ::crunch::traits::Serializer for MyEvent {
+     fn serialize(&self) -> Result<Vec<u8>, ::crunch::errors::SerializeError> {
+     Ok(self.encode_to_vec())
+     }
+    }
+    impl ::crunch::traits::Deserializer for MyEvent {
+     fn deserialize(raw: Vec<u8>) -> Result<Self, ::crunch::errors::DeserializeError>
+     where
+     Self: Sized,
+     {
+     let output = Self::decode(raw.as_slice()).map_err(|e| ::crunch::errors::DeserializeError::ProtoErr(e))?;
+     Ok(output)
+     }
+    }
+    
+    impl crunch::traits::Event for MyEvent {
+     fn event_info() -> ::crunch::traits::EventInfo {
+     ::crunch::traits::EventInfo {
+     domain: "my-domain",
+     entity_type: "my-entity-type",
+     event_name: "my-event-name",
+     }
+     }
+    }
+    }
+}

examples/basic-setup/src/main.rs

@@ -1,49 +1,30 @@
-mod crunch;
+mod gencrunch;
 
-use ::crunch::traits::{Deserializer, Event, EventInfo, Serializer};
+use gencrunch::basic::{includes::my_include::MyInclude, my_event::MyEvent};
-
-struct MyEvent {}
-
-impl Serializer for MyEvent {
-    fn serialize(&self) -> Result<Vec<u8>, ::crunch::errors::SerializeError> {
-        todo!()
-    }
-}
-impl Deserializer for MyEvent {
-    fn deserialize(_raw: Vec<u8>) -> Result<Self, ::crunch::errors::DeserializeError>
-    where
-        Self: Sized,
-    {
-        todo!()
-    }
-}
-
-impl Event for MyEvent {
-    fn event_info() -> ::crunch::traits::EventInfo {
-        EventInfo {
-            domain: "my-domain",
-            entity_type: "my-entity-type",
-            event_name: "my-event-name",
-        }
-    }
-}
 
 #[tokio::main]
 async fn main() -> anyhow::Result<()> {
-    crunch::basic::my_event::MyEvent {
+    let crunch = crunch::Builder::default().build()?;
-        name: "some-name".into(),
-        include: Some(crunch::basic::includes::my_include::MyInclude {
-            name: "some-name".into(),
-        }),
-    };
-
-    let crunch = ::crunch::builder::Builder::default().build()?;
 
     crunch
-        .subscribe(|_item: MyEvent| async move { Ok(()) })
+        .subscribe(|item: MyEvent| async move {
+            println!("received item: {:?}", item);
+
+            Ok(())
+        })
         .await?;
 
-    crunch.publish(MyEvent {}).await?;
+    crunch
+        .publish(MyEvent {
+            name: "some-name".into(),
+            include: Some(MyInclude {
+                name: "some-name".into(),
+            }),
+        })
+        .await?;
+
+    // Sleep a while to let subscriber catch item
+    tokio::time::sleep(std::time::Duration::from_secs(1)).await;
 
     Ok(())
 }