Draft: Reliable ordered message protocol

.gitignore

@@ -12,3 +12,4 @@ Cargo.lock
 
 # MSVC Windows builds of rustc generate these, which store debugging information
 *.pdb
+./.idea/*

Cargo.toml

@@ -1,12 +1,13 @@
 [workspace]
-
 resolver = "2"
+
 members = [
     "citadel-internal-service",
     "citadel-internal-service-types",
     "citadel-internal-service-connector",
     "citadel-internal-service-macros",
     "service",
+    "citadel-messaging",
 ]
 
 exclude = [
@@ -20,6 +21,7 @@ citadel-internal-service-types = { path = "./citadel-internal-service-types", de
 citadel-internal-service = { path = "./citadel-internal-service", default-features = false, version = "0.1.0" }
 citadel-internal-service-connector = { path = "./citadel-internal-service-connector", default-features = false, version = "0.1.0" }
 citadel-internal-service-macros = { path = "./citadel-internal-service-macros", default-features = false, version = "0.1.0" }
+citadel-messaging = { path = "./citadel-messaging", default-features = false, version = "0.1.0" }
 
 # Avarok deps
 citadel_sdk = { git = "https://github.com/Avarok-Cybersecurity/Citadel-Protocol/" }
@@ -42,4 +44,4 @@ anyhow = "1.0.71"
 async-recursion = { version = "1.0.4" }
 parking_lot = { version = "0.12.1" }
 structopt = { version = "0.3.26" }
-lazy_static = "1.4.0"
+lazy_static = "1.4.0"

citadel-internal-service-macros/src/lib.rs

@@ -1,5 +1,6 @@
 use proc_macro::TokenStream;
 use quote::quote;
+use syn::spanned::Spanned;
 use syn::{parse_macro_input, Data, DataEnum, DeriveInput, Ident};
 
 #[proc_macro_derive(IsError)]
@@ -12,6 +13,103 @@ pub fn is_notification_derive(input: TokenStream) -> TokenStream {
     generate_function(input, "Notification", "is_notification")
 }
 
+// Create a proc macro that generates a function that goes through each enum variant, looks at the first and only item in the variant, and creates a function called request_id(&self) -> Option<&Uuid>, that looks at the field "request_id" in the variant and returns a reference to it if it exists.
+#[proc_macro_derive(RequestId)]
+pub fn request_id_derive(input: TokenStream) -> TokenStream {
+    generate_field_function(input, "request_id", "request_id")
+}
+
+fn generate_field_function(
+    input: TokenStream,
+    field_name: &str,
+    function_name: &str,
+) -> TokenStream {
+    // Parse the input tokens into a syntax tree
+    let input = parse_macro_input!(input as DeriveInput);
+
+    // Extract the identifier and data from the input
+    let name = &input.ident;
+    let data = if let Data::Enum(data) = input.data {
+        data
+    } else {
+        // This macro only supports enums
+        panic!("{function_name} can only be derived for enums");
+    };
+
+    // Convert the function name to a tokenstream
+    let function_name = Ident::new(function_name, name.span());
+
+    // Generate match arms for each enum variant
+    let match_arms = generate_field_match_arms(name, &data, field_name);
+
+    // Generate the implementation of the `is_error` method
+    let expanded = quote! {
+        impl #name {
+            pub fn #function_name(&self) -> Option<&Uuid> {
+                match self {
+                    #(#match_arms)*
+                }
+            }
+        }
+    };
+
+    // Convert into a TokenStream and return it
+    TokenStream::from(expanded)
+}
+
+fn generate_field_match_arms(
+    name: &Ident,
+    data_enum: &DataEnum,
+    field_name: &str,
+) -> Vec<proc_macro2::TokenStream> {
+    data_enum
+        .variants
+        .iter()
+        .map(|variant| {
+            let variant_ident = &variant.ident;
+            let field = variant.fields.iter().next().unwrap();
+            let field_name = field_name.to_string();
+            let field_name = Ident::new(&field_name, field.span());
+
+            // Determine if the enum is of the form Enum::Variant(inner) or Enum::Variant { inner, .. }
+            let is_tuple_variant = variant
+                .fields
+                .iter()
+                .next()
+                .map_or(false, |field| field.ident.is_none());
+            if is_tuple_variant {
+                // See if the type is a Uuid or an Option<Uuid>
+                if let syn::Type::Path(type_path) = &field.ty {
+                    if type_path.path.segments.len() == 1 {
+                        if type_path.path.segments[0].ident == "Uuid" {
+                            return quote! {
+                                #name::#variant_ident(inner) => Some(&inner.#field_name),
+                            };
+                        }
+                    }
+                }
+
+                // See if "inner" has a field called "request_id"
+                if field.ident.is_none() {
+                    return quote! {
+                    #name::#variant_ident(_) => None,
+                    };
+                }
+
+                // Match against each variant, ignoring any inner data
+                quote! {
+                    #name::#variant_ident(inner, ..) => inner.#field_name.as_ref(),
+                }
+            } else {
+                // Match against each variant, ignoring any inner data
+                quote! {
+                    #name::#variant_ident { #field_name, .. } => Some(#field_name.as_ref()),
+                }
+            }
+        })
+        .collect()
+}
+
 fn generate_function(input: TokenStream, contains: &str, function_name: &str) -> TokenStream {
     // Parse the input tokens into a syntax tree
     let input = parse_macro_input!(input as DeriveInput);

citadel-internal-service-types/src/lib.rs

@@ -1,6 +1,5 @@
 use bytes::BytesMut;
-use citadel_internal_service_macros::{IsError, IsNotification};
-use citadel_sdk::prelude::PreSharedKey;
+use citadel_internal_service_macros::{IsError, IsNotification, RequestId};
 pub use citadel_types::prelude::{
     ConnectMode, MemberState, MessageGroupKey, ObjectTransferStatus, SecBuffer, SecurityLevel,
     SessionSecuritySettings, TransferType, UdpMode, UserIdentifier, VirtualObjectMetadata,
@@ -11,8 +10,10 @@ use std::net::SocketAddr;
 use std::path::PathBuf;
 use std::time::Duration;
 use uuid::Uuid;
+// TODO: Move PreSharedKey into citadel-types
+use citadel_sdk::prelude::PreSharedKey;
 
-pub mod service;
+pub mod messaging_layer;
 
 #[derive(Serialize, Deserialize, Debug, Clone)]
 pub struct ConnectSuccess {
@@ -565,7 +566,7 @@ pub struct FileTransferTickNotification {
     pub status: ObjectTransferStatus,
 }
 
-#[derive(Serialize, Deserialize, Debug, Clone, IsError, IsNotification)]
+#[derive(Serialize, Deserialize, Debug, Clone, IsError, IsNotification, RequestId)]
 pub enum InternalServiceResponse {
     ConnectSuccess(ConnectSuccess),
     ConnectFailure(ConnectFailure),
@@ -646,7 +647,7 @@ pub enum InternalServiceResponse {
     ListRegisteredPeersFailure(ListRegisteredPeersFailure),
 }
 
-#[derive(Serialize, Deserialize, Debug, Clone)]
+#[derive(Serialize, Deserialize, Debug, Clone, RequestId)]
 pub enum InternalServiceRequest {
     Connect {
         // A user-provided unique ID that will be returned in the response

citadel-internal-service-types/src/messaging_layer.rs

@@ -0,0 +1,35 @@
+use crate::InternalServiceResponse;
+use citadel_types::prelude::SecurityLevel;
+use serde::{Deserialize, Serialize};
+use uuid::Uuid;
+
+// What the consumer receives from the underlying messaging service
+#[derive(Serialize, Deserialize)]
+pub enum MessengerUpdate {
+    Message { message: CWMessage },
+    Other { response: InternalServiceResponse },
+}
+
+#[derive(Serialize, Deserialize)]
+pub enum CWProtocolMessage {
+    Message { message: CWMessage },
+    MessageAck { id: u64 },
+    Poll { ids: Vec<u64> },
+    PollResponse { messages: Vec<CWMessage> },
+}
+
+pub struct OutgoingCWMessage {
+    pub cid: u64,
+    pub peer_cid: Option<u64>,
+    pub contents: Vec<u8>,
+    pub request_id: Uuid,
+    pub security_level: Option<SecurityLevel>,
+}
+
+#[derive(Serialize, Deserialize, Clone)]
+pub struct CWMessage {
+    pub id: u64,
+    pub cid: u64,
+    pub peer_cid: Option<u64>,
+    pub contents: Vec<u8>,
+}

citadel-messaging/Cargo.toml

@@ -0,0 +1,16 @@
+[package]
+name = "citadel-messaging"
+version = "0.1.0"
+edition = "2021"
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[dependencies]
+citadel-internal-service-connector = { workspace = true }
+citadel-internal-service-types = { workspace = true }
+serde = { workspace = true }
+bincode2 = { workspace = true }
+uuid = { version = "1.8.0", features = ["v4"] }
+tokio = { workspace = true }
+log = "0.4.21"
+futures = { workspace = true }