No InterruptedException with synchronous BlockingObservable

[ypresto]

I'm using Observable for backport of Java 8's java.util.stream (i.e. no more loops) for Android project.

List<Foo> list2 = Observable.from(list)
        .map(...)
        .filter(...)
        .toList().toBlocking().single()

But it sometimes emits InterruptedException at BlockingObservable.
#1804 (comment)

As the BlocingObservable is placed in map() of another observable with subscribeOn(Schedulers.io()), and it is unsubscribed from main thread, perhaps Future.cancel(true) is called on unsubscribing. (#1914)

This PR allows BlockingObsevable not to be interrupted when source observable emits synchronously, by checking current latch or queue state before awaiting for them.

[akarnokd]

So what happens is that you use some blocking calls inside an observable sequence and it gets interrupted because of the underlying task is cancelled?

I believe it is an indication that the flow can be assembled non-blockingly instead. I've been fiddling with the idea that once RxJava scheduler threads become identifiable, toBlocking() would throw/onError if it is attempted to use it inside such schedulers.

[ypresto]

So what happens is that you use some blocking calls inside an observable sequence and it gets interrupted because of the underlying task is cancelled?

Yes. But in my case, blocking call is inside of another (synchronous) method and I don't want to make it non-blocking, for testability and convenience.

For example, in instagram-like app, there are entities to represent photo/movie and comment.
Both entities has entity translation logic and they can be written in for-loop or toList().toBlocking().single().

// MediaFileModel
public MediaFileEntity entityFromDao(MediaFileDao dao, List<CommentDao> commentDaoList) {
    List<CommentEntity> comments = Observable.from(commentDaoList)
            .map(mCommentModel::entityFromDao) // No more loops!
            .filter(not(mCommentModel::isDeleted))
            .toList().toBlocking().single(); // called from where..?
    MediaFileEntity mediaFile = new MediaFileEntity();
    mediaFile.setUuid(dao.getUuid());
    mediaFile.setComments(comments);
    ...
    return mediaFile;
}

// MediaFileStore
public Observable<List<MediaFileEntity>> allEntities() {
    return queryAll() // async operation
            .map(daoList -> {
                    Observable.from(daoList)
                            .map(mMediaFileModel::entityFromDao) // is calling toBlocking() or not...?
                            .toList()
            );
}

I think so too that asynchronous (with blocking wait) operation should be assembled non-blockingly.
But I think it is just convenient if nesting of synchronous (without blocking wait) operation is allowed. :)

[akarnokd]

The changes look okay to me.

[ypresto]

Thanks! 🎉 🎉

[ypresto]

For your (who found this PR by searching) information:

RxJava is quite slow for synchronous operation.
Especially for zip(), it has significant overhead of waiting for async (and even if it is sync actually) observables.

So it'll better to use Java 8 Stream API or its backport for sync stream operations.
Lightweight-Stream-API has only <500 methods and also suitable for Android.