RxとパフォーマンスとユニットテストとMoles再び

// 1000回イベントが発火(発火の度に長さ3000のbyte配列が得られる)を模写
static IObservable<byte[]> DummyEventsRaised()
{
    return Observable.Repeat(new byte[3000], 1000, Scheduler.Immediate);
}
 
// 配列をバラす処理にObservable.Rangeを用いた場合
static IObservable<byte> TestObservableRange()
{
    return Observable.Create<byte>(observer =>
    {
        return DummyEventsRaised()
            .Subscribe(xs =>
            {
                Observable.Range(0, xs.Length, Scheduler.Immediate).ForEach(x => observer.OnNext(xs[x]));
            });
    });
}
 
// 配列をバラす処理にEnumerable.Rangeを用いた場合(ForEachはIxのもの)
static IObservable<byte> TestEnumerableRange()
{
    return Observable.Create<byte>(observer =>
    {
        return DummyEventsRaised()
            .Subscribe(xs =>
            {
                Enumerable.Range(0, xs.Length).ForEach(x => observer.OnNext(xs[x]));
            });
    });
}
 
// SelectManyでバラす場合
static IObservable<byte> TestSelectMany()
{
    return DummyEventsRaised().SelectMany(xs => xs);
}
 
static void Main(string[] args)
{
    // ベンチマーク補助関数
    Action<Action, string> bench = (action, label) =>
    {
        var sw = Stopwatch.StartNew();
        action();
        Console.WriteLine("{0,-12}{1}", label, sw.Elapsed);
    };
 
    // 配列をばらすケースは再度連結する(ToList)
    bench(() => TestObservableRange().ToList().Subscribe(), "Ob.Range");
    bench(() => TestEnumerableRange().ToList().Subscribe(), "En.Range");
    bench(() => TestSelectMany().ToList().Subscribe(), "SelectMany");
    // 配列をばらして連結せず直接処理する場合
    bench(() => TestSelectMany().Subscribe(), "DirectRx");
    // byte[]をばらさず直接処理する場合
    bench(() => DummyEventsRaised().Subscribe(xs => { foreach (var x in xs);}), "DirectLoop");
 
    // 実行結果
    // Ob.Range    00:00:02.2619670
    // En.Range    00:00:00.2600460
    // SelectMany  00:00:00.2701137
    // DirectRx    00:00:00.0852836
    // DirectLoop  00:00:00.0152816
}

public static class SerialPortExtensions
{
    // 面倒くさいけれど単純なFromEventでのイベントのRx化
    public static IObservable<SerialDataReceivedEventArgs> DataReceivedAsObservable(this SerialPort serialPort)
    {
        return Observable.FromEvent<SerialDataReceivedEventHandler, SerialDataReceivedEventArgs>(
            h => (sender, e) => h(e), h => serialPort.DataReceived += h, h => serialPort.DataReceived -= h);
    }
 
    public static IObservable<SerialErrorReceivedEventArgs> ErrorReceivedAsObservable(this SerialPort serialPort)
    {
        return Observable.FromEvent<SerialErrorReceivedEventHandler, SerialErrorReceivedEventArgs>(
            h => (sender, e) => h(e), h => serialPort.ErrorReceived += h, h => serialPort.ErrorReceived -= h);
    }
 
    // DataReceived(プラスbyte[]化)とErrorReceivedを合成する
    public static IObservable<byte[]> ObserveReceiveBytes(this SerialPort serialPort)
    {
        var received = serialPort.DataReceivedAsObservable()
            .TakeWhile(e => e.EventType != SerialData.Eof) // これでOnCompletedを出す
            .Select(e =>
            {
                var buf = new byte[serialPort.BytesToRead];
                serialPort.Read(buf, 0, buf.Length);
                return buf;
            });
 
        var error = serialPort.ErrorReceivedAsObservable()
            .Take(1) // 届いたらすぐに例外だすので長さ1として扱う（どうせthrowするなら関係ないけど一応）
            .Do(x => { throw new Exception(x.EventType.ToString()); });
 
        return received.TakeUntil(error); // receivedが完了した時に同時にerrorをデタッチする必要があるのでMergeではダメ
    }
}

