pub fn build<G, B, I, D, T, R>(
        scope: G,
        source_builder: B,
    ) -> (Box<dyn std::any::Any + Send + Sync>, Stream<G, (D, T, R)>)
    where
        G: Scope<Timestamp = T>,
        B: FnOnce(SyncActivator) -> I,
        I: Iterator<Item = Message<D, T, R>> + 'static,
        D: ExchangeData + Hash,
        T: ExchangeData + Hash + Timestamp + Lattice,
        R: ExchangeData + Hash,
    {
        // Read messages are either updates or progress messages.
        // Each may contain duplicates, and we must take care to deduplicate information before introducing it to an accumulation.
        // This includes both emitting updates, and setting expectations for update counts.
        //
        // Updates need to be deduplicated by (data, time), and we should exchange them by such.
        // Progress needs to be deduplicated by time, and we should exchange them by such.
        //
        // The first cut of this is a dataflow graph that looks like (flowing downward)
        //
        // 1. MESSAGES:
        //      Reads `Message` stream; maintains capabilities.
        //      Sends `Updates` to UPDATES stage by hash((data, time, diff)).
        //      Sends `Progress` to PROGRESS stage by hash(time), each with lower, upper bounds.
        //      Shares capabilities with downstream operator.
        // 2. UPDATES:
        //      Maintains and deduplicates updates.
        //      Ships updates once frontier advances.
        //      Ships counts to PROGRESS stage, by hash(time).
        // 3. PROGRESS:
        //      Maintains outstanding message counts by time. Tracks frontiers.
        //      Tracks lower bounds of messages and progress frontier. Broadcasts changes to FEEDBACK stage
        // 4. FEEDBACK:
        //      Shares capabilities with MESSAGES; downgrades to track input from PROGRESS.
        //
        // Each of these stages can be arbitrarily data-parallel, and FEEDBACK *must* have the same parallelism as RAW.
        // Limitations: MESSAGES must broadcast lower and upper bounds to PROGRESS and PROGRESS must broadcast its changes
        // to FEEDBACK. This may mean that scaling up PROGRESS could introduce quadratic problems. Though, both of these
        // broadcast things are meant to be very reduced data.

        use crate::hashable::Hashable;
        use timely::dataflow::channels::pact::Exchange;
        use timely::dataflow::operators::generic::builder_rc::OperatorBuilder;
        use timely::progress::frontier::MutableAntichain;
        use timely::progress::ChangeBatch;

        // Some message distribution logic depends on the number of workers.
        let workers = scope.peers();

        let mut token = None;
        // Frontier owned by the FEEDBACK operator and consulted by the MESSAGES operators.
        let mut antichain = MutableAntichain::new();
        antichain.update_iter(Some((T::minimum(), workers as i64)));
        let shared_frontier = Rc::new(RefCell::new(antichain));
        let shared_frontier2 = shared_frontier.clone();

        // Step 1: The MESSAGES operator.
        let mut messages_op = OperatorBuilder::new("CDCV2_Messages".to_string(), scope.clone());
        let address = messages_op.operator_info().address;
        let activator = scope.sync_activator_for(address.to_vec());
        let activator2 = scope.activator_for(Rc::clone(&address));
        let drop_activator = DropActivator { activator: Arc::new(scope.sync_activator_for(address.to_vec())) };
        let mut source = source_builder(activator);
        let (mut updates_out, updates) = messages_op.new_output();
        let (mut progress_out, progress) = messages_op.new_output();
        messages_op.build(|capabilities| {

            // A Weak that communicates whether the returned token has been dropped.
            let drop_activator_weak = Arc::downgrade(&drop_activator.activator);

            token = Some(drop_activator);

            // Read messages from some source; shuffle them to UPDATES and PROGRESS; share capability with FEEDBACK.
            let mut updates_caps = CapabilitySet::from_elem(capabilities[0].clone());
            let mut progress_caps = CapabilitySet::from_elem(capabilities[1].clone());
            // Capture the shared frontier to read out frontier updates to apply.
            let local_frontier = shared_frontier.clone();
            //
            move |_frontiers| {
                // First check to ensure that we haven't been terminated by someone dropping our tokens.
                if drop_activator_weak.upgrade().is_none() {
                    // Give up our capabilities
                    updates_caps.downgrade(&[]);
                    progress_caps.downgrade(&[]);
                    // never continue, even if we are (erroneously) activated again.
                    return;
                }

                // Consult our shared frontier, and ensure capabilities are downgraded to it.
                let shared_frontier = local_frontier.borrow();
                updates_caps.downgrade(&shared_frontier.frontier());
                progress_caps.downgrade(&shared_frontier.frontier());

                // Next check to see if we have been terminated by the source being complete.
                if !updates_caps.is_empty() && !progress_caps.is_empty() {
                    let mut updates = updates_out.activate();
                    let mut progress = progress_out.activate();

                    // TODO(frank): this is a moment where multi-temporal capabilities need to be fixed up.
                    // Specifically, there may not be one capability valid for all updates.
                    let mut updates_session = updates.session(&updates_caps[0]);
                    let mut progress_session = progress.session(&progress_caps[0]);

                    // We presume the iterator will yield if appropriate.
                    for message in source.by_ref() {
                        match message {
                            Message::Updates(mut updates) => {
                                updates_session.give_container(&mut updates);
                            }
                            Message::Progress(progress) => {
                                // We must send a copy of each progress message to all workers,
                                // but we can partition the counts across workers by timestamp.
                                let mut to_worker = vec![Vec::new(); workers];
                                for (time, count) in progress.counts {
                                    to_worker[(time.hashed() as usize) % workers]
                                        .push((time, count));
                                }
                                for (worker, counts) in to_worker.into_iter().enumerate() {
                                    progress_session.give((
                                        worker,
                                        Progress {
                                            lower: progress.lower.clone(),
                                            upper: progress.upper.clone(),
                                            counts,
                                        },
                                    ));
                                }
                            }
                        }
                    }
                }
            }
        });

        // Step 2: The UPDATES operator.
        let mut updates_op = OperatorBuilder::new("CDCV2_Updates".to_string(), scope.clone());
        let mut input = updates_op.new_input(&updates, Exchange::new(|x: &(D, T, R)| x.hashed()));
        let (mut changes_out, changes) = updates_op.new_output();
        let (mut counts_out, counts) = updates_op.new_output();
        updates_op.build(move |_capability| {
            // Deduplicates updates, and ships novel updates and the counts for each time.
            // For simplicity, this operator ships updates as they are discovered to be new.
            // This has the defect that on load we may have two copies of the data (shipped,
            // and here for deduplication).
            //
            // Filters may be pushed ahead of this operator, but because of deduplication we
            // may not push projections ahead of this operator (at least, not without fields
            // that are known to form keys, and even then only carefully).
            let mut pending = std::collections::HashMap::new();
            let mut change_batch = ChangeBatch::<T>::new();
            move |frontiers| {
                // Thin out deduplication buffer.
                // This is the moment in a more advanced implementation where we might send
                // the data for the first time, maintaining only one copy of each update live
                // at a time in the system.
                pending.retain(|(_row, time), _diff| frontiers[0].less_equal(time));

                // Deduplicate newly received updates, sending new updates and timestamp counts.
                let mut changes = changes_out.activate();
                let mut counts = counts_out.activate();
                while let Some((capability, updates)) = input.next() {
                    let mut changes_session = changes.session(&capability);
                    let mut counts_session = counts.session(&capability);
                    for (data, time, diff) in updates.iter() {
                        if frontiers[0].less_equal(time) {
                            if let Some(prior) = pending.insert((data.clone(), time.clone()), diff.clone()) {
                                assert_eq!(&prior, diff);
                            } else {
                                change_batch.update(time.clone(), -1);
                                changes_session.give((data.clone(), time.clone(), diff.clone()));
                            }
                        }
                    }
                    if !change_batch.is_empty() {
                        counts_session.give_iterator(change_batch.drain());
                    }
                }
            }
        });

        // Step 3: The PROGRESS operator.
        let mut progress_op = OperatorBuilder::new("CDCV2_Progress".to_string(), scope.clone());
        let mut input = progress_op.new_input(
            &progress,
            Exchange::new(|x: &(usize, Progress<T>)| x.0 as u64),
        );
        let mut counts =
            progress_op.new_input(&counts, Exchange::new(|x: &(T, i64)| (x.0).hashed()));
        let (mut frontier_out, frontier) = progress_op.new_output();
        progress_op.build(move |_capability| {
            // Receive progress statements, deduplicated counts. Track lower frontier of both and broadcast changes.

            use timely::order::PartialOrder;
            use timely::progress::{frontier::AntichainRef, Antichain};

            let mut progress_queue = Vec::new();
            let mut progress_frontier = Antichain::from_elem(T::minimum());
            let mut updates_frontier = MutableAntichain::new();
            let mut reported_frontier = Antichain::from_elem(T::minimum());

            move |_frontiers| {
                let mut frontier = frontier_out.activate();

                // If the frontier changes we need a capability to express that.
                // Any capability should work; the downstream listener doesn't care.
                let mut capability: Option<Capability<T>> = None;

                // Drain all relevant update counts in to the mutable antichain tracking its frontier.
                while let Some((cap, counts)) = counts.next() {
                    updates_frontier.update_iter(counts.iter().cloned());
                    capability = Some(cap.retain());
                }
                // Drain all progress statements into the queue out of which we will work.
                while let Some((cap, progress)) = input.next() {
                    progress_queue.extend(progress.iter().map(|x| (x.1).clone()));
                    capability = Some(cap.retain());
                }

                // Extract and act on actionable progress messages.
                // A progress message is actionable if `self.progress_frontier` is beyond the message's lower bound.
                while let Some(position) = progress_queue.iter().position(|p| {
                    <_ as PartialOrder>::less_equal(
                        &AntichainRef::new(&p.lower),
                        &progress_frontier.borrow(),
                    )
                }) {
                    // Extract progress statement.
                    let mut progress = progress_queue.remove(position);
                    // Discard counts that have already been incorporated.
                    progress
                        .counts
                        .retain(|(time, _count)| progress_frontier.less_equal(time));
                    // Record any new reports of expected counts.
                    updates_frontier
                        .update_iter(progress.counts.drain(..).map(|(t, c)| (t, c as i64)));
                    // Extend self.progress_frontier by progress.upper.
                    let mut new_frontier = Antichain::new();
                    for time1 in progress.upper {
                        for time2 in progress_frontier.elements() {
                            new_frontier.insert(time1.join(time2));
                        }
                    }
                    progress_frontier = new_frontier;
                }

                // Determine if the lower bound of frontiers have advanced, and transmit updates if so.
                let mut lower_bound = progress_frontier.clone();
                lower_bound.extend(updates_frontier.frontier().iter().cloned());
                if lower_bound != reported_frontier {
                    let capability =
                        capability.expect("Changes occurred, without surfacing a capability");
                    let mut changes = ChangeBatch::new();
                    changes.extend(lower_bound.elements().iter().map(|t| (t.clone(), 1)));
                    changes.extend(reported_frontier.elements().iter().map(|t| (t.clone(), -1)));
                    let mut frontier_session = frontier.session(&capability);
                    for peer in 0..workers {
                        frontier_session.give((peer, changes.clone()));
                    }
                    reported_frontier = lower_bound.clone();
                }
            }
        });

        // Step 4: The FEEDBACK operator.
        let mut feedback_op = OperatorBuilder::new("CDCV2_Feedback".to_string(), scope.clone());
        let mut input = feedback_op.new_input(
            &frontier,
            Exchange::new(|x: &(usize, ChangeBatch<T>)| x.0 as u64),
        );
        feedback_op.build(move |_capability| {
            // Receive frontier changes and share the net result with MESSAGES.
            move |_frontiers| {
                let mut antichain = shared_frontier2.borrow_mut();
                let mut must_activate = false;
                while let Some((_cap, frontier_changes)) = input.next() {
                    for (_self, input_changes) in frontier_changes.iter() {
                        // Apply the updates, and observe if the lower bound has changed.
                        if antichain.update_iter(input_changes.unstable_internal_updates().iter().cloned()).next().is_some() {
                            must_activate = true;
                        }
                    }
                }
                // If the lower bound has changed, we must activate MESSAGES.
                if must_activate { activator2.activate(); }
            }
        });

        (Box::new(token.unwrap()), changes)
    }
Constructs a stream of updates from a source of messages.

The stream is built in the supplied `scope` and continues to run until
the returned `Box<Any>` token is dropped. The `source_builder` argument
is invoked with a `SyncActivator` that will re-activate the source.
differential-dataflow/src/capture.rs