#nullable enable

using System.Reactive;

using System.Reactive.Linq;

using System.Reactive.Subjects;

using System.Threading.Tasks;

using System;

​

var subjA = new Subject<int>();

var subjB = new Subject<Unit>();

​

subjA.Select(n =>

        Observable.Return(n).Delay(TimeSpan.FromMilliseconds(200))

        .TakeUntil(subjB))

    .Switch()

    .Subscribe(s => Console.WriteLine(s));

​

await Task.WhenAll(EmitA(), EmitB(), Task.Delay(2000));

​

async Task EmitA()

{

    subjA!.OnNext(1);

    await Task.Delay(100);

    subjA!.OnNext(2);

    await Task.Delay(500);

    subjA!.OnNext(3);

}

​

async Task EmitB()

{

    await Task.Delay(150);

    subjB!.OnNext(Unit.Default);

}