async fn publish_updates<T, U>(
        mut self,
        target: T,
        tx: watch::Sender<U>,
        mut stream: InnerStream<U>,
    ) where
        T: Clone + Send + Sync + 'static,
        U: Send + Sync + 'static,
        S: Service<T, Response = InnerRsp<U>, Error = tonic::Status>,
        S::Future: Send,
    {
        loop {
            tokio::select! {
                biased;

                // If all of the receivers are dropped, stop the task.
                _ = tx.closed() => {
                    trace!("Receivers dropped");
                    return;
                },

                // Otherwise, continue to get new profile versions and update the watch. The stream
                // may be re-instantiated each time
                res = self.recovering_next(&target, &mut stream) => match res {
                    Ok(profile) => {
                        // If sending the update fails, then we'll just look and hit the closed case above.
                        let _ = tx.send(profile);
                    }
                    Err(status) => {
                        debug!(%status, "Profile stream failed");
                        return;
                    }
                },
            }
        }
    }
// Publishes updates on `tx` from the stream, recovering and applying backoff backoff as
// necessary.
Publishes updates on `tx` from the stream, recovering and applying backoff backoff as
necessary.
Publishes updates on ` tx ` from the stream recovering and applying backoff backoff as necessary .
async fn publish_updates<T, U>(
        mut self,
        target: T,
        tx: watch::Sender<U>,
        mut stream: InnerStream<U>,
    ) where
        T: Clone + Send + Sync + 'static,
        U: Send + Sync + 'static,
        S: Service<T, Response = InnerRsp<U>, Error = tonic::Status>,
        S::Future: Send,
    {
        loop {
            tokio::select! {
                biased;
                // If all of the receivers are dropped, stop the task.
                _ = tx.closed() => {
                    trace!("Receivers dropped");
                    return;
                },
                // Otherwise, continue to get new profile versions and update the watch. The stream
                // may be re-instantiated each time
                res = self.recovering_next(&target, &mut stream) => match res {
                    Ok(profile) => {
                        // If sending the update fails, then we'll just look and hit the closed case above.
                        let _ = tx.send(profile);
                    }
                    Err(status) => {
                        debug!(%status, "Profile stream failed");
                        return;
                    }
                },
            }
        }
    }
async fn publish_updates < T , U > ( mut self , target : T , tx : watch :: Sender < U > , mut stream : InnerStream < U > , ) where T : Clone + Send + Sync + ' static , U : Send + Sync + ' static , S : Service < T , Response = InnerRsp < U > , Error = tonic :: Status > , S :: Future : Send , { loop { tokio :: select ! { biased // If all of the receivers are dropped, stop the task. tx closed ( ) { trace ( " " ) return } // Otherwise, continue to get new profile versions and update the watch. The stream // may be re-instantiated each time res self recovering_next ( target mut stream ) match res { Ok ( profile ) { // If sending the update fails, then we'll just look and hit the closed case above. let tx send ( profile ) } Err ( status ) { debug ( status " " ) return } } } } }
Publishes updates on `tx` from the stream, recovering and applying backoff backoff as
necessary.
Publishes updates on ` tx ` from the stream recovering and applying backoff backoff as necessary .
mut self
target
T
tx
watch::Sender<U>
stream
InnerStream<U>
mut self
target
T
tx
watch::Sender<U>
stream
InnerStream<U>