Rxswift: Can we stop the rx observable stream

[trungp]

Hi there,

I have an stream like below:

.map {
    // Do something to get data with some condition
}
.subscribe {
   // Do something to update data source for a table
}
.addDisposable(myDisposable)

In the map block, I have some condition to check/validate data and can we stop the stream on the map block?, I dont want the stream continuous running to subscribe block in some case (If the condition check is not true).

Please help!

Thanks,

[mbalex99]

You want the takeWhile operator. It is similar to the filter operator except that it will complete the stream and call any Anonymous disposables if they're running. The take~ family of operators is geared towards interruption.

    [1,2,3,4,5,6,7].toObservable()
            .takeWhile { (val) -> Bool in
                return val < 4
            }
            .subscribe(onNext: { (val) -> Void in
                print("Value: ", val)
            }, onError: { (err) -> Void in
                print("Error", err)
            }, onCompleted: { () -> Void in
                print("completed")
            }) { () -> Void in
                print("disposed");
            }

This will print

Value: 1
Value: 2
Value: 3
completed
disposed

Let me know if this the proper behavior that you want.

[trungp]

Great,
Thank @mbalex99 for your answer!

[AndroidDeveloperLB]

What if I have a different RX function that I use?

Example:

 Single.fromCallable<String> {
//here some condition that I want to be able to break
return ...
}.flatMap ...