AndroidObservable not work

[darktiny]

private static final String TAG = "MainActivity";

private Subscription subscription;

@Override
protected void onCreate(Bundle savedInstanceState) {
    super.onCreate(savedInstanceState);
    setContentView(R.layout.activity_main);
    // worked
    subscription = AndroidObservable.fromActivity(this,
            getObservable().subscribeOn(Schedulers.newThread())).subscribe(
            this);
    // Not work
    // subscription = AndroidObservable.fromActivity(this, getObservable())
    // .subscribeOn(Schedulers.newThread()).subscribe(this);
    // worked
    // subscription = getObservable().subscribeOn(Schedulers.newThread())
    // .observeOn(AndroidSchedulers.mainThread()).subscribe(this);
}

private Observable<Integer> getObservable() {
    return Observable.create(new OnSubscribeFunc<Integer>() {

        @Override
        public Subscription onSubscribe(Observer<? super Integer> obsever) {
            for (int i = 0; i < 10; i++) {
                obsever.onNext(i);
            }
            obsever.onCompleted();
            return Subscriptions.empty();
        }
    });
}

@Override
protected void onDestroy() {
    super.onDestroy();
    subscription.unsubscribe();
}

@Override
public void onCompleted() {
    Log.d(TAG, "finished");
}

@Override
public void onError(Throwable error) {
}

@Override
public void onNext(Integer i) {
    Log.i(TAG, i + "");
}

What I mean of not work is onNext() and onCompleted() not get called.

[mttkay]

I'll take this one.

The reason it's not working is we're doing a check for whether fromFragment is actually called from the main UI thread, and terminate the Observable if not. However, this check happens in the OnSubscribeFunc so will fail if the wrapped Observable is scheduled to run elsewhere.

The check should be moved to the invocation side, i.e. to the fromFragment helper.

[jmodrako]

I also got this issue. When it'll be fixed?

[mttkay]

I'm waiting for 0.17 to land; it contains major changes to the Observable API. Will pick it up after.

[jmodrako]

Thank for reply! You and your team made a bunch of great work, thanks man!

[benjchristensen]

@mttkay On this subject, can you confirm that 0.17 (master branch) is working for you on Android. In particular the Android Scheduler was changed by me and I'm not setup to test on real devices.

[mttkay]

I'll have a look ASAP. Has the scheduler rewrite landed yet? It was still
in a PR last time I checked

[benjchristensen]

Yes, it has been merged. Thank you.

[mttkay]

So, I gave this some more thought and I'm not sure we should actually "fix" this. The reason is, fromFragment schedules notifications on the main UI thread. If you do not apply this to the final composition of your inner observables, then the result might not be what you might expect. If you have long running operations, they will now be propagated on the main UI thread, and this can congest the Android message queue. For an activity, since you unsubscribe in onDestroy, and since onDestroy is triggered by finish, and since finish will also post a message to the Android message queue, you might defer releasing the references the operator holds to the activity.

After all, these helpers are meant for just one use case: to observe a fully constructed given sequence on the main thread while being able to unsubscribe such that no references to the activity or fragment will leak, plus some additional sanity checks for fragments. I guess that's why I put that guard in to begin with.

That being said, I think one problem is the naming. I would suggest the following:

• rename OperationObserveFromAndroidComponent to OperationObserveOnAndroidComponent
• rename fromFragment and fromActivity to observeOnFragment and observeOnActivity
• remove the UI thread check from the operator and pull it up into these helpers; that way, your app will fail immediately with a sanity check rather than obscuring it and piping it through an Rx error handler, which you might no even have supplied

Thoughts on this?

@benjchristensen
I did some quick testing in a sample app that I use as a sandbox for experiments and it looks good. Unit tests also look good. Would like to further test in our application, but if there's pressure to release 0.17 earlier than later I'd almost say go ahead, I didn't see any smoke signals, and smaller issues can still be addressed in a minor version bump. Ship early ship often and so on :-)

[benjchristensen]

@mttkay Thanks for doing that testing for me. I agree that you'll want further testing before production deployment, but the fact that it's good enough to pass right now means there are no major API signature or fatal issues. I am trying to wrap up a few last things so we can release in the next week give or take.

[mttkay]

Sounds good

[Leland-Takamine]

@mttkay This code sample below is is taken from your blog post Functional Reactive Programming on Android With RxJava. The code here seems to imply that the fromFragment and fromActivity methods can be used with .subscribeOn(Schedulers.newThread()), but from the discussion above it sounds like that is not the intention. So I'm wondering if the blog post is just out of date? Please let me know if I'm misunderstanding something here.

class MyFragment extends Fragment implements Observer<File> {
  private Subscription subscription;

  @Override
  protected void onCreate(Bundle savedInstanceState) {
    subscription = AndroidObservables.fromFragment(this, downloadFileObservable())
                          .subscribeOn(Schedulers.newThread())
                          .subscribe(this);
  }

  private Observable<File> downloadFileObservable() { /* as above */ }

  @Override
  protected void onDestroy() {
    subscription.unsubscribe();
  }

  public void onNext(File file) {
    Toast.makeText(getActivity(),
        "Downloaded: " + file.getAbsolutePath(),
        Toast.LENGTH_SHORT)
        .show();
  }

  public void onCompleted() {}

  public void onError(Throwable error) {
    Toast.makeText(getActivity(),
        "Download failed: " + error.getMessage(),
        Toast.LENGTH_SHORT)
        .show();
  }
}

[mttkay]

Yes that code snippet I wrote before adding the assertUiThread guard. Like I said, I'll revisit this soon but want to see 0.17 land first.

[zsxwing]

@mttkay Can you take a look at this PR #880? I'm worried that deferring the unsubscribe action may cause some problems. Maybe a better solution is:

if(isUIThread) {
    do unsubscribe
}
else {
    AndroidSchedulers.mainThread().schedule( () -> do unsubscribe )
}

[mttkay]

That PR already landed. Can we maybe agree that a PR gets at least 2 +1s before we merge it back? I'm not super happy with the changes in there. (see my comments in #880)