是否有任何waituntil运算符或rxswift中的类似操作员来等待一个值？

Question

我正在使用rxswift来处理许多Async AvassetWriterInput写操作，但是在文件中写入更多缓冲区之前，我需要等待IsreadyFormoremediadata，我该如何处理？

基本上，可观察到的可观察到的是Asyncwritersubject发出的大量缓冲区，我想按照我收到的顺序编写所有缓冲区。

我有这个主题：

private var asyncWriter = ReplaySubject<(AVAssetWriterInput,CMSampleBuffer)>.create(bufferSize: 1)

我使用此代码散发了它的值：

asyncWriter.onNext((videoWriterInput, buffer))

我在这里订阅它以聆听：

disposable = asyncWriter.asObservable()
    .takeWhile {
        (writerPointer, _) in
            writerPointer.isReadyForMoreMediaData
        }.observeOn(MainScheduler.asyncInstance)
        .subscribe(onNext: { (writerPointer, buffer) in
            writerPointer.append(buffer)
    })

Answer 1

以下是一些有关如何处理背压的一般信息。

这将向CMSPlame Buffer写入cmsample buffer WriterPointer每秒可达100次。当isreadyFormoreMediAdata为false时，它将存储示例缓冲区，直到bool再次为true为止。

func example(asyncWriter: Observable<CMSampleBuffer>, writerPointer: AVAssetWriterInput) -> Disposable {
    enum Action {
        case buffer(CMSampleBuffer)
        case isReady(Bool)
    }
    var isReadyForMoreMediaData: Observable<Bool> {
        Observable<Int>.interval(.milliseconds(10), scheduler: MainScheduler.instance)
            .flatMap { [writerPointer] _ in Observable.just(writerPointer.isReadyForMoreMediaData) }
    }
    return Observable.merge(
        isReadyForMoreMediaData.map { Action.isReady($0) },
        asyncWriter.map { Action.buffer($0) }
    )
        .scan(into: (buffer: [CMSampleBuffer](), trigger: false, emission: CMSampleBuffer?.none), accumulator: { current, new in
            switch new {
            case let .buffer(buff):
                if current.trigger {
                    if current.buffer.isEmpty {
                        current.emission = buff
                    }
                    else {
                        current.emission = current.buffer[0]
                        current.buffer.removeFirst()
                        current.buffer.append(buff)
                    }
                }
                else {
                    current.buffer.append(buff)
                }
            case let .isReady(trig):
                current.trigger = trig
                if trig && !current.buffer.isEmpty {
                    current.emission = current.buffer[0]
                    current.buffer.removeFirst()
                }
            }
        })
        .compactMap { $0.emission }
        .observe(on: MainScheduler.instance)
        .subscribe(onNext: { buffer in
            writerPointer.append(buffer)
        })
}