title: Simple way to concurrently subscribe to observable collection with limited subscribers
body: <p>I've been trying to implement a simple producer-consumer pattern using Rx and observable collections. I also need to be able to throttle the number of subscribers easily. I have seen lots of references to LimitedConcurrencyLevelTaskScheduler in parallel extensions but I don't seem to be able to get this to use multiple threads.</p>
  
  <p>I think I'm doing something silly so I was hoping someone could explain what. In the unit test below, I expect multiple (2) threads to be used to consume the strings in the blocking collection. What am I doing wrong?</p>
  
  <pre><code>[TestClass]
  public class LimitedConcurrencyLevelTaskSchedulerTestscs
  {
      private ConcurrentBag&lt;string&gt; _testStrings = new ConcurrentBag&lt;string&gt;();
      ConcurrentBag&lt;int&gt; _threadIds= new ConcurrentBag&lt;int&gt;();
  
      [TestMethod]
      public void WhenConsumingFromBlockingCollection_GivenLimitOfTwoThreads_TwoThreadsAreUsed()
      {
  
          // Setup the command queue for processing combinations
          var commandQueue = new BlockingCollection&lt;string&gt;();
  
          var taskFactory = new TaskFactory(new LimitedConcurrencyLevelTaskScheduler(2));
          var scheduler = new TaskPoolScheduler(taskFactory);
  
          commandQueue.GetConsumingEnumerable()
              .ToObservable(scheduler)
              .Subscribe(Go, ex =&gt; { throw ex; });
  
          var iterationCount = 100;
          for (int i = 0; i &lt; iterationCount; i++)
          {
              commandQueue.Add(string.Format("string {0}", i));
          }
          commandQueue.CompleteAdding();
  
          while (!commandQueue.IsCompleted)
          {
              Thread.Sleep(100);
          }
  
          Assert.AreEqual(iterationCount, _testStrings.Count);
          Assert.AreEqual(2, _threadIds.Distinct().Count());
      }
  
      private void Go(string testString)
      {
          _testStrings.Add(testString);
          _threadIds.Add(Thread.CurrentThread.ManagedThreadId);
      }
  }
  </code></pre>