rx-swift 中的 repeat-while 运算符

Question

我有一个函数可以向服务器发出 API 请求。我想遍历它直到它 returns false（没有更多数据）。

func getData(id: Int) -> Observable<Bool> {
     return Observable.create { observer in
           // Alamofire request
           // parse data
           // if can decode, 
           //   return true and increment page's property 
           // otherwise false
           // error, if there's a problem
     }
}

第一次尝试：我试过使用 takeWhile，例如：getData(id).takeWhile {[=15=]}。它只迭代我的函数 1x。

第二次尝试：使用范围。这里的问题是，即使我的 getData 函数出错，循环没有停止，而是继续！

Observable.range(start: 1, count: 100)
  .enumerated()
  .flatMapLatest({ _ in
        self.getData(someID)
  })
  .subscribe(onNext: { _ in
       // save to DB
       observer.onNext(true)
       observer.onCompleted()
   }, onError: { error in
       observer.onError(error)
   })
   .disposed(by: self.disposeBag)

有办法做到吗，rx 风格？

Answer 1

是这样的吗？

let callApiTrigger = BehaviorRelay<Bool>(value: true)

let callApiEnoughTimes = callApiTrigger.asObservable()
    .takeWhile { [=10=] }
    .flatMap { _ in
        return getData(someId)
    }
    .do(onNext: { (apiResult: Bool) in
        callApiTrigger.accept(apiResult)
    })

Answer 2

takeWhile 和 take(X) 不起作用的原因是它们没有重新订阅 Observable。网络请求 observable 通常最多发出 one 值。

您要查找的内容需要某种形式的递归/重新订阅。如果你想要硬核 Rx，我建议你对 retry 运算符进行逆向工程以满足你的需要。虽然我认为自己对 RxSwift 有经验，但这似乎是一座太远的桥梁。

幸运的是，我想出了一个递归方法，它也能正常工作:)

class PageService {
    typealias Page = (pageNumber: Int, items: [String]?)

    private func getPage(_ pageNumber: Int) -> Observable<Page> {
        let pageToReturn = Page(pageNumber: pageNumber, items: (pageNumber < 3) ? ["bert, henk"] : nil)

        return Observable<Page>
            .just(pageToReturn)
            .delay(0.5, scheduler: MainScheduler.instance)
    }

    func allPagesFollowing(pageNumber: Int) -> Observable<Page> {
        let objectToReturnInCaseOfError = Page(pageNumber: pageNumber + 1, items: nil)

        return getPage(pageNumber + 1)
            // in order to error out gracefully, you could catch the network error and interpret it as 0 results
            .catchErrorJustReturn(objectToReturnInCaseOfError)
            .flatMap { page -> Observable<Page> in
                // if this page has no items, do not continue the recursion
                guard page.items != nil else { return .empty() }

                // glue this single page together with all the following pages
                return Observable<Page>.just(page)
                    .concat(self.allPagesFollowing(pageNumber: page.pageNumber))
        }
    }
}

_ = PageService().allPagesFollowing(pageNumber: 0)
    .debug("get page")
    .subscribe()

这将打印：

2018-03-30 11:56:24.707: get page -> subscribed
2018-03-30 11:56:25.215: get page -> Event next((pageNumber: 1, data: Optional(["bert, henk"])))
2018-03-30 11:56:25.718: get page -> Event next((pageNumber: 2, data: Optional(["bert, henk"])))
2018-03-30 11:56:26.223: get page -> Event completed
2018-03-30 11:56:26.223: get page -> isDisposed

rx-swift 中的 repeat-while 运算符

repeat-while operator in rx-swift

reactive-programming

ios

swift

rx-swift