Testing Rx & TcpClient

Question

After quite some thought and rummaging around on the interwebs, I've come up with (what I think is) quite a nifty encapsulation of turning TcpIp into an Observable stream of data:

public static IObservable<string> CreateFromTcpIp(IPAddress ip, int port)
{
    return 
        Observable.Using(
            () => new TcpClient(),                               //create client
            client => client.ConnectObservable(ip, port)         //connect async
                      .Select(_ => client.GetStream())            // get stream
                      .Select(stream => new BinaryReader(stream)) // get reader 
                      .SelectMany(reader => reader.ToObservable()) // see below
                      .Retry()                                // if error, retry 
            )
            .Publish().RefCount(); //only subscribe once for all subscribers
}

Where ConnectObservable is just client.ConnectAsync(host, port).ToObservable(), and reader.ToObservable looks like this:

private static IObservable<string> ToObservable(this BinaryReader source)
{
   return Observable.Create<string>(
       observer =>
       {
           return Observable.Generate(
               source,
               reader => true, //just keep on going
               reader => reader,
               reader => reader.ReadString())
           .Subscribe(observer);
       });
}

The problem I'm facing is how to go about testing this. I don't want to implement/wrap interfaces all the way down that call stack.

I've been trying to create a 'server' using TcpListener inside my test class, but am coming up against various (possibly unrelated) problems getting the subscribing and serving to co-exist (Subscribing to the the results of my CreateFromTcpIp never seems to return the IDisposable, inserting delays in my tests seem to end up locking the assembly so I can't rebuild).

Is there an alternative approach that I'm missing?

EDIT The problem of Subscribe not returning is related to doing a Retry on a faulting Observable. I've asked another question here.

Answer 1

I suppose you could test the BinaryReader ToObservable like this (if you make it accessible). The fact that this expects an exception possibly highlights some issues, although I wrote it to pass the code as written (uses nunit):

[Test]
public void TestStreamToObservable()
{

    var expectedText = new List<string>
    {
        "A good test is simple.",
        "A rolling stone gathers no moss.",
        "Test properly"
    };

    var stream = new MemoryStream();
    var writer = new BinaryWriter(stream);
    expectedText.ForEach(writer.Write);
    writer.Flush();

    stream.Seek(0, SeekOrigin.Begin);
    var reader = new BinaryReader(stream);

    var resultText = new List<string>();

    Assert.Throws<EndOfStreamException>(
        () => reader.ToObservable().Subscribe(resultText.Add));

    CollectionAssert.AreEquivalent(expectedText, resultText);
}

As for the rest, you've kind of scuppered yourself for anything but integration testing by creating a new TcpClient(). A new of anything concrete like that will always spell trouble for testing. You could refactor and pull the Retry().Publish().RefCount() out into a separate method that operates on IObservable<string>. Then leave the TcpClient bits to integration testing only.

Answer 2

OK, for posterity, here are some of the things I've learnt while working through this problem. Mostly it's about the subtleties of using Rx:

• If you don't observe OnError, it will throw as an exception.
• On top of that, if you've done a ToObservable on a Task and you don't observe OnError, it will only throw when you unsubscribe
• Also, there's no point doing a Retry downstream from a Task: once it's thrown an exception, that's the Result, and it won't try again (too me a while to work that out!).
• In the case where my subscription was never returning, I needed to SubscribeOn a different thread (Scheduler.Default).
• Another bug was hinted at in James' answer: the ToObservable on the BinaryReader throws an exception. What is important here is that it is an Exception, not an OnError. So there's no Retry possible.

(Little code snippet to prove this to ourselves):

Observable.Generate(0, i => i < 4, i => i + 1, 
                    i => { if(i > 2) throw new Exception("sorry"); return i; })
          .Subscribe(i => Console.WriteLine(i), 
                     ex => Console.WriteLine(ex.Message));

Given this, I've recoded my ToObservable to use this (not so pretty):

return Observable.Create<string>(obs =>
{
    var subscribed = true;
    var errored = false;
    string result = String.Empty;
    while(subscribed && !errored)
    {
        try
        {
            result = source.ReadString();
        }
        catch(Exception ex)
        {
            errored = true;
            obs.OnError(ex);
        }
        obs.OnNext(result); 
    }
    return Disposable.Create(() => subscribed = false);
});

• For further development, I wanted a Retry that was slightly less aggressive. This led me to this excellent answer. (It seems that not matter what I manage to write in Rx, there is always a more elegant way to do it)
• James' Spy extension method was very helpful. Thoroughly recommended.