Started writing the code generation portion of btproto.

Cargo.lock

@@ -313,6 +313,17 @@ dependencies = [
  "tokio",
 ]
 
+[[package]]
+name = "btfs"
+version = "0.1.0"
+dependencies = [
+ "btlib",
+ "btproto",
+ "btrun",
+ "btsector",
+ "serde",
+]
+
 [[package]]
 name = "btfsd"
 version = "0.1.0"
@@ -415,8 +426,11 @@ dependencies = [
 name = "btproto"
 version = "0.1.0"
 dependencies = [
+ "btlib",
+ "btrun",
  "proc-macro2",
  "quote",
+ "serde",
  "syn 2.0.42",
 ]
 
@@ -456,6 +470,16 @@ dependencies = [
  "uuid",
 ]
 
+[[package]]
+name = "btsector"
+version = "0.1.0"
+dependencies = [
+ "btlib",
+ "btproto",
+ "btrun",
+ "serde",
+]
+
 [[package]]
 name = "btserde"
 version = "0.1.0"

crates/btfs/Cargo.toml

@@ -0,0 +1,13 @@
+[package]
+name = "btfs"
+version = "0.1.0"
+edition = "2021"
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[dependencies]
+btlib = { path = "../btlib" }
+btrun = { path = "../btrun" }
+btproto = { path = "../btproto" }
+btsector = { path = "../btsector" }
+serde = { version = "^1.0.136", features = ["derive"] }

crates/btrun/src/fs_proto.rs → crates/btfs/src/lib.rs

@@ -1,8 +1,9 @@
 use btproto::protocol;
+use btrun::{ActorName, CallMsg};
+use btsector::FileId; 
+use btlib::Result;
 use serde::{Deserialize, Serialize};
 
-use crate::{sector_proto::FileId, ActorName, CallMsg};
-
 #[derive(Serialize, Deserialize)]
 pub struct Open {
     id: FileId,
@@ -11,7 +12,6 @@ pub struct Open {
 impl CallMsg for Open {
     type Reply = ActorName;
 }
-
 protocol! {
     let name = FsProtocol;
     let states = [
@@ -32,7 +32,6 @@ protocol! {
 
     FileInit?Activate -> FileInit;
     FileInit?Open -> Opened;
-    //PoopedPants?Notification -> Changing;
 
     FileHandle[Opened] -> FileHandle[Opened], >Opened!FileOp;
     Opened?FileOp -> Opened, >FileHandle!FileOp::Reply;
@@ -41,3 +40,20 @@ protocol! {
     FileHandle[Opened] -> End, >Opened!Close;
     Opened?Close -> End;
 }
+
+#[derive(Serialize, Deserialize)]
+pub struct Query;
+
+impl CallMsg for Query {
+    type Reply = ();
+}
+
+#[derive(Serialize, Deserialize)]
+pub struct FileOp;
+
+impl CallMsg for FileOp {
+    type Reply = ();
+}
+
+#[derive(Serialize, Deserialize)]
+pub struct Close;

crates/btproto/Cargo.toml

@@ -9,6 +9,11 @@ proc-macro = true
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
+btrun = { path = "../btrun" }
 proc-macro2 = "1.0.71"
 quote = "1.0.33"
 syn = { version = "2.0.42", features = ["extra-traits"] }
+serde = { version = "^1.0.136", features = ["derive"] }
+
+[dev-dependencies]
+btlib = { path = "../btlib" }

crates/btproto/src/generation.rs

@@ -1,20 +1,141 @@
+use std::collections::{HashMap, HashSet};
+
+use crate::parsing::{Message, Transition};
+use btrun::{Activate, End};
+
 use super::Protocol;
-use proc_macro2::TokenStream;
+use proc_macro2::{Ident, TokenStream};
+use quote::{format_ident, quote, ToTokens};
 
-impl Protocol {
-    pub(crate) fn generate(&self) -> TokenStream {
-        let mut stream = self.generate_message_enum();
-        stream.extend(self.generate_state_traits());
-        stream
+impl ToTokens for Protocol {
+    fn to_tokens(&self, tokens: &mut TokenStream) {
+        tokens.extend(self.generate_message_enum());
+        tokens.extend(self.generate_state_traits());
     }
+}
 
+impl Protocol {
     fn generate_message_enum(&self) -> TokenStream {
-        // TODO: Generate message enum.
-        TokenStream::new()
+        let mut msgs: HashSet<&Message> = HashSet::new();
+        for transition in self.transitions.iter() {
+            // We only need to insert received messages because every sent message has a
+            // corresponding receiver thanks to the validator.
+            if let Some(msg) = &transition.in_msg {
+                msgs.insert(msg);
+            }
+        }
+        let variants = msgs.iter().map(|msg| msg.variant());
+        let msg_types = msgs.iter().map(|msg| msg.type_tokens());
+        let enum_name = format_ident!("{}Msgs", self.name_def.name);
+        quote! {
+            pub enum #enum_name {
+                #( #variants(#msg_types) ),*
+            }
+        }
     }
 
     fn generate_state_traits(&self) -> TokenStream {
-        // TODO: Generate state traits.
-        TokenStream::new()
+        let mut traits: HashMap<&Ident, Vec<&Transition>> = HashMap::new();
+        for transition in self.transitions.iter() {
+            let vec = traits
+                .entry(&transition.in_state.state_trait)
+                .or_insert_with(Vec::new);
+            vec.push(transition);
+        }
+        let mut tokens = TokenStream::new();
+        for (trait_ident, transitions) in traits {
+            quote! {
+                pub trait #trait_ident {
+                    #( #transitions )*
+                }
+            }
+            .to_tokens(&mut tokens);
+        }
+        tokens
+    }
+}
+
+impl Message {
+    /// Returns the tokens which represent the type of this message.
+    fn type_tokens(&self) -> TokenStream {
+        // We generate a fully-qualified path to the Activate message so that it doesn't need to be
+        // imported for every protocol definition.
+        let msg_type = if self.msg_type == Activate::ident() {
+            quote! { ::btrun::Activate }
+        } else {
+            let msg_type = &self.msg_type;
+            quote! { #msg_type }
+        };
+        if self.is_reply {
+            quote! {
+                <#msg_type as ::btrun::CallMsg>::Reply
+            }
+        } else {
+            quote! {
+                #msg_type
+            }
+        }
+    }
+}
+
+impl ToTokens for Transition {
+    fn to_tokens(&self, tokens: &mut TokenStream) {
+        let transition = if let Some(msg) = &self.in_msg {
+            self.generate_message_handler(msg)
+        } else {
+            self.generate_sender_method()
+        };
+        tokens.extend(transition);
+    }
+}
+
+impl Transition {
+    fn generate_message_handler(&self, msg: &Message) -> TokenStream {
+        let msg_type = if msg.msg_type == Activate::ident() {
+            quote! { ::btrun::Activate }
+        } else {
+            msg.msg_type.to_token_stream()
+        };
+        let method_ident = format_ident!("handle_{}", msg.msg_type.to_lowercase());
+        let output_pairs: Vec<_> = self
+            .out_states
+            .as_ref()
+            .iter()
+            .map(|state| {
+                let state_trait = &state.state_trait;
+                if state_trait == End::ident() {
+                    (quote! {}, quote! { ::btrun::End })
+                } else {
+                    let assoc_type = format_ident!("{}Output{}", msg.variant(), state_trait);
+                    let output_decl = quote! { type #assoc_type: #state_trait; };
+                    let output_type = quote! { Self::#assoc_type };
+                    (output_decl, output_type)
+                }
+            })
+            .collect();
+        let output_decls = output_pairs.iter().map(|(decl, _)| decl);
+        let output_types = output_pairs.iter().map(|(_, output_type)| output_type);
+        let future_name = format_ident!("Handle{}Fut", msg.variant());
+        quote! {
+            #( #output_decls )*
+            type #future_name: ::std::future::Future<Output = Result<( #( #output_types ),* )>>;
+            fn #method_ident(self, msg: #msg_type) -> Self::#future_name;
+        }
+    }
+
+    fn generate_sender_method(&self) -> TokenStream {
+        quote! {}
+    }
+}
+
+trait ToLowercase {
+    fn to_lowercase(&self) -> String;
+}
+
+impl ToLowercase for Ident {
+    fn to_lowercase(&self) -> String {
+        let mut buf = self.to_string();
+        buf.make_ascii_lowercase();
+        buf
     }
 }

crates/btproto/src/lib.rs

@@ -1,5 +1,6 @@
 extern crate proc_macro;
 use proc_macro::TokenStream;
+use quote::ToTokens;
 use syn::parse_macro_input;
 
 mod parsing;
@@ -38,5 +39,5 @@ macro_rules! unwrap_or_compile_err {
 pub fn protocol(input: TokenStream) -> TokenStream {
     let input = parse_macro_input!(input as Protocol);
     unwrap_or_compile_err!(input.validate());
-    input.generate().into()
+    input.to_token_stream().into()
 }

crates/btproto/src/parsing.rs

@@ -1,5 +1,6 @@
 //! Types for parsing the protocol grammar.
 
+use quote::format_ident;
 use syn::{bracketed, parenthesized, parse::Parse, punctuated::Punctuated, token, Ident, Token};
 
 /// This type represents the top-level production for the protocol grammar.
@@ -27,6 +28,7 @@ pub(crate) struct NameDef {
 }
 
 impl NameDef {
+    const NAME_IDENT: &str = "name";
     const NAME_IDENT_ERR: &str = "invalid name declaration identifier";
 }
 
@@ -34,7 +36,7 @@ impl Parse for NameDef {
     /// name_def : "let" "name" '=' Ident ';' ;
     fn parse(input: syn::parse::ParseStream) -> syn::Result<Self> {
         input.parse::<Token![let]>()?;
-        if Ident::parse(input)? != "name" {
+        if Ident::parse(input)? != Self::NAME_IDENT {
             return Err(input.error(Self::NAME_IDENT_ERR));
         }
         input.parse::<Token![=]>()?;
@@ -50,6 +52,7 @@ pub(crate) struct StatesDef {
 }
 
 impl StatesDef {
+    const STATES_ARRAY_IDENT: &str = "states";
     const ARRAY_IDENT_ERR: &str = "invalid states array identifier. Expected 'states'.";
 }
 
@@ -57,7 +60,7 @@ impl Parse for StatesDef {
     /// states_def : "let" "states" '=' ident_array ';' ;
     fn parse(input: syn::parse::ParseStream) -> syn::Result<Self> {
         input.parse::<Token![let]>()?;
-        if Ident::parse(input)? != "states" {
+        if Ident::parse(input)? != Self::STATES_ARRAY_IDENT {
             return Err(input.error(Self::ARRAY_IDENT_ERR));
         };
         input.parse::<Token![=]>()?;
@@ -277,10 +280,20 @@ pub(crate) struct Message {
     pub(crate) msg_type: Ident,
     pub(crate) is_reply: bool,
     pub(crate) owned_states: IdentArray,
+    variant: Option<Ident>,
 }
 
 impl Message {
     const REPLY_ERR: &str = "expected 'Reply'";
+
+    /// Returns the name of the message enum variant to enclose this message type.
+    pub(crate) fn variant(&self) -> &Ident {
+        if let Some(variant) = &self.variant {
+            variant
+        } else {
+            &self.msg_type
+        }
+    }
 }
 
 impl Parse for Message {
@@ -288,13 +301,16 @@ impl Parse for Message {
     fn parse(input: syn::parse::ParseStream) -> syn::Result<Self> {
         let msg_type = Ident::parse(input)?;
         let is_reply = input.peek(Token![::]);
-        if is_reply {
+        let variant = if is_reply {
             input.parse::<Token![::]>()?;
             let reply = Ident::parse(input)?;
             if reply != "Reply" {
                 return Err(syn::Error::new(reply.span(), Self::REPLY_ERR));
             }
-        }
+            Some(format_ident!("{msg_type}Reply"))
+        } else {
+            None
+        };
         let owned_states = if input.peek(token::Bracket) {
             IdentArray::parse(input)?
         } else {
@@ -304,6 +320,7 @@ impl Parse for Message {
             msg_type,
             is_reply,
             owned_states,
+            variant,
         })
     }
 }
@@ -752,10 +769,16 @@ let states = [{}];
             is_reply: bool,
             owned_states: impl Iterator<Item = &'static str>,
         ) -> Self {
+            let variant = if is_reply {
+                Some(format_ident!("{}Reply", msg_type))
+            } else {
+                None
+            };
             Self {
                 msg_type: ident(msg_type),
                 is_reply,
                 owned_states: IdentArray::new(owned_states),
+                variant,
             }
         }
     }

crates/btproto/src/validation.rs

@@ -2,6 +2,8 @@ use std::collections::HashSet;
 
 use proc_macro2::{Ident, Span};
 
+use btrun::{Activate, End};
+
 use crate::{
     error::MaybeErr,
     parsing::{DestinationState, Message, State},
@@ -19,11 +21,10 @@ impl Protocol {
 
     const UNDECLARED_STATE_ERR: &str = "State was not declared.";
     const UNUSED_STATE_ERR: &str = "State was declared but never used.";
-    const END_STATE: &str = "End";
 
     /// Verifies that every state which is used has been declared, except for the End state.
     fn all_states_declared_and_used(&self) -> MaybeErr {
-        let end = Ident::new(Self::END_STATE, Span::call_site());
+        let end = Ident::new(End::ident(), Span::call_site());
         let declared: HashSet<&Ident> = self
             .states_def
             .states
@@ -53,7 +54,7 @@ impl Protocol {
             .collect();
         let unused: MaybeErr = declared
             .difference(&used)
-            .filter(|ident| **ident != Self::END_STATE)
+            .filter(|ident| **ident != End::ident())
             .map(|ident| syn::Error::new(ident.span(), Self::UNUSED_STATE_ERR))
             .collect();
         undeclared.combine(unused)
@@ -61,7 +62,6 @@ impl Protocol {
 
     const UNMATCHED_SENDER_ERR: &str = "No receiver found for message type.";
     const UNMATCHED_RECEIVER_ERR: &str = "No sender found for message type.";
-    const ACTIVATE_MSG: &str = "Activate";
 
     /// Ensures that the recipient state for every sent message has a receiving transition
     /// defined, and every receiver has a sender (except for the Activate message which is sent
@@ -72,7 +72,7 @@ impl Protocol {
         for transition in self.transitions.iter() {
             if let Some(msg) = &transition.in_msg {
                 receivers.insert((&transition.in_state, msg));
-                if msg.msg_type == Self::ACTIVATE_MSG {
+                if msg.msg_type == Activate::ident() {
                     // The Activate message is sent by the run time, so a sender is created to
                     // represent it.
                     senders.insert((&transition.in_state, msg));

crates/btproto/tests/protocol_tests.rs

@@ -1,115 +1,52 @@
+use std::future::{ready, Ready};
+
+use btlib::Result;
 use btproto::protocol;
+use btrun::{Activate, CallMsg, End};
+use serde::{Deserialize, Serialize};
+
+#[derive(Serialize, Deserialize)]
+pub struct Ping;
 
-struct End;
+impl CallMsg for Ping {
+    type Reply = ();
+}
 
-struct Ping;
+/// Tests that the given variable is of the given type.
+macro_rules! assert_type {
+    ($var:expr, $ty:ty) => {{
+        let _: $ty = $var;
+    }};
+}
 
 #[test]
 fn minimal_syntax() {
-    // This is the token stream that should be passed to the protocol macro:
-    // Ident { ident: "let", span: #0 bytes(91..94) }
-    // Ident { ident: "name", span: #0 bytes(95..99) }
-    // Punct { ch: '=', spacing: Alone, span: #0 bytes(100..101) }
-    // Ident { ident: "Minimal", span: #0 bytes(102..109) }
-    // Punct { ch: ';', spacing: Alone, span: #0 bytes(109..110) }
-    // Ident { ident: "let", span: #0 bytes(119..122) }
-    // Ident { ident: "states", span: #0 bytes(123..129) }
-    // Punct { ch: '=', spacing: Alone, span: #0 bytes(130..131) }
-    // Group { delimiter: Bracket,
-    //    stream: TokenStream [Ident { ident: "Init", span: #0 bytes(133..137) }],
-    //    span: #0 bytes(132..138)
-    // }
-    // Punct { ch: ';', spacing: Alone, span: #0 bytes(138..139) }
-    // Ident { ident: "Init", span: #0 bytes(148..152) }
-    // Punct { ch: '?', spacing: Alone, span: #0 bytes(152..153) }
-    // Ident { ident: "Activate", span: #0 bytes(153..161) }
-    // Punct { ch: '-', spacing: Joint, span: #0 bytes(162..163) }
-    // Punct { ch: '>', spacing: Alone, span: #0 bytes(163..164) }
-    // Ident { ident: "End", span: #0 bytes(165..168) }
-    // Punct { ch: ';', spacing: Alone, span: #0 bytes(168..169) }
     protocol! {
-        let name = Minimal;
+        let name = MinimalTest;
         let states = [Init];
         Init?Activate -> End;
     }
+
+    let msg: Option<MinimalTestMsgs> = None;
+    match msg {
+        Some(MinimalTestMsgs::Activate(act)) => assert_type!(act, Activate),
+        None => (),
+    }
+
+    struct InitState;
+
+    impl Init for InitState {
+        type HandleActivateFut = Ready<Result<End>>;
+        fn handle_activate(self, _msg: btrun::Activate) -> Self::HandleActivateFut {
+            ready(Ok(End))
+        }
+    }
 }
 
 #[test]
 fn reply() {
-    // Ident { ident: "let", span: #0 bytes(1529..1532) }
-    // Ident { ident: "name", span: #0 bytes(1533..1537) }
-    // Punct { ch: '=', spacing: Alone, span: #0 bytes(1538..1539) }
-    // Ident { ident: "Reply", span: #0 bytes(1540..1545) }
-    // Punct { ch: ';', spacing: Alone, span: #0 bytes(1545..1546) }
-    // Ident { ident: "let", span: #0 bytes(1555..1558) }
-    // Ident { ident: "states", span: #0 bytes(1559..1565) }
-    // Punct { ch: '=', spacing: Alone, span: #0 bytes(1566..1567) }
-    // Group {
-    //    delimiter: Bracket,
-    //    stream: TokenStream [
-    //        Ident { ident: "ServerInit", span: #0 bytes(1582..1592) },
-    //        Punct { ch: ',', spacing: Alone, span: #0 bytes(1592..1593) },
-    //        Ident { ident: "Listening", span: #0 bytes(1594..1603) },
-    //        Punct { ch: ',', spacing: Alone, span: #0 bytes(1603..1604) },
-    //        Ident { ident: "Client", span: #0 bytes(1617..1623) },
-    //        Punct { ch: ',', spacing: Alone, span: #0 bytes(1623..1624) },
-    //        Ident { ident: "Waiting", span: #0 bytes(1625..1632) },
-    //        Punct { ch: ',', spacing: Alone, span: #0 bytes(1632..1633) }
-    //    ],
-    //    span: #0 bytes(1568..1643)
-    // }
-    // Punct { ch: ';', spacing: Alone, span: #0 bytes(1643..1644) }
-    // Ident { ident: "ServerInit", span: #0 bytes(1653..1663) }
-    // Punct { ch: '?', spacing: Alone, span: #0 bytes(1663..1664) }
-    // Ident { ident: "Activate", span: #0 bytes(1664..1672) }
-    // Punct { ch: '-', spacing: Joint, span: #0 bytes(1673..1674) }
-    // Punct { ch: '>', spacing: Alone, span: #0 bytes(1674..1675) }
-    // Ident { ident: "Listening", span: #0 bytes(1676..1685) }
-    // Punct { ch: ';', spacing: Alone, span: #0 bytes(1685..1686) }
-    // Ident { ident: "Client", span: #0 bytes(1695..1701) }
-    // Punct { ch: '-', spacing: Joint, span: #0 bytes(1702..1703) }
-    // Punct { ch: '>', spacing: Alone, span: #0 bytes(1703..1704) }
-    // Ident { ident: "Waiting", span: #0 bytes(1705..1712) }
-    // Punct { ch: ',', spacing: Alone, span: #0 bytes(1712..1713) }
-    // Punct { ch: '>', spacing: Alone, span: #0 bytes(1714..1715) }
-    // Ident { ident: "service", span: #0 bytes(1715..1722) }
-    // Group {
-    //     delimiter: Parenthesis,
-    //     stream: TokenStream [
-    //         Ident { ident: "Listening", span: #0 bytes(1723..1732) }
-    //     ],
-    //    span: #0 bytes(1722..1733)
-    // }
-    // Punct { ch: '!', spacing: Alone, span: #0 bytes(1733..1734) }
-    // Ident { ident: "Ping", span: #0 bytes(1734..1738) }
-    // Punct { ch: ';', spacing: Alone, span: #0 bytes(1738..1739) }
-    // Ident { ident: "Listening", span: #0 bytes(1748..1757) }
-    // Punct { ch: '?', spacing: Alone, span: #0 bytes(1757..1758) }
-    // Ident { ident: "Ping", span: #0 bytes(1758..1762) }
-    // Punct { ch: '-', spacing: Joint, span: #0 bytes(1763..1764) }
-    // Punct { ch: '>', spacing: Alone, span: #0 bytes(1764..1765) }
-    // Ident { ident: "Listening", span: #0 bytes(1766..1775) }
-    // Punct { ch: ',', spacing: Alone, span: #0 bytes(1775..1776) }
-    // Punct { ch: '>', spacing: Alone, span: #0 bytes(1777..1778) }
-    // Ident { ident: "Waiting", span: #0 bytes(1778..1785) }
-    // Punct { ch: '!', spacing: Alone, span: #0 bytes(1785..1786) }
-    // Ident { ident: "Ping", span: #0 bytes(1786..1790) }
-    // Punct { ch: ':', spacing: Joint, span: #0 bytes(1790..1791) }
-    // Punct { ch: ':', spacing: Alone, span: #0 bytes(1791..1792) }
-    // Ident { ident: "Reply", span: #0 bytes(1792..1797) }
-    // Punct { ch: ';', spacing: Alone, span: #0 bytes(1797..1798) }
-    // Ident { ident: "Waiting", span: #0 bytes(1807..1814) }
-    // Punct { ch: '?', spacing: Alone, span: #0 bytes(1814..1815) }
-    // Ident { ident: "Ping", span: #0 bytes(1815..1819) }
-    // Punct { ch: ':', spacing: Joint, span: #0 bytes(1819..1820) }
-    // Punct { ch: ':', spacing: Alone, span: #0 bytes(1820..1821) }
-    // Ident { ident: "Reply", span: #0 bytes(1821..1826) }
-    // Punct { ch: '-', spacing: Joint, span: #0 bytes(1827..1828) }
-    // Punct { ch: '>', spacing: Alone, span: #0 bytes(1828..1829) }
-    // Ident { ident: "End", span: #0 bytes(1830..1833) }
-    // Punct { ch: ';', spacing: Alone, span: #0 bytes(1833..1834) }
     protocol! {
-        let name = Reply;
+        let name = ReplyTest;
         let states = [
             ServerInit, Listening,
             Client, Waiting,
@@ -119,4 +56,47 @@ fn reply() {
         Listening?Ping -> Listening, >Waiting!Ping::Reply;
         Waiting?Ping::Reply -> End;
     }
+
+    let msg: Option<ReplyTestMsgs> = None;
+    match msg {
+        Some(ReplyTestMsgs::Activate(act)) => assert_type!(act, Activate),
+        Some(ReplyTestMsgs::Ping(ping)) => assert_type!(ping, Ping),
+        Some(ReplyTestMsgs::PingReply(reply)) => assert_type!(reply, <Ping as CallMsg>::Reply),
+        None => (),
+    }
+
+    struct ServerInitState;
+
+    impl ServerInit for ServerInitState {
+        type ActivateOutputListening = ListeningState;
+        type HandleActivateFut = Ready<Result<ListeningState>>;
+        fn handle_activate(self, _msg: btrun::Activate) -> Self::HandleActivateFut {
+            ready(Ok(ListeningState))
+        }
+    }
+
+    struct ListeningState;
+
+    impl Listening for ListeningState {
+        type PingOutputListening = Self;
+        type HandlePingFut = Ready<Result<Self>>;
+        fn handle_ping(self, _msg: Ping) -> Self::HandlePingFut {
+            ready(Ok(self))
+        }
+    }
+
+    struct ClientState;
+
+    impl Client for ClientState {
+        // TODO: Need to generate methods that the client must implement.
+    }
+
+    struct WaitingState;
+
+    impl Waiting for WaitingState {
+        type HandlePingReplyFut = Ready<Result<End>>;
+        fn handle_ping(self, _msg: Ping) -> Self::HandlePingReplyFut {
+            ready(Ok(End))
+        }
+    }
 }

crates/btrun/Cargo.toml

@@ -9,7 +9,6 @@ edition = "2021"
 btlib = { path = "../btlib" }
 bttp = { path = "../bttp" }
 btserde = { path = "../btserde" }
-btproto = { path = "../btproto" }
 tokio = { version = "1.23.0", features = ["rt-multi-thread"] }
 futures = "0.3.25"
 serde = { version = "^1.0.136", features = ["derive"] }
@@ -24,3 +23,4 @@ log = "0.4.17"
 btlib-tests = { path = "../btlib-tests" }
 env_logger = { version = "0.9.0" }
 ctor = { version = "0.1.22" }
+btproto = { path = "../btproto" }

crates/btrun/src/lib.rs

@@ -1,8 +1,5 @@
 #![feature(impl_trait_in_assoc_type)]
 
-pub mod fs_proto;
-pub mod sector_proto;
-
 use std::{
     any::Any,
     collections::HashMap,
@@ -32,10 +29,10 @@ use uuid::Uuid;
 macro_rules! declare_runtime {
     ($name:ident, $ip_addr:expr, $creds:expr) => {
         ::lazy_static::lazy_static! {
-            static ref $name: &'static Runtime = {
+            static ref $name: &'static ::btrun::Runtime = {
                 ::lazy_static::lazy_static! {
-                    static ref RUNTIME: Runtime =  Runtime::_new($creds).unwrap();
-                    static ref RECEIVER: Receiver = _new_receiver($ip_addr, $creds, &*RUNTIME);
+                    static ref RUNTIME: ::btrun::Runtime =  ::btrun::Runtime::_new($creds).unwrap();
+                    static ref RECEIVER: ::bttp::Receiver = _new_receiver($ip_addr, $creds, &*RUNTIME);
                 }
                 // By dereferencing RECEIVER we ensure it is started.
                 let _ = &*RECEIVER;
@@ -289,9 +286,15 @@ impl Display for RuntimeError {
 
 impl std::error::Error for RuntimeError {}
 
-#[allow(dead_code)]
 /// Represents the terminal state of an actor, where it stops processing messages and halts.
-struct End;
+pub struct End;
+
+impl End {
+    /// Returns the identifier for this type which is expected in protocol definitions.
+    pub fn ident() -> &'static str {
+        stringify!(End)
+    }
+}
 
 #[allow(dead_code)]
 /// Delivered to an actor implementation when it starts up.
@@ -302,19 +305,36 @@ pub struct Activate {
     act_id: Uuid,
 }
 
+impl Activate {
+    pub fn new(rt: &'static Runtime, act_id: Uuid) -> Self {
+        Self { rt, act_id }
+    }
+
+    /// Returns the identifier expected for this message in protocol definitions.
+    pub fn ident() -> &'static str {
+        stringify!(Activate)
+    }
+}
+
 /// Deserializes replies sent over the wire.
-struct ReplyCallback<T> {
+pub struct ReplyCallback<T> {
     _phantom: PhantomData<T>,
 }
 
 impl<T: CallMsg> ReplyCallback<T> {
-    fn new() -> Self {
+    pub fn new() -> Self {
         Self {
             _phantom: PhantomData,
         }
     }
 }
 
+impl<T: CallMsg> Default for ReplyCallback<T> {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
 impl<T: CallMsg> DeserCallback for ReplyCallback<T> {
     type Arg<'de> = WireReply<'de> where T: 'de;
     type Return = Result<T::Reply>;
@@ -491,12 +511,18 @@ const MAILBOX_LIMIT: usize = 32;
 
 /// The type of messages sent over the wire between runtimes.
 #[derive(Serialize, Deserialize)]
-struct WireMsg<'a> {
+pub struct WireMsg<'a> {
     to: ActorName,
     from: ActorName,
     payload: &'a [u8],
 }
 
+impl<'a> WireMsg<'a> {
+    pub fn new(to: ActorName, from: ActorName, payload: &'a [u8]) -> Self {
+        Self { to, from, payload }
+    }
+}
+
 impl<'a> bttp::CallMsg<'a> for WireMsg<'a> {
     type Reply<'r> = WireReply<'r>;
 }
@@ -504,7 +530,7 @@ impl<'a> bttp::CallMsg<'a> for WireMsg<'a> {
 impl<'a> bttp::SendMsg<'a> for WireMsg<'a> {}
 
 #[derive(Serialize, Deserialize)]
-enum WireReply<'a> {
+pub enum WireReply<'a> {
     Ok(&'a [u8]),
     Err(&'a str),
 }
@@ -664,379 +690,3 @@ impl Drop for ActorHandle {
         self.abort();
     }
 }
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-
-    use btlib::{
-        crypto::{ConcreteCreds, CredStore, CredsPriv},
-        log::BuilderExt,
-    };
-    use btlib_tests::TEST_STORE;
-    use btproto::protocol;
-    use btserde::to_vec;
-    use bttp::BlockAddr;
-    use ctor::ctor;
-    use lazy_static::lazy_static;
-    use std::{
-        net::{IpAddr, Ipv4Addr},
-        sync::atomic::{AtomicU8, Ordering},
-        time::{Duration, Instant},
-    };
-    use tokio::runtime::Builder;
-
-    const RUNTIME_ADDR: IpAddr = IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1));
-    lazy_static! {
-        static ref RUNTIME_CREDS: Arc<ConcreteCreds> = TEST_STORE.node_creds().unwrap();
-    }
-    declare_runtime!(RUNTIME, RUNTIME_ADDR, RUNTIME_CREDS.clone());
-
-    lazy_static! {
-        /// A tokio async runtime.
-        ///
-        /// When the `#[tokio::test]` attribute is used on a test, a new current thread runtime
-        /// is created for each test
-        /// (source: https://docs.rs/tokio/latest/tokio/attr.test.html#current-thread-runtime).
-        /// This creates a problem, because the first test thread to access the `RUNTIME` static
-        /// will initialize its `Receiver` in its runtime, which will stop running at the end of
-        /// the test. Hence subsequent tests will not be able to send remote messages to this
-        /// `Runtime`.
-        ///
-        /// By creating a single async runtime which is used by all of the tests, we can avoid this
-        /// problem.
-        static ref ASYNC_RT: tokio::runtime::Runtime = Builder::new_current_thread()
-            .enable_all()
-            .build()
-            .unwrap();
-    }
-
-    /// The log level to use when running tests.
-    const LOG_LEVEL: &str = "warn";
-
-    #[ctor]
-    fn ctor() {
-        std::env::set_var("RUST_LOG", format!("{},quinn=WARN", LOG_LEVEL));
-        env_logger::Builder::from_default_env().btformat().init();
-    }
-
-    #[derive(Serialize, Deserialize)]
-    struct EchoMsg(String);
-
-    impl CallMsg for EchoMsg {
-        type Reply = EchoMsg;
-    }
-
-    async fn echo(
-        _rt: &'static Runtime,
-        mut mailbox: mpsc::Receiver<Envelope<EchoMsg>>,
-        _act_id: Uuid,
-    ) {
-        while let Some(envelope) = mailbox.recv().await {
-            let (msg, replier, ..) = envelope.split();
-            if let Some(replier) = replier {
-                if let Err(_) = replier.send(msg) {
-                    panic!("failed to send reply");
-                }
-            }
-        }
-    }
-
-    #[test]
-    fn local_call() {
-        ASYNC_RT.block_on(async {
-            const EXPECTED: &str = "hello";
-            let name = RUNTIME.activate(echo).await;
-            let from = ActorName::new(name.path().clone(), Uuid::default());
-
-            let reply = RUNTIME
-                .call(name.clone(), from, EchoMsg(EXPECTED.into()))
-                .await
-                .unwrap();
-
-            assert_eq!(EXPECTED, reply.0);
-
-            RUNTIME.take(&name).await.unwrap();
-        })
-    }
-
-    #[test]
-    fn remote_call() {
-        ASYNC_RT.block_on(async {
-            const EXPECTED: &str = "hello";
-            let actor_name = RUNTIME.activate(echo).await;
-            let bind_path = Arc::new(RUNTIME_CREDS.bind_path().unwrap());
-            let block_addr = Arc::new(BlockAddr::new(RUNTIME_ADDR, bind_path));
-            let transmitter = Transmitter::new(block_addr, RUNTIME_CREDS.clone())
-                .await
-                .unwrap();
-            let buf = to_vec(&EchoMsg(EXPECTED.to_string())).unwrap();
-            let wire_msg = WireMsg {
-                to: actor_name.clone(),
-                from: RUNTIME.actor_name(Uuid::default()),
-                payload: &buf,
-            };
-
-            let reply = transmitter
-                .call(wire_msg, ReplyCallback::<EchoMsg>::new())
-                .await
-                .unwrap()
-                .unwrap();
-
-            assert_eq!(EXPECTED, reply.0);
-
-            RUNTIME.take(&actor_name).await.unwrap();
-        });
-    }
-
-    /// Tests the `num_running` method.
-    ///
-    /// This test uses its own runtime and so can use the `#[tokio::test]` attribute.
-    #[tokio::test]
-    async fn num_running() {
-        declare_runtime!(
-            LOCAL_RT,
-            // This needs to be different from the address where `RUNTIME` is listening.
-            IpAddr::from([127, 0, 0, 2]),
-            TEST_STORE.node_creds().unwrap()
-        );
-        assert_eq!(0, LOCAL_RT.num_running().await);
-        let name = LOCAL_RT.activate(echo).await;
-        assert_eq!(1, LOCAL_RT.num_running().await);
-        LOCAL_RT.take(&name).await.unwrap();
-        assert_eq!(0, LOCAL_RT.num_running().await);
-    }
-
-    // The following code is a proof-of-concept for what types should be generated for a
-    // simple ping-pong protocol:
-    //
-    protocol! {
-        let name = PingPongProtocol;
-        let states = [
-            ClientInit, SentPing,
-            ServerInit, Listening,
-        ];
-        ClientInit?Activate -> SentPing, >service(Listening)!Ping;
-        ServerInit?Activate -> Listening;
-        Listening?Ping -> End, >SentPing!Ping::Reply;
-        SentPing?Ping::Reply -> End;
-    }
-    //
-    // In words, the protocol is described as follows.
-    // 1. The ClientInit state receives the Activate message. It returns the SentPing state and a
-    //    Ping message to be sent to the Listening state.
-    // 2. The ServerInit state receives the Activate message. It returns the Listening state.
-    // 3. When the Listening state receives the Ping message it returns the End state and a
-    //    Ping::Reply message to be sent to the SentPing state.
-    // 4. When the SentPing state receives the Ping::Reply message it returns the End state.
-    //
-    // The End state represents an end to the session described by the protocol. When an actor
-    // transitions to the End state its function returns.
-    // The generated actor implementation is the sender of the Activate message.
-    // When a state is expecting a Reply message, an error occurs if the message is not received
-    // in a timely manner.
-
-    #[derive(Serialize, Deserialize)]
-    struct Ping;
-    impl CallMsg for Ping {
-        type Reply = PingReply;
-    }
-
-    // I was tempted to name this "Pong", but the proc macro wouldn't think to do that.
-    #[derive(Serialize, Deserialize)]
-    struct PingReply;
-
-    trait ClientInit {
-        type AfterActivate: SentPing;
-        type HandleActivateFut: Future<Output = Result<(Self::AfterActivate, Ping)>>;
-        fn handle_activate(self, msg: Activate) -> Self::HandleActivateFut;
-    }
-
-    trait ServerInit {
-        type AfterActivate: Listening;
-        type HandleActivateFut: Future<Output = Result<Self::AfterActivate>>;
-        fn handle_activate(self, msg: Activate) -> Self::HandleActivateFut;
-    }
-
-    trait Listening {
-        type HandlePingFut: Future<Output = Result<(End, PingReply)>>;
-        fn handle_ping(self, msg: Ping) -> Self::HandlePingFut;
-    }
-
-    trait SentPing {
-        type HandleReplyFut: Future<Output = Result<End>>;
-        fn handle_reply(self, msg: PingReply) -> Self::HandleReplyFut;
-    }
-
-    #[derive(Serialize, Deserialize)]
-    enum PingProtocolMsg {
-        Ping(Ping),
-        PingReply(PingReply),
-    }
-    impl CallMsg for PingProtocolMsg {
-        type Reply = PingProtocolMsg;
-    }
-    impl SendMsg for PingProtocolMsg {}
-
-    struct ClientInitState;
-
-    impl ClientInit for ClientInitState {
-        type AfterActivate = ClientState;
-        type HandleActivateFut = impl Future<Output = Result<(Self::AfterActivate, Ping)>>;
-        fn handle_activate(self, _msg: Activate) -> Self::HandleActivateFut {
-            ready(Ok((ClientState, Ping)))
-        }
-    }
-
-    struct ClientState;
-
-    impl SentPing for ClientState {
-        type HandleReplyFut = Ready<Result<End>>;
-        fn handle_reply(self, _msg: PingReply) -> Self::HandleReplyFut {
-            ready(Ok(End))
-        }
-    }
-
-    #[allow(dead_code)]
-    enum PingClientState {
-        Init(ClientInitState),
-        SentPing(ClientState),
-        End(End),
-    }
-
-    struct ServerInitState;
-
-    struct ServerState;
-
-    impl ServerInit for ServerInitState {
-        type AfterActivate = ServerState;
-        type HandleActivateFut = Ready<Result<Self::AfterActivate>>;
-        fn handle_activate(self, _msg: Activate) -> Self::HandleActivateFut {
-            ready(Ok(ServerState))
-        }
-    }
-
-    impl Listening for ServerState {
-        type HandlePingFut = impl Future<Output = Result<(End, PingReply)>>;
-        fn handle_ping(self, _msg: Ping) -> Self::HandlePingFut {
-            ready(Ok((End, PingReply)))
-        }
-    }
-
-    #[allow(dead_code)]
-    enum PingServerState {
-        ServerInit(ServerInitState),
-        Listening(ServerState),
-        End(End),
-    }
-
-    async fn ping_server(
-        counter: Arc<AtomicU8>,
-        rt: &'static Runtime,
-        mut mailbox: mpsc::Receiver<Envelope<PingProtocolMsg>>,
-        act_id: Uuid,
-    ) {
-        let mut state = {
-            let init = ServerInitState;
-            let state = init.handle_activate(Activate { rt, act_id }).await.unwrap();
-            PingServerState::Listening(state)
-        };
-        while let Some(envelope) = mailbox.recv().await {
-            let (msg, replier, _from) = envelope.split();
-            match (state, msg) {
-                (PingServerState::Listening(listening_state), PingProtocolMsg::Ping(msg)) => {
-                    let (new_state, reply) = listening_state.handle_ping(msg).await.unwrap();
-                    state = PingServerState::End(new_state);
-                    if let Err(_) = replier.unwrap().send(PingProtocolMsg::PingReply(reply)) {
-                        panic!("Failed to send Ping reply.");
-                    }
-                }
-                (_prev_state, _) => {
-                    panic!("Ping protocol violation.");
-                    // A real implementation should assign the previous state and log the error.
-                    // state = prev_state;
-                }
-            }
-            if let PingServerState::End(_) = state {
-                break;
-            }
-        }
-        counter.fetch_sub(1, Ordering::SeqCst);
-    }
-
-    async fn ping_client(
-        counter: Arc<AtomicU8>,
-        server_name: ActorName,
-        rt: &'static Runtime,
-        _mailbox: mpsc::Receiver<Envelope<PingProtocolMsg>>,
-        act_id: Uuid,
-    ) {
-        let init = ClientInitState;
-        let (state, msg) = init.handle_activate(Activate { rt, act_id }).await.unwrap();
-        let from = rt.actor_name(act_id);
-        let reply = rt
-            .call(server_name, from, PingProtocolMsg::Ping(msg))
-            .await
-            .unwrap();
-        if let PingProtocolMsg::PingReply(msg) = reply {
-            state.handle_reply(msg).await.unwrap();
-        } else {
-            panic!("Incorrect message type sent in reply to Ping.");
-        }
-        counter.fetch_sub(1, Ordering::SeqCst);
-    }
-
-    #[test]
-    fn ping_pong_test() {
-        ASYNC_RT.block_on(async {
-            let counter = Arc::new(AtomicU8::new(2));
-            let server_name = {
-                let counter = counter.clone();
-                RUNTIME
-                    .activate(move |rt, mailbox, act_id| ping_server(counter, rt, mailbox, act_id))
-                    .await
-            };
-            let client_name = {
-                let server_name = server_name.clone();
-                let counter = counter.clone();
-                RUNTIME
-                    .activate(move |rt, mailbox, act_id| {
-                        ping_client(counter, server_name, rt, mailbox, act_id)
-                    })
-                    .await
-            };
-
-            let deadline = Instant::now() + Duration::from_millis(500);
-            while counter.load(Ordering::SeqCst) > 0 && Instant::now() < deadline {
-                tokio::time::sleep(Duration::from_millis(20)).await;
-            }
-            // Check that both tasks finished successfully and we didn't just timeout.
-            assert_eq!(0, counter.load(Ordering::SeqCst));
-
-            // TODO: Should actor which return be removed from the runtime automatically?
-            RUNTIME.take(&server_name).await.unwrap();
-            RUNTIME.take(&client_name).await.unwrap();
-        });
-    }
-
-    // Here's another protocol example. This is the Customer and Travel Agency protocol used as an
-    // example in the survey paper "Behavioral Types in Programming Languages."
-    // Note that the Choosing state can send messages at any time, not just in response to another
-    // message because there is a transition from Choosing that doesn't use the receive operator
-    // (`?`).
-    protocol! {
-        let name = TravelAgency;
-        let states = [
-            AgencyInit, Listening,
-            Choosing,
-        ];
-        AgencyInit?Activate -> Listening;
-        Choosing -> Choosing, >service(Listening)!Query, service(Listening)!Accept, service(Listening)!Reject;
-        Listening?Query -> Listening, >Choosing!Query::Reply;
-        Choosing?Query::Reply -> Choosing;
-        Listening?Accept -> End, >Choosing!Accept::Reply;
-        Choosing?Accept::Reply -> End;
-        Listening?Reject -> End, >Choosing!Reject::Reply;
-        Choosing?Reject::Reply -> End;
-    }
-}

crates/btrun/tests/runtime_tests.rs

@@ -0,0 +1,413 @@
+#![feature(impl_trait_in_assoc_type)]
+
+use btrun::*;
+
+use btlib::{
+    Result,
+    crypto::{ConcreteCreds, CredStore, CredsPriv},
+    log::BuilderExt,
+};
+use btlib_tests::TEST_STORE;
+use btproto::protocol;
+use btserde::to_vec;
+use bttp::{BlockAddr, Transmitter};
+use ctor::ctor;
+use lazy_static::lazy_static;
+use std::{
+    net::{IpAddr, Ipv4Addr},
+    sync::{
+        Arc,
+        atomic::{AtomicU8, Ordering}
+    },
+    time::{Duration, Instant},
+    future::{Future, ready, Ready},
+};
+use tokio::{
+    runtime::Builder,
+    sync::mpsc,
+};
+use serde::{Serialize, Deserialize};
+use uuid::Uuid;
+
+const RUNTIME_ADDR: IpAddr = IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1));
+lazy_static! {
+    static ref RUNTIME_CREDS: Arc<ConcreteCreds> = TEST_STORE.node_creds().unwrap();
+}
+declare_runtime!(RUNTIME, RUNTIME_ADDR, RUNTIME_CREDS.clone());
+
+lazy_static! {
+    /// A tokio async runtime.
+    ///
+    /// When the `#[tokio::test]` attribute is used on a test, a new current thread runtime
+    /// is created for each test
+    /// (source: https://docs.rs/tokio/latest/tokio/attr.test.html#current-thread-runtime).
+    /// This creates a problem, because the first test thread to access the `RUNTIME` static
+    /// will initialize its `Receiver` in its runtime, which will stop running at the end of
+    /// the test. Hence subsequent tests will not be able to send remote messages to this
+    /// `Runtime`.
+    ///
+    /// By creating a single async runtime which is used by all of the tests, we can avoid this
+    /// problem.
+    static ref ASYNC_RT: tokio::runtime::Runtime = Builder::new_current_thread()
+        .enable_all()
+        .build()
+        .unwrap();
+}
+
+/// The log level to use when running tests.
+const LOG_LEVEL: &str = "warn";
+
+#[ctor]
+fn ctor() {
+    std::env::set_var("RUST_LOG", format!("{},quinn=WARN", LOG_LEVEL));
+    env_logger::Builder::from_default_env().btformat().init();
+}
+
+#[derive(Serialize, Deserialize)]
+struct EchoMsg(String);
+
+impl CallMsg for EchoMsg {
+    type Reply = EchoMsg;
+}
+
+async fn echo(
+    _rt: &'static Runtime,
+    mut mailbox: mpsc::Receiver<Envelope<EchoMsg>>,
+    _act_id: Uuid,
+) {
+    while let Some(envelope) = mailbox.recv().await {
+        let (msg, replier, ..) = envelope.split();
+        if let Some(replier) = replier {
+            if let Err(_) = replier.send(msg) {
+                panic!("failed to send reply");
+            }
+        }
+    }
+}
+
+#[test]
+fn local_call() {
+    ASYNC_RT.block_on(async {
+        const EXPECTED: &str = "hello";
+        let name = RUNTIME.activate(echo).await;
+        let from = ActorName::new(name.path().clone(), Uuid::default());
+
+        let reply = RUNTIME
+            .call(name.clone(), from, EchoMsg(EXPECTED.into()))
+            .await
+            .unwrap();
+
+        assert_eq!(EXPECTED, reply.0);
+
+        RUNTIME.take(&name).await.unwrap();
+    })
+}
+
+#[test]
+fn remote_call() {
+    ASYNC_RT.block_on(async {
+        const EXPECTED: &str = "hello";
+        let actor_name = RUNTIME.activate(echo).await;
+        let bind_path = Arc::new(RUNTIME_CREDS.bind_path().unwrap());
+        let block_addr = Arc::new(BlockAddr::new(RUNTIME_ADDR, bind_path));
+        let transmitter = Transmitter::new(block_addr, RUNTIME_CREDS.clone())
+            .await
+            .unwrap();
+        let buf = to_vec(&EchoMsg(EXPECTED.to_string())).unwrap();
+        let wire_msg = WireMsg::new(
+            actor_name.clone(),
+            RUNTIME.actor_name(Uuid::default()),
+            &buf,
+        );
+
+        let reply = transmitter
+            .call(wire_msg, ReplyCallback::<EchoMsg>::new())
+            .await
+            .unwrap()
+            .unwrap();
+
+        assert_eq!(EXPECTED, reply.0);
+
+        RUNTIME.take(&actor_name).await.unwrap();
+    });
+}
+
+/// Tests the `num_running` method.
+///
+/// This test uses its own runtime and so can use the `#[tokio::test]` attribute.
+#[tokio::test]
+async fn num_running() {
+    declare_runtime!(
+        LOCAL_RT,
+        // This needs to be different from the address where `RUNTIME` is listening.
+        IpAddr::from([127, 0, 0, 2]),
+        TEST_STORE.node_creds().unwrap()
+    );
+    assert_eq!(0, LOCAL_RT.num_running().await);
+    let name = LOCAL_RT.activate(echo).await;
+    assert_eq!(1, LOCAL_RT.num_running().await);
+    LOCAL_RT.take(&name).await.unwrap();
+    assert_eq!(0, LOCAL_RT.num_running().await);
+}
+
+mod ping_pong {
+    use super::*;
+
+    // The following code is a proof-of-concept for what types should be generated for a
+    // simple ping-pong protocol:
+    protocol! {
+        let name = PingPongProtocol;
+        let states = [
+            ClientInit, SentPing,
+            ServerInit, Listening,
+        ];
+        ClientInit?Activate -> SentPing, >service(Listening)!Ping;
+        ServerInit?Activate -> Listening;
+        Listening?Ping -> End, >SentPing!Ping::Reply;
+        SentPing?Ping::Reply -> End;
+    }
+    //
+    // In words, the protocol is described as follows.
+    // 1. The ClientInit state receives the Activate message. It returns the SentPing state and a
+    //    Ping message to be sent to the Listening state.
+    // 2. The ServerInit state receives the Activate message. It returns the Listening state.
+    // 3. When the Listening state receives the Ping message it returns the End state and a
+    //    Ping::Reply message to be sent to the SentPing state.
+    // 4. When the SentPing state receives the Ping::Reply message it returns the End state.
+    //
+    // The End state represents an end to the session described by the protocol. When an actor
+    // transitions to the End state its function returns.
+    // The generated actor implementation is the sender of the Activate message.
+    // When a state is expecting a Reply message, an error occurs if the message is not received
+    // in a timely manner.
+
+    #[derive(Serialize, Deserialize)]
+    pub struct Ping;
+    impl CallMsg for Ping {
+        type Reply = PingReply;
+    }
+
+    // I was tempted to name this "Pong", but the proc macro wouldn't think to do that.
+    #[derive(Serialize, Deserialize)]
+    pub struct PingReply;
+
+    trait ClientInit2 {
+        type AfterActivate: SentPing2;
+        type HandleActivateFut: Future<Output = Result<(Self::AfterActivate, Ping)>>;
+        fn handle_activate(self, msg: Activate) -> Self::HandleActivateFut;
+    }
+
+    trait ServerInit2 {
+        type AfterActivate: Listening2;
+        type HandleActivateFut: Future<Output = Result<Self::AfterActivate>>;
+        fn handle_activate(self, msg: Activate) -> Self::HandleActivateFut;
+    }
+
+    trait Listening2 {
+        type HandlePingFut: Future<Output = Result<(End, PingReply)>>;
+        fn handle_ping(self, msg: Ping) -> Self::HandlePingFut;
+    }
+
+    trait SentPing2 {
+        type HandleReplyFut: Future<Output = Result<End>>;
+        fn handle_reply(self, msg: PingReply) -> Self::HandleReplyFut;
+    }
+
+    #[derive(Serialize, Deserialize)]
+    enum PingProtocolMsg {
+        Ping(Ping),
+        PingReply(PingReply),
+    }
+    impl CallMsg for PingProtocolMsg {
+        type Reply = PingProtocolMsg;
+    }
+    impl SendMsg for PingProtocolMsg {}
+
+    struct ClientInitState;
+
+    impl ClientInit2 for ClientInitState {
+        type AfterActivate = ClientState;
+        type HandleActivateFut = impl Future<Output = Result<(Self::AfterActivate, Ping)>>;
+        fn handle_activate(self, _msg: Activate) -> Self::HandleActivateFut {
+            ready(Ok((ClientState, Ping)))
+        }
+    }
+
+    struct ClientState;
+
+    impl SentPing2 for ClientState {
+        type HandleReplyFut = Ready<Result<End>>;
+        fn handle_reply(self, _msg: PingReply) -> Self::HandleReplyFut {
+            ready(Ok(End))
+        }
+    }
+
+    #[allow(dead_code)]
+    enum PingClientState {
+        Init(ClientInitState),
+        SentPing(ClientState),
+        End(End),
+    }
+
+    struct ServerInitState;
+
+    struct ServerState;
+
+    impl ServerInit2 for ServerInitState {
+        type AfterActivate = ServerState;
+        type HandleActivateFut = Ready<Result<Self::AfterActivate>>;
+        fn handle_activate(self, _msg: Activate) -> Self::HandleActivateFut {
+            ready(Ok(ServerState))
+        }
+    }
+
+    impl Listening2 for ServerState {
+        type HandlePingFut = impl Future<Output = Result<(End, PingReply)>>;
+        fn handle_ping(self, _msg: Ping) -> Self::HandlePingFut {
+            ready(Ok((End, PingReply)))
+        }
+    }
+
+    #[allow(dead_code)]
+    enum PingServerState {
+        ServerInit(ServerInitState),
+        Listening(ServerState),
+        End(End),
+    }
+
+    async fn ping_server(
+        counter: Arc<AtomicU8>,
+        rt: &'static Runtime,
+        mut mailbox: mpsc::Receiver<Envelope<PingProtocolMsg>>,
+        act_id: Uuid,
+    ) {
+        let mut state = {
+            let init = ServerInitState;
+            let state = init.handle_activate(Activate::new(rt, act_id)).await.unwrap();
+            PingServerState::Listening(state)
+        };
+        while let Some(envelope) = mailbox.recv().await {
+            let (msg, replier, _from) = envelope.split();
+            match (state, msg) {
+                (PingServerState::Listening(listening_state), PingProtocolMsg::Ping(msg)) => {
+                    let (new_state, reply) = listening_state.handle_ping(msg).await.unwrap();
+                    state = PingServerState::End(new_state);
+                    if let Err(_) = replier.unwrap().send(PingProtocolMsg::PingReply(reply)) {
+                        panic!("Failed to send Ping reply.");
+                    }
+                }
+                (_prev_state, _) => {
+                    panic!("Ping protocol violation.");
+                    // A real implementation should assign the previous state and log the error.
+                    // state = prev_state;
+                }
+            }
+            if let PingServerState::End(_) = state {
+                break;
+            }
+        }
+        counter.fetch_sub(1, Ordering::SeqCst);
+    }
+
+    async fn ping_client(
+        counter: Arc<AtomicU8>,
+        server_name: ActorName,
+        rt: &'static Runtime,
+        _mailbox: mpsc::Receiver<Envelope<PingProtocolMsg>>,
+        act_id: Uuid,
+    ) {
+        let init = ClientInitState;
+        let (state, msg) = init.handle_activate(Activate::new(rt, act_id)).await.unwrap();
+        let from = rt.actor_name(act_id);
+        let reply = rt
+            .call(server_name, from, PingProtocolMsg::Ping(msg))
+            .await
+            .unwrap();
+        if let PingProtocolMsg::PingReply(msg) = reply {
+            state.handle_reply(msg).await.unwrap();
+        } else {
+            panic!("Incorrect message type sent in reply to Ping.");
+        }
+        counter.fetch_sub(1, Ordering::SeqCst);
+    }
+
+    #[test]
+    fn ping_pong_test() {
+        ASYNC_RT.block_on(async {
+            let counter = Arc::new(AtomicU8::new(2));
+            let server_name = {
+                let counter = counter.clone();
+                RUNTIME
+                    .activate(move |rt, mailbox, act_id| ping_server(counter, rt, mailbox, act_id))
+                    .await
+            };
+            let client_name = {
+                let server_name = server_name.clone();
+                let counter = counter.clone();
+                RUNTIME
+                    .activate(move |rt, mailbox, act_id| {
+                        ping_client(counter, server_name, rt, mailbox, act_id)
+                    })
+                    .await
+            };
+
+            let deadline = Instant::now() + Duration::from_millis(500);
+            while counter.load(Ordering::SeqCst) > 0 && Instant::now() < deadline {
+                tokio::time::sleep(Duration::from_millis(20)).await;
+            }
+            // Check that both tasks finished successfully and we didn't just timeout.
+            assert_eq!(0, counter.load(Ordering::SeqCst));
+
+            // TODO: Should actor which return be removed from the runtime automatically?
+            RUNTIME.take(&server_name).await.unwrap();
+            RUNTIME.take(&client_name).await.unwrap();
+        });
+    }
+}
+
+
+mod travel_agency {
+    use super::*;
+
+    // Here's another protocol example. This is the Customer and Travel Agency protocol used as an
+    // example in the survey paper "Behavioral Types in Programming Languages."
+    // Note that the Choosing state can send messages at any time, not just in response to another
+    // message because there is a transition from Choosing that doesn't use the receive operator
+    // (`?`).
+    protocol! {
+        let name = TravelAgency;
+        let states = [
+            AgencyInit, Listening,
+            Choosing,
+        ];
+        AgencyInit?Activate -> Listening;
+        Choosing -> Choosing, >service(Listening)!Query, service(Listening)!Accept, service(Listening)!Reject;
+        Listening?Query -> Listening, >Choosing!Query::Reply;
+        Choosing?Query::Reply -> Choosing;
+        Listening?Accept -> End, >Choosing!Accept::Reply;
+        Choosing?Accept::Reply -> End;
+        Listening?Reject -> End, >Choosing!Reject::Reply;
+        Choosing?Reject::Reply -> End;
+    }
+
+    #[derive(Serialize, Deserialize)]
+    pub struct Query;
+
+    impl CallMsg for Query {
+        type Reply = ();
+    }
+
+    #[derive(Serialize, Deserialize)]
+    pub struct Reject;
+
+    impl CallMsg for Reject {
+        type Reply = ();
+    }
+
+    #[derive(Serialize, Deserialize)]
+    pub struct Accept;
+
+    impl CallMsg for Accept {
+        type Reply = ();
+    }
+}

crates/btsector/Cargo.toml

@@ -0,0 +1,12 @@
+[package]
+name = "btsector"
+version = "0.1.0"
+edition = "2021"
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[dependencies]
+btlib = { path = "../btlib" }
+btrun = { path = "../btrun" }
+btproto = { path = "../btproto" }
+serde = { version = "^1.0.136", features = ["derive"] }

crates/btrun/src/sector_proto.rs → crates/btsector/src/lib.rs

@@ -1,9 +1,10 @@
 //! Types which define the protocol used by the sector layer.
 
-use btlib::{crypto::merkle_stream::VariantMerkleTree, BlockMeta, Inode};
+use btrun::CallMsg;
+use btlib::{Result, crypto::merkle_stream::VariantMerkleTree, BlockMeta, Inode};
 use btproto::protocol;
 
-use crate::{CallMsg, Deserialize, Serialize};
+use serde::{Serialize, Deserialize};
 
 #[derive(Serialize, Deserialize)]
 pub struct SectorMsg {