-
Notifications
You must be signed in to change notification settings - Fork 62
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
refactor various no-ops into NoOpWrapper and remove runtime timeouts
NoOpWrapper encapsulates a common pattern of dynamically disabling validations for various connector types. Remove timeouts from unary RPCs directly invoked against the runtime. These timeouts are moving into the `agent`, which is the only place we actually care to time out these requests. When running locally with `flowctl`, for example, we never want to time out.
- Loading branch information
1 parent
5dbe743
commit d7d789f
Showing
9 changed files
with
87 additions
and
128 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
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
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 |
|---|---|---|
| @@ -1,59 +1,42 @@ | ||
| use super::{LogHandler, Runtime}; | ||
| use futures::{stream::BoxStream, StreamExt, TryStreamExt}; | ||
| use proto_flow::{capture, derive, materialize}; | ||
| use std::time::Duration; | ||
|
|
||
| impl<L: LogHandler> Runtime<L> { | ||
| pub async fn unary_capture( | ||
| self, | ||
| request: capture::Request, | ||
| timeout: Duration, | ||
| ) -> anyhow::Result<capture::Response> { | ||
| let response = self.serve_capture(unary_in(request)); | ||
| unary_out(response, timeout).await | ||
| unary_out(response).await | ||
| } | ||
|
|
||
| pub async fn unary_derive( | ||
| self, | ||
| request: derive::Request, | ||
| timeout: Duration, | ||
| ) -> anyhow::Result<derive::Response> { | ||
| pub async fn unary_derive(self, request: derive::Request) -> anyhow::Result<derive::Response> { | ||
| let response = self.serve_derive(unary_in(request)); | ||
| unary_out(response, timeout).await | ||
| unary_out(response).await | ||
| } | ||
|
|
||
| pub async fn unary_materialize( | ||
| self, | ||
| request: materialize::Request, | ||
| timeout: Duration, | ||
| ) -> anyhow::Result<materialize::Response> { | ||
| let response = self.serve_materialize(unary_in(request)).boxed(); | ||
| unary_out(response, timeout).await | ||
| unary_out(response).await | ||
| } | ||
| } | ||
|
|
||
| fn unary_in<R: Send + 'static>(request: R) -> BoxStream<'static, anyhow::Result<R>> { | ||
| futures::stream::once(async move { Ok(request) }).boxed() | ||
| } | ||
|
|
||
| async fn unary_out<S, R>(response_rx: S, timeout: Duration) -> anyhow::Result<R> | ||
| async fn unary_out<S, R>(response_rx: S) -> anyhow::Result<R> | ||
| where | ||
| S: futures::Stream<Item = anyhow::Result<R>>, | ||
| { | ||
| let response = async move { | ||
| let mut responses: Vec<R> = response_rx.try_collect().await?; | ||
|
|
||
| if responses.len() != 1 { | ||
| anyhow::bail!("unary request didn't return a response"); | ||
| } | ||
| Ok(responses.pop().unwrap()) | ||
| }; | ||
| let mut responses: Vec<R> = response_rx.try_collect().await?; | ||
|
|
||
| tokio::select! { | ||
| response = response => response, | ||
| _ = tokio::time::sleep(timeout) => { | ||
| Err(tonic::Status::deadline_exceeded(r#"Timeout while waiting for the connector's response. | ||
| Please verify any network configuration and retry."#))? | ||
| } | ||
| if responses.len() != 1 { | ||
| anyhow::bail!("unary request didn't return a response"); | ||
| } | ||
| Ok(responses.pop().unwrap()) | ||
| } |
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