title: Reactive Extensions - De-coupling each onNext call within an observable
body: <p>Please go easy with me, I am learning this topic also, and am really enjoying it. So here goes...</p>
  
  <p>I create an observable reading from a stream, when enough data comes in and represents a "IMyEvent" type that I need to fire out, I will build that type and call the observer.OnNext.</p>
  
  <p>This stream is a response stream from a server, where commands get sent to this server either from myself or externally, so I can read this stream and react based on this.</p>
  
  <p>For each command I send, I subscribe to the observable made from this stream. I can see if my command has completed successfully. I also subscribe to this stream externally and to react on other events that may occur.</p>
  
  <p>Lets pick an example, lets say someone externally joins the <strong>conference</strong> of this server, the server will send data down this stream, then my Observable catches this and calls onNext of the observer. I want to react based on this and lock the <strong>conference</strong>. <strong>Edited here...</strong> What I mean by locking the environment is that I send a Lock command to the server and the server knows to "lock itself" from allowing anyone else to join the <strong>conference</strong>.</p>
  
  <p>I have a client that is always subscribing for the "SomeoneJoinsEvent" (they are subscribed to the observable stream) while executing this onNext within this subscription I also fire a command to lock the <strong>conference</strong>. This command then also subscribes temporarily(I use timeout operator) to the same observable stream, but I notice I am blocked/locked here.</p>
  
  <p>I can see that when I am executing the onNext it doesn't continue reading the stream to watch for more IMyEvents.</p>
  
  <p>So just my ideas/thoughts/brainfarts follow...
  Is there a special way I can somehow decouple this Observable stream, so it continually reads and fires OnNext to all its subscribers and doesn't wait for them to complete? Or exposing something in the middle that is safely decoupled from the main Observable stream.</p>
  
  <ol>
  <li>I was thinking I could subscribe to this Observable stream internally and create 2 new observables, but I thought I would still be propagating the problem further, when I subscribe to the new observables. Maybe I am wrong here see brainfart 2 and 3.</li>
  <li>After reading <a href="https://stackoverflow.com/questions/20018503/c-sharp-reactive-extensions-what-happens-when-onnext-take-long-time-and-observab">this</a> I also thought maybe I should be subscribing to this Observable stream on a separate thread and add each IMyEvent to 2 queues maybe this then decoupling it and allowing me to subscribe to each queue separately, one queue subscription for my internal calls that wait for commands to complete, and one subscription for my external calls, somehow this still feels wrong too, what if I have many external subscribers to this Observable stream and one of them blocks?</li>
  <li>I think I am a little confused here, now I am thinking that my main Observable stream is .Publish.RefCount, is all I need to do is to create a new observable each time I want to receive IMyEvent (this being the same as creating a queue, like brain fart 2), that subscribe to the main connectable observable? So the internal commands subscribe to 1 observable waiting for the correct event, and 1 observable is exposed externally, I haven't thought about or tried this one yet.</li>
  </ol>
  
  <p>Is it possible someone could help me here?
  I hope this makes sense what I am trying to do.</p>
  
  <p>Here is an example of my code...</p>
  
  <pre><code>//This is the observable that I create from the stream of events
  public IObservable&lt;IMyEvent&gt; MyEvents()
  {
      if (_myEventObservable != null)
      {
          return _myEventObservable;
      }
  
      // Here is one observable that reads data as it comes in off the asynchronous stream.
      _myEventObservable = Observable.Create&lt;IMyEvent&gt;(async observer =&gt;
      {
          var myServerEventStream = await GetStreamFromMyServer(_myAuthenticationConfiguration, _token, _httpClientWrapper);
          var streamReaderAsyncState = _streamReaderAsyncStateFactory.CreateStreamReadAsyncState(myServerEventStream, 255);
  
          var currentStateDisposable = new SerialDisposable();
          var iterator = new Action&lt;IStreamReaderAsyncState, Action&lt;IStreamReaderAsyncState&gt;&gt;((state, self) =&gt;
          {
              try
              {
                  currentStateDisposable.Disposable = state.StreamObservable().Subscribe(
                      //OnNext
                      bytesRead =&gt;
                      {
                          if (bytesRead == 0)
                          {
                              observer.OnCompleted();
                          }
                          else
                          {
                              //In here it does all the magic to put all the data together and only call OnNext on the observer when there is a complete IMyEvent.
                              //It is just a plain observer.OnNext(IMyEvent whatever event we have build up)
                              _messageParser.Parse(new DataChunk(state.Buffer, bytesRead), _token, observer);
                              self(state);
                          }
                      });
              }
              catch (Exception e)
              {
                  observer.OnError(e);
              }
          });
  
          var schedulerDisposable = TaskPoolScheduler.Default.Schedule(streamReaderAsyncState, iterator);
          return new CompositeDisposable(myServerEventStream, schedulerDisposable, currentStateDisposable);
      }).Publish().RefCount();
  
      return _myEventObservable;
  }
  
  //Just for fuller visibility on the stream reader async state, here is the class that is returned from the "_streamReaderAsyncStateFactory.CreateStreamReadAsyncState(myServerEventStream, 255)" call   
  internal class StreamReaderAsyncState : IStreamReaderAsyncState
  {
      private readonly IObservable&lt;int&gt; _readAsyncObservable;
  
      public StreamReaderAsyncState(Stream stream, int bufferSize)
      {
          Buffer = new byte[bufferSize];
          _readAsyncObservable = Observable.FromAsync(() =&gt; stream.ReadAsync(Buffer, 0, bufferSize));
      }
  
      public byte[] Buffer { get; private set; }
  
      public IObservable&lt;int&gt; StreamObservable()
      {
          return _readAsyncObservable;
      }
  }
  
  
  //Externally I subscribe to this like so...
  MyEvents.OfType&lt;SomoneJoinsEvent&gt;
  .Subscribe(
      //I read somewhere that I shouldn't be making this async and using a selectmany with the async, but I am unsure.
      async myEvent =&gt; {
          await LockEnvironment()
      }
  )
  
  //The LockEnvironment call
  //The myCommandPost is the LockEnvironment Command that is passed in.
  private async Task&lt;CommandResponse&lt;TCommandResponseDto&gt;&gt; PostCommandAndWaitForEvent&lt;TEventToWaitFor, TCommandResponseDto&gt;(IMyCommandPost myCommandPost)
      where TEventToWaitFor : IMyEvent
  {
      //So my intention here is to zip together the result of the post command with the TEventToWaitFor and return the first one. Otherwise if it takes too long it will return the result of the Timeout operator.
      return await MyEvents()
                      .OfType&lt;TEventToWaitFor&gt;()
                      //This myCommandPost.PostCommand... returns a Task&lt;CommandResponse&lt;TCommandResponseDto&gt;&gt;
                      .Zip(myCommandPost.PostCommand&lt;TCommandResponseDto&gt;().ToObservable(), (myEvent, myCommandResponse) =&gt; myCommandResponse)
                      .Timeout(new TimeSpan(0, 0, _myAuthenticationConfiguration.TimeoutToWaitForCommands), TimedOutLookingForMyEvent&lt;TCommandResponseDto&gt;())
                      .FirstAsync();
  }
  
  //The timeout observable
  private IObservable&lt;CommandResponse&lt;TCommandResponseDto&gt;&gt; TimedOutLookingForMyEvent&lt;TCommandResponseDto&gt;()
  {
      return Observable.Create&lt;CommandResponse&lt;TCommandResponseDto&gt;&gt;(observable =&gt;
      {
          observable.OnNext(new CommandResponse&lt;TCommandResponseDto&gt;());
          return Disposable.Empty;
      });
  }
  </code></pre>
  
  <p><strong>Also edited here, adding what I did for the event parser...</strong></p>
  
  <pre><code>internal class MyEventParser : IMessageParser
  {
      private readonly ILogService _logService;
      private readonly IMyEventFactory _MyEventFactory;
      private readonly StringBuilder _data = new StringBuilder();
      private const string EventDelimiter = "\n\n";
      private readonly Regex _eventDelimiterRegex = new Regex("\n{3,}");
  
      public MyEventParser(ILogService logService, IMyEventFactory myEventFactory)
      {
          _logService = logService;
          _myEventFactory = myEventFactory;
      }
  
      public void Parse(DataChunk dataChunk, string token, IObserver&lt;IMyEvent&gt; myEventObserver)
      {
          _data.Append(dataChunk);
          CleanUpEventDelimiterInData();
          var numberOfSubstrings = CountNumberOfSubstrings(EventDelimiter, _data.ToString());
          if (numberOfSubstrings == 0)
          {
              return;
          }
  
          var events = _data.ToString().Split(new[]{EventDelimiter}, StringSplitOptions.RemoveEmptyEntries);
  
          events.Take(numberOfSubstrings).Foreach(x =&gt;
          {
              _logService.InfoFormat("MyEventParser - {0} - OnNext: \n\n{1}\n\n", token.Substring(token.Length -10), x);
              myEventObserver.OnNext(_myEventFactory.Create(x));
          });
  
          //Clean up data of what has already been fired.
          if (events.Count() == numberOfSubstrings)
          {
              _data.Clear();
          }
          else
          {
              _data.Clear();
              _data.Append(events.Last());
          }
      }
  
      private void CleanUpEventDelimiterInData()
      {
          var eventDelimitersFixed = _eventDelimiterRegex.Replace(_data.ToString(), EventDelimiter);
          _data.Clear();
          _data.Append(eventDelimitersFixed);
      }
  
      private int CountNumberOfSubstrings(string subString, string source)
      {
          var i = 0;
          var count = 0;
          while ((i = source.IndexOf(subString, i, StringComparison.InvariantCulture)) != -1)
          {
              i += subString.Length;
              count++;
          }
  
          return count;
      }
  }
  </code></pre>
  
  <p>Thanks for all your help in advance :-)</p>