pub fn build<G, BS, D, T, R>(
        stream: &Stream<G, (D, T, R)>,
        sink_hash: u64,
        updates_sink: Weak<RefCell<BS>>,
        progress_sink: Weak<RefCell<BS>>,
    ) where
        G: Scope<Timestamp = T>,
        BS: Writer<Message<D,T,R>> + 'static,
        D: ExchangeData + Hash + Serialize + for<'a> Deserialize<'a>,
        T: ExchangeData + Hash + Serialize + for<'a> Deserialize<'a> + Timestamp + Lattice,
        R: ExchangeData + Hash + Serialize + for<'a> Deserialize<'a>,
    {
        // First we record the updates that stream in.
        // We can simply record all updates, under the presumption that the have been consolidated
        // and so any record we see is in fact guaranteed to happen.
        let mut builder = OperatorBuilder::new("UpdatesWriter".to_owned(), stream.scope());
        let reactivator = stream.scope().activator_for(builder.operator_info().address);
        let mut input = builder.new_input(stream, Pipeline);
        let (mut updates_out, updates) = builder.new_output();

        builder.build_reschedule(
            move |_capability| {
                let mut timestamps = <ChangeBatch<_>>::new();
                let mut send_queue = std::collections::VecDeque::new();
                move |_frontiers| {
                    let mut output = updates_out.activate();

                    // We want to drain inputs always...
                    input.for_each(|capability, updates| {
                        // Write each update out, and record the timestamp.
                        for (_data, time, _diff) in updates.iter() {
                            timestamps.update(time.clone(), 1);
                        }

                        // Now record the update to the writer.
                        send_queue.push_back(Message::Updates(std::mem::take(updates)));

                        // Transmit timestamp counts downstream.
                        output
                            .session(&capability)
                            .give_iterator(timestamps.drain());
                    });

                    // Drain whatever we can from the queue of bytes to send.
                    // ... but needn't do anything more if our sink is closed.
                    if let Some(sink) = updates_sink.upgrade() {
                        let mut sink = sink.borrow_mut();
                        while let Some(message) = send_queue.front() {
                            if let Some(duration) = sink.poll(message) {
                                // Reschedule after `duration` and then bail.
                                reactivator.activate_after(duration);
                                return true;
                            } else {
                                send_queue.pop_front();
                            }
                        }
                        // Signal incompleteness if messages remain to be sent.
                        !sink.done() || !send_queue.is_empty()
                    } else {
                        // We have been terminated, but may still receive indefinite data.
                        send_queue.clear();
                        // Signal that there are no outstanding writes.
                        false
                    }
                }
            },
        );

        // We use a lower-level builder here to get access to the operator address, for rescheduling.
        let mut builder = OperatorBuilder::new("ProgressWriter".to_owned(), stream.scope());
        let reactivator = stream.scope().activator_for(builder.operator_info().address);
        let mut input = builder.new_input(&updates, Exchange::new(move |_| sink_hash));
        let should_write = stream.scope().index() == (sink_hash as usize) % stream.scope().peers();

        // We now record the numbers of updates at each timestamp between lower and upper bounds.
        // Track the advancing frontier, to know when to produce utterances.
        let mut frontier = Antichain::from_elem(T::minimum());
        // Track accumulated counts for timestamps.
        let mut timestamps = <ChangeBatch<_>>::new();
        // Stash for serialized data yet to send.
        let mut send_queue = std::collections::VecDeque::new();
        let mut retain = Vec::new();

        builder.build_reschedule(|_capabilities| {
            move |frontiers| {
                let mut input = FrontieredInputHandle::new(&mut input, &frontiers[0]);

                // We want to drain inputs no matter what.
                // We could do this after the next step, as we are certain these timestamps will
                // not be part of a closed frontier (as they have not yet been read). This has the
                // potential to make things speedier as we scan less and keep a smaller footprint.
                input.for_each(|_capability, counts| {
                    timestamps.extend(counts.iter().cloned());
                });

                if should_write {
                    if let Some(sink) = progress_sink.upgrade() {
                        let mut sink = sink.borrow_mut();

                        // If our frontier advances strictly, we have the opportunity to issue a progress statement.
                        if <_ as PartialOrder>::less_than(
                            &frontier.borrow(),
                            &input.frontier.frontier(),
                        ) {
                            let new_frontier = input.frontier.frontier();

                            // Extract the timestamp counts to announce.
                            let mut announce = Vec::new();
                            for (time, count) in timestamps.drain() {
                                if !new_frontier.less_equal(&time) {
                                    announce.push((time, count as usize));
                                } else {
                                    retain.push((time, count));
                                }
                            }
                            timestamps.extend(retain.drain(..));

                            // Announce the lower bound, upper bound, and timestamp counts.
                            let progress = Progress {
                                lower: frontier.elements().to_vec(),
                                upper: new_frontier.to_vec(),
                                counts: announce,
                            };
                            send_queue.push_back(Message::Progress(progress));

                            // Advance our frontier to track our progress utterance.
                            frontier = input.frontier.frontier().to_owned();

                            while let Some(message) = send_queue.front() {
                                if let Some(duration) = sink.poll(message) {
                                    // Reschedule after `duration` and then bail.
                                    reactivator.activate_after(duration);
                                    // Signal that work remains to be done.
                                    return true;
                                } else {
                                    send_queue.pop_front();
                                }
                            }
                        }
                        // Signal incompleteness if messages remain to be sent.
                        !sink.done() || !send_queue.is_empty()
                    } else {
                        timestamps.clear();
                        send_queue.clear();
                        // Signal that there are no outstanding writes.
                        false
                    }
                } else { false }
            }
        });
    }
Constructs a sink, for recording the updates in `stream`.

It is crucial that `stream` has been consolidated before this method, which
will *not* perform the consolidation on the stream's behalf. If this is not
performed before calling the method, the recorded output may not be correctly
reconstructed by readers.
differential-dataflow/src/capture.rs