@alex.krupa send a PR

An example:

override fun onSatelliteClicked() {
        mapSubject.take(1).subscribe(this::toggleSatelliteMode)
    }

@Janez Štupar

ktor

is based on Kotlin coroutines. Instead of

readable.subscrible

you should do

readable.consumeEach

(add

kotlinx-coroutines-rx1/2

to your dependencies).

Isnt there some stuff about writes needing to be sequential? Id swear there were some exceptions about database being locked etc.

depends on what combining logic you need, and 90% of time you one just one emit, so you need take 1, and then its the same

Hi All, I am new to Rx world and stuck into implementing polling mechanism in RxJava2 for Android. I have to call a webservice max x times and at the interval of y seconds on the success or failure of that webservice. By interval I mean wait for y seconds before making another call. For network call call back mechanism is being used.I also need to show UI based on servive response.Can someone here can help me in implementing the same? It would be highly appreciated.

also probably there should be locks around the init? because I do expose the SyncManager to app where you can call sync, so callsite can probably call init() again and fck the thing up

Flatmap

@kenkyee Rx concepts are same on every platform plus kotlin syntax is similar to swift.

Is there a canonical way to subscribe to `Observable`s across

Fragment

boundaries?

My book Learning RxJava is free today. Grab it while you can! https://www.packtpub.com/packt/offers/free-learning

Hey there, how can I use

scan

where the accumulation function has 2 arguments of different types?

I’m trying to do below referencing the following article: https://blog.danlew.net/2015/03/02/dont-break-the-chain/

fun <T> applySchedulers(): FlowableTransformer<T, T> {
        return FlowableTransformer {
            it.observeOn(mainScheduler).subscribeOn(computationScheduler)
        }
    }

I’m try to operate on ‘BehaviorProcessor<List<XXX>>’.

behaviorProcessor.compose(applySchedulers())

works but trying to assign the result to Flowable<List<XXX>> causes an error.

Type Inference failed. Expected Type mismatch.
required: Flowable<List<XXX>>, found: Flowable<List<XXX>!>!

How do I resolve the type inference error?

yes but you need to pass the progress as parameter

i recommend not using viewmodels (the library), but for the general concept of view models you should be posting back to the main thread before they are cached

and you need that keep track of that boolean to know if you are in pip mode state

can someone help me refactoring this to

using

?

contentObservable(resolver, uri, handler)
                .flatMapSingle {
                    Single.fromCallable { resolver.query(uri, projection, selection, selectionArgs, sortOrder) }
                }
                .map<Optional<Contact>> { optionalOf(mapContact(it)) }

where that fromCallable returns Single<Cursor>, so obviously it will not close the cursor if stream gets unsubscribed after that and before mapContact where I actually close the cursor (now), therefore leaking the cursor right?

-- anyone has a clue about

using

usage?

Anyone still using RxJava 1 for some reason?

Also how stable is RxKotlin & how wide spread is it?

why not? isnt that what viewmodels do?

class ViewModel {
   private val relay = BehaviourRelay.create()
   val observable get() = relay

   init {
     repository.
        .operators...
        .subscribe(relay)
   }
}

class Activity {
   onCreate {
      viewModel
         .observable
         .subscribe { }
   }
}

btw how do you make viewmodel´s observable not disposed between config changes without relays, only via replayed hot observable?

refCount

would dispose it after first UI onDestroy, since its the only subscriber

I wanna ask about thread safety in rxjava, where do I need to add it manually? I have a hunch that subscribers, i.e. the onNext lambdas are not, right?

-- if I create scheduler via

Schedulers.from(Executors.newSingleThreadExecutor { Thread(it, "SipCallManagerThread") })

, do I need to shutdown it explicitly like a native resource or is leaving it to be GCed okay?

Is there a

switchMap

which will guarantee atleast one emission of the mapped observable? Or rather

concatMap

that will take 1 || takeUntil new upstream emit?

I am new to Rx. I have to POST total Y number of items in a batches of X items at a time via retrofit. Which Observable, operator, etc. should I be using? I want Rx to send X items at a time until total Y number of items have been sent.

Whos responsibility should it be? Lets say ForegroundGpsProvider, and it would expose that switchMap as a foregroundCurrentPositionObservable. Good. Now however I want to run a sync, sync is not rx api, so we need to subscribe and do it there

This message was deleted.

Is there a solution to diamond problem in combineLatest in rxjava? (when source observables of the combineLatest are driven by common observable -- you get unexpected results since first source will emit first, therefore paired with previous value of second source)