Function write

recon/src/protocol.rs:324–427

(
    mut sync_id: Option<String>,
    sink: S,
    mut to_writer_rx: mpsc::Receiver<ToWriter<M>>,
    init: Option<M>,
    finish: Option<M>,
    wip_limit: usize,
    metrics: Metrics,
)

#[instrument(skip_all, fields(sync_id))]
async fn write<M, S>(
    mut sync_id: Option<String>,
    sink: S,
    mut to_writer_rx: mpsc::Receiver<ToWriter<M>>,
    init: Option<M>,
    finish: Option<M>,
    wip_limit: usize,
    metrics: Metrics,
) -> Result<()>
where
    S: Sink<ReconMessage<M>, Error = anyhow::Error>,
    MessageLabels: for<'a> From<&'a M>,
    M: std::fmt::Debug,
{
    if let Some(sync_id) = &sync_id {
        // Record sync_id on the tracing span
        tracing::Span::current().record("sync_id", sync_id);
    }
    pin_mut!(sink);

    // Use a stack so we get depth first traversal
    let mut message_stack = Vec::new();
    // Track in progress
    let mut in_progress = 0;

    if let Some(init) = init {
        trace!(?init, "sending init");
        metrics.record(&MessageSent(&init));
        sink.send(ReconMessage::new(sync_id.clone(), init)).await?;
    }

    // Write messages to the remote until there are no more messages to send.
    while let Some(action) = to_writer_rx.recv().await {
        metrics.record(&ProtocolWriteLoop);
        match action {
            ToWriter::SyncId(remote_sync_id) => {
                // Record sync_id on the tracing span
                tracing::Span::current().record("sync_id", &remote_sync_id);
                sync_id = Some(remote_sync_id);
            }
            ToWriter::SendAll(messages) => {
                sink.send_all(&mut messages.map(|msg| {
                    msg.map(|message| {
                        metrics.record(&MessageSent(&message));
                        trace!(?message, "sending message");
                        ReconMessage::new(sync_id.clone(), message)
                    })
                }))
                .await?;
            }
            ToWriter::WIPCompleted(count) => {
                in_progress -= count;
            }
            ToWriter::SendWIPLimited(mut messages) => {
                // Send all message that fit within the work in progress limit.
                if in_progress < wip_limit {
                    let capacity = wip_limit - in_progress;
                    let mut sent = false;