// Licensed to the .NET Foundation under one or more agreements. // The .NET Foundation licenses this file to you under the Apache 2.0 License. // See the LICENSE file in the project root for more information. using System; using System.Collections.Generic; using System.Linq; using System.Text; using System.Threading.Tasks; using System.Reactive; using System.Reactive.Concurrency; using System.Reactive.Linq; using Microsoft.Reactive.Testing; using Xunit; using ReactiveTests.Dummies; using System.Reflection; using System.Threading; using System.Reactive.Disposables; using System.Reactive.Subjects; namespace ReactiveTests.Tests { public class AmbTest : ReactiveTest { [Fact] public void Amb_ArgumentChecking() { ReactiveAssert.Throws(() => Observable.Amb((IObservable[])null)); ReactiveAssert.Throws(() => Observable.Amb((IEnumerable>)null)); ReactiveAssert.Throws(() => Observable.Amb(null, DummyObservable.Instance)); ReactiveAssert.Throws(() => Observable.Amb(DummyObservable.Instance, null)); } [Fact] public void Amb_Never2() { var scheduler = new TestScheduler(); var l = scheduler.CreateHotObservable( OnNext(150, 1) ); var r = scheduler.CreateHotObservable( OnNext(150, 1) ); var res = scheduler.Start(() => l.Amb(r) ); res.Messages.AssertEqual( ); l.Subscriptions.AssertEqual( Subscribe(200, 1000) ); r.Subscriptions.AssertEqual( Subscribe(200, 1000) ); } [Fact] public void Amb_Never3() { var scheduler = new TestScheduler(); var n1 = scheduler.CreateHotObservable( OnNext(150, 1) ); var n2 = scheduler.CreateHotObservable( OnNext(150, 1) ); var n3 = scheduler.CreateHotObservable( OnNext(150, 1) ); var res = scheduler.Start(() => new[] { n1, n2, n3 }.Amb() ); res.Messages.AssertEqual( ); n1.Subscriptions.AssertEqual( Subscribe(200, 1000) ); n2.Subscriptions.AssertEqual( Subscribe(200, 1000) ); n3.Subscriptions.AssertEqual( Subscribe(200, 1000) ); } [Fact] public void Amb_Never3_Params() { var scheduler = new TestScheduler(); var n1 = scheduler.CreateHotObservable( OnNext(150, 1) ); var n2 = scheduler.CreateHotObservable( OnNext(150, 1) ); var n3 = scheduler.CreateHotObservable( OnNext(150, 1) ); var res = scheduler.Start(() => Observable.Amb(n1, n2, n3) ); res.Messages.AssertEqual( ); n1.Subscriptions.AssertEqual( Subscribe(200, 1000) ); n2.Subscriptions.AssertEqual( Subscribe(200, 1000) ); n3.Subscriptions.AssertEqual( Subscribe(200, 1000) ); } [Fact] public void Amb_NeverEmpty() { var scheduler = new TestScheduler(); var n = scheduler.CreateHotObservable( OnNext(150, 1) ); var e = scheduler.CreateHotObservable( OnNext(150, 1), OnCompleted(225) ); var res = scheduler.Start(() => n.Amb(e) ); res.Messages.AssertEqual( OnCompleted(225) ); n.Subscriptions.AssertEqual( Subscribe(200, 225) ); e.Subscriptions.AssertEqual( Subscribe(200, 225) ); } [Fact] public void Amb_EmptyNever() { var scheduler = new TestScheduler(); var n = scheduler.CreateHotObservable( OnNext(150, 1) ); var e = scheduler.CreateHotObservable( OnNext(150, 1), OnCompleted(225) ); var res = scheduler.Start(() => e.Amb(n) ); res.Messages.AssertEqual( OnCompleted(225) ); n.Subscriptions.AssertEqual( Subscribe(200, 225) ); e.Subscriptions.AssertEqual( Subscribe(200, 225) ); } [Fact] public void Amb_RegularShouldDisposeLoser() { var scheduler = new TestScheduler(); var o1 = scheduler.CreateHotObservable( OnNext(150, 1), OnNext(210, 2), OnCompleted(240) ); var o2 = scheduler.CreateHotObservable( OnNext(150, 1), OnNext(220, 3), OnCompleted(250) ); var res = scheduler.Start(() => o1.Amb(o2) ); res.Messages.AssertEqual( OnNext(210, 2), OnCompleted(240) ); o1.Subscriptions.AssertEqual( Subscribe(200, 240) ); o2.Subscriptions.AssertEqual( Subscribe(200, 210) ); } [Fact] public void Amb_WinnerThrows() { var scheduler = new TestScheduler(); var ex = new Exception(); var o1 = scheduler.CreateHotObservable( OnNext(150, 1), OnNext(210, 2), OnError(220, ex) ); var o2 = scheduler.CreateHotObservable( OnNext(150, 1), OnNext(220, 3), OnCompleted(250) ); var res = scheduler.Start(() => o1.Amb(o2) ); res.Messages.AssertEqual( OnNext(210, 2), OnError(220, ex) ); o1.Subscriptions.AssertEqual( Subscribe(200, 220) ); o2.Subscriptions.AssertEqual( Subscribe(200, 210) ); } [Fact] public void Amb_LoserThrows() { var scheduler = new TestScheduler(); var ex = new Exception(); var o1 = scheduler.CreateHotObservable( OnNext(150, 1), OnNext(220, 2), OnError(230, ex) ); var o2 = scheduler.CreateHotObservable( OnNext(150, 1), OnNext(210, 3), OnCompleted(250) ); var res = scheduler.Start(() => o1.Amb(o2) ); res.Messages.AssertEqual( OnNext(210, 3), OnCompleted(250) ); o1.Subscriptions.AssertEqual( Subscribe(200, 210) ); o2.Subscriptions.AssertEqual( Subscribe(200, 250) ); } [Fact] public void Amb_ThrowsBeforeElectionLeft() { var scheduler = new TestScheduler(); var ex = new Exception(); var o1 = scheduler.CreateHotObservable( OnNext(150, 1), OnError(210, ex) ); var o2 = scheduler.CreateHotObservable( OnNext(150, 1), OnNext(220, 3), OnCompleted(250) ); var res = scheduler.Start(() => o1.Amb(o2) ); res.Messages.AssertEqual( OnError(210, ex) ); o1.Subscriptions.AssertEqual( Subscribe(200, 210) ); o2.Subscriptions.AssertEqual( Subscribe(200, 210) ); } [Fact] public void Amb_ThrowsBeforeElectionRight() { var scheduler = new TestScheduler(); var ex = new Exception(); var o1 = scheduler.CreateHotObservable( OnNext(150, 1), OnNext(220, 3), OnCompleted(250) ); var o2 = scheduler.CreateHotObservable( OnNext(150, 1), OnError(210, ex) ); var res = scheduler.Start(() => o1.Amb(o2) ); res.Messages.AssertEqual( OnError(210, ex) ); o1.Subscriptions.AssertEqual( Subscribe(200, 210) ); o2.Subscriptions.AssertEqual( Subscribe(200, 210) ); } } }