Method flat_map_fallible

src/timely-util/src/operator.rs:272–303

(
        self,
        name: &str,
        mut logic: L,
    )

    // so the simple `|d1| logic(d1)` closure is load-bearing
    #[allow(clippy::redundant_closure)]
    fn flat_map_fallible<DCB, ECB, D2, E, I, L>(
        self,
        name: &str,
        mut logic: L,
    ) -> (
        Stream<'scope, T, DCB::Container>,
        Stream<'scope, T, ECB::Container>,
    )
    where
        DCB: ContainerBuilder + PushInto<D2>,
        ECB: ContainerBuilder + PushInto<E>,
        I: IntoIterator<Item = Result<D2, E>>,
        L: for<'a> FnMut(C1::Item<'a>) -> I + 'static,
    {
        self.unary_fallible::<DCB, ECB, _, _>(Pipeline, name, move |_, _| {
            Box::new(move |input, ok_output, err_output| {
                input.for_each_time(|time, data| {
                    let mut ok_session = ok_output.session_with_builder(&time);
                    let mut err_session = err_output.session_with_builder(&time);
                    for r in data
                        .flat_map(DrainContainer::drain)
                        .flat_map(|d1| logic(d1))
                    {
                        match r {
                            Ok(d2) => ok_session.give(d2),
                            Err(e) => err_session.give(e),
                        }
                    }
                })
            })
        })
    }

    fn expire_stream_at(self, name: &str, expiration: T) -> Stream<'scope, T, C1> {
        let name = format!("expire_stream_at({name})");