缓冲 Observable 直到另一个 Observable 完成

Question

我正在使用 RxSwift 包装移动应用程序的服务器同步过程。我有一个 Observable<RemoteEvent>，它包装了一个 websocket 连接，并将收到的每条消息作为 Event 发出。同样，我有一个 Observable<SynchronizationResult> 包装了一个 API 同步过程。一旦我的应用程序打开 WebSocket 连接，服务器就会发送一条 hello 消息。收到该消息后，我想启动同步过程并缓冲所有事件，直到同步完成。这就是我挣扎的地方。目前，我有：

self.eventStreamService.observe(connection).scan((nil, [])) { (state, event) -> (Observable<RemoteEvent>?, [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 -> Observable<RemoteEvent> in
        (state.1 + [event]).toObservable()
      },
    []
  )
}
.flatMapLatest { [=10=].0 ?? Observable.empty() }

尽管这相当丑陋，但它也有一个重大错误：任何传入事件导致同步 Observable 被重新订阅，然后重新启动整个同步过程。我相信一定有更好的方法来做到这一点。

Answer 1

您可以通过以下方式获得所需的功能：

// this is a stub for the purpose of the example
let interval = Observable<Int>.interval(1, scheduler: MainScheduler.instance)
let websocketEvents = interval
    .map { i -> String in
        if i == 1 {
            return "hello"
        } else {
            return String(i)
        }
    }
    .replayAll()

websocketEvents.connect()

func performSync() -> Observable<Void> {
    return Observable<Void>.create { o in
        print("starting sync")
        // actually start sync with server
        // ....
        delay(2.0) {
            print("sync finished")
            o.onNext(())
        }
        return NopDisposable.instance
    }
}

// websocket events as they come, regardless of sync status
websocketEvents
    .subscribeNext { e in
        print("websocket event received: \(e)")
    }

// all websocket events, buffered and only emitted post-sync
websocketEvents
    .filter { [=10=] == "hello" }
    .flatMapLatest { _ in performSync() }
    .flatMapLatest { _ in websocketEvents }
    .subscribeNext { e in
        print("websocket event post sync: \(e)")
    }

这将输出：

websocket event received: 0
websocket event received: hello
starting sync
websocket event received: 2
websocket event received: 3
sync finished
websocket event post sync: 0
websocket event post sync: hello
websocket event post sync: 2
websocket event post sync: 3
websocket event received: 4
websocket event post sync: 4
websocket event received: 5
websocket event post sync: 5

缓冲 Observable 直到另一个 Observable 完成

Buffer Observable Until Another Observable Completes

swift

rx-swift

reactivex