Rx Test

rx_test.cs

using System;
using System.Reactive.Linq;
using System.Reactive.Subjects;
using Microsoft.VisualStudio.TestTools.UnitTesting;

namespace ReactivePipes.Tests
{
    /// <summary>
    /// Summary description for UnitTest1
    /// </summary>
    [TestClass]
    public class PipeTests
    {
        [TestMethod]
        public void TestMethod1()
        {
            var pipeline = new Subject<IObservable<int>>();

            var intPipe = new IntPipe();
            var stringPipe = new StringPipe();
            var outputPipe = new StringPipe();

            IObservable<string> output;

            pipeline
                .SelectMany(msg => intPipe.Handle(msg))
                .SelectMany(msg => stringPipe.Handle(msg)) // Argument char not assignable to parameter string
                .Subscribe(msg => output = msg); // Cannot convert source type 'char' to type 'string'

            for (int i = 0; i < 10; i++)
            {
                pipeline.OnNext(Observable.Return(i));
            }
        }
    }

    public class IntPipe : Pipe<int, string>
    {
        public override IObservable<string> Handle(int message)
        {
            return Observable.Return(message.ToString());
        }
    }

    public class StringPipe : Pipe<string, string>
    {
        public override IObservable<string> Handle(string message)
        {
            return Observable.Return(message);
        }

        public void Handle(string msg)
        {
            // Handle the message.
        }
    }
}

///
/// Orrrrrrr....
///

namespace ReactivePipes.Tests
{
    /// <summary>
    /// Summary description for UnitTest1
    /// </summary>
    [TestClass]
    public class PipeTests
    {
        [TestMethod]
        public void TestMethod1()
        {
            var pipeline = new Subject<IObservable<int>>();

            var intPipe = new IntPipe();
            var stringPipe = new StringPipe();
            var outputPipe = new StringPipe();

            IObservable<string> output;

            pipeline
                .Select(msg => intPipe.Handle(msg))
                .Select(msg => stringPipe.Handle(msg)) // Argument char not assignable to parameter string
		.Select(msg = new Message(msg))
		.Select(msg => modifyMessagePipe.Handle(msg))
                .Subscribe(msg => output = msg); // Cannot convert source type 'char' to type 'string'

            for (int i = 0; i < 10; i++)
            {
                pipeline.OnNext(Observable.Return(i));
            }
        }
    }

    public class IntPipe : Pipe<int, string>
    {
        public override string Handle(int message)
        {
            return message.ToString();
        }
    }

    public class StringPipe : Pipe<string, string>
    {
        public override string Handle(string message)
        {
            return message;
        }

        public void Handle(string msg)
        {
            // Handle the message.
        }
    }

    public class ModifyMessagePipe : Pipe<Message, Message>
    {
        public override string Handle(Message theMessage)
        {
		theMessage.Foo = "Bar";
		return theMessage;
        }
    }

}