Add subscribeBy extensions for proxy classes. #127
Conversation
Unfortunately, since the SubscribeProxy classes don't extend the observable class they proxy to, extension functions written for the observable class can't be used. One commonly used set of extensions is RxKotlin's subscribeBy. This commit adds copies of those extensions that work on SubscribeProxies.
|
|
|
While I like the idea, I think this would be better served as a recipe/example in the sample app rather than the public supported API. It's simple enough that people can add it in their own projects as they see fit, and keep the kotlin artifacts as lean as possible. What do you think? |
|
I think that keeping the artifacts small is a generally good idea, and these extensions could be considered outside the scope of the library. It seems like these particular extensions might be worth including due to the fact that they are the only ones in RxKotlin that don't work with AutoDispose, and they are fairly widely used. But it's up to you. |
|
My gut says it's better to make these a separate thing - erring on the side of restricting the API surface area in autodispose itself. If you want to put them in the sample though, I'd be happy to merge that and point to them as a recipe for RxKotlin users. This also sounds like a good library idea for you if you wanted to maintain it yourself :) |
|
Ok, I moved the extensions to the sample recipes and updated the activity to use them. Is that what you had in mind? |
| @@ -0,0 +1,54 @@ | |||
| package com.uber.autodispose.recipes | |||
|
|
|||
| import com.uber.autodispose.* | |||
| private val onErrorStub: (Throwable) -> Unit = { RxJavaPlugins.onError(OnErrorNotImplementedException(it)) } | ||
| private val onCompleteStub: () -> Unit = {} | ||
|
|
||
|
|
| import io.reactivex.exceptions.OnErrorNotImplementedException | ||
| import io.reactivex.plugins.RxJavaPlugins | ||
|
|
||
| private val onNextStub: (Any) -> Unit = {} |
There was a problem hiding this comment.
Let's add a file doc above this that explains how this is an example of how you can tack on extension functions to the proxy interface
/*
* Stuff!
*/| import io.reactivex.plugins.RxJavaPlugins | ||
|
|
||
| private val onNextStub: (Any) -> Unit = {} | ||
| private val onErrorStub: (Throwable) -> Unit = { RxJavaPlugins.onError(OnErrorNotImplementedException(it)) } |
There was a problem hiding this comment.
I'd like to propose diverging from how RxKotlin does this. The reason is that RxJava actually does special handling for missing error implementations via LambdaConsumerIntrospection, that supplying this would not catch. I'd say for the cases where only a next/success/complete consumer/action is passed in, let's call through to the unhandled one in rxjava. Will comment inline with specifics
There was a problem hiding this comment.
Hm, interesting. I added RxKotlin's current implementation before LambdaConsumerIntrospection was in place, and it matched RxJava's behavior at the time. I'll look into your PR in detail later, but do you think RxKotlin should also be updated?
There was a problem hiding this comment.
Yeah I'd say it probably should. Would be a non-breaking change, but it would allow rxjava to properly expose the introspection details
| onError: (Throwable) -> Unit = onErrorStub, | ||
| onComplete: () -> Unit = onCompleteStub, | ||
| onNext: (T) -> Unit = onNextStub | ||
| ): Disposable = subscribe(onNext, onError, onComplete) |
There was a problem hiding this comment.
Adjust for LambdaConsumerIntrospection
if (onError === onErrorStub && onComplete === onCompleteStub) {
return subscribe(onNext)
} else {
return subscribe(onNext, onError, onComplete)
}| onError: (Throwable) -> Unit = onErrorStub, | ||
| onComplete: () -> Unit = onCompleteStub, | ||
| onNext: (T) -> Unit = onNextStub | ||
| ): Disposable = subscribe(onNext, onError, onComplete) |
There was a problem hiding this comment.
Adjust for LambdaConsumerIntrospection
if (onError === onErrorStub && onComplete === onCompleteStub) {
return subscribe(onNext)
} else {
return subscribe(onNext, onError, onComplete)
}| fun <T : Any> SingleSubscribeProxy<T>.subscribeBy( | ||
| onError: (Throwable) -> Unit = onErrorStub, | ||
| onSuccess: (T) -> Unit = onNextStub | ||
| ): Disposable = subscribe(onSuccess, onError) |
There was a problem hiding this comment.
Adjust for LambdaConsumerIntrospection
if (onError === onErrorStub) {
return subscribe(onSuccess)
} else {
return subscribe(onSuccess, onError)
}| onError: (Throwable) -> Unit = onErrorStub, | ||
| onComplete: () -> Unit = onCompleteStub, | ||
| onSuccess: (T) -> Unit = onNextStub | ||
| ): Disposable = subscribe(onSuccess, onError, onComplete) |
There was a problem hiding this comment.
Adjust for LambdaConsumerIntrospection
if (onError === onErrorStub && onComplete === onCompleteStub) {
return subscribe(onSuccess)
} else {
return subscribe(onSuccess, onError, onComplete)
}| fun CompletableSubscribeProxy.subscribeBy( | ||
| onError: (Throwable) -> Unit = onErrorStub, | ||
| onComplete: () -> Unit = onCompleteStub | ||
| ): Disposable = subscribe(onComplete, onError) |
There was a problem hiding this comment.
Adjust for LambdaConsumerIntrospection
if (onError === onErrorStub) {
return subscribe(onComplete)
} else {
return subscribe(onComplete, onError)
}| @@ -45,7 +46,7 @@ class KotlinActivity : AppCompatActivity() { | |||
| Observable.interval(1, TimeUnit.SECONDS) | |||
| .doOnDispose { Log.i(TAG, "Disposing subscription from onCreate()") } | |||
| .autoDisposeWith(scopeProvider) | |||
| .subscribe { num -> Log.i(TAG, "Started in onCreate(), running until onDestroy(): " + num) } | |||
There was a problem hiding this comment.
nit: let's keep the num, as it was an intentionally more specific name
|
I'm not sure I understand what you mean by filling out method bodies? |
|
As in instead of: fun <T : Any> ObservableSubscribeProxy<T>.subscribeBy(
onError: (Throwable) -> Unit = onErrorStub,
onComplete: () -> Unit = onCompleteStub,
onNext: (T) -> Unit = onNextStub
): Disposable =
if (onError === onErrorStub && onComplete === onCompleteStub) {
subscribe(onNext)
} else {
subscribe(onNext, onError, onComplete)
}Do this fun <T : Any> ObservableSubscribeProxy<T>.subscribeBy(
onError: (Throwable) -> Unit = onErrorStub,
onComplete: () -> Unit = onCompleteStub,
onNext: (T) -> Unit = onNextStub
): Disposable {
return if (onError === onErrorStub && onComplete === onCompleteStub) {
subscribe(onNext)
} else {
subscribe(onNext, onError, onComplete)
}
} |
As we discussed at Droidcon, since the SubscribeProxy classes don't extend the
observable class they proxy to, extension functions written for the
observable class can't be used.
One commonly used set of extensions is RxKotlin's subscribeBy. This
commit adds copies of those extensions that work on SubscribeProxies.