kotlinlang #rx

I'm trying to create a continuous stream of items where I'd always get the last one (aka. cache) using the BehaviorSubject (or ReplySubject) but whenever the api call stream terminates, so does my main stream. How can I prevent that? One way would be linking them manually and ignoring the onComplete but then I loose the main feature of the Subject... Let me explain with an example. Here I only get the result the first time.

-.kt

Mohamed Ibrahim

04/16/2019, 7:33 PM

Hello, I have a case which I need to retrieve data from local first then show it, after that retrieve data from API, then override local data, then show the updated version. I started with concat operator and it doesn't work I think because first time there is no local data to get

nic

04/18/2019, 7:21 PM

Hey there, tried searching for this, but wasn't able to pull any results. What's the best way to take an Observable, FlatMapSingle to a Single Observable, but wait until we receive the result of the Single Observable, and decide to either complete or move onto the next one. For example:

getSomeObservable().flatMapSingle { t -> performAPICallWithData(t) }

I don't want to

performAPICallWithData

until the first one has completed, and I want to use the result from

performAPICallWithData

to determine wether I should perform the next one or complete everything

oday

04/20/2019, 3:33 PM

unless SingleSource is something that’s not the developer calling the method

ursus

04/20/2019, 10:11 PM

------ Guys, how can I "pass through" emit and also trigger new observable?

ursus

04/23/2019, 8:31 PM

@Mohamed Ibrahim what the issue? whole point of retryWhen is that it will resubscribe the source when the function emits anything, and will give up if throwable is emitted

ursus

04/23/2019, 8:32 PM

@nic flatmap?

locke

04/29/2019, 4:39 AM

Is there a better way to store an observable result for replay than replay subject? I have a stream that I want to create and execute the first time it's called, and then afterwards subscribing to that observable would just return the items that were originally emitted during the first call, rather than redo all of the processing. Describing this sure sounds like describing the lazy pattern (do work once, return repeatedly) but Googling the idea just gives you lazy stream generation, not results.

Mohamed Ibrahim

04/30/2019, 11:50 AM

I have a publish relay which emits items regularly, I just want to get the last tow items each time it emits, the case is to evaluate UI state according to previous and current emissions

alexsullivan114

04/30/2019, 6:03 PM

I'm using

Observable.interval

to count down from 30 seconds to zero. After that, I want to use

flatMap

and continue another Rx chain. I'm not sure what the best way is to actually know when

after that

is. Using

flatMap

directly after the

interval

call obviously doesn't work since the

flatMap

gets called after each tick of the interval. RIght now I'm using

materialize

and checking if the notification

isOnComplete

. Is that the best way to handle this?

ursus

05/02/2019, 6:05 AM

Is there a way to force a subclass to subscribe to observable, sort of like abstract methods?

ursus

05/05/2019, 5:17 PM

Does anyone know why RxJava fight so much not having subscribers?. I want to just have the action finish regardless of anyone listening; and if they are they get result; else doesnt matter. Sort of like event buses used to do

iex

05/06/2019, 1:31 AM

How do I prevent subscribing a subject multiple times to an observable?

subscribe(subject)

doesn't return a disposable.

alexsullivan114

05/06/2019, 5:33 PM

Ray Wenderlich is having a sale on a few Android books - one of which is is about reactive programming with Kotlin on Android. Check it out if you're interested! https://www.raywenderlich.com/2887766-advanced-spring-fling-sale-ends-this-friday-may-10

Arkadii Ivanov

05/12/2019, 1:49 PM

https://badootech.badoo.com/reaktive-a-multiplatform-library-for-reactive-kotlin-android-ios-77b6ff25adb1

👏 1

😍 2

05/13/2019, 3:14 PM

Why doesn't this mention Flow which seems like the obvious choice for multiplatform, Kotlin, null-safe Rx?

ursus

05/26/2019, 3:51 AM

How do you guys deal with pagination and database observables? I have refresh() in the domain layer which calls the api, and writes to the database

fun refresh(page: Int) : Completable {
    .. call api
   database.write(...)
}

(edited) and now I want to display progressbar while this is happening + progressbar at the end if page > 1 (edited) the trouble with this is that presentation layer has a query observable opened liek this

ViewModel {

   database.fooQuery()
     .subscribe { ... post to adapter }
}

which means that this will emit at the time database.write is called so If map refresh to events of Started, Success, Error Success will come after the database is refreshed which to me is weird, since it should be done in one go, but that means refresh would need to return that data too (edited) tldr; writing to the database is side-effecting - how to deal with that?

