-
Notifications
You must be signed in to change notification settings - Fork 3
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
chore(event_ingestion): Try to adapt to new design (#159)
* chore(event_ingestion): Try to adapt to new design * add `sync` flag for `tokio` dependency (#160) * add sync feature to tokio crate * add changeset * chore: bump versions before release (#161) Co-authored-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com> * chore: fix typo in publish.yml update_ruby_lockfile step (#162) * chore(ruby): update lockfile (#163) * chore: hide event ingestion under a feature flag (#164) * chore: bump versions before release (#165) Co-authored-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com> * chore(ruby): update lockfile (#166) * enable feature, use tokio timer advance * more code review comments * fix failing tests * DRY up --------- Co-authored-by: Leo Romanovsky <[email protected]> Co-authored-by: github-actions[bot] <41898282+github-actions[bot]@users.noreply.github.com> Co-authored-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com> Co-authored-by: Oleksii Shmalko <[email protected]>
- Loading branch information
5 people
authored
Jan 17, 2025
1 parent
f3e1006
commit 903eae4
Showing
29 changed files
with
752 additions
and
370 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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,142 @@ | ||
| use std::time::Duration; | ||
|
|
||
| use tokio::{sync::mpsc, time::Instant}; | ||
|
|
||
| use super::BatchedMessage; | ||
|
|
||
| /// Auto-flusher forwards all messages from `uplink` to `downlink` unchanged and inserts extra flush | ||
| /// requests if it hasn't seen one within the given `period`. In other words, it makes sure that the | ||
| /// channel is flushed at least every `period`. | ||
| pub(super) async fn auto_flusher<T>( | ||
| mut uplink: mpsc::Receiver<BatchedMessage<T>>, | ||
| downlink: mpsc::Sender<BatchedMessage<T>>, | ||
| period: Duration, | ||
| ) -> Option<()> { | ||
| 'flushed: loop { | ||
| // Process first message. | ||
| let msg = uplink.recv().await?; | ||
| let flushed = msg.flush.is_some(); | ||
| downlink.send(msg).await.ok()?; | ||
|
|
||
| // No need to time if we just flushed. | ||
| if flushed { | ||
| continue; | ||
| } | ||
|
|
||
| let flush_at = Instant::now() + period; | ||
| // loop till we reach flush_at or see a flushed message. | ||
| loop { | ||
| tokio::select! { | ||
| _ = tokio::time::sleep_until(flush_at) => { | ||
| downlink.send(BatchedMessage { batch: Vec::new(), flush: Some(()) }).await.ok()?; | ||
| continue 'flushed; | ||
| }, | ||
| msg = uplink.recv() => { | ||
| let msg = msg?; | ||
| let flushed = msg.flush.is_some(); | ||
| downlink.send(msg).await.ok()?; | ||
| if flushed { | ||
| continue 'flushed; | ||
| } | ||
| } | ||
| } | ||
| } | ||
| } | ||
| } | ||
|
|
||
| #[cfg(test)] | ||
| mod tests { | ||
| use crate::event_ingestion::auto_flusher; | ||
| use crate::event_ingestion::batched_message::BatchedMessage; | ||
| use tokio::sync::mpsc; | ||
| use tokio::time::{Duration}; | ||
|
|
||
| #[tokio::test(start_paused = true)] | ||
| async fn test_auto_flusher() { | ||
| let (uplink_tx, uplink_rx) = mpsc::channel(10); | ||
| let (downlink_tx, mut downlink_rx) = mpsc::channel(10); | ||
| let flush_period = Duration::from_millis(100); | ||
| tokio::spawn(auto_flusher::auto_flusher( | ||
| uplink_rx, | ||
| downlink_tx, | ||
| flush_period, | ||
| )); | ||
|
|
||
| uplink_tx | ||
| .send(BatchedMessage { | ||
| batch: vec![1, 2, 3], | ||
| flush: None, | ||
| }) | ||
| .await | ||
| .unwrap(); | ||
| uplink_tx | ||
| .send(BatchedMessage { | ||
| batch: vec![4, 5, 6], | ||
| flush: None, | ||
| }) | ||
| .await | ||
| .unwrap(); | ||
|
|
||
| // Verify that the messages are forwarded to downlink | ||
| assert_eq!( | ||
| downlink_rx.recv().await, | ||
| Some(BatchedMessage { | ||
| batch: vec![1, 2, 3], | ||
| flush: None | ||
| }) | ||
| ); | ||
| assert_eq!( | ||
| downlink_rx.recv().await, | ||
| Some(BatchedMessage { | ||
| batch: vec![4, 5, 6], | ||
| flush: None | ||
| }) | ||
| ); | ||
|
|
||
| // Wait for the flush period to trigger an auto-flush | ||
| tokio::time::advance(flush_period * 2).await; | ||
|
|
||
| // Verify the auto-flush behavior | ||
| assert_eq!( | ||
| downlink_rx.recv().await, | ||
| Some(BatchedMessage { | ||
| batch: Vec::new(), | ||
| flush: Some(()) | ||
| }) | ||
| ); | ||
|
|
||
| // Send a flushed message explicitly | ||
| uplink_tx | ||
| .send(BatchedMessage { | ||
| batch: vec![], | ||
| flush: Some(()), | ||
| }) | ||
| .await | ||
| .unwrap(); | ||
|
|
||
| // Verify that the flushed message is forwarded immediately | ||
| assert_eq!( | ||
| downlink_rx.recv().await, | ||
| Some(BatchedMessage { | ||
| batch: vec![], | ||
| flush: Some(()) | ||
| }) | ||
| ); | ||
|
|
||
| // Ensure the loop continues and processes further messages | ||
| uplink_tx | ||
| .send(BatchedMessage { | ||
| batch: vec![7, 8, 9], | ||
| flush: None, | ||
| }) | ||
| .await | ||
| .unwrap(); | ||
| assert_eq!( | ||
| downlink_rx.recv().await, | ||
| Some(BatchedMessage { | ||
| batch: vec![7, 8, 9], | ||
| flush: None | ||
| }) | ||
| ); | ||
| } | ||
| } |
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,39 @@ | ||
| /// Batched message contain a batch of data and may optionally require processors to flush any processing. | ||
| #[derive(Debug, PartialEq)] | ||
| pub(super) struct BatchedMessage<T> { | ||
| pub batch: Vec<T>, | ||
| /// `None` means the message does not require a flush. | ||
| /// `Some` contains a list of watchers. | ||
| pub flush: Option<()>, | ||
| } | ||
|
|
||
| impl<T> BatchedMessage<T> { | ||
| /// Create a new empty message. | ||
| pub fn empty() -> BatchedMessage<T> { | ||
| BatchedMessage { | ||
| batch: Vec::new(), | ||
| flush: None, | ||
| } | ||
| } | ||
|
|
||
| /// Create a new message with a batch of data and optionally a list of watchers. | ||
| pub fn new(batch: Vec<T>, flush: Option<()>) -> BatchedMessage<T> { | ||
| BatchedMessage { | ||
| batch, | ||
| flush, | ||
| } | ||
| } | ||
|
|
||
| pub fn requires_flush(&self) -> bool { | ||
| self.flush.is_some() | ||
| } | ||
|
|
||
| // Mark the message as successfully flushed, consuming it and notifying any interested parties. | ||
| // pub fn flushed(self) { | ||
| // if let Some(flush) = self.flush { | ||
| // for f in flush { | ||
| // f.send(()); | ||
| // } | ||
| // } | ||
| // } | ||
| } |
Oops, something went wrong.