pub fn flat_map_batches<I, L>(stream: &Stream<G, Tr::Batch>, mut logic: L) -> Collection<G, I::Item, Tr::Diff>
    where
        I: IntoIterator,
        I::Item: Data,
        L: FnMut(Tr::Key<'_>, Tr::Val<'_>) -> I+'static,
    {
        stream.unary(Pipeline, "AsCollection", move |_,_| move |input, output| {
            input.for_each(|time, data| {
                let mut session = output.session(&time);
                for wrapper in data.iter() {
                    let batch = &wrapper;
                    let mut cursor = batch.cursor();
                    while let Some(key) = cursor.get_key(batch) {
                        while let Some(val) = cursor.get_val(batch) {
                            for datum in logic(key, val) {
                                cursor.map_times(batch, |time, diff| {
                                    session.give((datum.clone(), time.into_owned(), diff.into_owned()));
                                });
                            }
                            cursor.step_val(batch);
                        }
                        cursor.step_key(batch);
                    }
                }
            });
        })
        .as_collection()
    }
Extracts elements from a stream of batches as a collection.

The supplied logic may produce an iterator over output values, allowing either
filtering or flat mapping as part of the extraction.

This method exists for streams of batches without the corresponding arrangement.
If you have the arrangement, its `flat_map_ref` method is equivalent to this.
differential-dataflow/src/operators/arrange/arrangement.rs