结合 2 个 PublishSubject's 并使用 Observable.combineLatest 发出

Question

我的服务中有 2 个单独的数据集合。

Featured和Standard内容。

我有 2 个 api 的调用来返回这些项目。它们可以单独使用，但是我也有用例，我想获取两组数据，根据条件提供一些扩充，然后将它们返回给消费者。

我希望我能做这样的事情：

class ContentService: ContentServiceType {
let featured = PublishSubject<[Content]>()
let standard = PublishSubject<[Content]>()
let content: Observable<(featured: [Content], standard: [Content])>
private let client: Client<ContentAPI>
private let disposeBag = DisposeBag()
init(client: Client<ContentAPI>) {
self.client = client
content = Observable
.combineLatest(featured, standard)
.map { (featured, standard) -> (featured: [Content], standard: [Content]) in
/*
Do some enrichment and create then return new, updated versions
*/
return (featured: updatedFeatured, standard: updatedStandard)
}.share()
}
func fetchStandardContent(page: Int = 0, size: Int = 100) -> Single<Void> {
let params = ["page": page, "size": size]
let request: Single<Content> = client.request(.getStandardContent(params))
return request.map { [unowned self] launchers in
self.standard.onNext(content.props)
return ()
}
}
func fetchFeaturedContent(page: Int = 0, size: Int = 100) -> Single<Void> {
let params = ["page": page, "size": size]
let request: Single<Content> = client.request(.getFeaturedContent(params))
return request.map { [unowned self] content in
self.featured.onNext(content.props)
return ()
}
}
}   

在我的应用程序的其他地方，我希望我能做一些类似的事情

contentSvc.content
.observeOn(MainScheduler.instance)
.subscribeOn(ConcurrentDispatchQueueScheduler(qos: .background))
.subscribe(onNext: { content in
/* do something w/ content */
}).disposed(by: disposeBag)

然后，每当调用contentSvc.fetchFeaturedContent或contentSvc.fetchStandardContent时，上面的contentSvc.content订阅者都会获得新数据。

相反content似乎没有发出任何值。

Answer 1

我相信combineLatest需要两个源在发出自身之前发出。

我可能会考虑使用BehaviorSubject或BehaviorRelay而不是PublishSubject。

Answer 2

我使用的是 BehaviorRelay 而不是 PublishSubject，因为当从多个流绑定到 PublishSubject(跨应用程序共享(时，如果这些流中的任何一个发送完整，则 PublishSubject 可能会终止。中继类永远不会产生错误或永远不会完成。

let featured = BehaviorRelay(value: [Content]())
let standard = BehaviorRelay(value: [Content]())
func getContent() -> Observable<(featured: [Content], standard: [Content])> {
return Observable
.combineLatest(
featured.asObservable(),
standard.asObservable(),
resultSelector: { (featured, standard) -> (featured: [Content], standard: [Content]) in
return (featured: featured, standard: standard)
}
)
}
func addElemToFeatured() {
featured.accept([Content(name: "abc")])
}
func addElemToStandard() {
standard.accept([Content(name: "xyz")])
}

从初始值设定项方法中的不同类调用 getContent(( 方法。也可以调用addElemToFeatured，addElemToStandard从不同的地方，如按钮动作。

listener!.getContent()
.subscribe(onNext: { (featured, standard) in
print(featured)
print(standard)
}).disposed(by: disposeBag)