Returns the replicas that are actively running the given object (ingestion or export).
(&mut self, id: &GlobalId)
| 654 | |
| 655 | /// Returns the replicas that are actively running the given object (ingestion or export). |
| 656 | fn active_replicas(&mut self, id: &GlobalId) -> Box<dyn Iterator<Item = &mut Replica> + '_> { |
| 657 | if let Some(ingestion_id) = self.ingestion_exports.get(id) { |
| 658 | match self.active_ingestions.get(ingestion_id) { |
| 659 | Some(ingestion) => Box::new(self.replicas.iter_mut().filter_map( |
| 660 | move |(replica_id, replica)| { |
| 661 | if ingestion.active_replicas.contains(replica_id) { |
| 662 | Some(replica) |
| 663 | } else { |
| 664 | None |
| 665 | } |
| 666 | }, |
| 667 | )), |
| 668 | None => { |
| 669 | // The ingestion has already been compacted away (aka. stopped). |
| 670 | Box::new(std::iter::empty()) |
| 671 | } |
| 672 | } |
| 673 | } else if let Some(ingestion) = self.active_ingestions.get(id) { |
| 674 | Box::new( |
| 675 | self.replicas |
| 676 | .iter_mut() |
| 677 | .filter_map(move |(replica_id, replica)| { |
| 678 | if ingestion.active_replicas.contains(replica_id) { |
| 679 | Some(replica) |
| 680 | } else { |
| 681 | None |
| 682 | } |
| 683 | }), |
| 684 | ) |
| 685 | } else if let Some(export) = self.active_exports.get(id) { |
| 686 | Box::new( |
| 687 | self.replicas |
| 688 | .iter_mut() |
| 689 | .filter_map(move |(replica_id, replica)| { |
| 690 | if export.active_replicas.contains(replica_id) { |
| 691 | Some(replica) |
| 692 | } else { |
| 693 | None |
| 694 | } |
| 695 | }), |
| 696 | ) |
| 697 | } else { |
| 698 | Box::new(self.replicas.values_mut()) |
| 699 | } |
| 700 | } |
| 701 | |
| 702 | /// Returns whether the given replica is actively running the given object (ingestion or export). |
| 703 | fn is_active_replica(&self, id: &GlobalId, replica_id: &ReplicaId) -> bool { |
no test coverage detected