title: Testing Rx & TcpClient
body: <p>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:</p>
  
  <pre><code>public static IObservable&lt;string&gt; CreateFromTcpIp(IPAddress ip, int port)
  {
      return 
          Observable.Using(
              () =&gt; new TcpClient(),                               //create client
              client =&gt; client.ConnectObservable(ip, port)         //connect async
                        .Select(_ =&gt; client.GetStream())            // get stream
                        .Select(stream =&gt; new BinaryReader(stream)) // get reader 
                        .SelectMany(reader =&gt; reader.ToObservable()) // see below
                        .Retry()                                // if error, retry 
              )
              .Publish().RefCount(); //only subscribe once for all subscribers
  }
  </code></pre>
  
  <p>Where <code>ConnectObservable</code> is just <code>client.ConnectAsync(host, port).ToObservable()</code>, and <code>reader.ToObservable</code> looks like this:</p>
  
  <pre><code>private static IObservable&lt;string&gt; ToObservable(this BinaryReader source)
  {
     return Observable.Create&lt;string&gt;(
         observer =&gt;
         {
             return Observable.Generate(
                 source,
                 reader =&gt; true, //just keep on going
                 reader =&gt; reader,
                 reader =&gt; reader.ReadString())
             .Subscribe(observer);
         });
  }
  </code></pre>
  
  <p>The problem I'm facing is how to go about testing this. I <em>don't</em> want to implement/wrap interfaces all the way down that call stack.</p>
  
  <p>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).</p>
  
  <p>Is there an alternative approach that I'm missing?</p>
  
  <p><strong>EDIT</strong> The problem of Subscribe not returning is related to doing a Retry on a faulting Observable. I've asked another question <a href="https://stackoverflow.com/q/20212951">here</a>. </p>