Implement workflow machines

Cargo.toml

@@ -6,19 +6,32 @@ edition = "2018"
 
 [lib]
 
+# TODO: Feature flag opentelem stuff
 [dependencies]
+anyhow = "1.0"
 async-trait = "0.1"
 derive_more = "0.99"
-prost = "0.6"
-prost-types = "0.6"
+env_logger = "0.8"
+futures = "0.3"
+log = "0.4"
+opentelemetry-jaeger = "0.10"
+prost = "0.7"
+prost-types = "0.7"
 thiserror = "1.0"
-tonic = "0.3"
+tokio = { version = "1.1", features = ["rt", "rt-multi-thread"] }
+tonic = "0.4"
+tracing = { version = "0.1", features = ["log"] }
+tracing-opentelemetry = "0.10"
+tracing-subscriber = "0.2"
 
 [dependencies.rustfsm]
 path = "fsm"
 
+[dev-dependencies]
+rstest = "0.6"
+
 [build-dependencies]
-tonic-build = "0.3"
+tonic-build = "0.4"
 
 [workspace]
 members = [".", "fsm"]

README.md

@@ -26,3 +26,12 @@ To format all code run:
 We are using [clippy](https://github.com/rust-lang/rust-clippy) for linting.
 You can run it using:
 `cargo clippy --all -- -D warnings`
+
+## Style Guidelines
+
+### Error handling
+Any error which is returned from a public interface should be well-typed, and we use 
+[thiserror](https://github.com/dtolnay/thiserror) for that purpose.
+
+Errors returned from things only used in testing are free to use 
+[anyhow](https://github.com/dtolnay/anyhow) for less verbosity.

build.rs

@@ -5,6 +5,15 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
         .build_server(false)
         .build_client(true)
         .out_dir("src/protos")
+        // Make conversions easier for some types
+        .type_attribute(
+            "temporal.api.history.v1.HistoryEvent.attributes",
+            "#[derive(::derive_more::From)]",
+        )
+        .type_attribute(
+            "temporal.api.command.v1.Command.attributes",
+            "#[derive(::derive_more::From)]",
+        )
         .compile(
             &[
                 "protos/local/core_interface.proto",

fsm/src/lib.rs

@@ -1,2 +1,2 @@
 pub use state_machine_procmacro::fsm;
-pub use state_machine_trait::{StateMachine, TransitionResult};
+pub use state_machine_trait::{MachineError, StateMachine, TransitionResult};

fsm/state_machine_procmacro/src/lib.rs

@@ -9,7 +9,7 @@ use syn::{
     parse_macro_input,
     punctuated::Punctuated,
     spanned::Spanned,
-    Error, Fields, Ident, Token, Type, Variant,
+    Error, Fields, Ident, Token, Type, Variant, Visibility,
 };
 
 /// Parses a DSL for defining finite state machines, and produces code implementing the
@@ -176,17 +176,27 @@ mod kw {
 }
 
 struct StateMachineDefinition {
+    visibility: Visibility,
     name: Ident,
     shared_state_type: Option<Type>,
     command_type: Ident,
     error_type: Ident,
     transitions: HashSet<Transition>,
 }
 
+impl StateMachineDefinition {
+    fn is_final_state(&self, state: &Ident) -> bool {
+        // If no transitions go from this state, it's a final state.
+        self.transitions.iter().find(|t| t.from == *state).is_none()
+    }
+}
+
 impl Parse for StateMachineDefinition {
     // TODO: Pub keyword
     fn parse(input: ParseStream) -> Result<Self> {
-        // First parse the state machine name, command type, and error type
+        // Parse visibility if present
+        let visibility = input.parse()?;
+        // parse the state machine name, command type, and error type
         let (name, command_type, error_type, shared_state_type) = parse_machine_types(&input).map_err(|mut e| {
             e.combine(Error::new(
                 e.span(),
@@ -200,6 +210,7 @@ impl Parse for StateMachineDefinition {
             input.parse_terminated(Transition::parse)?;
         let transitions = transitions.into_iter().collect();
         Ok(Self {
+            visibility,
             name,
             shared_state_type,
             transitions,
@@ -316,6 +327,7 @@ impl Parse for Transition {
 
 impl StateMachineDefinition {
     fn codegen(&self) -> TokenStream {
+        let visibility = self.visibility.clone();
         // First extract all of the states into a set, and build the enum's insides
         let states: HashSet<_> = self
             .transitions
@@ -328,6 +340,7 @@ impl StateMachineDefinition {
             }
         });
         let name = &self.name;
+        let name_str = &self.name.to_string();
         let state_enum_name = Ident::new(&format!("{}State", name), name.span());
         // If user has not defined any shared state, use the unit type.
         let shared_state_type = self
@@ -336,24 +349,41 @@ impl StateMachineDefinition {
             .unwrap_or_else(|| syn::parse_str("()").unwrap());
         let machine_struct = quote! {
             #[derive(Clone)]
-            pub struct #name {
+            #visibility struct #name {
                 state: #state_enum_name,
                 shared_state: #shared_state_type
             }
         };
         let states_enum = quote! {
             #[derive(::derive_more::From, Clone)]
-            pub enum #state_enum_name {
+            #visibility enum #state_enum_name {
                 #(#state_variants),*
             }
         };
+        let state_is_final_match_arms = states.iter().map(|s| {
+            let val = if self.is_final_state(s) {
+                quote! { true }
+            } else {
+                quote! { false }
+            };
+            quote! { #state_enum_name::#s(_) => #val }
+        });
+        let states_enum_impl = quote! {
+            impl #state_enum_name {
+                fn is_final(&self) -> bool {
+                    match self {
+                        #(#state_is_final_match_arms),*
+                    }
+                }
+            }
+        };
 
         // Build the events enum
         let events: HashSet<Variant> = self.transitions.iter().map(|t| t.event.clone()).collect();
         let events_enum_name = Ident::new(&format!("{}Events", name), name.span());
         let events: Vec<_> = events.into_iter().collect();
         let events_enum = quote! {
-            pub enum #events_enum_name {
+            #visibility enum #events_enum_name {
                 #(#events),*
             }
         };
@@ -451,6 +481,10 @@ impl StateMachineDefinition {
                 type Event = #events_enum_name;
                 type Command = #cmd_type;
 
+                fn name(&self) -> &str {
+                  #name_str
+                }
+
                 fn on_event(self, event: #events_enum_name)
                   -> ::rustfsm::TransitionResult<Self> {
                     match self.state {
@@ -469,6 +503,10 @@ impl StateMachineDefinition {
                     &self.shared_state
                 }
 
+                fn on_final_state(&self) -> bool {
+                    self.state.is_final()
+                }
+
                 fn from_parts(shared: Self::SharedState, state: Self::State) -> Self {
                     Self { shared_state: shared, state }
                 }
@@ -484,6 +522,7 @@ impl StateMachineDefinition {
             #transition_type_alias
             #machine_struct
             #states_enum
+            #states_enum_impl
             #events_enum
             #trait_impl
         };

fsm/state_machine_trait/src/lib.rs

@@ -56,11 +56,18 @@ pub trait StateMachine: Sized {
         }
     }
 
+    fn name(&self) -> &str;
+
+    /// Returns the current state of the machine
     fn state(&self) -> &Self::State;
     fn set_state(&mut self, new_state: Self::State);
 
+    /// Returns the current shared state of the machine
     fn shared_state(&self) -> &Self::SharedState;
 
+    /// Returns true if the machine's current state is a final one
+    fn on_final_state(&self) -> bool;
+
     /// Given the shared data and new state, create a new instance.
     fn from_parts(shared: Self::SharedState, state: Self::State) -> Self;
 }
@@ -160,6 +167,18 @@ where
         }
     }
 
+    /// Produce a transition with commands relying on [Default] for the destination state's value
+    pub fn commands<CI, DestState>(commands: CI) -> Self
+    where
+        CI: IntoIterator<Item = S::Command>,
+        DestState: Into<S::State> + Default,
+    {
+        Self::OkNoShare {
+            commands: commands.into_iter().collect(),
+            new_state: DestState::default().into(),
+        }
+    }
+
     /// Produce a transition with no commands relying on [Default] for the destination state's
     /// value
     pub fn default<DestState>() -> Self

src/lib.rs

@@ -1,3 +1,6 @@
+#[macro_use]
+extern crate tracing;
+
 mod machines;
 mod pollers;
 pub mod protos;