feat: Expose endpoint to control streams #430
Conversation
morgsmccauley
changed the base branch from
main
to
418-created-dedicated-block-stream-per-indexer
morgsmccauley
force-pushed
the
420-expose-endpoint-to-control-streams
branch
3 times, most recently
from
e1ac0e6
to
5511837
Compare
| @@ -0,0 +1,5 @@ | |||
| fn main() -> Result<(), Box<dyn std::error::Error>> { | |||
There was a problem hiding this comment.
Generates the Rust types from the proto file on cargo build
| @@ -0,0 +1,17 @@ | |||
| use tonic::Request; | |||
There was a problem hiding this comment.
These examples use the exposed Rust types to interact with the gRPC service - these not only serve as examples, but also make it easy to quickly debug by running them
| // pub code: String, | ||
| // pub start_block_height: Option<u64>, | ||
| // pub schema: Option<String>, | ||
| // pub provisioned: bool, |
There was a problem hiding this comment.
No longer used/needed - this is more for the Runner side of things which will be passed from the Control Plane directly
| pub indexer_rule: IndexerRule, | ||
| } | ||
|
|
||
| impl IndexerConfig { | ||
| pub fn get_full_name(&self) -> String { | ||
| format!("{}/{}", self.account_id, self.function_name) | ||
| } | ||
|
|
||
| pub fn get_hash_id(&self) -> String { |
There was a problem hiding this comment.
Consistent/deterministic ID to use for the BlockStream so it can be persisted across restarts
| tonic::include_proto!("blockstreamer"); | ||
| } | ||
|
|
||
| pub use blockstreamer::*; |
There was a problem hiding this comment.
Exposes the Rust Client types
|
|
||
| Ok(Response::new(response)) | ||
| } | ||
| } |
There was a problem hiding this comment.
I should probably add tests here
morgsmccauley
changed the title
|
|
||
| #[tokio::main] | ||
| async fn main() -> Result<(), Box<dyn std::error::Error>> { | ||
| let mut client = BlockStreamerClient::connect("http://[::1]:10000").await?; |
There was a problem hiding this comment.
Do we want to create a client for each request? Can we just use one client instead?
There was a problem hiding this comment.
This is a cargo example, a standalone binary which can be executed via cargo --example list_streams. I created it so we can easily send requests to the server, rather than creating verbose cli commands.
block-streamer/src/block_stream.rs
Outdated
| std::cmp::max(last_block_in_index, last_indexed_block) | ||
| }); | ||
|
|
||
| tracing::debug!( |
There was a problem hiding this comment.
I see. So, here is where we'd end up with duplicated reads since each stream has its own lake stream? I can see why we would need to rethink the caching as you had briefly mentioned in the weekly meeting.
There was a problem hiding this comment.
Yep exactly - we'd need to create some shared client so it can cache internally, but that's a future problem :)
block-streamer/src/redis.rs
Outdated
| @@ -0,0 +1,142 @@ | |||
| pub use redis::{self, aio::ConnectionManager, FromRedisValue, ToRedisArgs}; | |||
There was a problem hiding this comment.
I feel like I saw these changes already. I don't know if a rebase is needed or something?
morgsmccauley
force-pushed
the
420-expose-endpoint-to-control-streams
branch
from
9322584
to
2d570f0
Compare
This PR exposes a gRPC endpoint for the Block Streamer service to provide control over individual Block Streams. There are currently 3 methods: start/stop/list. The functionality across these methods is quite basic, I didn't want to spend too much time fleshing them out as I'm still not certain on how exactly they will be used. I expect them to Change once the Control Plane starts using them.