feat: grpc driver project proto implemented

Cargo.toml

@@ -16,6 +16,9 @@ tracing = "0.1.40"
 tracing-subscriber = {version = "0.3.18", features = ["env-filter"]}
 rand = "0.8.5"
 dotenv = "0.15.0"
+prost = "0.12.6"
+tonic-reflection = "0.11.0"
+futures = "0.3.30"
 
 [[bin]]
 name = "daemon"
@@ -31,3 +34,5 @@ path = "src/lib.rs"
 [dev-dependencies]
 mockall = "0.12.1"
 
+[build-dependencies]
+tonic-build = "0.11.0"

build.rs

@@ -0,0 +1,4 @@
+fn main() -> Result<(), Box<dyn std::error::Error>> {
+    tonic_build::compile_protos("proto/v1alpha/project.proto")?;
+    Ok(())
+}

proto/v1alpha/project.proto

@@ -0,0 +1,16 @@
+syntax = "proto3";
+
+package fabric.project.v1alpha;
+
+message CreateProjectRequest {
+  string name = 1; 
+}
+
+message CreateProjectResponse {
+  string name = 1; 
+  string slug = 2; 
+}
+
+service ProjectService {
+  rpc CreateProject(CreateProjectRequest) returns (CreateProjectResponse);
+}

src/bin/rpc.rs

@@ -1,6 +1,6 @@
 use anyhow::Result;
 use dotenv::dotenv;
-use tracing::Level;
+use tracing::{info, Level};
 use tracing_subscriber::{fmt, layer::SubscriberExt, util::SubscriberInitExt, EnvFilter};
 
 #[tokio::main]
@@ -17,9 +17,12 @@ async fn main() -> Result<()> {
         .with(env_filter)
         .init();
 
-    fabric::drivers::grpc::server().await?;
+    let grpc_driver = tokio::spawn(async { fabric::drivers::grpc::server().await });
 
-    fabric::drivers::event::subscribe().await?;
+    let event_driver = tokio::spawn(async { fabric::drivers::event::subscribe().await });
+
+    info!("rpc services running");
+    let _result = futures::future::join(grpc_driver, event_driver).await;
 
     Ok(())
 }

src/domain/management/events.rs

@@ -14,6 +14,6 @@ pub enum Event {
 }
 
 #[async_trait::async_trait]
-pub trait EventBridge {
+pub trait EventBridge: Send + Sync {
     async fn dispatch(&self, event: Event) -> Result<()>;
 }

src/domain/management/project.rs

@@ -43,7 +43,7 @@ impl From<Namespace> for Project {
 }
 
 #[async_trait::async_trait]
-pub trait ProjectCache {
+pub trait ProjectCache: Send + Sync {
     async fn create(&self, project: &Project) -> Result<()>;
     async fn find_by_slug(&self, slug: &str) -> Result<Option<Project>>;
 }

src/drivers/event/mod.rs

@@ -4,7 +4,6 @@ use kafka::{
     consumer::Consumer,
 };
 use std::{path::Path, sync::Arc};
-use tracing::info;
 
 use crate::{
     domain::management::{events::Event, project::create_cache},
@@ -27,8 +26,6 @@ pub async fn subscribe() -> Result<()> {
         .with_offset_storage(Some(GroupOffsetStorage::Kafka))
         .create()?;
 
-    info!("Event Driver started listening");
-
     loop {
         let mss = consumer.poll()?;
         if mss.is_empty() {

src/drivers/grpc/mod.rs

@@ -1,31 +1,36 @@
 use anyhow::Result;
-use rand::distributions::Alphanumeric;
-use rand::Rng;
+use std::net::SocketAddr;
+use std::str::FromStr;
 use std::{path::Path, sync::Arc};
+use tonic::transport::Server;
 
-use crate::domain::management;
-use crate::domain::management::project::Project;
 use crate::driven::cache::{project::SqliteProjectCache, SqliteCache};
 use crate::driven::kafka::KafkaEventBridge;
 
+mod project;
+
+pub mod proto {
+    pub mod project {
+        tonic::include_proto!("fabric.project.v1alpha");
+    }
+}
+
 pub async fn server() -> Result<()> {
     let sqlite_cache = Arc::new(SqliteCache::new(Path::new("dev.db")).await?);
     let project_cache = Arc::new(SqliteProjectCache::new(sqlite_cache));
 
     let event_bridge = Arc::new(KafkaEventBridge::new(&["localhost:9092".into()], "events")?);
 
-    let slug: String = rand::thread_rng()
-        .sample_iter(&Alphanumeric)
-        .take(7)
-        .map(char::from)
-        .collect();
+    let project_inner = project::ProjectServiceImpl::new(project_cache, event_bridge);
+    let project_service =
+        proto::project::project_service_server::ProjectServiceServer::new(project_inner);
 
-    let project = Project {
-        name: format!("test name {slug}"),
-        slug,
-    };
+    let address = SocketAddr::from_str("0.0.0.0:5000")?;
 
-    management::project::create(project_cache, event_bridge, project).await?;
+    Server::builder()
+        .add_service(project_service)
+        .serve(address)
+        .await?;
 
     Ok(())
 }