Equivalent of RxJS switchMap in ReactiveX/Rx.NET

Question

In RxJS, there is a switchMap function. Is there an equivalent in ReactiveX/Rx.NET? I don't see one in the transforming documentation.

Cassy

Answer 1

Edit

switch is the equivalent. http://reactivex.io/documentation/operators/switch.html

In short, switchMap and switch will cancel any previous streams whereas flatMap will not.

Vetter

flatMap and switchMap are definitely not equivalent – Epizootic

switchMap does cancel previous streams, flatMap doesnt – Sabrina

Answer 2

There is no single SwitchMany in Rx.NET equivalent to switchMap in Rx.js. You need to use separate Select and Switch functions.

Observable.Interval(TimeSpan.FromMinutes(1))
    .Select(_ => Observable.Return(10))
    .Switch()

Documentation: https://msdn.microsoft.com/en-us/library/hh229197(v=vs.103).aspx

Craving

Answer 3

From http://reactivex.io/documentation/operators/switch.html

Switch operator subscribes to an Observable that emits Observables. Each time it observes one of these emitted Observables, the Observable returned by Switch unsubscribes from the previously-emitted Observable begins emitting items from the latest Observable.

As MorleyDev pointed, the .NET implementation is https://learn.microsoft.com/en-us/previous-versions/dotnet/reactive-extensions/hh229197(v=vs.103), so the equivalent of RxJS switchMap in Rx.NET is a combination of Switch and Select operators:

// RxJS
observableOfObservables.pipe(
    switchMap(next => transform(next))
    ...
)

// RX.Net
observableOfObservables
    .Switch()
    .Select(next => transform(next))
    ...

Redress

I don't think you have the order correct. The Switch happens on inner observables but you haven't got any until you call Select. The above answer also has them the other way round. – Balaton

Answer 4

I created some C# extensions similar to the switchMap, concatMap and mergeMap:

public static class ReactiveExtensions
{
    public static IObservable<TOutput> SwitchSelect<TInput, TOutput>(this IObservable<TInput> source, Func<TInput, IObservable<TOutput>> selector) => source
        .Select(selector)
        .Switch();

    public static IObservable<TOutput> ConcatSelect<TInput, TOutput>(this IObservable<TInput> source, Func<TInput, IObservable<TOutput>> selector) => source
         .Select(selector)
         .Concat();

    public static IObservable<TOutput> MergeSelect<TInput, TOutput>(this IObservable<TInput> source, Func<TInput, IObservable<TOutput>> selector) => source
        .Select(selector)
        .Merge();
}

Then you can use it like:

Observable
   .Timer(someTimer)
   .SwitchSelect(_ => Observable.FromAsync(() => foo.RunAsyncMethod()))
   .Subscribe();

Zane