feat: support service generation

Signed-off-by: Brooks Townsend <brooksmtownsend@gmail.com>

Author: brooksmtownsend

Cargo.lock

@@ -0,0 +1 @@
+src/generated/*

crates/control-interface-proto/.gitignore

@@ -10,11 +10,19 @@ repository.workspace = true
 [dependencies]
 # Required for generating
 proto = { workspace = true }
-# Required for using
+# Required for using the types
 prost = { workspace = true, features = ["derive", "std"] }
 
+# TEMPORARY: not required for this crate, required for consumers
+# maybe consider keeping this and the generated code? idk
+anyhow = { workspace = true }
+async-nats = { workspace = true, features = ["ring"] }
+bytes = { workspace = true }
+futures = { workspace = true }
+
 [build-dependencies]
 prost-build = { workspace = true, features = ["format"] }
+nats-service-generator = { path = "../nats-service-generator" }
 
 [dev-dependencies]
 async-nats = { workspace = true, features = ["ring"] }

crates/control-interface-proto/Cargo.toml

@@ -1,3 +1,5 @@
+use nats_service_generator::NatsServiceGenerator;
+
 fn main() -> std::io::Result<()> {
     // Find all the .proto files
     let protos = std::fs::read_dir("src")?
@@ -12,6 +14,10 @@ fn main() -> std::io::Result<()> {
             })
         })
         .collect::<Vec<_>>();
-    prost_build::compile_protos(&protos, &["src/"])?;
+
+    prost_build::Config::new()
+        .out_dir("src/generated")
+        .service_generator(Box::new(NatsServiceGenerator))
+        .compile_protos(&protos, &["src/"])?;
     Ok(())
 }

crates/control-interface-proto/build.rs

@@ -2,6 +2,11 @@ syntax = "proto3";
 
 package wasmcloud.types;
 
+service ControlInterfaceService {
+    // Request/Response
+    rpc StartComponent(StartComponentRequest) returns (StartComponentResponse);
+}
+
 // CTL message to start component
 message StartComponentRequest {
     // OCI image reference

crates/control-interface-proto/src/ctl.proto

@@ -2,134 +2,80 @@
 // It is important to maintain the same structure as in the proto.
 pub mod wasmcloud {
     pub mod ctl {
-        include!(concat!(env!("OUT_DIR"), "/wasmcloud.types.rs"));
+        include!("generated/wasmcloud.types.rs");
     }
 }
 
 #[cfg(test)]
 mod test {
     use super::wasmcloud;
+    use crate::wasmcloud::ctl::{ControlInterfaceServiceClient, ControlInterfaceServiceServer};
 
-    use bytes::BytesMut;
-    use prost::Message;
-    use tokio::task::JoinSet;
-
-    // (Host-side) implementation of the ControlInterface
-    pub trait ControlServer {
-        fn start_component(
-            &self,
-            request: wasmcloud::ctl::StartComponentRequest,
-        ) -> impl futures::Future<Output = Result<wasmcloud::ctl::StartComponentResponse, String>> + Send;
-    }
-    pub struct WasmcloudControlServer {}
-    impl ControlServer for WasmcloudControlServer {
+    struct Host;
+    impl ControlInterfaceServiceServer for Host {
         async fn start_component(
             &self,
             request: wasmcloud::ctl::StartComponentRequest,
-        ) -> Result<wasmcloud::ctl::StartComponentResponse, String> {
+        ) -> anyhow::Result<wasmcloud::ctl::StartComponentResponse> {
             let component_id = format!("{}-{}", request.reference, request.max_instances);
             Ok(wasmcloud::ctl::StartComponentResponse {
                 component_id,
                 message: "everything went smoothly and we're going to be okay".to_string(),
             })
         }
     }
-    async fn start_server<S>(server: S, client: async_nats::Client) -> Result<JoinSet<()>, String>
-    where
-        S: ControlServer + Send + 'static,
-    {
-        let mut sub = client
-            .subscribe("wasmbus.ctl.proto.>")
-            .await
-            .expect("to subscribe to start component");
-        use futures::StreamExt;
-        let mut tasks = JoinSet::new();
-        tasks.spawn(async move {
-            while let Some(message) = sub.next().await {
-                if message.subject.contains("start.component") {
-                    let request = wasmcloud::ctl::StartComponentRequest::decode(message.payload)
-                        .expect("to decode request");
-                    let reply = server
-                        .start_component(request)
-                        .await
-                        .expect("to start component");
-                    if let Some(reply_to) = message.reply {
-                        let mut buf = BytesMut::with_capacity(reply.encoded_len());
-                        reply
-                            .encode(&mut buf)
-                            .expect("to encode without reaching capacity");
-                        client
-                            .publish(reply_to, buf.into())
-                            .await
-                            .expect("to send reply");
-                    }
-                }
-            }
-        });
 
-        Ok(tasks)
-    }
+    // Test and validate host functionality without NATS
+    #[tokio::test]
+    async fn unit_test_proto() {
+        let host = Host {};
+        let request = wasmcloud::ctl::StartComponentRequest {
+            reference: "test".to_string(),
+            max_instances: 1,
+        };
+        let response = host
+            .start_component(request)
+            .await
+            .expect("host to handle request");
 
-    // (wash/wadm/client-side) implementation of the ControlInterface
-    pub trait ControlClient {
-        async fn start_component(
-            &self,
-            request: wasmcloud::ctl::StartComponentRequest,
-        ) -> Result<wasmcloud::ctl::StartComponentResponse, String>;
-    }
-    pub struct WasmcloudControlClient {
-        client: async_nats::Client,
-    }
-    impl WasmcloudControlClient {
-        pub fn new(client: async_nats::Client) -> Self {
-            Self { client }
-        }
-    }
-    impl ControlClient for WasmcloudControlClient {
-        async fn start_component(
-            &self,
-            request: wasmcloud::ctl::StartComponentRequest,
-        ) -> Result<wasmcloud::ctl::StartComponentResponse, String> {
-            let mut buf = BytesMut::with_capacity(request.encoded_len());
-            request
-                .encode(&mut buf)
-                .expect("to encode without reaching capacity");
-            let reply = self
-                .client
-                .request("wasmbus.ctl.proto.start.component", buf.into())
-                .await
-                .expect("to send request");
-            wasmcloud::ctl::StartComponentResponse::decode(reply.payload).map_err(|e| e.to_string())
-        }
+        assert_eq!(response.component_id, "test-1");
+        assert_eq!(
+            response.message,
+            "everything went smoothly and we're going to be okay"
+        );
     }
 
+    // Test and validate host functionality with NATS
     #[tokio::test]
     async fn end_to_end_proto() {
-        let server = WasmcloudControlServer {};
         let nats_client = async_nats::connect("nats://127.0.0.1:4222")
             .await
             .expect("should connect to NATS");
-        let _task = start_server(server, nats_client)
-            .await
-            .expect("to start server");
-
-        let client = WasmcloudControlClient::new(
-            async_nats::connect("nats://127.0.0.1:4222")
+        let host = Host {};
+        let server = tokio::spawn({
+            let nats_client = nats_client.clone();
+            wasmcloud::ctl::start_server(host, nats_client)
                 .await
-                .expect("should connect to NATS"),
-        );
-        let reply = client
-            .start_component(wasmcloud::ctl::StartComponentRequest {
+                .expect("to subscribe and start server")
+        });
+
+        // Let subscriptions get set up
+        tokio::time::sleep(std::time::Duration::from_secs(1)).await;
+
+        let reply = nats_client
+            .start_component(crate::wasmcloud::ctl::StartComponentRequest {
                 reference: "test".to_string(),
                 max_instances: 1,
             })
             .await
-            .expect("reply to be okay");
+            .expect("should start component");
 
         assert_eq!(reply.component_id, "test-1");
         assert_eq!(
             reply.message,
             "everything went smoothly and we're going to be okay"
         );
+
+        server.abort();
     }
 }

crates/control-interface-proto/src/lib.rs

@@ -0,0 +1,12 @@
+[package]
+name = "nats-service-generator"
+version = "0.1.0"
+authors.workspace = true
+categories.workspace = true
+edition.workspace = true
+license.workspace = true
+repository.workspace = true
+
+[dependencies]
+convert_case = "0.7.1"
+prost-build = { workspace = true, features = ["format"] }