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

Question

我正在尝试使用 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 吗？

Answer 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)