Наблюдаемый буфер, пока не завершится другой наблюдаемый

Вот как можно получить нужную функциональность:

// 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 { $0 == "hello" }
    .flatMapLatest { _ in performSync() }
    .flatMapLatest { _ in websocketEvents }
    .subscribeNext { e in
        print("websocket event post sync: \(e)")
    }

Это выведет:

событие websocket получено: 0
событие websocket получено: привет
начало синхронизации
получено событие websocket: 2
получено событие websocket: 3
синхронизация завершена
синхронизация сообщения события websocket: 0
Синхронизация сообщений о событиях websocket: привет
Синхронизация сообщений событий websocket: 2
Синхронизация сообщений событий websocket: 3
полученных событий websocket: 4
Синхронизация сообщений событий websocket: 4
полученных событий websocket: 5
Синхронизация сообщений о событиях websocket: 5