Merge with Dynamic Maximum Concurrency

[mattpodwysocki]

Copied from https://rx.codeplex.com/workitem/43

Rx defines an overload of the Merge operator with a maxConcurrent parameter that is useful for controlling the level of concurrency through the number of active subscriptions to inner observables; however, the maxConcurrent parameter is static and in some cases it's impossible to write a query that changes the parameter dynamically. Unsubscribing and re-subscribing to Merge won't work for a hot source due to the fact that Merge uses an internal buffer. Unsubscribing will lose all of the inner observables that have already been buffered but aren't yet active, and since the source is hot, this causes permanent data loss.

Consider adding a Merge overload such as the one shown in the following discussion:

http://social.msdn.microsoft.com/Forums/en-US/62743ffd-befd-474a-8f0a-19dbaec7a926/design-question-throttling-hits-on-external-systems

It has the following signature:

IObservable<TSource> Merge<TSource>(this IObservable<IObservable<TSource>> source, IObservable<int> maxConcurrent)

[RxDave]

Some more details:
http://davesexton.com/blog/post/Reactive-Reactive-Read-all-about-it.aspx

[akarnokd]

What should the operator do when a smaller maxConcurrent value is signalled after a larger one? Stop consuming the excess active ones or let them finish and don't subscribe to new ones until the active source count goes below the current limit?

[RxDave]

Here's the comment from the dynamic Merge operator in the linked discussion:

// When the maximum concurrency value is raised, try to dequeue and subscribe to as many 
// available observables as needed to fill the new bounds.
//
// When the maximum concurrency value is lowered, do not dispose of any active observables.
// 
// Lowering the maximum concurrency value does not imply cancelation; it only implies that 
// future inner observables will be deferred until the number of active observables drops
// sufficiently.  The number of active observables may currently be higher than the current 
// maximum concurrency value, though prematurely disposing of active observables may result 
// in irreversible data loss.

[akarnokd]

This requires deep changes and possibly done for a lot of merge variants. This is a really old issue, do you still want this or would an implementation with existing building blocks suffice?