meta_srv/
event.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 async_trait::async_trait;
16use client::inserter::{Context, Inserter};
17use common_catalog::consts::{DEFAULT_CATALOG_NAME, DEFAULT_PRIVATE_SCHEMA_NAME};
18use common_error::ext::BoxedError;
19use common_event_recorder::error::{InsertEventsSnafu, Result};
20use common_event_recorder::{Event, EventHandler, build_row_inserts_request, group_events_by_type};
21use snafu::ResultExt;
22
23pub mod gc;
24pub mod region_migration;
25pub mod repartition;
26pub(crate) mod wal_prune;
27
28/// EventHandlerImpl is the default event handler implementation in metasrv.
29/// It sends the received events to the frontend instances.
30pub struct EventHandlerImpl {
31    inserter: Box<dyn Inserter>,
32}
33
34impl EventHandlerImpl {
35    pub fn new(inserter: Box<dyn Inserter>) -> Self {
36        Self { inserter }
37    }
38}
39
40#[async_trait]
41impl EventHandler for EventHandlerImpl {
42    async fn handle(&self, events: &[Box<dyn Event>]) -> Result<()> {
43        let event_groups = group_events_by_type(events);
44
45        for (_, events) in event_groups {
46            let requests = build_row_inserts_request(&events)?;
47            self.inserter
48                .insert_rows(
49                    &Context {
50                        catalog: DEFAULT_CATALOG_NAME,
51                        schema: DEFAULT_PRIVATE_SCHEMA_NAME,
52                    },
53                    requests,
54                )
55                .await
56                .map_err(BoxedError::new)
57                .context(InsertEventsSnafu)?;
58        }
59
60        Ok(())
61    }
62}
meta_srv/event.rs

meta_srv/
event.rs
