title: Rx DotNet - Chaining observables with threading
body: <p>I'm trying to introduce Rx in my Xamarin mobile app and I'd like to chain a sequence of calls during my login phase on app startup.</p>
  
  <blockquote>
    <p><strong>TL;DR;</strong> How do I run 2 / 3 observables one after the other to retrieve data and setup threading properly.</p>
  </blockquote>
  
  <p>More in detail I perform the following:</p>
  
  <ol>
  <li>Login the User and retrieve related data <code>LoginUser() : IObservable&lt;User&gt;</code></li>
  <li>Based on User's type the ID to either <code>RetrieveRemoteA() : IObservable&lt;A&gt;</code> or <code>RetrieveRemoteB() : IObservable&lt;B&gt;</code></li>
  <li>Once data are available (either of type <code>A</code> or <code>B</code>), I update the UI.</li>
  </ol>
  
  <p>This is a sort of diagram to describe the flow (explained above):</p>
  
  <p><a href="https://i.stack.imgur.com/Jeye4.png" rel="nofollow noreferrer"><img src="https://i.stack.imgur.com/Jeye4.png" alt="enter image description here"></a></p>
  
  <p>Considering that I want to avoid calling new observables (sources) from within the <code>Subscribe()</code> of the previous one, this is what I've implemente for retrieving my data and update the UI (the flow in the image has been serialized in the code below).</p>
  
  <pre><code>IObservable&lt;User&gt; loginUserObservable = LogInUser(currentUser);
  
  loginUserObservable
      .SubscribeOn(ThreadPoolScheduler.Instance)
      .SelectMany(
          (user) =&gt;
          {
              if (user.Type == UserType.A)
                  return RetrieveRemoteA(user.UserId); // outputs IObservable&lt;A&gt;
  
              return Observable.Return(new A());
          },
          (user, a) =&gt;
          {
              B b = null;
              return new { user, a, b }; // Create anonymous type to keet track of 'user'
          })
      .SelectMany(
          (xType) =&gt;
          {
              if (xType.user.Type == UserType.B)
                  return RetrieveRemoteB(xType.user.UserId); // outputs IObservable&lt;B&gt;
  
              return Observable.Return(new B());
          },
          (xType, bData) =&gt;
          {
              var user = xType.user;
              var a = xType.a;
              var b = b;
  
              return new { user, a, b };
          })
      .ObserveOn(ImmediateScheduler.Instance)
      .Select((xType) =&gt;
      {
          if (xType.user.Type == UserType.A)
          {
              A a = xType.a;
              B b = null;
              return new { a, b };
          }
          else {
              A a = null;
              B b = xType.b;
              return new { a, b };
          }
      })
      .Subscribe((result) =&gt;
      {
          if (result.a != null)
          {
              Console.WriteLine($"ID: {result.a.Id}");
          }
          else {
              Console.WriteLine($"ID: {result.b.Id}");
          }
      });
  </code></pre>
  
  <p>Once running, it seems that the flow get stacked within <code>RetrieveRemoteA(user.UserId)</code>, even if the method finishes without errors.</p>
  
  <pre><code>public IObservable&lt;A&gt; RetrieveRemoteA(string userId)
  {
      return Observable.FromAsync&lt;A&gt;(async () =&gt;
      {
          A a = await CustomAPI(userId) 
          return a;
      }
  }
  </code></pre>
  
  <p>When I implement the flow sequentially by calling each new observable within the Subscribe of the previous one, it works properly (but it is not the right way to do it).</p>
  
  <p>I think this is an issue with Threads or either with a wrong Rx implementation of mine.</p>
  
  <p>Do you have any clue, please?</p>