pub fn execute_lifespan(
        &self,
        lifespan: Lifespan,
        available_components: HashSet<SettingType>,
        switchboard_client: SwitchboardClient,
        service_context: ServiceContextHandle,
        sequential: bool,
    ) -> futures::channel::oneshot::Receiver<Result<(), Error>> {
        let (completion_tx, completion_rx) =
            futures::channel::oneshot::channel::<Result<(), Error>>();
        let agents = self.agents.clone();

        // Due to waiting on acknowledgements, we must spawn a separate task.
        fasync::spawn(async move {
            let mut pending_acks = Vec::new();

            for agent in agents {
                let invocation_context = InvocationContext {
                    lifespan: lifespan,
                    available_components: available_components.clone(),
                    service_context: service_context.clone(),
                    switchboard_client: switchboard_client.clone(),
                };

                // Create ack channel.
                let (response_tx, response_rx) =
                    futures::channel::oneshot::channel::<Result<(), Error>>();
                match agent.lock().await.invoke(Invocation {
                    context: invocation_context,
                    ack_sender: Arc::new(Mutex::new(Some(response_tx))),
                }) {
                    Ok(handled) => {
                        // did not process result.
                        if !handled {
                            continue;
                        }

                        // Wait for invocation ack is sequential, otherwise store receiver to
                        // be waited on later.
                        if sequential {
                            let result = process_invocation_ack(response_rx).await;
                            // This cannot be part of process_invocation_ack since completion_tx
                            // would be moved.
                            if result.is_err() {
                                completion_tx.send(result).ok();
                                return;
                            }
                        } else {
                            pending_acks.push(response_rx);
                        }
                    }
                    _ => {
                        completion_tx.send(Err(format_err!("failed to invoke agent"))).ok();
                        return;
                    }
                }
            }

            // Pending acks should only be present for non sequential execution. In
            // this case wait for each to complete.
            for ack in pending_acks {
                let result = process_invocation_ack(ack).await;
                if result.is_err() {
                    completion_tx.send(result).ok();
                    return;
                }
            }

            // Acknowledge completion of the execution.
            completion_tx.send(Ok(())).ok();
        });

        return completion_rx;
    }
/// Invokes each registered agent for a given lifespan. If sequential is true,
/// invocations will only proceed to the next agent once the current
/// invocation has been successfully acknowledged. When sequential is false,
/// agents will receive their invocations without waiting. However, the
/// overall completion (signaled through the receiver returned by the method),
/// will not return until all invocations have been acknowledged.
Invokes each registered agent for a given lifespan. If sequential is true,
invocations will only proceed to the next agent once the current
invocation has been successfully acknowledged. When sequential is false,
agents will receive their invocations without waiting. However, the
overall completion (signaled through the receiver returned by the method),
will not return until all invocations have been acknowledged.
Invokes each registered agent for a given lifespan . If sequential is true invocations will only proceed to the next agent once the current invocation has been successfully acknowledged . When sequential is false agents will receive their invocations without waiting . However the overall completion ( signaled through the receiver returned by the method ) will not return until all invocations have been acknowledged .
pub fn execute_lifespan(
        &self,
        lifespan: Lifespan,
        available_components: HashSet<SettingType>,
        switchboard_client: SwitchboardClient,
        service_context: ServiceContextHandle,
        sequential: bool,
    ) -> futures::channel::oneshot::Receiver<Result<(), Error>> {
        let (completion_tx, completion_rx) =
            futures::channel::oneshot::channel::<Result<(), Error>>();
        let agents = self.agents.clone();
        // Due to waiting on acknowledgements, we must spawn a separate task.
        fasync::spawn(async move {
            let mut pending_acks = Vec::new();
            for agent in agents {
                let invocation_context = InvocationContext {
                    lifespan: lifespan,
                    available_components: available_components.clone(),
                    service_context: service_context.clone(),
                    switchboard_client: switchboard_client.clone(),
                };
                // Create ack channel.
                let (response_tx, response_rx) =
                    futures::channel::oneshot::channel::<Result<(), Error>>();
                match agent.lock().await.invoke(Invocation {
                    context: invocation_context,
                    ack_sender: Arc::new(Mutex::new(Some(response_tx))),
                }) {
                    Ok(handled) => {
                        // did not process result.
                        if !handled {
                            continue;
                        }
                        // Wait for invocation ack is sequential, otherwise store receiver to
                        // be waited on later.
                        if sequential {
                            let result = process_invocation_ack(response_rx).await;
                            // This cannot be part of process_invocation_ack since completion_tx
                            // would be moved.
                            if result.is_err() {
                                completion_tx.send(result).ok();
                                return;
                            }
                        } else {
                            pending_acks.push(response_rx);
                        }
                    }
                    _ => {
                        completion_tx.send(Err(format_err!("failed to invoke agent"))).ok();
                        return;
                    }
                }
            }
            // Pending acks should only be present for non sequential execution. In
            // this case wait for each to complete.
            for ack in pending_acks {
                let result = process_invocation_ack(ack).await;
                if result.is_err() {
                    completion_tx.send(result).ok();
                    return;
                }
            }
            // Acknowledge completion of the execution.
            completion_tx.send(Ok(())).ok();
        });
        return completion_rx;
    }
