Expose observable source for ObservableSubscribeProxy

[jaychang0917]

fun <T : Any> Observable<T>.mySubscribe(f: () -> Unit): Disposable {
  return doOnSubscribe { f() }
    .subscribe(Consumer {}, Consumer {})
}

With .to(ObservableScoper(AndroidLifecycleScopeProvider.from(this))), we can't take advantage of kotlin's extension to use my custom mySubscribe method since it is a ObservableSubscribeProxy type.

Do you think it is better to expose the observable source for ObservableSubscribeProxy?

[ZacSweers]

In the face of it, I don't think so. It has an intern observable but disposal of intermediary streams is not something we want to encourage. Disposal only travels up, so everything below it would be unbound (this caused a lot of issues when using something via compose()).

What would the API look like?

[jaychang0917]

I get your consideration.

What would the API look like?

I expect the API of ObservableSubscribeProxy interface to be

public interface ObservableSubscribeProxy<T> {
  ...
  Observable<? extends T> getSource();
}

then I can do

fun <T : Any> ObservableSubscribeProxy<T>.mySubscribe(f: () -> Unit): Disposable {
  return getSource().doOnSubscribe { f() }
    .subscribe(Consumer {}, Consumer {})
}

Any alternative solutions?

Cheers.

[ZacSweers]

hmm. I see the value from a flexibility, but your example actually ends up not using AutoDispose's functionality. Under the hood, this works because we decorate the received observer or lambda observers with a custom autodisposing one. In your example, you're just hitting RxJava's standard subscribe() method and forfeiting AutoDispose's.

You could adjust it to use an autodispose one, but now you no longer have access to the scope, so we'd need to then provide that somewhere too, so now we're talking about two extra methods (source() and scope()) just to be functional.

Two ideas that come to mind:

• Depending on what you want to do, a plugin similar to RxJava's onSubscribe hooks might give you the flexibility you're looking for. It would be per-runtime though, and not as flexible as extension functions in terms of visibility/scoping/etc.
• We expose a method on SubscribeProxy that outlets to another interface that exposes the above two elements and lets you handle it how you want (and hopefully remember to still call a real autodispose subscribe under the hood). In Kotlin, you could then tack on whatever extension methods you want, but it would be largely useless in Java and annoying to show in autocomplete. If we could somehow hide it from java but available in kotlin, I'd be pretty interested.

interface ObservableSubscribeProxy<T> {
  // Currently existing methods...

  // New method
  ObservableExtensionsHook<T> extensions()
}

interface ObservableExtensionsHook<T> {
  Observable<T> source()
  Maybe<?> scope()
}

// Your example
fun <T : Any> ObservableExtensionsHook<T>.mySubscribe(f: () -> Unit): Disposable {
  return source().doOnSubscribe { f() }
    .autoDisposeWith(scope())
    .subscribe(Consumer {}, Consumer {})
}

// In use
Observable.just(1)
  .autoDisposeWith(this)
  .extensions()
  .mySubscribe(...)

Basically, I like this, but ObservableExtensionsHook<T> extensions() is practically useless to java consumers and I wouldn't want to pollute the API.

[jaychang0917]

I manage to solve my question.

interface MyProxy<T> {
  fun sourceObservable(): Observable<T>
  fun lifecycleOwner(): LifecycleOwner
}

class MyProxyImpl<T>(val source: Observable<T>, val lifecycleOwner: LifecycleOwner): MyProxy<T> {
  override fun sourceObservable(): Observable<T> = source
  override fun lifecycleOwner(): LifecycleOwner = lifecycleOwner
}

fun <T> Observable<T>.autoDispose(lifecycleOwner: LifecycleOwner): MyProxy<T> {
  return MyProxyImpl(this, lifecycleOwner)
}

fun <T : Any> MyProxy<T>.mySubscribe(f: () -> Unit): Disposable {
  return sourceObservable().doOnSubscribe { f() }
    .autoDisposeWith(lifecycleOwner())
    .subscribe(Consumer {}, Consumer {})
}