| 480 | } |
| 481 | |
| 482 | async fn process_events(events: &mut Vec<EventInsert>, event_store: &Arc<M>, node_id: NodeId) { |
| 483 | trace!(count = events.len(), "process_events"); |
| 484 | if events.is_empty() { |
| 485 | return; |
| 486 | } |
| 487 | let mut oneshots = HashMap::with_capacity(events.len()); |
| 488 | let mut items = Vec::with_capacity(events.len()); |
| 489 | events.drain(..).for_each(|req: EventInsert| { |
| 490 | oneshots |
| 491 | .entry(req.id.clone()) |
| 492 | .or_insert(vec![]) |
| 493 | .push(req.tx); |
| 494 | items.push(ApiItem::new(req.id, req.data)); |
| 495 | }); |
| 496 | tracing::trace!("calling insert many with {} items.", items.len()); |
| 497 | match event_store.insert_many(items, node_id).await { |
| 498 | Ok(results) => { |
| 499 | tracing::debug!("insert many returned {} results.", results.len()); |
| 500 | for result in results { |
| 501 | let id = result.id(); |
| 502 | if let Some(txs) = oneshots.get_mut(id) { |
| 503 | // Expect one result per oneshot channel |
| 504 | if let Some(tx) = txs.pop() { |
| 505 | if let Err(e) = tx.send(Ok(result.clone())) { |
| 506 | tracing::warn!( |
| 507 | "failed to send success response to api listener: {:?}", |
| 508 | e |
| 509 | ); |
| 510 | } |
| 511 | } else { |
| 512 | tracing::warn!( |
| 513 | "no more channels to respond to API listener for duplicate event ID: {:?}", |
| 514 | id |
| 515 | ); |
| 516 | } |
| 517 | } else { |
| 518 | tracing::warn!( |
| 519 | "lost channel to respond to API listener for event ID: {:?}", |
| 520 | id |
| 521 | ); |
| 522 | } |
| 523 | } |
| 524 | } |
| 525 | Err(e) => { |
| 526 | tracing::warn!("failed to insert events: {e}"); |
| 527 | for txs in oneshots.into_values() { |
| 528 | for tx in txs { |
| 529 | if let Err(e) = tx.send(Err(anyhow::anyhow!("Failed to insert event: {e}"))) |
| 530 | { |
| 531 | tracing::warn!( |
| 532 | "failed to send failed response to api listener: {:?}", |
| 533 | e |
| 534 | ); |
| 535 | } |
| 536 | } |
| 537 | } |
| 538 | } |
| 539 | }; |