Method send

src/storage-controller/src/instance.rs:326–373

Sends a command to this storage instance.

(&mut self, command: StorageCommand)

    /// Sends a command to this storage instance.
    pub fn send(&mut self, command: StorageCommand) {
        // Record the command so that new replicas can be brought up to speed.
        self.history.push(command.clone());

        match command.clone() {
            StorageCommand::RunIngestion(ingestion) => {
                // First absorb into our state, because this might change
                // scheduling decisions, which need to be respected just below
                // when sending commands.
                self.absorb_ingestion(*ingestion.clone());

                for replica in self.active_replicas(&ingestion.id) {
                    replica.send(StorageCommand::RunIngestion(ingestion.clone()));
                }
            }
            StorageCommand::RunSink(sink) => {
                // First absorb into our state, because this might change
                // scheduling decisions, which need to be respected just below
                // when sending commands.
                self.absorb_export(*sink.clone());

                for replica in self.active_replicas(&sink.id) {
                    replica.send(StorageCommand::RunSink(sink.clone()));
                }
            }
            StorageCommand::AllowCompaction(id, frontier) => {
                // First send out commands and then absorb into our state since
                // absorbing them might remove entries from active_ingestions.
                for replica in self.active_replicas(&id) {
                    replica.send(StorageCommand::AllowCompaction(
                        id.clone(),
                        frontier.clone(),
                    ));
                }

                self.absorb_compaction(id, frontier);
            }
            command => {
                for replica in self.replicas.values_mut() {
                    replica.send(command.clone());
                }
            }
        }

        if command.installs_objects() && self.replicas.is_empty() {
            self.update_paused_statuses();
        }
    }

    /// Updates internal state based on incoming ingestion commands.
    ///