Method batch

indexer/src/main.rs:160–229

(&mut self, messages: impl Iterator<Item = SharedMessage> + Send)

#[async_trait::async_trait]
impl Processor for DatabaseSink {
    async fn batch(&mut self, messages: impl Iterator<Item = SharedMessage> + Send) -> Result<()> {
        let mut files = Vec::new();
        let mut bps = Vec::new();
        let mut events = Vec::new();
        let mut types = Vec::new();
        let mut allocs = Vec::new();
        let mut flush = false;

        for message in messages {
            match message.header().stream_key().name() {
                INFO_STREAM => {
                    save_debugger_info(&self.db, translate::debugger_info(deser_info(&message)))
                        .await?
                }
                FILE_STREAM => files.push(deser(&message)),
                BREAKPOINT_STREAM => bps.push(deser(&message)),
                EVENT_STREAM => events.push({
                    let event = EventStream::read_from(message.message().into_bytes().into());
                    translate::type_info(&event, |ty| types.push(ty));
                    let mut parent_frame_id = None;
                    match &event {
                        Event::FunctionCall {
                            thread_id,
                            frame_id,
                            ..
                        } => {
                            let stack = self.stack.entry(*thread_id).or_default();
                            parent_frame_id = stack.last().copied();
                            stack.push(*frame_id);
                        }
                        Event::FunctionReturn {
                            thread_id,
                            frame_id,
                            ..
                        } => {
                            assert_eq!(
                                *frame_id,
                                self.stack
                                    .get_mut(thread_id)
                                    .expect("Thread not found")
                                    .pop()
                                    .expect("Stack frame empty")
                            );
                        }
                        _ => (),
                    }
                    let mut event = translate::event(message.timestamp(), event);
                    event.parent_frame_id = sea_orm::Set(parent_frame_id.map(|s| s as i64));
                    event
                }),
                ALLOCATION_STREAM => allocs.push(deser(&message)),
                _ => anyhow::bail!("Unexpected stream key {}", message.stream_key()),
            }
            self.count += 1;
            flush |= self.count % 10000 == 0;
        }

        insert_files(&self.db, files.into_iter().map(translate::source_file)).await?;