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

Question

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

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

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

这是我当前的代码:

    // 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)(query: "A", page: 1)
(RequestPage #1)(query: "B", page: 1)
(RequestPage #1)(query: "B", page: 2)
(RequestPage #1)(query: "C", page: 1)
(RequestPage #1)(query: "C", page: 2)
(RequestPage #1)(query: "C", page: 3)