Mohamed Ibrahim

06/02/2019, 3:10 PM

Could anyone tell us the difference between the Kotlin Flow api and Rxjava, is it a Kotlin implementation for the reactive extension or is it a different thing

06/04/2019, 12:42 PM

because the library is built with Java 6 and lacks type_use annotations to denote that the generic is non-nullable

👍 1

ursus

06/07/2019, 6:21 AM

I really believe it should have been different types

ubu

06/07/2019, 10:31 AM

Hi there. I'm looking for a way to collect `Observable`'s emissions in bundles of data, some sort of history of the last values. I know it sound like

buffer()

operator, but I just can't get it right. When having

emitted one after another with some delay, I would like to have an

Observable

emitting:

[A]

[A, B]

[A,B,C]

[B, C, D]

[C, D, E]

[D, E, F]

. We are only interested in last values, so I suppose, we could use here a

Flowable

. Thanks.

ivano

06/13/2019, 7:55 AM

In java is more syntactic sugarazied than in Kotlin?

ursus

06/14/2019, 4:43 PM

there is no switchMap? That like my 1# operator 😕

Mohamed Ibrahim

06/16/2019, 1:36 PM

I have a Behavior Relay to control

MediaPlayer

(in Activity) from

viewModel

, and I have another for settings .. now I want the player relay to only

skipUntil

the settings emit an enable sound item. from docs

skipUntil

only waits for an Observable to emit at least an item. but what if need to check on that item too, should I use something else or there is something missing

myanmarking

06/18/2019, 3:07 PM

observable.create(BatteryLevelObservable()) is the same thing no ?

myanmarking

06/19/2019, 9:22 AM

Observable.create<Unit> { System.out.println("Yo") }

class PrintObservable: ObservableOnSubscribe<Unit> {
    override fun subscribe(emitter: ObservableEmitter<Unit>) {
        System.out.println("Yo")
    }
}

Observable.create(PrintObservable())

alexsullivan114

06/19/2019, 1:26 PM

Is the

BehaviorRelay

in the

ViewModel

or in the fragment/activity?

liminal

06/21/2019, 4:43 AM

Trying to justify using RxKotlin on top of RxJava 2. So far I like the plusAssign operator to add to CompositeDisposable as well as clarity of writing out of onNext, onError, etc. anything else that you like? I wonder how RxKotlin will affect RxJava 2 to RxJava 3 conversion

ursus

06/21/2019, 3:25 PM

private fun doPostMessage(...) {
    var messageId: Long? = null
    try {
		messageId = messageRepository.insertSendingTextMessage(...)

        val apiMessage = apiClient.postMessage(...)

        messageRepository.updateSentMessageAndParents(apiMessage)

    } catch (ex: Exception) {
        if (messageId != null) messageRepository.updateMessageState(messageId, MessageState.ERROR)
        throw ex
    }
}

ursus

06/21/2019, 3:26 PM

but how should I do the catch block where it sees the messageId param?

Title

ursus

06/21/2019, 3:26 PM

but how should I do the catch block where it sees the messageId param?

kioba

06/21/2019, 3:36 PM

messageRepository
.insertSendingTextMessage(...)
  .flatMap { messageId ->
    apiClient.postMessage(...)
      .flatMap{ apiMessage -> messageRepository.updateSentMessageAndParents(apiMessage) }
      .onErrorResumeNext { error -> 
        messageRepository.updateMessageState(messageId, MessageState.ERROR) }
        .flatMap{ Observable.error(error) }
      }// ONLY if you want to trigger the error but My suggestion would be to return a Succes() / Error() sealed class.
  }

the

.flatMap{ Observable.error(error) }

at the end is there only because when you catch an exception you also throw it again. but it is up to you if you want to keep it or not 🙂

ursus

06/21/2019, 3:55 PM

Yea nesting, I thought about that. Thanks. Btw now, how would you add the

if (messageId != null) messageRepository.updateMessageState(messageId, MessageState.ERROR)

catch?

just throw from the

insertSendingTextMessage

, right?

kioba

06/21/2019, 3:58 PM

right now if

insertSendingTextMessage

fails, it never executes the inner part so the

onErrorResumeNext { updateMessageState() }

never gets called.

ursus

06/21/2019, 3:58 PM

yes, nesting for visiblity gave it to you automatically, nice, thanks

kioba

06/21/2019, 3:59 PM

yes exactly as you said, any time 👍

View count: 2