diff options
Diffstat (limited to 'crates/common')
-rw-r--r-- | crates/common/batcher/Cargo.toml | 2 | ||||
-rw-r--r-- | crates/common/batcher/src/batch.rs | 37 | ||||
-rw-r--r-- | crates/common/batcher/src/batchable.rs | 4 | ||||
-rw-r--r-- | crates/common/batcher/src/batcher.rs | 49 | ||||
-rw-r--r-- | crates/common/batcher/src/config.rs | 2 | ||||
-rw-r--r-- | crates/common/batcher/src/driver.rs | 36 | ||||
-rw-r--r-- | crates/common/clock/Cargo.toml | 5 | ||||
-rw-r--r-- | crates/common/clock/src/lib.rs | 8 |
8 files changed, 70 insertions, 73 deletions
diff --git a/crates/common/batcher/Cargo.toml b/crates/common/batcher/Cargo.toml index d5f3587d..1ab8976c 100644 --- a/crates/common/batcher/Cargo.toml +++ b/crates/common/batcher/Cargo.toml @@ -8,7 +8,7 @@ rust-version = "1.58.1" # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html [dependencies] -chrono = "0.4" +time = "0.3" tokio = { version = "1.12", features = ["sync", "time"] } [dev-dependencies] diff --git a/crates/common/batcher/src/batch.rs b/crates/common/batcher/src/batch.rs index 8c06d5c6..8de20ee2 100644 --- a/crates/common/batcher/src/batch.rs +++ b/crates/common/batcher/src/batch.rs @@ -1,7 +1,7 @@ use crate::batchable::Batchable; -use chrono::{DateTime, Utc}; use std::collections::HashMap; use std::iter::once; +use time::OffsetDateTime; #[must_use] #[derive(Debug)] @@ -13,13 +13,13 @@ pub enum BatchAdd<B: Batchable> { #[derive(Debug)] pub struct Batch<B: Batchable> { - batch_start: DateTime<Utc>, - batch_end: DateTime<Utc>, + batch_start: OffsetDateTime, + batch_end: OffsetDateTime, events: HashMap<B::Key, B>, } impl<B: Batchable> Batch<B> { - pub fn new(batch_start: DateTime<Utc>, batch_end: DateTime<Utc>, event: B) -> Batch<B> { + pub fn new(batch_start: OffsetDateTime, batch_end: OffsetDateTime, event: B) -> Batch<B> { let mut events = HashMap::new(); events.insert(event.key(), event); @@ -30,11 +30,11 @@ impl<B: Batchable> Batch<B> { } } - pub fn batch_start(&self) -> DateTime<Utc> { + pub fn batch_start(&self) -> OffsetDateTime { self.batch_start } - pub fn batch_end(&self) -> DateTime<Utc> { + pub fn batch_end(&self) -> OffsetDateTime { self.batch_end } @@ -55,7 +55,7 @@ impl<B: Batchable> Batch<B> { BatchAdd::Added } - fn split(&mut self, existing_event_time: DateTime<Utc>, event: B) -> Batch<B> { + fn split(&mut self, existing_event_time: OffsetDateTime, event: B) -> Batch<B> { let split_point = midpoint(existing_event_time, event.event_time()); let mut new_batch_events = HashMap::new(); @@ -92,20 +92,19 @@ impl<B: Batchable> Batch<B> { } } -fn midpoint(event_time1: DateTime<Utc>, event_time2: DateTime<Utc>) -> DateTime<Utc> { - let gap = event_time1.signed_duration_since(event_time2); +fn midpoint(event_time1: OffsetDateTime, event_time2: OffsetDateTime) -> OffsetDateTime { + let gap = event_time1 - event_time2; event_time2 + gap / 2 } #[cfg(test)] mod tests { use super::*; - use chrono::TimeZone; #[test] fn add() { - let batch_start = Utc.timestamp_millis(0); - let batch_end = Utc.timestamp_millis(100); + let batch_start = OffsetDateTime::from_unix_timestamp(0).unwrap(); + let batch_end = OffsetDateTime::from_unix_timestamp(100).unwrap(); let event1 = TestBatchEvent::new(1, 40); let event2 = TestBatchEvent::new(2, 60); @@ -120,8 +119,8 @@ mod tests { #[test] fn split() { - let batch_start = Utc.timestamp_millis(0); - let batch_end = Utc.timestamp_millis(100); + let batch_start = OffsetDateTime::from_unix_timestamp(0).unwrap(); + let batch_end = OffsetDateTime::from_unix_timestamp(100).unwrap(); let event1 = TestBatchEvent::new(1, 40); let event2 = TestBatchEvent::new(1, 60); @@ -142,8 +141,8 @@ mod tests { #[test] fn duplicate() { - let batch_start = Utc.timestamp_millis(0); - let batch_end = Utc.timestamp_millis(100); + let batch_start = OffsetDateTime::from_unix_timestamp(0).unwrap(); + let batch_end = OffsetDateTime::from_unix_timestamp(100).unwrap(); let event1 = TestBatchEvent::new(1, 40); let event2 = TestBatchEvent::new(1, 40); @@ -158,12 +157,12 @@ mod tests { #[derive(Debug, Clone, Eq, PartialEq)] struct TestBatchEvent { key: u64, - event_time: DateTime<Utc>, + event_time: OffsetDateTime, } impl TestBatchEvent { fn new(key: u64, event_time: i64) -> TestBatchEvent { - let event_time = Utc.timestamp_millis(event_time); + let event_time = OffsetDateTime::from_unix_timestamp(event_time).unwrap(); TestBatchEvent { key, event_time } } } @@ -175,7 +174,7 @@ mod tests { self.key } - fn event_time(&self) -> DateTime<Utc> { + fn event_time(&self) -> OffsetDateTime { self.event_time } } diff --git a/crates/common/batcher/src/batchable.rs b/crates/common/batcher/src/batchable.rs index 3585fc2f..caa926ee 100644 --- a/crates/common/batcher/src/batchable.rs +++ b/crates/common/batcher/src/batchable.rs @@ -1,6 +1,6 @@ -use chrono::{DateTime, Utc}; use std::fmt::Debug; use std::hash::Hash; +use time::OffsetDateTime; /// Implement this interface for the items that you want batched. /// No items with the same key will go in the same batch. @@ -13,5 +13,5 @@ pub trait Batchable { fn key(&self) -> Self::Key; /// The time at which this item was created. This time is used to group items into a batch. - fn event_time(&self) -> DateTime<Utc>; + fn event_time(&self) -> OffsetDateTime; } diff --git a/crates/common/batcher/src/batcher.rs b/crates/common/batcher/src/batcher.rs index 237e1b57..92b70996 100644 --- a/crates/common/batcher/src/batcher.rs +++ b/crates/common/batcher/src/batcher.rs @@ -1,12 +1,12 @@ use crate::batch::{Batch, BatchAdd}; use crate::batchable::Batchable; use crate::config::BatchConfig; -use chrono::{DateTime, Utc}; +use time::OffsetDateTime; #[derive(Debug, Eq, PartialEq)] pub(crate) enum BatcherOutput<B> { Batch(Vec<B>), - Timer(DateTime<Utc>), + Timer(OffsetDateTime), } /// Provides the core implementation of the batching algorithm. @@ -27,7 +27,7 @@ impl<B: Batchable> Batcher<B> { pub(crate) fn event( &mut self, - processing_time: DateTime<Utc>, + processing_time: OffsetDateTime, event: B, ) -> Vec<BatcherOutput<B>> { let event_time = event.event_time(); @@ -63,8 +63,8 @@ impl<B: Batchable> Batcher<B> { fn output_for_batch_end( &mut self, - processing_time: DateTime<Utc>, - batch_end: DateTime<Utc>, + processing_time: OffsetDateTime, + batch_end: OffsetDateTime, ) -> Vec<BatcherOutput<B>> { let batch_timeout = batch_end + self.config.delivery_jitter(); if processing_time < batch_timeout { @@ -77,7 +77,7 @@ impl<B: Batchable> Batcher<B> { } } - pub(crate) fn time(&mut self, time: DateTime<Utc>) -> Vec<Vec<B>> { + pub(crate) fn time(&mut self, time: OffsetDateTime) -> Vec<Vec<B>> { let batches = std::mem::take(&mut self.batches); let (open_batches, closed_batches) = batches @@ -92,7 +92,7 @@ impl<B: Batchable> Batcher<B> { .collect() } - fn is_open(&self, batch: &Batch<B>, time: DateTime<Utc>) -> bool { + fn is_open(&self, batch: &Batch<B>, time: OffsetDateTime) -> bool { batch.batch_end() + self.config.delivery_jitter() > time } @@ -106,7 +106,7 @@ impl<B: Batchable> Batcher<B> { batches } - fn find_target_batch(&mut self, event_time: DateTime<Utc>) -> Option<&mut Batch<B>> { + fn find_target_batch(&mut self, event_time: OffsetDateTime) -> Option<&mut Batch<B>> { for batch in &mut self.batches { if batch.batch_start() <= event_time && event_time <= batch.batch_end() { return Some(batch); @@ -131,14 +131,14 @@ impl<B: Batchable> Batcher<B> { Batch::new(batch_start, batch_end, event) } - fn previous_batch(&self, event_time: DateTime<Utc>) -> Option<&Batch<B>> { + fn previous_batch(&self, event_time: OffsetDateTime) -> Option<&Batch<B>> { self.batches .iter() .filter(|batch| batch.batch_end() < event_time) .max_by(|batch1, batch2| batch1.batch_end().cmp(&batch2.batch_end())) } - fn next_batch(&self, event_time: DateTime<Utc>) -> Option<&Batch<B>> { + fn next_batch(&self, event_time: OffsetDateTime) -> Option<&Batch<B>> { self.batches .iter() .filter(|batch| batch.batch_start() > event_time) @@ -151,9 +151,8 @@ mod tests { use super::*; use crate::batchable::Batchable; use crate::config::BatchConfigBuilder; - use chrono::offset::TimeZone; - use chrono::{DateTime, Duration, Utc}; use std::collections::BTreeMap; + use time::Duration; #[test] fn single_event_batch() { @@ -424,7 +423,7 @@ mod tests { #[derive(Debug, Clone, Eq, PartialEq)] struct TestBatchEvent { - event_time: DateTime<Utc>, + event_time: OffsetDateTime, key: String, value: u64, } @@ -436,7 +435,7 @@ mod tests { self.key.clone() } - fn event_time(&self) -> DateTime<Utc> { + fn event_time(&self) -> OffsetDateTime { self.event_time } } @@ -448,11 +447,11 @@ mod tests { } struct BatcherTest { - start_time: DateTime<Utc>, + start_time: OffsetDateTime, batcher: Batcher<TestBatchEvent>, - inputs: BTreeMap<DateTime<Utc>, EventOrTimer>, - flush_time: Option<DateTime<Utc>>, - expected_batches: BTreeMap<DateTime<Utc>, Vec<Vec<TestBatchEvent>>>, + inputs: BTreeMap<OffsetDateTime, EventOrTimer>, + flush_time: Option<OffsetDateTime>, + expected_batches: BTreeMap<OffsetDateTime, Vec<Vec<TestBatchEvent>>>, } impl BatcherTest { @@ -463,7 +462,7 @@ mod tests { .message_leap_limit(message_leap_limit) .build(); - let start_time = Utc.timestamp_millis(0); + let start_time = OffsetDateTime::from_unix_timestamp(0).unwrap(); let batcher = Batcher::new(batcher_config); BatcherTest { @@ -538,10 +537,10 @@ mod tests { fn handle_outputs( &mut self, - t: DateTime<Utc>, + t: OffsetDateTime, outputs: Vec<BatcherOutput<TestBatchEvent>>, - all_batches: &mut BTreeMap<DateTime<Utc>, Vec<Vec<TestBatchEvent>>>, - flush_time: Option<DateTime<Utc>>, + all_batches: &mut BTreeMap<OffsetDateTime, Vec<Vec<TestBatchEvent>>>, + flush_time: Option<OffsetDateTime>, ) { let mut batches = vec![]; @@ -577,14 +576,14 @@ mod tests { } } - fn create_instant(&self, time: i64) -> DateTime<Utc> { + fn create_instant(&self, time: i64) -> OffsetDateTime { self.start_time + Duration::milliseconds(time) } } fn verify( - expected_batches: BTreeMap<DateTime<Utc>, Vec<Vec<TestBatchEvent>>>, - mut actual_batches: BTreeMap<DateTime<Utc>, Vec<Vec<TestBatchEvent>>>, + expected_batches: BTreeMap<OffsetDateTime, Vec<Vec<TestBatchEvent>>>, + mut actual_batches: BTreeMap<OffsetDateTime, Vec<Vec<TestBatchEvent>>>, ) { assert_eq!( actual_batches.keys().collect::<Vec<_>>(), diff --git a/crates/common/batcher/src/config.rs b/crates/common/batcher/src/config.rs index 3041d772..9c70279a 100644 --- a/crates/common/batcher/src/config.rs +++ b/crates/common/batcher/src/config.rs @@ -1,4 +1,4 @@ -use chrono::Duration; +use time::Duration; /// The parameters for the batching process. #[derive(Debug, Clone)] diff --git a/crates/common/batcher/src/driver.rs b/crates/common/batcher/src/driver.rs index 0f7c0ed5..3eeb5dac 100644 --- a/crates/common/batcher/src/driver.rs +++ b/crates/common/batcher/src/driver.rs @@ -1,9 +1,9 @@ use crate::batchable::Batchable; use crate::batcher::Batcher; use crate::batcher::BatcherOutput; -use chrono::{DateTime, Utc}; use std::collections::BTreeSet; use std::time::Duration; +use time::OffsetDateTime; use tokio::sync::mpsc::error::SendError; use tokio::sync::mpsc::{Receiver, Sender}; @@ -33,13 +33,13 @@ pub struct BatchDriver<B: Batchable> { batcher: Batcher<B>, input: Receiver<BatchDriverInput<B>>, output: Sender<BatchDriverOutput<B>>, - timers: BTreeSet<DateTime<Utc>>, + timers: BTreeSet<OffsetDateTime>, } enum TimeTo { Unbounded, Future(std::time::Duration), - Past(DateTime<Utc>), + Past(OffsetDateTime), } impl<B: Batchable> BatchDriver<B> { @@ -65,7 +65,7 @@ impl<B: Batchable> BatchDriver<B> { TimeTo::Future(timeout) => self.recv(Some(timeout)), TimeTo::Past(timer) => { self.timers.remove(&timer); - self.time(Utc::now()).await?; + self.time(OffsetDateTime::now_utc()).await?; continue; } }; @@ -95,17 +95,20 @@ impl<B: Batchable> BatchDriver<B> { match self.timers.iter().next() { None => TimeTo::Unbounded, Some(timer) => { - let signed_duration = timer.signed_duration_since(Utc::now()); - match signed_duration.to_std() { - Ok(d) => TimeTo::Future(d), - Err(_) => TimeTo::Past(*timer), + let signed_duration = *timer - OffsetDateTime::now_utc(); + if signed_duration.is_negative() { + return TimeTo::Past(*timer); } + TimeTo::Future(std::time::Duration::new( + signed_duration.abs().whole_seconds() as u64, + 0, + )) } } } async fn event(&mut self, event: B) -> Result<(), SendError<BatchDriverOutput<B>>> { - for action in self.batcher.event(Utc::now(), event) { + for action in self.batcher.event(OffsetDateTime::now_utc(), event) { match action { BatcherOutput::Batch(batch) => { self.output.send(BatchDriverOutput::Batch(batch)).await?; @@ -119,7 +122,7 @@ impl<B: Batchable> BatchDriver<B> { Ok(()) } - async fn time(&mut self, timer: DateTime<Utc>) -> Result<(), SendError<BatchDriverOutput<B>>> { + async fn time(&mut self, timer: OffsetDateTime) -> Result<(), SendError<BatchDriverOutput<B>>> { for batch in self.batcher.time(timer) { self.output.send(BatchDriverOutput::Batch(batch)).await?; } @@ -143,7 +146,6 @@ mod tests { use crate::batcher::Batcher; use crate::config::BatchConfigBuilder; use crate::driver::BatchDriver; - use chrono::{DateTime, Utc}; use std::time::Duration; use tokio::sync::mpsc::error::SendError; use tokio::sync::mpsc::{channel, Receiver, Sender}; @@ -161,7 +163,7 @@ mod tests { async fn flush_one_batch() -> Result<(), SendError<BatchDriverInput<TestBatchEvent>>> { let (input_send, mut output_recv) = spawn_driver(); - let event1 = TestBatchEvent::new(1, Utc::now()); + let event1 = TestBatchEvent::new(1, OffsetDateTime::now_utc()); input_send.send(BatchDriverInput::Event(event1)).await?; input_send.send(BatchDriverInput::Flush).await?; @@ -175,12 +177,12 @@ mod tests { async fn two_batches_with_timer() -> Result<(), SendError<BatchDriverInput<TestBatchEvent>>> { let (input_send, mut output_recv) = spawn_driver(); - let event1 = TestBatchEvent::new(1, Utc::now()); + let event1 = TestBatchEvent::new(1, OffsetDateTime::now_utc()); input_send.send(BatchDriverInput::Event(event1)).await?; assert_recv_batch(&mut output_recv, vec![event1]).await; - let event2 = TestBatchEvent::new(2, Utc::now()); + let event2 = TestBatchEvent::new(2, OffsetDateTime::now_utc()); input_send.send(BatchDriverInput::Event(event2)).await?; assert_recv_batch(&mut output_recv, vec![event2]).await; @@ -237,11 +239,11 @@ mod tests { #[derive(Debug, Copy, Clone, Eq, PartialEq)] struct TestBatchEvent { key: u64, - event_time: DateTime<Utc>, + event_time: OffsetDateTime, } impl TestBatchEvent { - fn new(key: u64, event_time: DateTime<Utc>) -> TestBatchEvent { + fn new(key: u64, event_time: OffsetDateTime) -> TestBatchEvent { TestBatchEvent { key, event_time } } } @@ -253,7 +255,7 @@ mod tests { self.key } - fn event_time(&self) -> DateTime<Utc> { + fn event_time(&self) -> OffsetDateTime { self.event_time } } diff --git a/crates/common/clock/Cargo.toml b/crates/common/clock/Cargo.toml index c5bd138e..8615da7e 100644 --- a/crates/common/clock/Cargo.toml +++ b/crates/common/clock/Cargo.toml @@ -6,7 +6,6 @@ edition = "2021" rust-version = "1.58.1" [dependencies] -time = { version = "0.3", features = ["serde-human-readable"] } -chrono = "0.4" -serde = { version = "1.0", features = ["derive"] } mockall = "0.10" +serde = { version = "1.0", features = ["derive"] } +time = { version = "0.3", features = ["serde-human-readable"] } diff --git a/crates/common/clock/src/lib.rs b/crates/common/clock/src/lib.rs index 8ae44bcd..7c87d4a0 100644 --- a/crates/common/clock/src/lib.rs +++ b/crates/common/clock/src/lib.rs @@ -1,9 +1,8 @@ -use chrono::{DateTime, FixedOffset, Local}; use mockall::automock; use serde::{Deserialize, Deserializer}; use time::{format_description::well_known::Rfc3339, OffsetDateTime}; -pub type Timestamp = DateTime<FixedOffset>; +pub type Timestamp = OffsetDateTime; #[automock] pub trait Clock: Sync + Send + 'static { @@ -15,8 +14,7 @@ pub struct WallClock; impl Clock for WallClock { fn now(&self) -> Timestamp { - let local_time_now = Local::now(); - local_time_now.with_timezone(local_time_now.offset()) + OffsetDateTime::now_utc() } } @@ -29,5 +27,5 @@ where let timestamp = String::deserialize(deserializer)?; OffsetDateTime::parse(timestamp.as_str(), &Rfc3339) .map_err(serde::de::Error::custom) - .map(|val| Some(val)) + .map(Some) } |