title: Why is my process still alive after an exception thrown in SelectMany whereas an exception in similar rx operator results in an unhandled exception?
body: <p>Here is a sample program that makes two subscriptions to console input (source observable is not relevant here). In the first subscription it uses Observable.SelectMany and in the second one a similar SelectMany operator that internally uses <a href="https://www.nuget.org/packages/System.Threading.Tasks.Dataflow/" rel="nofollow noreferrer">System.Threading.Tasks.Dataflow</a> package. An exception is thrown for certain inputs in each of them. The exception is correctly forwarded to Observer onError which rethrows it in the default Subscribe implementation. The observed behaviour is that in case of an exception in SelectMany the process keeps running and in case of an exception in SelectManyPreseveOrder the process is terminated with an unhandled exception. 
  What is the cause of the different behaviour? Is there a way to achieve the 'more friendly' behaviour in the SelectManyPreserveOrder operator?
  It's a .net 4.6.1 console application using Rx.Linq 2.2.5 and System.Threading.Tasks.Dataflow 4.10.0:</p>
  
  <pre><code>class Program
  {
      static async Task Main()
      {
          AppDomain.CurrentDomain.UnhandledException += (sender, args) =&gt; Console.WriteLine("App domain unhandled exception");
          TaskScheduler.UnobservedTaskException += (sender, args) =&gt; Console.WriteLine("Unobserved task exception");
  
          var consoleInput = Helper.ConsoleInput();
  
          consoleInput.SelectMany(async input =&gt;
              {
                  await Task.Delay(50).ConfigureAwait(false);
                  if (input == "1")
                      throw new Exception("This exception is swallowed");
                  return input;
              })
              .Subscribe(s =&gt; Console.WriteLine($"SelectMany: {s}"));
  
          consoleInput.SelectManyPreserveOrder(async input =&gt;
              {
                  await Task.Delay(50).ConfigureAwait(false);
                  if (input == "2")
                      throw new Exception("This exception kills the process");
                  return input;
              })
              .Subscribe(s =&gt; Console.WriteLine($"SelectMany (TPL Dataflow): {s}"));
  
          await Task.Delay(TimeSpan.FromMinutes(10)).ConfigureAwait(false);
      }
  }
  
  public static class ObservableExtension
  {
      public static IObservable&lt;TResult&gt; SelectManyPreserveOrder&lt;TSource, TResult&gt;(this IObservable&lt;TSource&gt; source, Func&lt;TSource, Task&lt;TResult&gt;&gt; selector, int maxParallelBatches = 1)
      {
          return source.FromTplDataflow(() =&gt;
              new TransformBlock&lt;TSource, TResult&gt;(selector,
                  new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = maxParallelBatches }));
      }
  
      public static IObservable&lt;TResult&gt; FromTplDataflow&lt;T, TResult&gt;(
          this IObservable&lt;T&gt; source, Func&lt;IPropagatorBlock&lt;T, TResult&gt;&gt; blockFactory)
      {
          return Observable.Defer(() =&gt;
          {
              var block = blockFactory();
              return Observable.Using(() =&gt;
              {
                  var disposable = source.Subscribe(block.AsObserver());
                  return Disposable.Create(dispose: () =&gt; disposable.Dispose());
              }, r =&gt; block.AsObservable());
          });
      }
  }
  
  public static class Helper
  {
      public static IObservable&lt;string&gt; ConsoleInput()
      {
          return
              Observable
                  .FromAsync(() =&gt; Console.In.ReadLineAsync())
                  .Repeat()
                  .Publish()
                  .RefCount()
                  .SubscribeOn(Scheduler.Default);
      }
  }
  </code></pre>
  
  <p>Interestingly enough the UnobservedTaskException handler is never called.</p>