that's

.buffer(2, 1).map(..)

or maybe

window(2, 1)

i always forget which is which

Perfect! Thank you

I have ported a rather big app to RxJava2, it compiles and runs, but my unit tests which depend on rx all are failing with this error:

java.lang.NoClassDefFoundError: Could not initialize class io.reactivex.android.schedulers.AndroidSchedulers

googling doesn't help me for some reason to find the answer

My main app is > 64K methods, but I'm not using multidex, rather I just run proguard. I suspect this might be the reason, but I'm not sure how multidex could actually be related to JVM-only unit tests...

so maybe that's not it...

Is proguard running against your test build? I’m not 100% how that works. did you try to keep io.reactivex.android.schedulers.AndroidSchedulers ?

hmm. not sure, I'll try to turn it off and see what happens when building tests... (as for keep - i don't think it's necessary because that class is surely used...)

You’re right, keep likely isn’t necessary.

Idk if this is relevant http://stackoverflow.com/questions/34621524/didnt-find-the-class-rx-android-schedulers-androidschedulers

yeah, I've seen this, but it is related to enabling multidex which I didn't do...

Right, good point.

omg I hope this the best channel in the world, I have some many question about rx

So here I go : Hi guys ! I’m wondering if there a way to do this in rx : I have an Rx Observable that will take a various time to call onNext, first time 5 seconds, next time 1ms Is there a way to subscribe, and if the OBS is taking more than 1ms, have a special call back/onNext() on another observable?

I have ported a rather big app to RxJava2, it compiles and runs, but my unit tests which depend on rx all are failing with this error: java.lang.NoClassDefFoundError: Could not initialize class io.reactivex.android.schedulers.AndroidSchedulers

I kinda figured this out, but still don't know what happens. So if I call this in

@BeforeClass

in my unit tests:

fun injectTestingRxSchedulers() {
  RxJavaPlugins.reset()
  RxJavaPlugins.setInitComputationSchedulerHandler { Schedulers.trampoline() }
  RxJavaPlugins.setComputationSchedulerHandler { Schedulers.trampoline() }
  RxAndroidPlugins.reset()
  RxAndroidPlugins.setInitMainThreadSchedulerHandler { Schedulers.trampoline() }
  RxAndroidPlugins.setMainThreadSchedulerHandler { Schedulers.trampoline() }
}

I receive a class def not found errors and also exception-in-initializer errors. But if I strip it down to only

fun injectTestingRxSchedulers() {
//  RxJavaPlugins.reset()
//  RxJavaPlugins.setInitComputationSchedulerHandler { Schedulers.trampoline() }
//  RxJavaPlugins.setComputationSchedulerHandler { Schedulers.trampoline() }
  RxAndroidPlugins.reset()
  RxAndroidPlugins.setInitMainThreadSchedulerHandler { Schedulers.trampoline() }
  RxAndroidPlugins.setMainThreadSchedulerHandler { Schedulers.trampoline() }
}

then no exceptions are thrown! So this is not a class path error. But some tricky hook-related one, I don't know them that well...

What happens when you run with that commented out block below RxAndroidPlugins fns? does initialization order matter somehow?

are you able to get any other class in

@BeforeClass

?

What happens when you run with that commented out block below RxAndroidPlugins fns? does initialization order matter somehow?

Perhaps it's related, but it doesn't help, still same errors... 🙂 I've heard that hooks are tricky, I guess I should inject schedulers instead, but the codebase is rather large, need some "quick" solution for now...

are you able to get any other class in @BeforeClass?

yes, seems like there's no problem with that

@jw - it was buffer. The following achieved what I needed

fun <T, R> Observable<T>.scanMap(func2: (T?, T) -> R): Observable<R> {
    return this.startWith(null as T?)
            .buffer(2, 1)
            .map { func2.invoke(it[0], it[1]) }
}

https://gist.github.com/ZakTaccardi/06512860c37c732da165f6b4c2be40a5

guys I keep getting this error

FATAL EXCEPTION: RxCachedThreadScheduler-1  java.io.InterruptedIOException: thread interrupted

if I dispose a pending okhttp3 call in an observable? but why does the app crash? should not observable catch the error and delegate it to onError?! I dont understand this..... 😞

I need help

@mingkangpan do you have a code sample? Do you have an

onError

?

you'll need an

onError

handler in your

.subscribe()

In @RxJava, can

autoConnect()

cause memory leaks due to persistent subscription to source when no more downstream subscribers are present? Especially with

replay()

?

Can the same issue happen with

cache()

as well?

Neither of those operators have a notion of being disposed at any time, so I'm curious if the GC will not collect them due to their persistent subscription...

Depends on the references the source has and where you store the observable. If you never release the reference, then yes it can cause a memory leak.