[TestClass]
public class SerialPortExtensionsTest : ReactiveTest
{
    [TestMethod, HostType("Moles")]
    public void ObserveReceiveBytesOnCompleted()
    {
        // EventArgsを捏造！
        var chars = new MSerialDataReceivedEventArgs() { EventTypeGet = () => SerialData.Chars };
        var eof = new MSerialDataReceivedEventArgs() { EventTypeGet = () => SerialData.Eof };
        // SerialPort::BytesToRead/SerialPort::Readで何もしない
        MSerialPort.AllInstances.BytesToReadGet = (self) => 0;
        MSerialPort.AllInstances.ReadByteArrayInt32Int32 = (self, buffer, offset, count) => 0;
 
        var scheduler = new TestScheduler();
 
        // 時間10, 20, 30, 40でSerialData.Charsのイベントを、時間50でEofのイベントを発行
        MSerialPortExtensions.DataReceivedAsObservableSerialPort = _ => scheduler.CreateHotObservable(
                OnNext(10, chars),
                OnNext(20, chars),
                OnNext(30, chars),
                OnNext(40, chars),
                OnNext(50, eof))
            .Select(x => (SerialDataReceivedEventArgs)x);
 
        // 走らせる(戻り値のbyte[]はどうでもいいので無視するためUnitに変換)
        var result = scheduler.Start(() => new SerialPort().ObserveReceiveBytes().Select(_ => Unit.Default), 0, 0, 100);
 
        result.Messages.Is(
            OnNext(10, Unit.Default),
            OnNext(20, Unit.Default),
            OnNext(30, Unit.Default),
            OnNext(40, Unit.Default),
            OnCompleted<Unit>(50));
    }
 
    [TestMethod, HostType("Moles")]
    public void ObserveReceiveBytesOnError()
    {
        // EventArgsを捏造！
        var chars = new MSerialDataReceivedEventArgs() { EventTypeGet = () => SerialData.Chars };
        var eof = new MSerialDataReceivedEventArgs() { EventTypeGet = () => SerialData.Eof };
        // SerialPort::BytesToRead/SerialPort::Readで何もしない
        MSerialPort.AllInstances.BytesToReadGet = (self) => 0;
        MSerialPort.AllInstances.ReadByteArrayInt32Int32 = (self, buffer, offset, count) => 0;
 
        var scheduler = new TestScheduler();
 
        // 時間10, 20, 30, 40でSerialData.Charsのイベントを、時間50でEofのイベントを発行
        MSerialPortExtensions.DataReceivedAsObservableSerialPort = _ => scheduler.CreateHotObservable(
                OnNext(10, chars),
                OnNext(20, chars),
                OnNext(30, chars),
                OnNext(40, chars),
                OnNext(50, eof))
            .Select(x => (SerialDataReceivedEventArgs)x);
 
        /* ↑までOnCompletedのものと共通 */
 
        // 時間35でErrorのイベントを発行
        MSerialPortExtensions.ErrorReceivedAsObservableSerialPort = _ => scheduler.CreateHotObservable(
            OnNext<SerialErrorReceivedEventArgs>(35, new MSerialErrorReceivedEventArgs()));
 
        // 走らせる(戻り値のbyte[]はどうでもいいので無視するためUnitに変換)
        var result = scheduler.Start(() => new SerialPort().ObserveReceiveBytes().Select(_ => Unit.Default), 0, 0, 100);
 
        // Exceptionの等値比較ができないので、バラしてAssertする
        result.Messages.Count.Is(4);
 
        result.Messages[0].Is(OnNext(10, Unit.Default));
        result.Messages[1].Is(OnNext(20, Unit.Default));
        result.Messages[2].Is(OnNext(30, Unit.Default));
 
        result.Messages[3].Value.Kind.Is(NotificationKind.OnError);
        result.Messages[3].Time.Is(35);
    }
}