// Processes a new source, generating the associated tracking data and inserting its job stream
// into the monitored job futures.
Processes a new source, generating the associated tracking data and inserting its job stream
into the monitored job futures.
Processes a new source generating the associated tracking data and inserting its job stream into the monitored job futures .
async fn process_source_event(&mut self, event: service::message::MessageEvent) {
        // Manager only expects to receive new job streams from events passed into this method.
        let Payload::Source(source) = Payload::try_from(event).expect("should convert to source");
        // Extract job stream from payload.
        let job_stream = source.lock().await.take().expect("should capture job stream");
        // Associate stream with a new id.
        let source_id = self.source_id_generator.generate();
        // Create a handler to manage jobs produced by this stream.
        let _ = self.sources.insert(source_id, source::Handler::new());
        // Add the stream to the monitored pool. Associate jobs with the source id along with
        // appending an empty value to the end for indicating when the stream has completed.
        let stream_fut = job_stream
            .map(move |val| (source_id, Some(val)))
            .chain(async move { (source_id, None) }.into_stream())
            .boxed()
            .into_future();
        self.job_futures.push(stream_fut);
        self.event_publisher.send_event(Event::Source(SourceEvent::Start(source_id)));
    }
async fn process_source_event ( & mut self , event : service :: message :: MessageEvent ) { // Manager only expects to receive new job streams from events passed into this method. let Payload :: Source ( source ) = Payload :: try_from ( event ) . expect ( " " ) ; // Extract job stream from payload. let job_stream = source . lock ( ) . await . take ( ) . expect ( " " ) ; // Associate stream with a new id. let source_id = self . source_id_generator . generate ( ) ; // Create a handler to manage jobs produced by this stream. let _ = self . sources . insert ( source_id , source :: Handler :: new ( ) ) ; // Add the stream to the monitored pool. Associate jobs with the source id along with // appending an empty value to the end for indicating when the stream has completed. let stream_fut = job_stream . map ( move | val | ( source_id , Some ( val ) ) ) . chain ( async move { ( source_id , None ) } . into_stream ( ) ) . boxed ( ) . into_future ( ) ; self . job_futures . push ( stream_fut ) ; self . event_publisher . send_event ( Event :: Source ( SourceEvent :: Start ( source_id ) ) ) ; }
Processes a new source, generating the associated tracking data and inserting its job stream
into the monitored job futures.
Processes a new source generating the associated tracking data and inserting its job stream into the monitored job futures .
&mut self
event
service::message::MessageEvent