-
Notifications
You must be signed in to change notification settings - Fork 117
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
add(rpc): Add a tonic server in zebra-rpc (#8674)
* adds a tonic server * Adds a test stub, moves method impls to their own modules, minor fixes. * Moves indexer rpc mod behind a feature, adds a config field for its listen address, and initializes the indexer RPC when zebrad starts * Skips tonic_build() in zebra-rpc build script unless indexer-rpcs feature is selected, simplifies indexer.proto file, makes tonic deps optional * formats zebra-rpc Cargo.toml * Adds tokio_stream dependency, adds chain_tip_change field to IndexerRPC, and implements a simple version of the chain_tip_change RPC method * passes latest chain tip to indexer::server::init from start cmd and updates vectors test * Update zebra-rpc/src/config.rs * fixes a race condition in trusted_chain_sync_handles_forks_correctly
- Loading branch information
Showing
15 changed files
with
371 additions
and
20 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,15 @@ | ||
| //! Compile proto files | ||
| fn main() -> Result<(), Box<dyn std::error::Error>> { | ||
| #[cfg(feature = "indexer-rpcs")] | ||
| { | ||
| use std::{env, path::PathBuf}; | ||
| let out_dir = env::var("OUT_DIR").map(PathBuf::from); | ||
| tonic_build::configure() | ||
| .type_attribute(".", "#[derive(serde::Deserialize, serde::Serialize)]") | ||
| .file_descriptor_set_path(out_dir.unwrap().join("indexer_descriptor.bin")) | ||
| .compile(&["proto/indexer.proto"], &[""])?; | ||
| } | ||
|
|
||
| Ok(()) | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,10 @@ | ||
| syntax = "proto3"; | ||
| package zebra.indexer.rpc; | ||
|
|
||
| // Used by methods that take no arguments. | ||
| message Empty {}; | ||
|
|
||
| service Indexer { | ||
| // Notifies listeners of chain tip changes | ||
| rpc ChainTipChange(Empty) returns (stream Empty); | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,13 @@ | ||
| //! A tonic RPC server for Zebra's indexer API. | ||
| #[cfg(test)] | ||
| mod tests; | ||
|
|
||
| pub mod methods; | ||
| pub mod server; | ||
|
|
||
| // The generated indexer proto | ||
| tonic::include_proto!("zebra.indexer.rpc"); | ||
|
|
||
| pub(crate) const FILE_DESCRIPTOR_SET: &[u8] = | ||
| tonic::include_file_descriptor_set!("indexer_descriptor"); |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,57 @@ | ||
| //! Implements `Indexer` methods on the `IndexerRPC` type | ||
| use std::pin::Pin; | ||
|
|
||
| use futures::Stream; | ||
| use tokio_stream::wrappers::ReceiverStream; | ||
| use tonic::{Response, Status}; | ||
| use tower::BoxError; | ||
|
|
||
| use zebra_chain::chain_tip::ChainTip; | ||
|
|
||
| use super::{indexer_server::Indexer, server::IndexerRPC, Empty}; | ||
|
|
||
| /// The maximum number of messages that can be queued to be streamed to a client | ||
| const RESPONSE_BUFFER_SIZE: usize = 10_000; | ||
|
|
||
| #[tonic::async_trait] | ||
| impl<ReadStateService, Tip> Indexer for IndexerRPC<ReadStateService, Tip> | ||
| where | ||
| ReadStateService: tower::Service< | ||
| zebra_state::ReadRequest, | ||
| Response = zebra_state::ReadResponse, | ||
| Error = BoxError, | ||
| > + Clone | ||
| + Send | ||
| + Sync | ||
| + 'static, | ||
| <ReadStateService as tower::Service<zebra_state::ReadRequest>>::Future: Send, | ||
| Tip: ChainTip + Clone + Send + Sync + 'static, | ||
| { | ||
| type ChainTipChangeStream = Pin<Box<dyn Stream<Item = Result<Empty, Status>> + Send>>; | ||
|
|
||
| async fn chain_tip_change( | ||
| &self, | ||
| _: tonic::Request<Empty>, | ||
| ) -> Result<Response<Self::ChainTipChangeStream>, Status> { | ||
| let (response_sender, response_receiver) = tokio::sync::mpsc::channel(RESPONSE_BUFFER_SIZE); | ||
| let response_stream = ReceiverStream::new(response_receiver); | ||
| let mut chain_tip_change = self.chain_tip_change.clone(); | ||
|
|
||
| tokio::spawn(async move { | ||
| // Notify the client of chain tip changes until the channel is closed | ||
| while let Ok(()) = chain_tip_change.best_tip_changed().await { | ||
| let tx = response_sender.clone(); | ||
| tokio::spawn(async move { tx.send(Ok(Empty {})).await }); | ||
| } | ||
|
|
||
| let _ = response_sender | ||
| .send(Err(Status::unavailable( | ||
| "chain_tip_change channel has closed", | ||
| ))) | ||
| .await; | ||
| }); | ||
|
|
||
| Ok(Response::new(Box::pin(response_stream))) | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,77 @@ | ||
| //! A tonic RPC server for Zebra's indexer API. | ||
| use std::net::SocketAddr; | ||
|
|
||
| use tokio::task::JoinHandle; | ||
| use tonic::transport::{server::TcpIncoming, Server}; | ||
| use tower::BoxError; | ||
| use zebra_chain::chain_tip::ChainTip; | ||
|
|
||
| use super::indexer_server::IndexerServer; | ||
|
|
||
| type ServerTask = JoinHandle<Result<(), BoxError>>; | ||
|
|
||
| /// Indexer RPC service. | ||
| pub struct IndexerRPC<ReadStateService, Tip> | ||
| where | ||
| ReadStateService: tower::Service< | ||
| zebra_state::ReadRequest, | ||
| Response = zebra_state::ReadResponse, | ||
| Error = BoxError, | ||
| > + Clone | ||
| + Send | ||
| + Sync | ||
| + 'static, | ||
| <ReadStateService as tower::Service<zebra_state::ReadRequest>>::Future: Send, | ||
| Tip: ChainTip + Clone + Send + Sync + 'static, | ||
| { | ||
| _read_state: ReadStateService, | ||
| pub(super) chain_tip_change: Tip, | ||
| } | ||
|
|
||
| /// Initializes the indexer RPC server | ||
| pub async fn init<ReadStateService, Tip>( | ||
| listen_addr: SocketAddr, | ||
| _read_state: ReadStateService, | ||
| chain_tip_change: Tip, | ||
| ) -> Result<(ServerTask, SocketAddr), BoxError> | ||
| where | ||
| ReadStateService: tower::Service< | ||
| zebra_state::ReadRequest, | ||
| Response = zebra_state::ReadResponse, | ||
| Error = BoxError, | ||
| > + Clone | ||
| + Send | ||
| + Sync | ||
| + 'static, | ||
| <ReadStateService as tower::Service<zebra_state::ReadRequest>>::Future: Send, | ||
| Tip: ChainTip + Clone + Send + Sync + 'static, | ||
| { | ||
| let indexer_service = IndexerRPC { | ||
| _read_state, | ||
| chain_tip_change, | ||
| }; | ||
|
|
||
| let reflection_service = tonic_reflection::server::Builder::configure() | ||
| .register_encoded_file_descriptor_set(crate::indexer::FILE_DESCRIPTOR_SET) | ||
| .build() | ||
| .unwrap(); | ||
|
|
||
| tracing::info!("Trying to open indexer RPC endpoint at {}...", listen_addr,); | ||
|
|
||
| let tcp_listener = tokio::net::TcpListener::bind(listen_addr).await?; | ||
| let listen_addr = tcp_listener.local_addr()?; | ||
| let incoming = TcpIncoming::from_listener(tcp_listener, true, None)?; | ||
|
|
||
| let server_task: JoinHandle<Result<(), BoxError>> = tokio::spawn(async move { | ||
| Server::builder() | ||
| .add_service(reflection_service) | ||
| .add_service(IndexerServer::new(indexer_service)) | ||
| .serve_with_incoming(incoming) | ||
| .await?; | ||
|
|
||
| Ok(()) | ||
| }); | ||
|
|
||
| Ok((server_task, listen_addr)) | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1 @@ | ||
| mod vectors; |
Oops, something went wrong.