Periodically Call an API with Rxswift

Periodically call an API with RxSwift

Welcome to StackOverflow!

There's quite a lot of operators required for this, and I would recommend to look them up on the ReactiveX Operator page, which I check every time I forget something.

First off, ensure MyModel conforms to Decodable so it can be constructed from a JSON response (see Codable).

let willEnterForegroundNotification = NotificationCenter.default.rx.notification(.UIApplicationWillEnterForeground)
let didEnterBackgroundNotification = NotificationCenter.default.rx.notification(.UIApplicationDidEnterBackground)

let myModelObservable = BehaviorRelay<MyModel?>(value: nil)

willEnterForegroundNotification
    // discard the notification object
    .map { _ in () }
    // emit an initial element to trigger the timer immediately upon subscription
    .startWith(())
    .flatMap { _ in 
        // create an interval timer which stops emitting when the app goes to the background
        return Observable<Int>.interval(10, scheduler: MainScheduler.instance)
            .takeUntil(didEnterBackgroundNotification)
    }
    .flatMapLatest { _ in 
        return RxAlamofire.requestData(.get, yourUrl)
            // get Data object from emitted tuple
            .map { $0.1 } 
            // ignore any network errors, otherwise the entire subscription is disposed
            .catchError { _ in .empty() } 
    } 
    // leverage Codable to turn Data into MyModel
    .map { try? JSONDecoder().decode(MyModel.self, from: $0) } }
    // operator from RxOptional to turn MyModel? into MyModel
    .filterNil() 
    .bind(to: myModelObservable)
    .disposed(by: disposeBag)

Then, you can just continue the data stream into your UI elements.

myModelObservable
    .map { $0.messagesCount }
    .map { "\($0) messages" }
    .bind(to: yourLabel.rx.text }
    .disposed(by: disposeBag)

I didn't run this code, so there might be some typos/missing conversions in here, but this should point you in the right direction. Feel free to ask for clarification. If are really new to Rx, I recommend going through the Getting Started guide. It's great! Rx is very powerful, but it took me a while to grasp.

Edit

As @daniel-t pointed out, the background/foreground bookkeeping is not necessary when using Observable<Int>.interval.

Make a network call every 10 seconds with RxSwift

There are a couple of problems with your Observable.create closure. You have to make sure that something is sent to the observer in every path, otherwise the Observable will call the function and then not emit anything and you will not know why.

Also, you want to minimize the amount of logic being performed in the closure passed to create. You are doing way too much in there.

So let's simplify the code in the create closure as much as possible first:

extension RestManager {
    func rx_makeRequest(withEndPoint endPoint: String, withHttpMethod method: HttpMethod) -> Observable<(response: MyHTTPURLResponse, data: Data)> {
        Observable.create { observer in
            self.makeRequest(withEndPoint: endPoint, withHttpMethod: method) { result in
                if let response = result.response, let data = result.data {
                    observer.onNext((response, data))
                    observer.onCompleted()
                }
                else {
                    observer.onError(result.error ?? RxError.unknown)
                }
            }
            return Disposables.create() // is there some way of canceling a request? If so, it should be done here.
        }
    }
}

This does the bare minimum. Just wraps the underlying callback and nothing else. Now your fetchMarkets call is much simpler:

class MarketService: MarketServiceProtocol  {
    func fetchMarkets() -> Observable <[Market]> {
        return RestManager.shared.rx_makeRequest(withEndPoint: "market/v2/get-summary?region=US", withHttpMethod: .get)
            .do(onNext: { result in
                guard 200...299 ~= result.response.httpStatusCode
                else { throw URLError.httpRequestFailed(response: result.response, data: result.data) }
            })
            .map { try JSONDecoder().decode(MarketResult.self, from: $0.data).marketSummaryAndSparkResponse.markets }
    }
}

Now to the meat of your question. How to make the network call every 10 seconds... Just wrap your network call in a flatMap like this:

