Permalink
Find file
Fetching contributors…
Cannot retrieve contributors at this time
138 lines (107 sloc) 3.92 KB
using System.Reactive.Linq;
using System.Reactive.Subjects;
using Xunit;
using System;
using System.Linq;
using System.Collections.Generic;
using ReactiveUI;
using System.IO;
using System.Text;
using ReactiveUI.Testing;
using ReactiveUI.Tests;
using System.Threading;
using Microsoft.Reactive.Testing;
using System.Threading.Tasks;
namespace ReactiveUI.Tests
{
public class MessageBusTest
{
[Fact]
public void MessageBusSmokeTest()
{
var input = new[] {1, 2, 3, 4};
var result = (new TestScheduler()).With(sched => {
var source = new Subject<int>();
var fixture = new MessageBus();
fixture.RegisterMessageSource(source, "Test");
Assert.False(fixture.IsRegistered(typeof (int)));
Assert.False(fixture.IsRegistered(typeof (int), "Foo"));
var output = fixture.Listen<int>("Test").CreateCollection();
input.Run(source.OnNext);
sched.Start();
return output;
});
input.AssertAreEqual(result);
}
[Fact]
public void ExplicitSendMessageShouldWorkEvenAfterRegisteringSource()
{
var fixture = new MessageBus();
fixture.RegisterMessageSource(Observable.Never<int>());
bool messageReceived = false;
fixture.Listen<int>().Subscribe(_ => messageReceived = true);
fixture.SendMessage(42);
Assert.True(messageReceived);
}
[Fact]
public void ListeningBeforeRegisteringASourceShouldWork()
{
var fixture = new MessageBus();
int result = -1;
fixture.Listen<int>().Subscribe(x => result = x);
Assert.Equal(-1, result);
fixture.SendMessage(42);
Assert.Equal(42, result);
}
[Fact]
public void GCShouldNotKillMessageService()
{
var bus = new MessageBus();
bool received_message = false;
var dispose = bus.Listen<int>().Subscribe(x => received_message = true);
bus.SendMessage(1);
Assert.True(received_message);
GC.Collect();
GC.WaitForPendingFinalizers();
received_message = false;
bus.SendMessage(2);
Assert.True(received_message);
}
[Fact]
public void RegisteringSecondMessageSourceShouldMergeBothSources()
{
var bus = new MessageBus();
var source1 = new Subject<int>();
var source2 = new Subject<int>();
var received_message1 = false;
var received_message2 = false;
bus.RegisterMessageSource(source1);
bus.Listen<int>().Subscribe(x => received_message1 = true);
bus.RegisterMessageSource(source2);
bus.Listen<int>().Subscribe(x => received_message2 = true);
source1.OnNext(1);
Assert.True(received_message1);
Assert.True(received_message2);
received_message1 = false;
received_message2 = false;
source2.OnNext(2);
Assert.True(received_message1);
Assert.True(received_message2);
}
[Fact]
public void MessageBusThreadingTest()
{
var mb = new MessageBus();
int? listenedThread = null;
int? otherThread = null;
int thisThread = Thread.CurrentThread.ManagedThreadId;
Task.Run(() => {
otherThread = Thread.CurrentThread.ManagedThreadId;
mb.Listen<int>().Subscribe(_ => listenedThread = Thread.CurrentThread.ManagedThreadId);
mb.SendMessage<int>(42);
}).Wait();
Assert.NotEqual(listenedThread.Value, thisThread);
Assert.Equal(listenedThread.Value, otherThread.Value);
}
}
}