RxSwift是否有类waitUntil操作符 等待就绪后写入媒体缓存

原有实现的核心问题是takeWhile属于序列终止类操作符，一旦检测到isReadyForMoreMediaData == false就会直接结束整个订阅序列，后续传入的所有缓存数据都会被丢弃，既没有等待状态恢复的逻辑，也无法保证写入顺序。

实现该需求需要满足两个核心约束：

• 所有缓存严格按照上游发送顺序串行写入，不允许并发、乱序
• 每次执行写入操作前，必须等待对应的AVAssetWriterInput处于可接收数据的状态，不满足时挂起等待，不丢弃后续缓存

具体实现 #

1. 定义专用串行调度队列 #

媒体写入属于IO耗时操作，不要放在主线程执行，单独创建串行队列保证所有写入操作串行执行：

private let writerProcessQueue = DispatchQueue(label: "com.mediaprocess.assetwriter.serial")
private lazy var writerScheduler = SerialDispatchQueueScheduler(queue: writerProcessQueue,
                                                               internalSerialQueueName: "com.mediaprocess.assetwriter.rx")

2. 替换原有订阅逻辑 #

使用concatMap操作符实现顺序排队逻辑：concatMap会严格遵循上游元素的发送顺序，前一个元素的处理逻辑未完成时，后续元素会自动在内部队列排队，不会提前执行。每个写入任务内部先等待输入就绪，再执行append操作，完成后才会触发下一个任务执行：

disposable = asyncWriter.asObservable()
    .observeOn(writerScheduler)
    .concatMap { (writerInput, sampleBuffer) -> Observable<Void> in
        // 已就绪直接写入，不需要等待
        if writerInput.isReadyForMoreMediaData {
            writerInput.append(sampleBuffer)
            return .just(())
        }

        // 未就绪时监听状态，等到第一次就绪就执行写入
        return writerInput.rx
            .observe(Bool.self, #keyPath(AVAssetWriterInput.isReadyForMoreMediaData))
            .compactMap { $0 }
            .filter { $0 }
            .take(1)
            .do(onNext: { _ in
                writerInput.append(sampleBuffer)
            })
            .map { _ in () }
    }
    .subscribe(
        onError: { err in
            // 处理写入错误，比如writer状态异常、buffer格式不匹配等
            debugPrint("AssetWriter写入失败: \(err)")
        },
        onCompleted: {
            // 所有写入完成后的收尾逻辑，比如标记写入完成、释放资源
        }
    )

关键注意事项 #

• 禁止使用flatMap、flatMapLatest替代concatMap：这两个操作符不会维护串行执行队列，会出现后到的缓存先写入的乱序问题，直接破坏媒体文件的时间轴正确性。
• 不要使用MainScheduler处理写入逻辑：媒体写入耗时较长，放在主线程会直接阻塞UI响应，引发卡顿。
• KVO监听的兼容性：isReadyForMoreMediaData是苹果官方明确支持KVO的属性，状态变化时会自动触发通知，不需要额外添加定时轮询逻辑；如果遇到特定系统版本KVO不触发的bug，可以将等待逻辑替换为10ms间隔的轻量轮询，整体排队逻辑不需要修改。
• 原有的ReplaySubject(bufferSize:1)配置可以保留：订阅未启动或者写入队列阻塞时，新传入的缓存会被Subject缓存，不会丢失。

内容的提问来源于stack exchange，提问作者Matheus Weber