Observable<Int>.interval(.seconds(10), scheduler: MainScheduler.instance)
    .flatMapLatest { _ in
        viewModel.fetchMarketViewModels()
    }
    .observe(on: MainScheduler.instance)
    .bind(to: tableView.rx.items(cellIdentifier: HomeTableViewCell.cellIdentifier)) { index, viewModel, cell in
        guard let cell = cell as? HomeTableViewCell else { return }
        cell.setupData(viewModel: viewModel)
    }
    .disposed(by: self.disposableBag)

Learn more about flatMap and its variants from this article.

Updating periodically with RxSwift

This is very similar to this question/answer.

You should use timer and then flatMapLatest:

Observable<Int>.timer(0, period: 20, scheduler: MainScheduler.instance)
    .flatMapLatest { _ in
        provider.request(.Issue)
    }
    .mapArray(Issue.self, keyPath: "issues")
    // ...

Update view after async API call with RxSwift

First, create a generic return object to wrap communication errors.

enum APIResult<T> {
    case success(T)
    case error(Error)
}

Then, convert your completion handler to return an Observable:

func getFoods() -> Observable<APIResult<[FoodType]>> {
    return Observable<APIResult<[FoodType]>>.create { observer -> Disposable in
        self.foodService.getAll(completionHandler: { result in
            switch result {
            case .Success(let foods):
                observer.onNext(.success(foods))
                break
            case .Failure(let error):
                observer.onNext(.error(error))
                break
            }
            observer.onCompleted()

            return Disposables.create()
        })
    }
}

Now simply process the observable as any other in RxSwift.

getFoods().subscribe(onNext: { result in
    switch result {
        case .success(let foods):
            print("Received foods: \(foods)")
            break
        case .error(let error):
            print("Received error: \(error)")
            break
    }
}.addDisposableTo(disposeBag)

---

Using these utility classes will help you mapping success results and split error and success signals to different observables. For example:

let foodsRequest = getFoods().splitSuccess

foodsRequest.error.subscribe(onNext: { error in
    print("Received error: \(error)")
})

foodsRequest.success.subscribe(onNext: { foods in
    print("Received foods: \(foods)")
}

---

You can also convert Realm objects to RxSwift observables:

let realm = try! Realm()
realm.objects(Lap).asObservable()
  .subscribeNext {[weak self] laps in
    self?.tableView.reloadData()
  }

Take a look at Using Realm Seamlessly in an RxSwift App for more information and examples.

RxSwift repeated action

Use flatMap:

var loop: Observable<Element> {
    return Observable<Int>.interval(5.0, scheduler: MainScheduler.instance).flatMap { _ in
        return networkRequest() // returns Observable<Element>
    }
}

Correct way to restart observable interval in RxSwift

What you're looking for is merge. You have two Observables, one of which is an interval and the other which represents preference changes. You want to merge those into one Observable with the elements from both, immediately as they come.

That would look like this:

// this should really come from somewhere else in your app
let preferencesChanged = PublishSubject<Void>()

// the `map` is so that the element type goes from `Int` to `Void`
// since the `merge` requires that the element types match
let timer = Observable<Int>.timer(0, period: 3, scheduler: MainScheduler.instance).map { _ in () }

Observable.of(timer, preferencesChanged)
    .merge()
    .flatMapLatest(makeRequest)
    .subscribeNext(setSummary)
    .addDisposableTo(disposeBag)

Also notice how I'm using timer instead of interval, since it allows us to specify when to fire for the first time, as well as the period for subsequent firings. That way, you don't need a startWith. However, both ways work. It's a matter of preference.

One more thing to note. This is outside of the scope of your question (and maybe you kept it simple for the sake of the question) but instead of subscribeNext(setSummary), you should consider keeping the result as an Observable and instead bindTo or drive the UI or DB (or whatever "summary" is).

---

Related Topics