its probably not pure right, I would use something with implicit "memory" like scan but it just doesnt work out, ill endup returning buffer from scan and then mapping to get the stuff from it, or returning a tuple of buffer and event, thats kind of silly

Anyone know how to

switchOnNext

against an

Observable<Completable>

? Essentially I want to keep a subscription to the most recently emitted

Completable

emitted by an

Observable

, but for some reason that's proving challenging.

Hey guys, I’m trying to add a retry mechanism to a web call (made via RxJava + Retrofit) and I’m not sure what I should use - retry/When/Until or takeUntil. My problem with retry is that as far as I can tell it retries only on error, but my case is that I might get a PENDING or IN_PROGRESS status in the response body (the response itself being successful). So I need to retry the call let’s say 5 times or until I get a COMPLETED in the response.body(). Can anyone help me with that?

it has the same behavior as publish that doesn't take a function except it's calling connect() and managing the subscription upstream automatically

interface Data

data class Response(
        val data: List<Data>,
        val hasNextPage: Boolean,
        val nextPageId: Int?
)

fun makeRequest(nextPageId: Int): Single<Response> = TODO()

fun getData(page: Int = 0): Flowable<Data> = makeRequest(page)
        .flatMapPublisher {
            Flowable.fromIterable(it.data)
                    .concatWith(it.nextPageId?.let(::getData) ?: Flowable.empty())
        }

@ubu not sure how you want to do it without recursion though

@arekolek, thanks. this is what I wanted to ask you about. Actually, does iterating over a list of data seem redundant? Why not just take last element of the collection?

I think its called spread spreadout something like that

Is there any way to say "switch back to caller thread"? For example, I want to wrap some callback call into a single, but the implementation of the callback is switching my thread to

Main

. As a result, the whole chain is now on the

Main

, which is not obvious, if you look at the declaration. Ex:

....
   .map { single_that_switches_the_thread_to_Main() }
   .map {}  // <--- NOT io
  .subscribeOn(io)

So, I'd like to be able to tell to this

single_that_switches_the_thread_to_Main()

to switch to the caller thread, inside of it, not on the usage side.

More a design question, but do you wrap all your events in

Result

class? In Repository as well? I see this idea popping up here and there and I just feel like it's a useless noise. There is a case when you woudn't want to break the stream on error (in UI layer, for example), but then I'd just catch the error and emit a special type of item. I'd rather do this, then wrap/unwrap the

Result

everywhere. Seems like google's influence with its primitive

LiveData

with no error handling support, where you have no choice other than adding exceptions to the data itself.

Same here. Any opinions?

id go for something like isInitialized.filter true.delay { ... }

Hello all, I have a question about the best approach to handling conditional/if-else situation. I have an API response that will return one of three classes:

Loading

,

Error

, or

Finished

— all of which inherit from a class called:

NetworkState

. If its Loading or Error, I want to update my UI appropriately. If its finished, I need to extract the data, do further transformations, then update the UI. I also need to perform a side-effect depending on one of the values returned. I can think of two ways to do this. Which of these is preferable? Is there a better third option? OPTION 1

fetchFromApi().
  .map {
    when (it) {
      is Loading -> transformToLoadingView()
      is Error -> transformToErrorView()
      is Finished -> mapDataToUI(it.data)
  }.subscribe { view -> updateUi(view) }

fun mapDataToUI(data: Data) {
  performSideEffect(data)
  transformDataToView(data)
}

OPTION 2

val sharedResponse = fetchFromApi().share()

Observable.merge(
  sharedResponse.ofType('Loading').map { loadingView },
  sharedResponse.ofType('Error').map { errorView },
  sharedResponse.ofType('Finished').map { transformDataToView(it) }
).subscribe { view -> updateUi(view) }

sharedResponse.ofType('Finished').subscribe { performSideEffect(it) }

Guys, super quick question. Any idea why this test is not passing? I would expect

firstOrError

to throw error if subject has no values.

@Test
    fun `assert tha behaviour subject behaves like i would expect to behave`() {
        BehaviorSubject.create<Boolean>()
                .firstOrError()
                .test()
                .assertComplete()
    }

Because it's a representation of a higher order function that lets you pass around observable operators as data

is it the same functional argument as for not using exceptions, i.e mapping them to some monad / adt?

@arekolek so you mean to rethrow basically?

mostly what I mean is like this

val request = apiRequestSingle()
	.map { data -> Result.Success(data) }
	.onErrorReturn { throwable -> Result.Error(throwable)}

request
	.flatMap { result ->
		if(result is ResultSuccess) {
			saveToDatabaseObservble(it)
				.map { result }
		} else {
			Observable.just(result)
		}
	}
	...

vs basically

apiRequestSingle()
   .flatMap { data -> saveToDatabaseObservable(data)
                      .map { data }
    }

Hey guys, could you recommend a good book or online course for Rx (also maybe for functional programming with Kotlin/Java)?

Every

doOnXXX()

is intended for side effect, e.g. logging

Also it shouldn't be switchMap x.toObservable bit switchMapCompletable

why, just have the view as parameter

sounds like you want

Completable

but updating form automatically as db changes, under user's fingers, prolly not a good idea?

If I have a network request with

replay(1).autoConnect()

and the stream errs, what is the best way to not cache that error? What I would like is basically to restart the stream.

If I have a call, that is potentially leaking (no way of removing the listener)

and dispose its subscriptions, nothing leaks?

is that what you are looking for?