pub fn execute_lifespan ( & self , lifespan : Lifespan , available_components : HashSet < SettingType > , switchboard_client : SwitchboardClient , service_context : ServiceContextHandle , sequential : bool , ) -> futures :: channel :: oneshot :: Receiver < Result < ( ) , Error > > { let ( completion_tx , completion_rx ) = futures :: channel :: oneshot :: channel :: < Result < ( ) , Error > > ( ) ; let agents = self . agents . clone ( ) ; // Due to waiting on acknowledgements, we must spawn a separate task. fasync :: spawn ( async move { let mut pending_acks = Vec :: new ( ) ; for agent in agents { let invocation_context = InvocationContext { lifespan : lifespan , available_components : available_components . clone ( ) , service_context : service_context . clone ( ) , switchboard_client : switchboard_client . clone ( ) , } ; // Create ack channel. let ( response_tx , response_rx ) = futures :: channel :: oneshot :: channel :: < Result < ( ) , Error > > ( ) ; match agent . lock ( ) . await . invoke ( Invocation { context : invocation_context , ack_sender : Arc :: new ( Mutex :: new ( Some ( response_tx ) ) ) , } ) { Ok ( handled ) => { // did not process result. if ! handled { continue ; } // Wait for invocation ack is sequential, otherwise store receiver to // be waited on later. if sequential { let result = process_invocation_ack ( response_rx ) . await ; // This cannot be part of process_invocation_ack since completion_tx // would be moved. if result . is_err ( ) { completion_tx . send ( result ) . ok ( ) ; return ; } } else { pending_acks . push ( response_rx ) ; } } _ => { completion_tx . send ( Err ( format_err ! ( " " ) ) ) . ok ( ) ; return ; } } } // Pending acks should only be present for non sequential execution. In // this case wait for each to complete. for ack in pending_acks { let result = process_invocation_ack ( ack ) . await ; if result . is_err ( ) { completion_tx . send ( result ) . ok ( ) ; return ; } } // Acknowledge completion of the execution. completion_tx . send ( Ok ( ( ) ) ) . ok ( ) ; } ) ; return completion_rx ; }
Invokes each registered agent for a given lifespan.
Invokes each registered agent for a given lifespan .
&self
lifespan
Lifespan
available_components
HashSet<SettingType>
switchboard_client
SwitchboardClient
service_context
ServiceContextHandle
sequential
bool
&self
lifespan
Lifespan
available_components
HashSet<SettingType>
switchboard_client
SwitchboardClient
service_context
ServiceContextHandle
sequential
bool