when.rs

fn when<
  I: 'static + Send,
  F: 'static + Send + Future<Item = I, Error = ()>,
  R: 'static + Send,
  K: 'static + Send + Fn(&[I]) -> Option<R>,
>(
  futures: Vec<F>,
  k: K,
) -> Box<dyn Future<Item = R, Error = ()> + Send> {
  fn when_rec<
    I: 'static + Send,
    R: 'static + Send,
    K: 'static + Send + Fn(&[I]) -> Option<R>,
  >(
    stream: Box<dyn Stream<Item = I, Error = ()> + Send>,
    mut acc: Vec<I>,
    k: K,
  ) -> Box<dyn Future<Item = R, Error = ()> + Send> {
    if let Some(result) = k(&acc) {
      Box::new(ok(result))
    } else {
      Box::new(stream.into_future().then(|result| match result {
        Ok((x, s)) => {
          if let Some(r) = x {
            acc.push(r);
            when_rec(s, acc, k)
          } else {
            Box::new(err(()))
          }
        }
        Err((_, s)) => when_rec(s, acc, k),
      }))
    }
  }

  when_rec(
    futures
      .into_iter()
      .map(|future| future.into_stream())
      .fold(Box::new(stream::empty()), |acc, x| Box::new(acc.select(x))),
    Vec::new(),
    k,
  )
}