4.x: possible new operators based on RxJava's feature set

I've collected the operators and overloads of theirs which are available in RxJava but (likely) not in Rx.NET yet to enable discussions about feature synchronization.

### Operators

Operator in RxJava | Description
-----------|--------------
`combineLatestDelayError` | Keep combining sources as long as possible, then signal the aggregate exception at the end.
`concatDelayError` | Keep concatenating and signal the aggregate exception at the end.
`concatEager` | Run many or all source observables concurrently, but provide the items from the sources in their original order.
`concatEagerDelayError` | Run many or all source observables concurrently, provide items in order and signal the aggregate exception at the end.
`error(Callable)` | Signal an onError with the generated exception per observer.
`fromCallable` | Invoke a function to generate a single item & completion per observer.
`generate` | Generate events in a callback fashion with an observer-abstraction and per observer state.
`intervalRange` | Produce a range of numbers with an initial delay for the first followed by a periodic signal of the rest.
`mergeDelayError` | Merges some or all sources into one sequence and signals an aggregate exception when all of them terminated.
`switchOnNextDelayError` | Switch to newer sources, signal the aggregate exception at the end.
`zip(delayError)` | Zips sources into one sequence and signals the error after the last row could be combined.
`blockingIterable` | Turn a sequence into a blocking enumerable sequence.
`blockingSubscribe` | Observe on the caller thread in a blocking fashion.
`cache` | Consume the upstream once, cache and replay all items to observers.
`collect` | Aggregate items into a single, custom container per observer with the help of a collector action and signal the container as a single result.
`compose` | Applies a transformation function from observable to observable when the sequence is constructed.
`concatMap` | Maps the upstream items into observables and concatenates them in order; an in-sequence version of `concat`. Can be implemented more efficiently than `concat(map(source, f))`.
`concatMapDelayError` | Maps the upstream items into observables and concatenates them in order, signaling an aggregate exception at the end.
`concatMapEager` | Maps the upstream items into observables, runs some or all of them eagerly but emits their items in order; an in-sequence version of `concatEager`.
`concatMapEagerDelayError` | Maps the upstream items into observables, runs some or all of them eagerly but emits their items in order while signaling an aggregate exception at the end.
`concatMapIterable` | Maps the upstream items into enumerations and emits their items in order; more efficient than `concatMap`ping an `Iterable` turned into an observable.
`doAfterNext` | Call an action with the current item after it has been emitted to the downstream but before the next value from upstream.
`doFinally` | Perform an action exactly once per observer when it either completes normally, with an error or gets disposed.
`doOnDispose` | Perform an action when the sequence gets disposed.
`doOnSubscribe` | Perform an action when the subscription happens but before items start flowing.
`flatMap(delayError)` | Maps the upstream items into observables, merges them into one sequence and signals the aggregate error at the end. Can be implemented more efficiently than `merge(map(source, f))`.
`forEachWhile` | Consumes the sequence via a predicate that when returns false, the sequence is disposed.
`reduceWith` | Start with a per observer initial value as the accumulator and apply a function that takes that accumulator, the upstream item and returns a new accumulator value, then emit the last accumulator value as a single result.
`repeat(BooleanSupplier)` | Repeatedly subscribe to a source if the boolean predicate at each completion allows that.
`repeatWhen` | Repeatedly subscribe to a source when after the current completion, a secondary per observer sequence signals an item.
`retry(Predicate)` | Retry a failed sequence if the predicate returns true for the exception
`retry(BiPredicate)` | Retry a failed sequence if the predicate returns true for the exception and the total retry count so far.
`retryUntil(BooleanSupplier)` | Retry a failed sequence if the boolean predicate allows it.
`retryWhen` | Retry a a failed sequence when the secondary per observer sequence signals an item in response to the current failure.
`sample(emitLast)` | Sample the source sequence and optionally emit the very last unsampled item when the sequence completes before the sampler signals again.
`scanWith` | Starts with a per observer generated initial accumulator value and emits a combination of this accumulator and the upstream item which also becomes the new accumulator value.
`switchIfEmpty` | If the main sequence turns out to be empty, a fallback sequence is used to stream events instead.
`switchMap` | Maps the upstream item into an observable and keeps switching to newer ones as those come along, disposing the previous ones. Can be implemented more efficiently than `switchOnNext(map(source, f))`.
`switchMapDelayError` | Maps the upstream item into an observable and keeps switching to newer ones as those come along, disposing the previous ones and signalling the aggregated exception at the end.
`takeUntil(Predicate)` | Emits items from the upstream and after each, it runs a predicate to determine if the sequence should stop just after the current item.
`observeOn(delayError)` | Observe items on another thread and optionally keep the total event order or have the errors cut ahead.
`unsubscribeOn` | When the sequence is diposed, the dispose call traveling upstream is executed on a scheduler.
`withLatestFrom(Observable...)` | Combines the items of the main source with the latest items of multiple other sources.
`autoConnect` | Automatically connects once to a `ConnectableObservable` after the specified number of observers have subscribed.
`refCount(n, time)` | Connect to a `ConnectableObservable` after the specified number of observers have subscribed and/or disconnect after a grace period (under review).

### Subjects

RxJava has one additional subject type: `UnicastSubject`. It buffers items until one observer subscribes some time later and then drains the buffer into the observer.

In addition, `Subject`s in RxJava have additional state-peeking methods available to all variants:

Method | Description
-----------|--------------
`hasObservers()` | (all subjects) `true` if there are currently any observers subscribed to the subject.
`hasComplete()` | (all subjects) `true` if the subject terminated normally.
`hasThrowable()` | (all subjects) `true` if the subject terminated exceptionally.
`getThrowable()` | (all subjects) Returns the exception if the subject terminated exceptionally.
`toSerialized()` | Makes the `onXXX` methods threadsafe to call from any thread.
`hasValue()` | `true` if a `BehaviorSubject` or `ReplaySubject` have items in their buffer.
`getValue()` | Returns the current item of a `BehaviorSubject`.
`getValues()` | Returns the currently buffered items in a `ReplaySubject`.

I'm not sure how interface evolution works in C#, so assuming it would break `System.Reactive.Interfaces` users, the methods could be added to a new interface derived from `ISubject` or just onto the various Rx `Subject` types.