Buffer Observable Until Another Observable Completes
<p>I'm using RxSwift to wrap a mobile app's server sync process. I have an <code>Observable&lt;RemoteEvent&gt;</code> that wraps a websocket connection and emits each message received as an <code>Event</code>. Similarly, I have an <code>Observable&lt;SynchronizationResult&gt;</code> that wraps an API sync process. Once my application opens the WebSocket connection, the server sends a <code>hello</code> message. After that message is received, I want to start the sync process and buffer any events until the sync is completed. This is where I'm struggling. Currently, I have:</p>

<pre><code>self.eventStreamService.observe(connection).scan((nil, [])) { (state, event) -&gt; (Observable&lt;RemoteEvent&gt;?, [RemoteEvent]) in
  guard event.type == "hello" else {
    return (state.0?.concat(Observable.just(event)), state.1 + [event])
  }

  // This is the sync operation
  return (
    self.synchronizationService
      .synchronize(ConnectionSynchronizationContext(connection: connection), lightweight: true)
      .toArray()
      .flatMap { results -&gt; Observable&lt;RemoteEvent&gt; in
        (state.1 + [event]).toObservable()
      },
    []
  )
}
.flatMapLatest { $0.0 ?? Observable.empty() }
</code></pre>

<p>Despite this being fairly ugly, it also has a significant bug: <em>any</em> incoming event results in the synchronization <code>Observable</code> being re-subscribed to, which then restarts the whole sync process. I'm sure there must be a better way to do this.</p>