diff --git a/Cargo.lock b/Cargo.lock index 55c05eb..0240cea 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -137,6 +137,12 @@ version = "0.21.7" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9d297deb1925b89f2ccc13d7635fa0714f12c87adce1c75356b39ca9b7178567" +[[package]] +name = "beef" +version = "0.5.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3a8241f3ebb85c056b509d4327ad0358fbbba6ffb340bf388f26350aeda225b1" + [[package]] name = "bitflags" version = "1.3.2" @@ -514,6 +520,38 @@ version = "0.4.20" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b5e6163cb8c49088c2c36f57875e58ccd8c87c7427f7fbd50ea6710b2f3f2e8f" +[[package]] +name = "logos" +version = "0.13.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c000ca4d908ff18ac99b93a062cb8958d331c3220719c52e77cb19cc6ac5d2c1" +dependencies = [ + "logos-derive", +] + +[[package]] +name = "logos-codegen" +version = "0.13.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dc487311295e0002e452025d6b580b77bb17286de87b57138f3b5db711cded68" +dependencies = [ + "beef", + "fnv", + "proc-macro2", + "quote", + "regex-syntax 0.6.29", + "syn", +] + +[[package]] +name = "logos-derive" +version = "0.13.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dbfc0d229f1f42d790440136d941afd806bc9e949e2bcb8faa813b0f00d1267e" +dependencies = [ + "logos-codegen", +] + [[package]] name = "matchit" version = "0.7.3" @@ -526,6 +564,28 @@ version = "2.7.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "523dc4f511e55ab87b694dc30d0f820d60906ef06413f93d4d7a1385599cc149" +[[package]] +name = "miette" +version = "7.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "baed61d13cc3723ee6dbed730a82bfacedc60a85d81da2d77e9c3e8ebc0b504a" +dependencies = [ + "miette-derive", + "thiserror", + "unicode-width", +] + +[[package]] +name = "miette-derive" +version = "7.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f301c3f54f98abc6c212ee722f5e5c62e472a334415840669e356f04850051ec" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + [[package]] name = "mime" version = "0.3.17" @@ -733,6 +793,18 @@ dependencies = [ "prost", ] +[[package]] +name = "protox-parse" +version = "0.6.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "033b939d76d358f7c32120c86c71f515bae45e64f2bde455200356557276276c" +dependencies = [ + "logos", + "miette", + "prost-types", + "thiserror", +] + [[package]] name = "quote" version = "1.0.35" @@ -790,7 +862,7 @@ dependencies = [ "aho-corasick", "memchr", "regex-automata", - "regex-syntax", + "regex-syntax 0.8.2", ] [[package]] @@ -801,9 +873,15 @@ checksum = "5bb987efffd3c6d0d8f5f89510bb458559eab11e4f869acb20bf845e016259cd" dependencies = [ "aho-corasick", "memchr", - "regex-syntax", + "regex-syntax 0.8.2", ] +[[package]] +name = "regex-syntax" +version = "0.6.29" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f162c6dd7b008981e4d40210aca20b4bd0f9b60ca9271061b07f78537722f2e1" + [[package]] name = "regex-syntax" version = "0.8.2" @@ -820,7 +898,9 @@ dependencies = [ "hyper-util", "prost", "prost-types", + "protox-parse", "tokio", + "tokio-stream", "tonic", "tonic-build", "tonic-reflection", @@ -942,6 +1022,26 @@ dependencies = [ "windows-sys 0.52.0", ] +[[package]] +name = "thiserror" +version = "1.0.57" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1e45bcbe8ed29775f228095caf2cd67af7a4ccf756ebff23a306bf3e8b47b24b" +dependencies = [ + "thiserror-impl", +] + +[[package]] +name = "thiserror-impl" +version = "1.0.57" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a953cb265bef375dae3de6663da4d3804eee9682ea80d8e2542529b73c531c81" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + [[package]] name = "tokio" version = "1.35.1" @@ -1136,6 +1236,12 @@ version = "1.0.12" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3354b9ac3fae1ff6755cb6db53683adb661634f67557942dea4facebec0fee4b" +[[package]] +name = "unicode-width" +version = "0.1.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e51733f11c9c4f72aa0c160008246859e340b00807569a0da0e7a1079b27ba85" + [[package]] name = "want" version = "0.3.1" diff --git a/Cargo.toml b/Cargo.toml index 5ec9096..97d02cb 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -16,6 +16,8 @@ tower = "0.4.13" hyper-util = {version = "0.1.3",features = ["tokio"]} http-body-util = "0.1.0" anyhow = "1.0.79" +tokio-stream = "0.1.14" +protox-parse = "0.6.0" [build-dependencies] tonic-build = "0.10.2" diff --git a/build.rs b/build.rs index 4f4fd27..428dd64 100644 --- a/build.rs +++ b/build.rs @@ -4,11 +4,4 @@ fn main() { let mut news = PathBuf::from(env!("CARGO_MANIFEST_DIR")); news.push("news.proto"); tonic_build::compile_protos(news).expect("Failed to compile protos"); - - let out_dir = PathBuf::from(std::env::var("OUT_DIR").unwrap()); - - tonic_build::configure() - .file_descriptor_set_path(out_dir.join("news_descriptor.bin")) - .compile(&["news.proto"], &["proto"]) - .unwrap(); } diff --git a/renovate.json b/renovate.json index 5db72dd..22a9943 100644 --- a/renovate.json +++ b/renovate.json @@ -1,6 +1,4 @@ { "$schema": "https://docs.renovatebot.com/renovate-schema.json", - "extends": [ - "config:recommended" - ] + "extends": ["config:recommended"] } diff --git a/src/main.rs b/src/main.rs index e2aa03f..5a06d9e 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,16 +1,19 @@ -use tonic::{transport::Server as TonicServer, Response, Status}; +mod server; use crate::news::news_service_server::NewsServiceServer; +use crate::server::Builder; use anyhow::Result; use news::news_service_server::NewsService; use news::{MultipleNewsId, News, NewsId, NewsList}; +use prost_types::FileDescriptorSet; +use std::fs::File; +use std::io::Read; use std::sync::{Arc, Mutex}; +use tonic::{transport::Server as TonicServer, Response, Status}; use tower::make::Shared; pub mod news { tonic::include_proto!("news"); // The package name specified in your .proto - pub(crate) const FILE_DESCRIPTOR_SET: &[u8] = - tonic::include_file_descriptor_set!("news_descriptor"); } #[derive(Debug, Default)] @@ -152,16 +155,24 @@ async fn main() -> Result<()> { let addr = ([127, 0, 0, 1], 50051).into(); let news_service = MyNewsService::new(); - let service = tonic_reflection::server::Builder::configure() - .register_encoded_file_descriptor_set(news::FILE_DESCRIPTOR_SET) - .build() - .unwrap(); + + let mut file = File::open("news.proto")?; + let mut content = String::new(); + file.read_to_string(&mut content)?; + + let news = protox_parse::parse("news.proto", &content)?; + let mut news_descriptor_set = FileDescriptorSet::default(); + news_descriptor_set.file.push(news); + + let service = Builder::configure() + .register_file_descriptor_set(news_descriptor_set) + .build()?; println!("NewsService server listening on {}", addr); let tonic_service = TonicServer::builder() - .add_service(NewsServiceServer::new(news_service)) .add_service(service) + .add_service(NewsServiceServer::new(news_service)) .into_service(); let make_svc = Shared::new(tonic_service); println!("Server listening on http://{}", addr); diff --git a/src/server.rs b/src/server.rs new file mode 100644 index 0000000..be8d7ff --- /dev/null +++ b/src/server.rs @@ -0,0 +1,384 @@ +#![allow(unused)] + +use prost::{DecodeError, Message}; +use prost_types::{ + DescriptorProto, EnumDescriptorProto, FieldDescriptorProto, FileDescriptorProto, + FileDescriptorSet, +}; +use std::collections::HashMap; +use std::fmt::{Display, Formatter}; +use std::sync::Arc; +use tokio::sync::mpsc; +use tokio_stream::{wrappers::ReceiverStream, StreamExt}; +use tonic::{Request, Response, Status, Streaming}; +use tonic_reflection::pb::server_reflection_request::MessageRequest; +use tonic_reflection::pb::server_reflection_response::MessageResponse; +pub use tonic_reflection::pb::server_reflection_server::{ + ServerReflection, ServerReflectionServer, +}; +use tonic_reflection::pb::{ + ExtensionNumberResponse, FileDescriptorResponse, ListServiceResponse, ServerReflectionRequest, + ServerReflectionResponse, ServiceResponse, +}; + +/// Represents an error in the construction of a gRPC Reflection Service. +#[derive(Debug)] +pub enum Error { + /// An error was encountered decoding a `prost_types::FileDescriptorSet` from a buffer. + DecodeError(prost::DecodeError), + /// An invalid `prost_types::FileDescriptorProto` was encountered. + InvalidFileDescriptorSet(String), +} + +impl From for Error { + fn from(e: DecodeError) -> Self { + Error::DecodeError(e) + } +} + +impl std::error::Error for Error {} + +impl Display for Error { + fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { + match self { + Error::DecodeError(_) => f.write_str("error decoding FileDescriptorSet from buffer"), + Error::InvalidFileDescriptorSet(s) => { + write!(f, "invalid FileDescriptorSet - {}", s) + } + } + } +} + +/// A builder used to construct a gRPC Reflection Service. +#[derive(Debug)] +pub struct Builder<'b> { + file_descriptor_sets: Vec, + encoded_file_descriptor_sets: Vec<&'b [u8]>, + include_reflection_service: bool, + + service_names: Vec, + use_all_service_names: bool, + symbols: HashMap>, +} + +impl<'b> Builder<'b> { + /// Create a new builder that can configure a gRPC Reflection Service. + pub fn configure() -> Self { + Builder { + file_descriptor_sets: Vec::new(), + encoded_file_descriptor_sets: Vec::new(), + include_reflection_service: true, + + service_names: Vec::new(), + use_all_service_names: true, + symbols: HashMap::new(), + } + } + + /// Registers an instance of `prost_types::FileDescriptorSet` with the gRPC Reflection + /// Service builder. + pub fn register_file_descriptor_set(mut self, file_descriptor_set: FileDescriptorSet) -> Self { + self.file_descriptor_sets.push(file_descriptor_set); + self + } + + /// Registers a byte slice containing an encoded `prost_types::FileDescriptorSet` with + /// the gRPC Reflection Service builder. + pub fn register_encoded_file_descriptor_set( + mut self, + encoded_file_descriptor_set: &'b [u8], + ) -> Self { + self.encoded_file_descriptor_sets + .push(encoded_file_descriptor_set); + self + } + + /// Serve the gRPC Reflection Service descriptor via the Reflection Service. This is enabled + /// by default - set `include` to false to disable. + pub fn include_reflection_service(mut self, include: bool) -> Self { + self.include_reflection_service = include; + self + } + + /// Advertise a fully-qualified gRPC service name. + /// + /// If not called, then all services present in the registered file descriptor sets + /// will be advertised. + pub fn with_service_name(mut self, name: impl Into) -> Self { + self.use_all_service_names = false; + self.service_names.push(name.into()); + self + } + + /// Build a gRPC Reflection Service to be served via Tonic. + pub fn build(mut self) -> Result, Error> { + if self.include_reflection_service { + self = self + .register_encoded_file_descriptor_set(tonic_reflection::pb::FILE_DESCRIPTOR_SET); + } + + for encoded in &self.encoded_file_descriptor_sets { + let decoded = FileDescriptorSet::decode(*encoded)?; + self.file_descriptor_sets.push(decoded); + } + + let all_fds = self.file_descriptor_sets.clone(); + let mut files: HashMap> = HashMap::new(); + + for fds in all_fds { + for fd in fds.file { + let name = match fd.name.clone() { + None => { + return Err(Error::InvalidFileDescriptorSet("missing name".to_string())); + } + Some(n) => n, + }; + + if files.contains_key(&name) { + continue; + } + + let fd = Arc::new(fd); + files.insert(name, fd.clone()); + + self.process_file(fd)?; + } + } + + let service_names = self + .service_names + .iter() + .map(|name| ServiceResponse { name: name.clone() }) + .collect(); + + Ok(ServerReflectionServer::new(ReflectionService { + state: Arc::new(ReflectionServiceState { + service_names, + files, + symbols: self.symbols, + }), + })) + } + + fn process_file(&mut self, fd: Arc) -> Result<(), Error> { + let prefix = &fd.package.clone().unwrap_or_default(); + + for msg in &fd.message_type { + self.process_message(fd.clone(), prefix, msg)?; + } + + for en in &fd.enum_type { + self.process_enum(fd.clone(), prefix, en)?; + } + + for service in &fd.service { + let service_name = extract_name(prefix, "service", service.name.as_ref())?; + if self.use_all_service_names { + self.service_names.push(service_name.clone()); + } + self.symbols.insert(service_name.clone(), fd.clone()); + + for method in &service.method { + let method_name = extract_name(&service_name, "method", method.name.as_ref())?; + self.symbols.insert(method_name, fd.clone()); + } + } + + Ok(()) + } + + fn process_message( + &mut self, + fd: Arc, + prefix: &str, + msg: &DescriptorProto, + ) -> Result<(), Error> { + let message_name = extract_name(prefix, "message", msg.name.as_ref())?; + self.symbols.insert(message_name.clone(), fd.clone()); + + for nested in &msg.nested_type { + self.process_message(fd.clone(), &message_name, nested)?; + } + + for en in &msg.enum_type { + self.process_enum(fd.clone(), &message_name, en)?; + } + + for field in &msg.field { + self.process_field(fd.clone(), &message_name, field)?; + } + + for oneof in &msg.oneof_decl { + let oneof_name = extract_name(&message_name, "oneof", oneof.name.as_ref())?; + self.symbols.insert(oneof_name, fd.clone()); + } + + Ok(()) + } + + fn process_enum( + &mut self, + fd: Arc, + prefix: &str, + en: &EnumDescriptorProto, + ) -> Result<(), Error> { + let enum_name = extract_name(prefix, "enum", en.name.as_ref())?; + self.symbols.insert(enum_name.clone(), fd.clone()); + + for value in &en.value { + let value_name = extract_name(&enum_name, "enum value", value.name.as_ref())?; + self.symbols.insert(value_name, fd.clone()); + } + + Ok(()) + } + + fn process_field( + &mut self, + fd: Arc, + prefix: &str, + field: &FieldDescriptorProto, + ) -> Result<(), Error> { + let field_name = extract_name(prefix, "field", field.name.as_ref())?; + self.symbols.insert(field_name, fd); + Ok(()) + } +} + +fn extract_name( + prefix: &str, + name_type: &str, + maybe_name: Option<&String>, +) -> Result { + match maybe_name { + None => Err(Error::InvalidFileDescriptorSet(format!( + "missing {} name", + name_type + ))), + Some(name) => { + if prefix.is_empty() { + Ok(name.to_string()) + } else { + Ok(format!("{}.{}", prefix, name)) + } + } + } +} + +#[derive(Debug)] +struct ReflectionServiceState { + service_names: Vec, + files: HashMap>, + symbols: HashMap>, +} + +impl ReflectionServiceState { + fn list_services(&self) -> MessageResponse { + MessageResponse::ListServicesResponse(ListServiceResponse { + service: self.service_names.clone(), + }) + } + + fn symbol_by_name(&self, symbol: &str) -> Result { + match self.symbols.get(symbol) { + None => Err(Status::not_found(format!("symbol '{}' not found", symbol))), + Some(fd) => { + let mut encoded_fd = Vec::new(); + if fd.clone().encode(&mut encoded_fd).is_err() { + return Err(Status::internal("encoding error")); + }; + + Ok(MessageResponse::FileDescriptorResponse( + FileDescriptorResponse { + file_descriptor_proto: vec![encoded_fd], + }, + )) + } + } + } + + fn file_by_filename(&self, filename: &str) -> Result { + match self.files.get(filename) { + None => Err(Status::not_found(format!("file '{}' not found", filename))), + Some(fd) => { + let mut encoded_fd = Vec::new(); + if fd.clone().encode(&mut encoded_fd).is_err() { + return Err(Status::internal("encoding error")); + } + + Ok(MessageResponse::FileDescriptorResponse( + FileDescriptorResponse { + file_descriptor_proto: vec![encoded_fd], + }, + )) + } + } + } +} + +#[derive(Debug)] +struct ReflectionService { + state: Arc, +} + +#[tonic::async_trait] +impl ServerReflection for ReflectionService { + type ServerReflectionInfoStream = ReceiverStream>; + + async fn server_reflection_info( + &self, + req: Request>, + ) -> Result, Status> { + let mut req_rx = req.into_inner(); + let (resp_tx, resp_rx) = mpsc::channel::>(1); + + let state = self.state.clone(); + + tokio::spawn(async move { + while let Some(req) = req_rx.next().await { + let req = match req { + Ok(req) => req, + Err(_) => { + return; + } + }; + + let resp_msg = match req.message_request.clone() { + None => Err(Status::invalid_argument("invalid MessageRequest")), + Some(msg) => match msg { + MessageRequest::FileByFilename(s) => state.file_by_filename(&s), + MessageRequest::FileContainingSymbol(s) => state.symbol_by_name(&s), + MessageRequest::FileContainingExtension(_) => { + Err(Status::not_found("extensions are not supported")) + } + MessageRequest::AllExtensionNumbersOfType(_) => { + // NOTE: Workaround. Some grpc clients (e.g. grpcurl) expect this method not to fail. + // https://github.com/hyperium/tonic/issues/1077 + Ok(MessageResponse::AllExtensionNumbersResponse( + ExtensionNumberResponse::default(), + )) + } + MessageRequest::ListServices(_) => Ok(state.list_services()), + }, + }; + + match resp_msg { + Ok(resp_msg) => { + let resp = ServerReflectionResponse { + valid_host: req.host.clone(), + original_request: Some(req.clone()), + message_response: Some(resp_msg), + }; + resp_tx.send(Ok(resp)).await.expect("send"); + } + Err(status) => { + resp_tx.send(Err(status)).await.expect("send"); + return; + } + } + } + }); + + Ok(Response::new(ReceiverStream::new(resp_rx))) + } +}