i.e. start timer from subscription

switchmap will prevent that

i thought of this

upstream.flatmap(foo -> timer().map(specialFoo).startWith(foo))

but yea id need some "null" value startwith for upstream

that still requires one item to emit

yea

timeout is pretty much precisely what I want .. how about catching its timeoutexception and resubscribing to upstream?

sure. you could probably do

upstream.publish(u -> merge(timer().map(specialFoo).takeUntil(u), u.take(1)).retryWhen { true })

or something

that looks good

im bit confused on publish .. its makes downstream hot, so unsubcribing wont stop it right?

btw doesnt retry work on errors? those takes will complete it so,

repeat

?

repeat, yes, my bad. publish prevents multiple subscriptions upstream because we subscribe over and over again, if it's already hot don't bother.

oh, thanks .. but doesnt publish create connectable observable which you need to connect() to and all that stuff?

or is that overload different?

that is different

i see, thank you very much .. btw since it prevents mutiple subscriptions .. its also prevents canceling upstream when downstream unsubscribes?

(like normal cold observables do?)

it will propagate unsubscription through itself to the downstream observable that is returned, it's only inside the lambda that subscribing/unsubscribing has no effect on the upstream

addNoteUseCase.add(note)

returns a

Completable

Note(text = note.nodeText)

can throw

IllegalArgumentException

. What is the recommended way to handle the error (if it occurs, the error is not caught in

onError

since the emission has not occurred? Or should validation be done outside of the

Note

constructor?

This should do it:

actions.ofType<Action.AddNote>()
                .map { Note(text = it.nodeText) }
                .switchMap { note ->
                    addNoteUseCase.add(note)
                            .subscribeOn(<http://Schedulers.io|Schedulers.io>())
                            .toSingleDefault<Change>(Change.NoteAddSuccess)
                            .onErrorReturn { Change.NoteAddError(it) }
                            .toObservable()
                }

thanks @arekolek. while it compiles, that does not seem to do the trick. it fails in step

.map { Note(text = it.nodeText) }

java.lang.IllegalArgumentException: Failed requirement. and the rest of the chain does not get executed.

Yeah, I see I misunderstood completely. This should be better:

actions.ofType<Action.AddNote>()
                .switchMapSingle { note ->
                    Single.fromCallable { Note(text = note.nodeText) }
                            .flatMapCompletable { addNoteUseCase.add(it) }
                            .subscribeOn(<http://Schedulers.io|Schedulers.io>())
                            .toSingleDefault<Change>(Change.NoteAddSuccess)
                            .onErrorReturn { Change.NoteAddError(it) }
                }

Hi guys. Is it a good practice to use

ReplaySubject

or

ReplayRelay

as some kind of cash mechanism in a repository exposing it as an

Observable

?

I’m not 100% sure, but I’m pretty sure

ReplaySubject

stores its history indefinitely.. so just keep that in mind. Also, the only time you get every result replayed is when you attach a subscriber to it. I’m not saying not to, there’s just probably a better mechanism available (like writing your own cache).

Relays where written for the unfortunate case that somebody can call onError or onComplete on your Subject, which will terminate it. Btw

replay().refCount()

should do it for caching (emitting last on subscribe)(

is it possible in kotlin to have a getter that returns a more generic type than its backing property to an external interface? Or is there a better way of solving this pattern? I find myself frequently writing stuff like

private val _events: ReplaySubject<Thing>
val events: Observable<Thing>
    get = _events

that's the pattern i follow

better to use a

val

to expose an observable? ex:

repo.events

vs

repo.events()

Is there any talk from the kotlin designers about having a more general public type for a field? I can't really figure out what the programming language term for that sort of construct would be, so i'm having a hard time finding open issues / discussions about it.

what would the syntax be?