fn send_records<I, Out, Err>(
    iter: I,
    tx: std::sync::mpsc::SyncSender<Result<Out, anyhow::Error>>,
    activator: Option<SyncActivator>,
) where
    I: IntoIterator<Item = Result<Out, Err>>,
    Err: Into<anyhow::Error>,
{
    let mut records = 0;
    for record in iter {
        records += 1;
        let record = record.map_err(Into::into);
        // TODO: each call to `send` allocates and performs some
        // atomic work; we could aim to batch up transmissions.
        if tx.send(record).is_err() {
            // The receiver went away, probably due to `DROP SOURCE`
            break;
        }
        // TODO: this is very spammy for the timely activator; it
        // appends an address to a list for each activation which
        // looks like it will be per-record in this case.
        if let Some(activator) = &activator {
            activator.activate().expect("activation failed");
        }
    }
    log::trace!("sent {} records to reader", records);
}
/// Sends a sequence of records and activates a timely operator for each.
Sends a sequence of records and activates a timely operator for each.
Sends a sequence of records and activates a timely operator for each .
fn send_records<I, Out, Err>(
    iter: I,
    tx: std::sync::mpsc::SyncSender<Result<Out, anyhow::Error>>,
    activator: Option<SyncActivator>,
) where
    I: IntoIterator<Item = Result<Out, Err>>,
    Err: Into<anyhow::Error>,
{
    let mut records = 0;
    for record in iter {
        records += 1;
        let record = record.map_err(Into::into);
        // TODO: each call to `send` allocates and performs some
        // atomic work; we could aim to batch up transmissions.
        if tx.send(record).is_err() {
            // The receiver went away, probably due to `DROP SOURCE`
            break;
        }
        // TODO: this is very spammy for the timely activator; it
        // appends an address to a list for each activation which
        // looks like it will be per-record in this case.
        if let Some(activator) = &activator {
            activator.activate().expect("activation failed");
        }
    }
    log::trace!("sent {} records to reader", records);
}
fn send_records < I , Out , Err > ( iter : I , tx : std :: sync :: mpsc :: SyncSender < Result < Out , anyhow :: Error > > , activator : Option < SyncActivator > , ) where I : IntoIterator < Item = Result < Out , Err > > , Err : Into < anyhow :: Error > , { let mut records = 0 ; for record in iter { records += 1 ; let record = record . map_err ( Into :: into ) ; // TODO: each call to `send` allocates and performs some // atomic work; we could aim to batch up transmissions. if tx . send ( record ) . is_err ( ) { // The receiver went away, probably due to `DROP SOURCE` break ; } // TODO: this is very spammy for the timely activator; it // appends an address to a list for each activation which // looks like it will be per-record in this case. if let Some ( activator ) = & activator { activator . activate ( ) . expect ( " " ) ; } } log :: trace ! ( " " records ) ; }
Sends a sequence of records and activates a timely operator for each.
Sends a sequence of records and activates a timely operator for each .
iter
I
tx
std::sync::mpsc::SyncSender<Result<Out, anyhow::Error>>
activator
Option<SyncActivator>
iter
I
tx
std::sync::mpsc::SyncSender<Result<Out, anyhow::Error>>
activator
Option<SyncActivator>