common_event_recorder/
recorder.rs

1// Copyright 2023 Greptime Team
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15use std::any::Any;
16use std::collections::{HashMap, HashSet};
17use std::fmt::Debug;
18use std::sync::Arc;
19use std::time::Duration;
20
21use api::v1::value::ValueData;
22use api::v1::{ColumnSchema, Row, RowInsertRequest, RowInsertRequests, Rows};
23use async_trait::async_trait;
24use backon::{BackoffBuilder, ExponentialBuilder};
25use common_telemetry::{debug, error, info, warn};
26use common_time::timestamp::{TimeUnit, Timestamp};
27use humantime::format_duration;
28use itertools::Itertools;
29use serde::{Deserialize, Serialize};
30use store_api::mito_engine_options::{APPEND_MODE_KEY, TTL_KEY};
31use tokio::sync::mpsc::{Receiver, Sender, channel};
32use tokio::task::JoinHandle;
33use tokio::time::sleep;
34use tokio_util::sync::CancellationToken;
35
36use crate::error::{MismatchedSchemaSnafu, Result};
37use crate::event_table::{
38    PAYLOAD_COLUMN, TIMESTAMP_COLUMN, TYPE_COLUMN, base_column_schemas, jsonb_value,
39};
40
41/// The default table name for storing the events.
42pub const DEFAULT_EVENTS_TABLE_NAME: &str = "events";
43
44/// The column name for the event type.
45pub const EVENTS_TABLE_TYPE_COLUMN_NAME: &str = TYPE_COLUMN.name();
46/// The column name for the event payload.
47pub const EVENTS_TABLE_PAYLOAD_COLUMN_NAME: &str = PAYLOAD_COLUMN.name();
48/// The column name for the event timestamp.
49pub const EVENTS_TABLE_TIMESTAMP_COLUMN_NAME: &str = TIMESTAMP_COLUMN.name();
50
51/// EventRecorderRef is the reference to the event recorder.
52pub type EventRecorderRef = Arc<dyn EventRecorder>;
53
54/// A shared event-type filter used by event producers and recorders.
55pub type EventTypeFilterRef = Arc<EventTypeFilter>;
56
57/// Restricts the event types that are recorded.
58#[derive(Debug, Clone, PartialEq, Eq, Default)]
59pub enum EventTypeFilter {
60    /// Records all current and future event types.
61    #[default]
62    All,
63    /// Records only the event types in the set.
64    Only(HashSet<String>),
65}
66
67impl EventTypeFilter {
68    /// Returns whether the filter retains `event_type`.
69    pub fn allows(&self, event_type: &str) -> bool {
70        match self {
71            Self::All => true,
72            Self::Only(event_types) => event_types.contains(event_type),
73        }
74    }
75}
76
77fn deserialize_event_types<'de, D>(
78    deserializer: D,
79) -> std::result::Result<EventTypeFilterRef, D::Error>
80where
81    D: serde::Deserializer<'de>,
82{
83    HashSet::<String>::deserialize(deserializer)
84        .map(|event_types| Arc::new(EventTypeFilter::Only(event_types)))
85}
86
87fn serialize_event_types<S>(
88    event_types: &EventTypeFilterRef,
89    serializer: S,
90) -> std::result::Result<S::Ok, S::Error>
91where
92    S: serde::Serializer,
93{
94    match event_types.as_ref() {
95        EventTypeFilter::All => serializer.serialize_none(),
96        EventTypeFilter::Only(event_types) => {
97            let mut event_types = event_types.iter().collect::<Vec<_>>();
98            event_types.sort_unstable();
99            event_types.serialize(serializer)
100        }
101    }
102}
103
104fn event_type_filter_is_all(event_types: &EventTypeFilterRef) -> bool {
105    matches!(event_types.as_ref(), EventTypeFilter::All)
106}
107
108/// The time interval for flushing batched events to the event handler.
109pub const DEFAULT_FLUSH_INTERVAL_SECONDS: Duration = Duration::from_secs(5);
110/// The default TTL(90 days) for the events table.
111const DEFAULT_EVENTS_TABLE_TTL: Duration = Duration::from_days(90);
112/// The default compaction time window for the events table.
113pub const DEFAULT_COMPACTION_TIME_WINDOW: Duration = Duration::from_days(1);
114// The capacity of the tokio channel for transmitting events to background processor.
115const DEFAULT_CHANNEL_SIZE: usize = 2048;
116// The size of the buffer for batching events before flushing to event handler.
117const DEFAULT_BUFFER_SIZE: usize = 100;
118// The maximum number of retry attempts when event handler processing fails.
119const DEFAULT_MAX_RETRY_TIMES: u64 = 3;
120
121/// Event trait defines the interface for events that can be recorded and persisted as the system table.
122/// By default, the event will be persisted as the system table with the following schema:
123///
124/// - `type`: the type of the event.
125/// - `payload`: the JSON bytes of the event.
126/// - `timestamp`: the timestamp of the event.
127///
128/// The event can also add the extra schema and row to the event by overriding the `extra_schema` and `extra_row` methods.
129pub trait Event: Send + Sync + Debug {
130    /// Returns the table name of the event.
131    fn table_name(&self) -> &str {
132        DEFAULT_EVENTS_TABLE_NAME
133    }
134
135    /// Returns the type of the event.
136    fn event_type(&self) -> &str;
137
138    /// Returns the timestamp of the event. Default to the current time.
139    fn timestamp(&self) -> Timestamp {
140        Timestamp::current_time(TimeUnit::Nanosecond)
141    }
142
143    /// Returns the event payload as a structured JSON value. It will be encoded as JSONB when stored.
144    fn json_payload(&self) -> Result<serde_json::Value> {
145        Ok(serde_json::Value::Null)
146    }
147
148    /// Add the extra schema to the event with the default schema.
149    fn extra_schema(&self) -> Vec<ColumnSchema> {
150        vec![]
151    }
152
153    /// Add the extra rows to the event with the default row.
154    fn extra_rows(&self) -> Result<Vec<Row>> {
155        Ok(vec![Row { values: vec![] }])
156    }
157
158    /// Returns the event as any type.
159    fn as_any(&self) -> &dyn Any;
160}
161
162/// Eventable trait defines the interface for objects that can be converted to [Event].
163pub trait Eventable: Send + Sync + Debug {
164    /// Converts the object to an [Event].
165    fn to_event(&self) -> Option<Box<dyn Event>> {
166        None
167    }
168}
169
170/// Groups events by its `event_type`.
171#[allow(clippy::borrowed_box)]
172pub fn group_events_by_type(events: &[Box<dyn Event>]) -> HashMap<&str, Vec<&Box<dyn Event>>> {
173    events
174        .iter()
175        .into_grouping_map_by(|event| event.event_type())
176        .collect()
177}
178
179/// Builds the row inserts request for the events that will be persisted to the events table. The `events` should have the same event type, or it will return an error.
180#[allow(clippy::borrowed_box)]
181pub fn build_row_inserts_request(events: &[&Box<dyn Event>]) -> Result<RowInsertRequests> {
182    // Ensure all the events are the same type.
183    validate_events(events)?;
184
185    // We already validated the events, so it's safe to get the first event to build the schema for the RowInsertRequest.
186    let event = &events[0];
187    let extra_schema = event.extra_schema();
188    let mut schema: Vec<ColumnSchema> = Vec::with_capacity(3 + extra_schema.len());
189    schema.extend(base_column_schemas());
190    schema.extend(extra_schema);
191
192    let mut rows: Vec<Row> = Vec::with_capacity(events.len());
193    for event in events {
194        let extra_rows = event.extra_rows()?;
195        for extra_row in extra_rows {
196            let mut values = Vec::with_capacity(3 + extra_row.values.len());
197            values.extend([
198                ValueData::StringValue(event.event_type().to_string()).into(),
199                jsonb_value(&event.json_payload()?),
200                ValueData::TimestampNanosecondValue(event.timestamp().value()).into(),
201            ]);
202            values.extend(extra_row.values);
203            rows.push(Row { values });
204        }
205    }
206
207    Ok(RowInsertRequests {
208        inserts: vec![RowInsertRequest {
209            table_name: event.table_name().to_string(),
210            rows: Some(Rows { schema, rows }),
211        }],
212    })
213}
214
215// Ensure the events with the same event type have the same extra schema.
216#[allow(clippy::borrowed_box)]
217fn validate_events(events: &[&Box<dyn Event>]) -> Result<()> {
218    // It's safe to get the first event because the events are already grouped by the event type.
219    let extra_schema = events[0].extra_schema();
220    for event in events {
221        if event.extra_schema() != extra_schema {
222            MismatchedSchemaSnafu {
223                expected: extra_schema.clone(),
224                actual: event.extra_schema(),
225            }
226            .fail()?;
227        }
228    }
229    Ok(())
230}
231
232/// EventRecorder trait defines the interface for recording events.
233pub trait EventRecorder: Send + Sync + Debug + 'static {
234    /// Records an event for persistence and processing by [EventHandler].
235    fn record(&self, event: Box<dyn Event>);
236
237    /// Returns the event types accepted by this recorder.
238    fn event_type_filter(&self) -> EventTypeFilterRef;
239
240    /// Cancels the event recorder.
241    fn close(&self);
242}
243
244/// EventHandlerOptions is the options for the event handler.
245#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
246pub struct EventHandlerOptions {
247    /// TTL for the events table that will be used to store the events.
248    pub ttl: Duration,
249    /// Append mode for the events table that will be used to store the events.
250    pub append_mode: bool,
251}
252
253impl Default for EventHandlerOptions {
254    fn default() -> Self {
255        Self {
256            ttl: DEFAULT_EVENTS_TABLE_TTL,
257            append_mode: true,
258        }
259    }
260}
261
262impl EventHandlerOptions {
263    /// Converts the options to the hints for the insert operation.
264    pub fn to_hints(&self) -> Vec<(&str, String)> {
265        vec![
266            (TTL_KEY, format_duration(self.ttl).to_string()),
267            (APPEND_MODE_KEY, self.append_mode.to_string()),
268        ]
269    }
270}
271
272/// EventHandler trait defines the interface for how to handle the event.
273#[async_trait]
274pub trait EventHandler: Send + Sync + 'static {
275    /// Processes and handles incoming events. The [DefaultEventHandlerImpl] implementation forwards events to frontend instances for persistence.
276    /// We use `&[Box<dyn Event>]` to avoid consuming the events, so the caller can buffer the events and retry if the handler fails.
277    async fn handle(&self, events: &[Box<dyn Event>]) -> Result<()>;
278}
279
280/// Configuration options for the event recorder.
281#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
282pub struct EventRecorderOptions {
283    /// TTL for the events table that will be used to store the events.
284    #[serde(default = "default_events_table_ttl", with = "humantime_serde")]
285    pub ttl: Duration,
286    /// Event types that the recorder retains. When omitted, all event types are retained.
287    #[serde(
288        default,
289        deserialize_with = "deserialize_event_types",
290        serialize_with = "serialize_event_types",
291        skip_serializing_if = "event_type_filter_is_all"
292    )]
293    pub event_types: EventTypeFilterRef,
294}
295
296fn default_events_table_ttl() -> Duration {
297    DEFAULT_EVENTS_TABLE_TTL
298}
299
300impl Default for EventRecorderOptions {
301    fn default() -> Self {
302        Self {
303            ttl: DEFAULT_EVENTS_TABLE_TTL,
304            event_types: Arc::new(EventTypeFilter::All),
305        }
306    }
307}
308
309/// Implementation of [EventRecorder] that records the events and processes them in the background by the [EventHandler].
310#[derive(Debug)]
311pub struct EventRecorderImpl {
312    // The channel to send the events to the background processor.
313    tx: Sender<Box<dyn Event>>,
314    // The event types this recorder accepts before sending to the background processor.
315    event_types: EventTypeFilterRef,
316    // The cancel token to cancel the background processor.
317    cancel_token: CancellationToken,
318    // The background processor to process the events.
319    handle: Option<JoinHandle<()>>,
320}
321
322impl EventRecorderImpl {
323    pub fn new(event_handler: Box<dyn EventHandler>) -> Self {
324        Self::with_event_type_filter(event_handler, Arc::new(EventTypeFilter::All))
325    }
326
327    /// Creates an event recorder with an event-type filter.
328    pub fn with_event_type_filter(
329        event_handler: Box<dyn EventHandler>,
330        event_types: EventTypeFilterRef,
331    ) -> Self {
332        let (tx, rx) = channel(DEFAULT_CHANNEL_SIZE);
333        let cancel_token = CancellationToken::new();
334
335        let mut recorder = Self {
336            tx,
337            event_types,
338            handle: None,
339            cancel_token: cancel_token.clone(),
340        };
341
342        let processor = EventProcessor::new(
343            rx,
344            event_handler,
345            DEFAULT_FLUSH_INTERVAL_SECONDS,
346            DEFAULT_MAX_RETRY_TIMES,
347        )
348        .with_cancel_token(cancel_token);
349
350        // Spawn a background task to process the events.
351        let handle = tokio::spawn(async move {
352            processor.process(DEFAULT_BUFFER_SIZE).await;
353        });
354
355        recorder.handle = Some(handle);
356
357        recorder
358    }
359}
360
361impl EventRecorder for EventRecorderImpl {
362    // Accepts an event and send it to the background handler.
363    fn record(&self, event: Box<dyn Event>) {
364        if !self.event_types.allows(event.event_type()) {
365            return;
366        }
367
368        if let Err(e) = self.tx.try_send(event) {
369            error!("Failed to send event to the background processor: {}", e);
370        }
371    }
372
373    fn event_type_filter(&self) -> EventTypeFilterRef {
374        self.event_types.clone()
375    }
376
377    // Closes the event recorder. It will stop the background processor and flush the buffer.
378    fn close(&self) {
379        self.cancel_token.cancel();
380    }
381}
382
383impl Drop for EventRecorderImpl {
384    fn drop(&mut self) {
385        if let Some(handle) = self.handle.take() {
386            handle.abort();
387            info!("Aborted the background processor in event recorder");
388        }
389    }
390}
391
392struct EventProcessor {
393    rx: Receiver<Box<dyn Event>>,
394    event_handler: Box<dyn EventHandler>,
395    max_retry_times: u64,
396    process_interval: Duration,
397    cancel_token: CancellationToken,
398}
399
400impl EventProcessor {
401    fn new(
402        rx: Receiver<Box<dyn Event>>,
403        event_handler: Box<dyn EventHandler>,
404        process_interval: Duration,
405        max_retry_times: u64,
406    ) -> Self {
407        Self {
408            rx,
409            event_handler,
410            max_retry_times,
411            process_interval,
412            cancel_token: CancellationToken::new(),
413        }
414    }
415
416    fn with_cancel_token(mut self, cancel_token: CancellationToken) -> Self {
417        self.cancel_token = cancel_token;
418        self
419    }
420
421    async fn process(mut self, buffer_size: usize) {
422        info!("Start the background processor in event recorder to handle the received events.");
423
424        let mut buffer = Vec::with_capacity(buffer_size);
425        let mut interval = tokio::time::interval(self.process_interval);
426
427        loop {
428            tokio::select! {
429                maybe_event = self.rx.recv() => {
430                    if let Some(maybe_event) = maybe_event {
431                        debug!("Received event: {:?}", maybe_event);
432
433                        if buffer.len() >= buffer_size {
434                            debug!(
435                                "Flushing events to the event handler because the buffer is full with {} events",
436                                buffer.len()
437                            );
438                            self.flush_events_to_handler(&mut buffer).await;
439                        }
440
441                        // Push the event to the buffer, the buffer will be flushed when the interval is triggered or received a closed signal.
442                        buffer.push(maybe_event);
443                    } else {
444                        // When received a closed signal, flush the buffer and exit the loop.
445                        self.flush_events_to_handler(&mut buffer).await;
446                        break;
447                    }
448                }
449                // Cancel the processor through the cancel token.
450                _ = self.cancel_token.cancelled() => {
451                    warn!("Received a cancel signal, flushing the buffer and exiting the loop");
452                    self.flush_events_to_handler(&mut buffer).await;
453                    break;
454                }
455                // When the interval is triggered, flush the buffer and send the events to the event handler.
456                _ = interval.tick() => {
457                    self.flush_events_to_handler(&mut buffer).await;
458                }
459            }
460        }
461    }
462
463    // NOTE: While we implement a retry mechanism for failed event handling, there is no guarantee that all events will be processed successfully.
464    async fn flush_events_to_handler(&self, buffer: &mut Vec<Box<dyn Event>>) {
465        if !buffer.is_empty() {
466            debug!("Flushing {} events to the event handler", buffer.len());
467
468            let mut backoff = ExponentialBuilder::default()
469                .with_min_delay(Duration::from_millis(
470                    DEFAULT_FLUSH_INTERVAL_SECONDS.as_millis() as u64 / self.max_retry_times.max(1),
471                ))
472                .with_max_delay(Duration::from_millis(
473                    DEFAULT_FLUSH_INTERVAL_SECONDS.as_millis() as u64,
474                ))
475                .with_max_times(self.max_retry_times as usize)
476                .build();
477
478            loop {
479                match self.event_handler.handle(buffer).await {
480                    Ok(()) => {
481                        debug!("Successfully handled {} events", buffer.len());
482                        break;
483                    }
484                    Err(e) => {
485                        if let Some(d) = backoff.next() {
486                            warn!(e; "Failed to handle events, retrying...");
487                            sleep(d).await;
488                            continue;
489                        } else {
490                            warn!(
491                                e; "Failed to handle events after {} retries",
492                                self.max_retry_times
493                            );
494                            break;
495                        }
496                    }
497                }
498            }
499        }
500
501        // Clear the buffer to prevent unbounded memory growth, regardless of whether event processing succeeded or failed.
502        buffer.clear();
503    }
504}
505
506#[cfg(test)]
507mod tests {
508    use std::collections::HashSet;
509    use std::sync::atomic::{AtomicUsize, Ordering};
510
511    use serde_json::json;
512
513    use super::*;
514
515    #[derive(Debug)]
516    struct TestEvent {}
517
518    impl Event for TestEvent {
519        fn event_type(&self) -> &str {
520            "test_event"
521        }
522
523        fn json_payload(&self) -> Result<serde_json::Value> {
524            Ok(json!({"procedure_id": "1234567890"}))
525        }
526
527        fn as_any(&self) -> &dyn Any {
528            self
529        }
530    }
531
532    struct TestEventHandlerImpl {}
533
534    #[async_trait]
535    impl EventHandler for TestEventHandlerImpl {
536        async fn handle(&self, events: &[Box<dyn Event>]) -> Result<()> {
537            let event = events
538                .first()
539                .unwrap()
540                .as_any()
541                .downcast_ref::<TestEvent>()
542                .unwrap();
543            assert_eq!(
544                event.json_payload().unwrap(),
545                json!({"procedure_id": "1234567890"}),
546            );
547            assert_eq!(event.event_type(), "test_event");
548            Ok(())
549        }
550    }
551
552    #[test]
553    fn test_event_type_filter_defaults_to_all() {
554        let options = toml::from_str::<EventRecorderOptions>("ttl = '90d'").unwrap();
555
556        assert!(options.event_types.allows("slow_query"));
557        assert!(options.event_types.allows("future_event"));
558    }
559
560    #[test]
561    fn test_event_type_filter_deserializes_explicit_empty_array() {
562        let options =
563            toml::from_str::<EventRecorderOptions>("ttl = '90d'\nevent_types = []").unwrap();
564
565        assert_eq!(
566            options.event_types.as_ref(),
567            &EventTypeFilter::Only(HashSet::new())
568        );
569    }
570
571    #[test]
572    fn test_event_recorder_options_default_ttl_for_partial_table() {
573        let options = toml::from_str::<EventRecorderOptions>("event_types = []").unwrap();
574
575        assert_eq!(DEFAULT_EVENTS_TABLE_TTL, options.ttl);
576        assert_eq!(
577            options.event_types.as_ref(),
578            &EventTypeFilter::Only(HashSet::new())
579        );
580    }
581
582    #[test]
583    fn test_event_type_filter_deserializes_selected_types() {
584        let options = toml::from_str::<EventRecorderOptions>(
585            "ttl = '90d'\nevent_types = ['create_database']",
586        )
587        .unwrap();
588
589        assert!(options.event_types.allows("create_database"));
590        assert!(!options.event_types.allows("drop_database"));
591    }
592
593    struct CountingEventHandler {
594        count: Arc<AtomicUsize>,
595    }
596
597    #[async_trait]
598    impl EventHandler for CountingEventHandler {
599        async fn handle(&self, _events: &[Box<dyn Event>]) -> Result<()> {
600            self.count.fetch_add(1, Ordering::Relaxed);
601            Ok(())
602        }
603    }
604
605    #[tokio::test]
606    async fn test_event_recorder_rejects_filtered_event_before_queueing() {
607        let count = Arc::new(AtomicUsize::new(0));
608        let event_type_filter = Arc::new(EventTypeFilter::Only(HashSet::new()));
609        let mut event_recorder = EventRecorderImpl::with_event_type_filter(
610            Box::new(CountingEventHandler {
611                count: count.clone(),
612            }),
613            event_type_filter.clone(),
614        );
615
616        assert!(Arc::ptr_eq(
617            &event_type_filter,
618            &event_recorder.event_type_filter()
619        ));
620
621        event_recorder.record(Box::new(TestEvent {}));
622        event_recorder.close();
623
624        if let Some(handle) = event_recorder.handle.take() {
625            assert!(handle.await.is_ok());
626        }
627        assert_eq!(count.load(Ordering::Relaxed), 0);
628    }
629
630    #[tokio::test]
631    async fn test_event_recorder() {
632        let mut event_recorder = EventRecorderImpl::new(Box::new(TestEventHandlerImpl {}));
633        event_recorder.record(Box::new(TestEvent {}));
634
635        // Sleep for a while to let the event be sent to the event handler.
636        sleep(Duration::from_millis(500)).await;
637
638        // Close the event recorder to flush the buffer.
639        event_recorder.close();
640
641        // Sleep for a while to let the background task process the event.
642        sleep(Duration::from_millis(500)).await;
643
644        if let Some(handle) = event_recorder.handle.take() {
645            assert!(handle.await.is_ok());
646        }
647    }
648
649    struct TestEventHandlerImplShouldPanic {}
650
651    #[async_trait]
652    impl EventHandler for TestEventHandlerImplShouldPanic {
653        async fn handle(&self, events: &[Box<dyn Event>]) -> Result<()> {
654            let event = events
655                .first()
656                .unwrap()
657                .as_any()
658                .downcast_ref::<TestEvent>()
659                .unwrap();
660
661            // Set the incorrect payload and event type to trigger the panic.
662            assert_eq!(
663                event.json_payload().unwrap(),
664                "{\"procedure_id\": \"should_panic\"}"
665            );
666            assert_eq!(event.event_type(), "should_panic");
667            Ok(())
668        }
669    }
670
671    #[tokio::test]
672    async fn test_event_recorder_should_panic() {
673        let mut event_recorder =
674            EventRecorderImpl::new(Box::new(TestEventHandlerImplShouldPanic {}));
675
676        event_recorder.record(Box::new(TestEvent {}));
677
678        // Sleep for a while to let the event be sent to the event handler.
679        sleep(Duration::from_millis(500)).await;
680
681        // Close the event recorder to flush the buffer.
682        event_recorder.close();
683
684        // Sleep for a while to let the background task process the event.
685        sleep(Duration::from_millis(500)).await;
686
687        if let Some(handle) = event_recorder.handle.take() {
688            assert!(handle.await.unwrap_err().is_panic());
689        }
690    }
691
692    #[derive(Debug)]
693    struct TestEventA {}
694
695    impl Event for TestEventA {
696        fn event_type(&self) -> &str {
697            "A"
698        }
699
700        fn as_any(&self) -> &dyn Any {
701            self
702        }
703    }
704
705    #[derive(Debug)]
706    struct TestEventB {}
707
708    impl Event for TestEventB {
709        fn table_name(&self) -> &str {
710            "table_B"
711        }
712
713        fn event_type(&self) -> &str {
714            "B"
715        }
716
717        fn as_any(&self) -> &dyn Any {
718            self
719        }
720    }
721
722    #[derive(Debug)]
723    struct TestEventC {}
724
725    impl Event for TestEventC {
726        fn table_name(&self) -> &str {
727            "table_C"
728        }
729
730        fn event_type(&self) -> &str {
731            "C"
732        }
733
734        fn as_any(&self) -> &dyn Any {
735            self
736        }
737    }
738
739    #[test]
740    fn test_group_events_by_type() {
741        let events: Vec<Box<dyn Event>> = vec![
742            Box::new(TestEventA {}),
743            Box::new(TestEventB {}),
744            Box::new(TestEventA {}),
745            Box::new(TestEventC {}),
746            Box::new(TestEventB {}),
747            Box::new(TestEventC {}),
748            Box::new(TestEventA {}),
749        ];
750
751        let event_groups = group_events_by_type(&events);
752        assert_eq!(event_groups.len(), 3);
753        assert_eq!(event_groups.get("A").unwrap().len(), 3);
754        assert_eq!(event_groups.get("B").unwrap().len(), 2);
755        assert_eq!(event_groups.get("C").unwrap().len(), 2);
756    }
757
758    #[test]
759    fn test_build_row_inserts_request() {
760        let events: Vec<Box<dyn Event>> = vec![
761            Box::new(TestEventA {}),
762            Box::new(TestEventB {}),
763            Box::new(TestEventA {}),
764            Box::new(TestEventC {}),
765            Box::new(TestEventB {}),
766            Box::new(TestEventC {}),
767            Box::new(TestEventA {}),
768        ];
769
770        let event_groups = group_events_by_type(&events);
771        assert_eq!(event_groups.len(), 3);
772        assert_eq!(event_groups.get("A").unwrap().len(), 3);
773        assert_eq!(event_groups.get("B").unwrap().len(), 2);
774        assert_eq!(event_groups.get("C").unwrap().len(), 2);
775
776        for (event_type, events) in event_groups {
777            let row_inserts_request = build_row_inserts_request(&events).unwrap();
778            if event_type == "A" {
779                assert_eq!(row_inserts_request.inserts.len(), 1);
780                assert_eq!(
781                    row_inserts_request.inserts[0].table_name,
782                    DEFAULT_EVENTS_TABLE_NAME
783                );
784                assert_eq!(
785                    row_inserts_request.inserts[0]
786                        .rows
787                        .as_ref()
788                        .unwrap()
789                        .rows
790                        .len(),
791                    3
792                );
793            } else if event_type == "B" {
794                assert_eq!(row_inserts_request.inserts.len(), 1);
795                assert_eq!(row_inserts_request.inserts[0].table_name, "table_B");
796                assert_eq!(
797                    row_inserts_request.inserts[0]
798                        .rows
799                        .as_ref()
800                        .unwrap()
801                        .rows
802                        .len(),
803                    2
804                );
805            } else if event_type == "C" {
806                assert_eq!(row_inserts_request.inserts.len(), 1);
807                assert_eq!(row_inserts_request.inserts[0].table_name, "table_C");
808                assert_eq!(
809                    row_inserts_request.inserts[0]
810                        .rows
811                        .as_ref()
812                        .unwrap()
813                        .rows
814                        .len(),
815                    2
816                );
817            } else {
818                panic!("Unexpected event type: {}", event_type);
819            }
820        }
821    }
822}
common_event_recorder/recorder.rs

common_event_recorder/
recorder.rs
