如何在 Reactive Extensions 中组合两个 observables 以对结果进行分页？

Posted 2023-03-10

【中文标题】如何在 Reactive Extensions 中组合两个 observables 以对结果进行分页？

【英文标题】：How to combine two observables in Reactive Extensions in order to paginate results?

【发布时间】：2016-09-02 20:18:18

【问题描述】：

我正在尝试使用 RxSwift 在 ios 应用程序中开发分页系统。用例很简单：用户可以在搜索字段中输入文本，然后应用执行分页请求。当他更改值时，会在第一页执行新请求（这意味着必须将 observable 的值重置为 1）。如果用户清除搜索字段（或输入少于 2 个字符的文本），则清除结果列表并重置当前页面。当用户滚动到列表底部时获取下一页。

这不是 swift 或 iOS 特定的情况，我想它可以使用 RxKotlin 或 RxJs 或任何其他响应式扩展以相同的方式编写。

我目前的尝试是为文本设置一个可观察对象，为当前页面设置一个可观察对象，并将它们组合起来，以便使用这两个参数执行请求。 我已经成功地完成了我正在寻找的事情，但使用全局属性来存储当前查询和当前页面。我想找到一种方法，只使用可观察对象发出的值而不必维护它们（我想代码会更简洁，更容易阅读和理解）。

这是我当前的代码：

    // self.nextPage is a Variable<Int>
    let moreObs: Observable<Int> = self.nextPage.asObservable()
        .distinctUntilChanged() // Emit only if page has changed.

    // self.searchTextObservable is a PublishedSubject<String> that receives the values from the textfield
    let searchObs: Observable<String> = self.searchTextObservable
        .throttle(0.4, scheduler: MainScheduler.instance) // Wait 400ms when the user stops writing.
        .distinctUntilChanged() // Emit only if query has changed.

    self.resultsObservable = Observable
        .combineLatest(searchObs, moreObs)  query, page in
            return ["q": query, "p": "\(page)"]
        
        .subscribeOn(MainScheduler.instance) // Emit on main thread.
        .observeOn(ConcurrentDispatchQueueScheduler(globalConcurrentQueueQOS: .Background)) // Perform on background thread.
        .map  params in
            if params["q"]!.characters.count > 2 
                return params
            
            return [:]
        
        .flatMap  params in 
          return params.isEmpty ?
            Observable.of([]) :
            self.search(params)
        
        .map  results in
            if results.count > 0 
                self.results.appendContentsOf(results)
             else 
                self.results = []
            
            return self.results
    

到目前为止，唯一不起作用的功能是对 nextPage 的值的重置操作。如果我在 searchObs 发出时将其强制为1：

let searchObs: Observable<String> = self.searchTextObservable
        .throttle(0.4, scheduler: MainScheduler.instance) // Wait 400ms when the user stops writing.
        .distinctUntilChanged() // Emit only if query has changed.
        .map query in
          self.nextPage.value = 1
          return query
        

然后我执行了 2 个请求。

我在滥用 Rx 吗？

【参考方案1】：

我不会使用combineLatest。您的页码取决于您当前的搜索文本，因此您应该使用flatMapLatest 链接它。这样一来，您无需自己负责维护其状态，而是让操作员链接为您重置。

let disposeBag = DisposeBag()

let searchText = PublishSubject<String>()  // search field text
let newPageNeeded = PublishSubject<Void>() // fires when a new page is needed

struct RequestPage 
    let query: String
    let page: Int


let requestNeeded = searchText.asObservable()
    .flatMapLatest  text in
        newPageNeeded.asObservable()
            .startWith(())
            .scan(RequestPage(query: text, page: 0))  request, _ in
                return RequestPage(query: text, page: request.page + 1)
            
    

requestNeeded
    .subscribeNext  print($0) 
    .addDisposableTo(disposeBag)

searchText.onNext("A")

searchText.onNext("B")
newPageNeeded.onNext(())

searchText.onNext("C")
newPageNeeded.onNext(())
newPageNeeded.onNext(())

这将输出：

(RequestPage #1)(查询：“A”，页面：1) (RequestPage #1)(查询：“B”，页面：1) (RequestPage #1)(查询：“B”，页面：2) (RequestPage #1)(查询：“C”，页面：1) (RequestPage #1)(查询：“C”，页面：2) (RequestPage #1)(query: "C", page: 3)

【讨论】：

这太完美了！