Back-pressure aware operators

[Atry]

Currently merge_all_threads can result in out-of-memory, when concurrent is not Infinity and the source submits items faster than the mapper handles items.

I wonder if we could change the type of concurrent from usize to a struct:

pub struct ConcurrentConfig<F, Item> {
  usize max_number_of_inner_observers;
  usize max_buffer_size;
  F backpressure_handler;
}

impl<F, Item> ConcurrentConfig<F, Item>
where
    F: FnMut(Item) -> () {
}

[M-Adoo]

I'd be glad to accept Back-pressure support for merge_all_threads op. To implement it, I would prefer to keep the concurrent type and add a with_backpressure method to config the callback for MergeAllOp/MergeAllOpThreads. So, we can keep the API compatible and make the Back-pressure be an optional feature.

[Atry]

How do you think about merge_scan, like https://rxjs.dev/api/index/function/mergeScan ?

I think merge_scan is useful to be used as a general producer/consumer operator.

Would you like to review a merge_scan_with_backpressure PR?

[M-Adoo]

Would you like to review a merge_scan_with_backpressure PR?

Of course.