/// Creates a new [Manager] with the given MessageHub. A reference to the service MessageHub is
/// provided so that it can be passed to [Jobs](Job) for communicating with the rest of the
/// service.
Creates a new [Manager] with the given MessageHub. A reference to the service MessageHub is
provided so that it can be passed to [Jobs](Job) for communicating with the rest of the
service.
Creates a new [ Manager ] with the given MessageHub . A reference to the service MessageHub is provided so that it can be passed to [ Jobs ] ( Job ) for communicating with the rest of the service .
pub(crate) async fn spawn(message_hub_delegate: &message::Delegate) -> message::Signature {
        // Create a top-level receptor in the MessageHub to accept new sources from.
        let receptor = message_hub_delegate
            .create(MessengerType::Unbound)
            .await
            .expect("messenger should be available")
            .1;
        // Create a channel for execution tasks to communicate when a Job has been completed.
        let (execution_completion_sender, execution_completion_receiver) =
            futures::channel::mpsc::unbounded::<(source::Id, job::Info, execution::Details)>();
        // Capture the top-level receptor's signature so it can be passed back
        // to the caller for sending new sources.
        let signature = receptor.get_signature();
        let event_publisher =
            event::Publisher::create(&message_hub_delegate, MessengerType::Unbound).await;
        let mut manager = Self {
            sources: HashMap::new(),
            job_futures: FuturesUnordered::new(),
            source_id_generator: source::IdGenerator::new(),
            execution_completion_sender,
            message_hub_delegate: message_hub_delegate.clone(),
            event_publisher,
        };
        // Spawn a task to run the main event loop, which handles the following events:
        // 1) Receiving new sources to process
        // 2) Accepting and processing new jobs from sources
        // 3) Executing jobs and handling the their results
        fasync::Task::spawn(async move {
            let nonce = fuchsia_trace::generate_nonce();
            trace!(nonce, "job_manager");
            let source_fuse = receptor.fuse();
            let execution_fuse = execution_completion_receiver.fuse();
            futures::pin_mut!(source_fuse, execution_fuse);
            loop {
                futures::select! {
                    source_event = source_fuse.select_next_some() => {
                        trace!(nonce, "process_source_event");
                        manager.process_source_event(source_event).await;
                    },
                    (source_id, job_info, details) = execution_fuse.select_next_some() => {
                        trace!(nonce, "process_completed_execution");
                        manager.process_completed_execution(source_id, job_info, details, nonce).await;
                    },
                    (job_info, stream) = manager.job_futures.select_next_some() => {
                        trace!(nonce, "process_job");
                        // Since the manager owns job_futures, we should never reach the end of
                        // the stream.
                        let (source_id, job) = job_info.expect("job should be present");
                        manager.process_job(source_id, job, stream, nonce).await;
                    }
                }
            }
        })
        .detach();
        signature
    }
pub ( crate ) async fn spawn ( message_hub_delegate : & message :: Delegate ) -> message :: Signature { // Create a top-level receptor in the MessageHub to accept new sources from. let receptor = message_hub_delegate . create ( MessengerType :: Unbound ) . await . expect ( " " ) . 1 ; // Create a channel for execution tasks to communicate when a Job has been completed. let ( execution_completion_sender , execution_completion_receiver ) = futures :: channel :: mpsc :: unbounded :: < ( source :: Id , job :: Info , execution :: Details ) > ( ) ; // Capture the top-level receptor's signature so it can be passed back // to the caller for sending new sources. let signature = receptor . get_signature ( ) ; let event_publisher = event :: Publisher :: create ( & message_hub_delegate , MessengerType :: Unbound ) . await ; let mut manager = Self { sources : HashMap :: new ( ) , job_futures : FuturesUnordered :: new ( ) , source_id_generator : source :: IdGenerator :: new ( ) , execution_completion_sender , message_hub_delegate : message_hub_delegate . clone ( ) , event_publisher , } ; // Spawn a task to run the main event loop, which handles the following events: // 1) Receiving new sources to process // 2) Accepting and processing new jobs from sources // 3) Executing jobs and handling the their results fasync :: Task :: spawn ( async move { let nonce = fuchsia_trace :: generate_nonce ( ) ; trace ! ( nonce " " ) ; let source_fuse = receptor . fuse ( ) ; let execution_fuse = execution_completion_receiver . fuse ( ) ; futures :: pin_mut ! ( source_fuse execution_fuse ) ; loop { futures :: select ! { source_event source_fuse select_next_some ( ) { trace ( nonce " " ) manager process_source_event ( source_event ) await } ( source_id job_info details ) execution_fuse select_next_some ( ) { trace ( nonce " " ) manager process_completed_execution ( source_id job_info details nonce ) await } ( job_info stream ) manager job_futures select_next_some ( ) { trace ( nonce " " ) // Since the manager owns job_futures, we should never reach the end of // the stream. let ( source_id job ) job_info expect ( " " ) manager process_job ( source_id job stream nonce ) await } } } } ) . detach ( ) ; signature }
Creates a new [Manager] with the given MessageHub.
Creates a new [ Manager ] with the given MessageHub .
message_hub_delegate
&message::Delegate