title: Observable.Using { WebSocket.ReceiveAsync(); } aborts and cancels early
body: <p>In the code below, the <code>Connect()</code> appears to work fine. However <code>webSocket.ReceiveAsync()</code> in <code>ReceiveMessage()</code> seems to set <code>cancelToken.IsCancellationRequested = true</code> and <code>webSocket.State = Aborted</code> the moment the server doesn't have anything to offer. At the same time, the same <code>ReceiveAsync()</code> call raises an (unhelpful) <code>OperationCanceledException</code>, with no inner exception.</p>
  
  <p>I've engaged the same server using what I thought was the same WebSocket interaction without this problem, but with simpler async code. It wasn't until I tried converting it to an observable that I had this problem.</p>
  
  <p>Does it get cancelled by Rx, or does it cancel itself? I found a mention of such a thing being caused in .NET WebSockets by unobserved exceptions (<a href="http://www.salmanq.com/blog/5-things-you-probably-didnt-know-about-net-websockets/2013/04/" rel="nofollow">http://www.salmanq.com/blog/5-things-you-probably-didnt-know-about-net-websockets/2013/04/</a>), but I couldn't find any evidence of this occurring after adding <code>TaskScheduler.UnobservedTaskException += ExceptionLogger</code>.</p>
  
  <p>Any ideas?</p>
  
  <pre><code>public static IObservable&lt;string&gt; GetWebSocketStream(Uri uri)
  {
      var buffer = new byte[2048];
      return Observable
          .Using(cancelToken =&gt; Connect(uri, cancelToken),
              (webSocket, cancelToken) =&gt; Task.FromResult(
                  Observable.Defer(
                      () =&gt; ReceiveMessage(webSocket, buffer, cancelToken).ToObservable()
                  ).Repeat()
              )
          ).Publish().RefCount();
  }
  
  private static async Task&lt;ClientWebSocket&gt; Connect(Uri uri, CancellationToken token)
  {
      var webSocket = new ClientWebSocket();
      await webSocket.ConnectAsync(uri, token).ConfigureAwait(false);
      return webSocket;
  }
  
  private static async Task&lt;string&gt; ReceiveMessage(ClientWebSocket webSocket, byte[] buffer, CancellationToken cancelToken)
  {
      var segment = new ArraySegment&lt;byte&gt;(buffer);
      var message = new StringBuilder();
      WebSocketReceiveResult result;
      do
      {
          result = await webSocket.ReceiveAsync(segment, cancelToken).ConfigureAwait(false);
          message.Append(Encoding.UTF8.GetString(segment.Array, 0, result.Count));
      } while (!cancelToken.IsCancellationRequested &amp;&amp; !result.EndOfMessage);
      return message.ToString();
  }
  </code></pre>
  
  <hr>
  
  <p><strong>EDIT:</strong> To clarify, my question/problem is probably related to BOTH WebSockets AND Rx. The similar code below, on the same device, remains open and receives data:</p>
  
  <pre><code>public static void Test() {
      var ts = new CancellationTokenSource();
      AsyncTest(new Uri("ws://server/test/"), ts.Token).Wait();
  }
  
  public static async Task AsyncTest(Uri uri, CancellationToken token) {
      var socket = await GetConnectedWebSocket(uri, token);
      var buffer = new byte[4000];
      while (!token.IsCancellationRequested)
      {
          try
          {
              var msg = await ReceiveMessage(socket, buffer, token);
              Debug.WriteLine(msg);
          }
          catch (Exception e)
          {
              Debug.WriteLine(e.Message);
              throw;
          }
      }
  }
  </code></pre>