async fn export(
        tx: &mpsc::Sender<ExportTraceServiceRequest>,
        spans: &mut S,
        accum: &mut Vec<Span>,
        node: &mut Option<Node>,
    ) -> Result<(), SpanRxClosed> {
        loop {
            // Collect spans into a batch.
            let collect = Self::collect_batch(spans, accum).await;

            // If we collected spans, flush them.
            if !accum.is_empty() {
                // Once a batch has been accumulated, ensure that the
                // request stream is ready to accept the batch.
                match tx.reserve().await {
                    Ok(tx) => {
                        let msg = ExportTraceServiceRequest {
                            spans: accum.drain(..).collect(),
                            node: node.take(),
                            ..Default::default()
                        };
                        trace!(
                            node = msg.node.is_some(),
                            spans = msg.spans.len(),
                            "Sending batch"
                        );
                        tx.send(msg);
                    }
                    Err(error) => {
                        // If the channel isn't open, start a new stream
                        // and retry sending the batch.
                        debug!(%error, "Request stream lost; restarting");
                        return Ok(());
                    }
                }
            }

            // If the span source was closed, end the task.
            if let Err(closed) = collect {
                debug!("Span channel lost");
                return Err(closed);
            }
        }
    }
/// Accumulate spans and send them on the export stream.
///
/// Returns an error when the proxy has closed the span stream.
Accumulate spans and send them on the export stream.
Returns an error when the proxy has closed the span stream.
Accumulate spans and send them on the export stream . Returns an error when the proxy has closed the span stream .
async fn export(
        tx: &mpsc::Sender<ExportTraceServiceRequest>,
        spans: &mut S,
        accum: &mut Vec<Span>,
        node: &mut Option<Node>,
    ) -> Result<(), SpanRxClosed> {
        loop {
            // Collect spans into a batch.
            let collect = Self::collect_batch(spans, accum).await;
            // If we collected spans, flush them.
            if !accum.is_empty() {
                // Once a batch has been accumulated, ensure that the
                // request stream is ready to accept the batch.
                match tx.reserve().await {
                    Ok(tx) => {
                        let msg = ExportTraceServiceRequest {
                            spans: accum.drain(..).collect(),
                            node: node.take(),
                            ..Default::default()
                        };
                        trace!(
                            node = msg.node.is_some(),
                            spans = msg.spans.len(),
                            "Sending batch"
                        );
                        tx.send(msg);
                    }
                    Err(error) => {
                        // If the channel isn't open, start a new stream
                        // and retry sending the batch.
                        debug!(%error, "Request stream lost; restarting");
                        return Ok(());
                    }
                }
            }
            // If the span source was closed, end the task.
            if let Err(closed) = collect {
                debug!("Span channel lost");
                return Err(closed);
            }
        }
    }
async fn export ( tx : & mpsc :: Sender < ExportTraceServiceRequest > , spans : & mut S , accum : & mut Vec < Span > , node : & mut Option < Node > , ) -> Result < ( ) , SpanRxClosed > { loop { // Collect spans into a batch. let collect = Self :: collect_batch ( spans , accum ) . await ; // If we collected spans, flush them. if ! accum . is_empty ( ) { // Once a batch has been accumulated, ensure that the // request stream is ready to accept the batch. match tx . reserve ( ) . await { Ok ( tx ) => { let msg = ExportTraceServiceRequest { spans : accum . drain ( .. ) . collect ( ) , node : node . take ( ) , .. Default :: default ( ) } ; trace ! ( node msg node is_some ( ) spans msg spans len ( ) " " ) ; tx . send ( msg ) ; } Err ( error ) => { // If the channel isn't open, start a new stream // and retry sending the batch. debug ! ( error " " ) ; return Ok ( ( ) ) ; } } } // If the span source was closed, end the task. if let Err ( closed ) = collect { debug ! ( " " ) ; return Err ( closed ) ; } } }
Accumulate spans and send them on the export stream.
Accumulate spans and send them on the export stream .
tx
&mpsc::Sender<ExportTraceServiceRequest>
spans
&mut S
accum
&mut Vec<Span>
node
&mut Option<Node>
tx
&mpsc::Sender<ExportTraceServiceRequest>
spans
&mut S
accum
&mut Vec<Span>
node
&mut Option<Node>