(
frame: &super::wire::SyncFrame,
inbound: &crate::control::array_sync::OriginArrayInbound,
)
| 576 | } |
| 577 | |
| 578 | async fn dispatch_array_frame( |
| 579 | frame: &super::wire::SyncFrame, |
| 580 | inbound: &crate::control::array_sync::OriginArrayInbound, |
| 581 | ) -> Option<super::wire::SyncFrame> { |
| 582 | match frame.msg_type { |
| 583 | SyncMessageType::ArrayDelta => { |
| 584 | if let Some(msg) = frame.decode_body::<ArrayDeltaMsg>() { |
| 585 | match inbound.handle_delta(&msg).await { |
| 586 | Ok(_) => None, |
| 587 | Err(Some(r)) => { |
| 588 | super::wire::SyncFrame::try_encode(SyncMessageType::ArrayReject, &r) |
| 589 | } |
| 590 | Err(None) => None, |
| 591 | } |
| 592 | } else { |
| 593 | None |
| 594 | } |
| 595 | } |
| 596 | SyncMessageType::ArrayDeltaBatch => { |
| 597 | if let Some(msg) = frame.decode_body::<ArrayDeltaBatchMsg>() { |
| 598 | let outcomes = inbound.handle_delta_batch(&msg).await; |
| 599 | outcomes.into_iter().find_map(|r| match r { |
| 600 | Err(Some(reject)) => { |
| 601 | super::wire::SyncFrame::try_encode(SyncMessageType::ArrayReject, &reject) |
| 602 | } |
| 603 | _ => None, |
| 604 | }) |
| 605 | } else { |
| 606 | None |
| 607 | } |
| 608 | } |
| 609 | SyncMessageType::ArraySnapshot => { |
| 610 | if let Some(msg) = frame.decode_body::<ArraySnapshotMsg>() { |
| 611 | match inbound.handle_snapshot_header(&msg) { |
| 612 | Ok(_) => None, |
| 613 | Err(Some(r)) => { |
| 614 | super::wire::SyncFrame::try_encode(SyncMessageType::ArrayReject, &r) |
| 615 | } |
| 616 | Err(None) => None, |
| 617 | } |
| 618 | } else { |
| 619 | None |
| 620 | } |
| 621 | } |
| 622 | SyncMessageType::ArraySnapshotChunk => { |
| 623 | if let Some(msg) = frame.decode_body::<ArraySnapshotChunkMsg>() { |
| 624 | match inbound.handle_snapshot_chunk(&msg).await { |
| 625 | Ok(_) => None, |
| 626 | Err(Some(r)) => { |
| 627 | super::wire::SyncFrame::try_encode(SyncMessageType::ArrayReject, &r) |
| 628 | } |
| 629 | Err(None) => None, |
| 630 | } |
| 631 | } else { |
| 632 | None |
| 633 | } |
| 634 | } |
| 635 | SyncMessageType::ArraySchema => { |
no test coverage detected