pub fn consolidate_stream(&self) -> Self {

        use timely::dataflow::channels::pact::Pipeline;
        use timely::dataflow::operators::Operator;
        use crate::collection::AsCollection;

        self.inner
            .unary::<ConsolidatingContainerBuilder<_>, _, _, _>(Pipeline, "ConsolidateStream", |_cap, _info| {

                move |input, output| {
                    input.for_each(|time, data| {
                        output.session_with_builder(&time).give_iterator(data.drain(..));
                    })
                }
            })
            .as_collection()
    }
Aggregates the weights of equal records.

Unlike `consolidate`, this method does not exchange data and does not
ensure that at most one copy of each `(data, time)` pair exists in the
results. Instead, it acts on each batch of data and collapses equivalent
`(data, time)` pairs found therein, suppressing any that accumulate to
zero.

# Examples

```
use differential_dataflow::input::Input;

::timely::example(|scope| {

let x = scope.new_collection_from(1 .. 10u32).1;

// nothing to assert, as no particular guarantees.
x.negate()
.concat(&x)
.consolidate_stream();
});
```
differential-dataflow/src/operators/consolidate.rs