How to cancel previous request using RxAlamofire?

Question

I want to cancel the previous request using RxAlamofire.
But I don't know where to call the cancel function.
I have a searchBar, and I call the API in function "textdidchange".
So, I want to cancel the previous request and call the API with new parameters.
Have any suggestion to help me?
Thanks.

func request(_ method: Alamofire.HTTPMethod, url:String, params:[String:Any] = [:], callback: @escaping (JSON) -> Void) {

    var headers:[String:String] = [String:String]()
    if token.isEmpty == false {
        headers["Authorization"] = "Bearer \(token)"
    }

    let configuration = URLSessionConfiguration.default
    configuration.httpAdditionalHeaders = SessionManager.defaultHTTPHeaders
    configuration.timeoutIntervalForRequest = 10.0

    _ = SessionManager(configuration: configuration)
        .rx.responseJSON(method,
                         url,
                         parameters: params,
                         encoding: ((method == .get) ? URLEncoding.default : JSONEncoding.default),
                         headers: headers)
        .subscribeOn(SerialDispatchQueueScheduler.init(qos: .background))
        .subscribe(onNext: { (r, data) in

            let json = JSON(data)

            if json["status"].stringValue == "success" {

                callback(json["responseData"])

            }else {

                callback(json)
            }
        }, onError: { (error) in

            callback(JSON(error))

        })
        .addDisposableTo(ResfulAPIDisposeBag)
}

Answer 1

We would need to add some delay, that will fire up the request after X seconds after the typing, but only if the phrase didn’t change.

let searchResults = searchBar.rx.text.orEmpty
    .debounce(0.5, scheduler: MainScheduler.instance)
    .distinctUntilChanged()
    //if search: doesn't finish, the observable is cancelled
    .flatMapLatest { query -> Observable<[Repository]> in
        if query.isEmpty {
            return .just([])
        }
        return search(query)
            .catchErrorJustReturn([])
    }
    .observeOn(MainScheduler.instance)

Answer 2

When you subscribe to an Observable, the resulting object is a Disposable which contains the subscription. This disposable can either be disposed manually (yourSubscription.dispose()), or you can add it to a DisposeBag. When the dispose bag is deallocated, all Disposables it contains are disposed. In your code, that would be ResfulAPIDisposeBag.

let subscription = Observable<Int>.interval(0.1, scheduler: MainScheduler.instance)
    .debug("manually disposed")
    .subscribe()

subscription.dispose()

Or, using a Dispose Bag:

var disposeBag = DisposeBag()

Observable<Int>.interval(0.1, scheduler: MainScheduler.instance)
    .debug("using a dispose bag")
    .subscribe()
    .disposed(by: disposeBag)

// deallocates the first disposeBag and disposes any Disposables inside
disposeBag = DisposeBag()

Note: .addDisposableTo(_) is deprecated in the latest version of RxSwift, use .disposed(by: _) instead.