Lift and Observer+Subscription #784

benjchristensen · 2014-01-25T00:48:33Z

Make Observer implement Subscription and rename bind to lift as per decisions in #775

Signatures are now:

// Observable.create
public final static <T> Observable<T> create(OnSubscribe<T> f)

// Observable.OnSubscribe typed function interface
public static interface OnSubscribe<T> extends Action1<Observer<? super T>>

// lift function
public <R> Observable<R> lift(final Func1<Observer<? super R>, Observer<? super T>> bind)

// Observer
public abstract class Observer<T> implements Subscription {
     public abstract void onNext(T t);
     public abstract void onError(Throwable e);
     public abstract void onCompleted();
     public final void add(Subscription s)
     public final void unsubscribe()
     public final boolean isUnsubscribed()
}

// Subject
public abstract class Subject<T, R> extends Observer<T> {
    public abstract Observable<R> toObservable();
}

This is a major set of changes to the internals, particularly the unit tests because Mockito has issues with abstract classes as opposed to Observer being an interface.

There are still some unit tests failing that I haven't yet figured out:

OperationJoinsTest.whenComplicated in rxjava-core
OperationConditionalsTest.testDoWhileManyTimes in rxjava-computation-expressions
OperationConditionalsTest.testWhileDoManyTimes in rxjava-computation-expressions

I intend on merging this sooner rather than later so everyone can be working off the same codebase, even though I do not consider this code ready for release, even once those unit tests are fixed.

Interestingly, these performance tests on my machine are much better:

// OperatorFromIterablePerformance.timeTenLongs()
v0.16
     * Run: 10 - 8,096,667 ops/sec
     * Run: 11 - 8,382,131 ops/sec
     * Run: 12 - 8,256,288 ops/sec
     * Run: 13 - 8,139,703 ops/sec
     * Run: 14 - 8,011,023 ops/sec

... after v0.17 work:

     * Run: 10 - 31,296,553 ops/sec
     * Run: 11 - 30,080,435 ops/sec
     * Run: 12 - 31,886,941 ops/sec
     * Run: 13 - 32,281,807 ops/sec
     * Run: 14 - 33,519,028 ops/sec


// OperatorMapPerformance.timeMapPlusOne();

v0.16

     * Run: 10 - 11,375,632 ops/sec
     * Run: 11 - 11,390,325 ops/sec
     * Run: 12 - 11,655,527 ops/sec
     * Run: 13 - 11,528,440 ops/sec
     * Run: 14 - 11,321,181 ops/sec

... after v0.17 work:

     * Run: 10 - 17,674,464 ops/sec
     * Run: 11 - 17,890,239 ops/sec
     * Run: 12 - 17,919,155 ops/sec
     * Run: 13 - 16,356,974 ops/sec
     * Run: 14 - 16,723,414 ops/sec

// OperatorMergePerformance.timeRepetitionsEmissionSynchronous()

v0.16

     * Run: 10 - 32,609,617 ops/sec
     * Run: 11 - 33,511,839 ops/sec
     * Run: 12 - 34,768,096 ops/sec
     * Run: 13 - 32,376,499 ops/sec
     * Run: 14 - 33,166,835 ops/sec

... after v0.17 work:

     * Run: 10 - 45,945,747 ops/sec
     * Run: 11 - 46,342,209 ops/sec
     * Run: 12 - 44,493,090 ops/sec
     * Run: 13 - 44,999,640 ops/sec
     * Run: 14 - 47,389,771 ops/sec

As per discussion at ReactiveX#775 (comment)

As per decision at ReactiveX#775 (comment)

Follow same pattern as rx.observables, rx.schedulers, rx.subjects, rx.subscriptions

Now that Observer is an abstract class, Mockito is having issues with it so unit tests are a mess.

- work around inability of Mockito to correctly mock an abstract class - 15 of 590 tests still failing

…ous to do so.

Remove possibility of infinite loop

… numbers from my machine. Can’t fully explain the increases in performance.

- confirmed the assertions, leaving broken until can be fixed (not sure what’s wrong)

benjchristensen · 2014-01-25T00:51:27Z

/cc @abersnaze @akarnokd @headinthebox

akarnokd · 2014-01-25T08:51:14Z

If I factor out the subjects toObservable calls to be the same, the test passes:

Observable<Integer> xso = xs.toObservable();
Observable<Integer> yso = ys.toObservable();
Observable<Integer> zso = zs.toObservable();

Observable<Integer> m = Observable.when(
      xso.and(yso).then(add2), // 1+4=5, 2+5=7, 3+6=9
      xso.and(zso).then(mul2), // 1*7=7, 2*8=16, 3*9=27
      yso.and(zso).then(sub2)  // 4-7=-3, 5-8=-3, 6-9=-3
);

Otherwise, it seems you had like 6 different and operations and source numbers were replicated to all of them. This way, there were multiple queues per source and did independent matching.

As for the conditionals, change the SerialSubscription to MultipleAssignmentSubscription in OperationConditionals#L235 and any dependent places.

@akarnokd

- now correctly creates only 1 Observable instance for the life of the Subject - this fixes the OperationJoinsTest - thanks @akarnokd for pointing out my mistake! - all rxjava-core unit tests are now passing

benjchristensen · 2014-01-25T16:34:44Z

@akarnokd Thanks for the pointer to the problem. I have fixed the Subject.toObservable implementations to correctly have only 1 Observable instance for the life of the Subject.

Looking at OperationConditionals ...

headinthebox · 2014-01-25T20:46:06Z

For Observable.join and Observable.groupJoin you must draw a marble diagram (at least I do) otherwise your brain will melt.

viktorklang · 2014-01-25T22:09:19Z

rxjava-core/src/main/java/rx/Observer.java

+
+    protected Observer(CompositeSubscription cs) {
+        if (cs == null) {
+            throw new IllegalArgumentException("The CompositeException can not be null");


s/CompositeException/CompositeSubscription

Hi Viktor, thanks for pointing that out! Fix just pushed.

All unit tests now pass.

benjchristensen · 2014-01-26T04:10:26Z

As for the conditionals, change the SerialSubscription to MultipleAssignmentSubscription in OperationConditionals#L235 and any dependent places.

That indeed fixes it, though I have yet to track down why the unsubscribe on each pass causes things to fail now whereas it didn't before.

cloudbees-pull-request-builder · 2014-01-26T04:16:26Z

RxJava-pull-requests #703 SUCCESS
This pull request looks good

benjchristensen · 2014-01-26T05:17:45Z

Merging so work can continue on master branch.

Lift and Observer+Subscription

benjchristensen added 21 commits January 22, 2014 14:37

Rename bind to lift

88f3613

As per discussion at ReactiveX#775 (comment)

Observer becomes Observer implements Subscription

050e62a

As per decision at ReactiveX#775 (comment)

Move Observer factory methods to Observers

b49943a

Follow same pattern as rx.observables, rx.schedulers, rx.subjects, rx.subscriptions

Subjects Fixed

015e2b4

Move Observers to rx.observers

510850e

Compiling. Unit Tests Failing.

5478e30

Now that Observer is an abstract class, Mockito is having issues with it so unit tests are a mess.

TestObserver

6d9d540

Operator Tests Retrofitted to use TestObserver

d2cfac1

- work around inability of Mockito to correctly mock an abstract class - 15 of 590 tests still failing

587 of 590 operator unit tests passing

a70afdf

Down to 7 of 893 unit tests failing

d37ce86

Fixed 1 Unit Test - Linked Observer/Subscription where seemingly obvi…

0ac1e6c

…ous to do so.

Fix Subject SubscriptionManager

4e4c165

Remove possibility of infinite loop

Clarify intent of unit test

8926255

OperationJoinTest - fixing

59bb911

Ignore JoinsTest.whenComplicated until questions answered

fdc4c60

Fix Exception Handling from SafeObserver and Subscriptions

028d189

Performance comparison

93bd190

… numbers from my machine. Can’t fully explain the increases in performance.

TestObserver fix for Scala UnitTest

023cb50

Fix Unit Tests

6a1c8f0

Restore Broken OperationJoinsTest

e24a23b

- confirmed the assertions, leaving broken until can be fixed (not sure what’s wrong)

Conditionals: Fix all but 2 tests

f26b7b2

Fix Subject.toObservable

f5eb8f7

- now correctly creates only 1 Observable instance for the life of the Subject - this fixes the OperationJoinsTest - thanks @akarnokd for pointing out my mistake! - all rxjava-core unit tests are now passing

viktorklang reviewed Jan 25, 2014
View reviewed changes

benjchristensen added 2 commits January 25, 2014 19:40

Fix Typo: s/CompositeException/CompositeSubscription

9ee0fdc

Fix OperationalConditional

6725754

All unit tests now pass.

benjchristensen added a commit that referenced this pull request Jan 26, 2014

Merge pull request #784 from benjchristensen/lift-observer

86993d3

Lift and Observer+Subscription

benjchristensen merged commit 86993d3 into ReactiveX:master Jan 26, 2014

benjchristensen deleted the lift-observer branch January 26, 2014 05:18

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Lift and Observer+Subscription #784

Lift and Observer+Subscription #784

benjchristensen commented Jan 25, 2014

benjchristensen commented Jan 25, 2014

akarnokd commented Jan 25, 2014

benjchristensen commented Jan 25, 2014

headinthebox commented Jan 25, 2014

viktorklang Jan 25, 2014

benjchristensen Jan 26, 2014

benjchristensen commented Jan 26, 2014

cloudbees-pull-request-builder commented Jan 26, 2014

benjchristensen commented Jan 26, 2014

Lift and Observer+Subscription #784

Lift and Observer+Subscription #784

Conversation

benjchristensen commented Jan 25, 2014

benjchristensen commented Jan 25, 2014

akarnokd commented Jan 25, 2014

benjchristensen commented Jan 25, 2014

headinthebox commented Jan 25, 2014

viktorklang Jan 25, 2014

Choose a reason for hiding this comment

benjchristensen Jan 26, 2014

Choose a reason for hiding this comment

benjchristensen commented Jan 26, 2014

cloudbees-pull-request-builder commented Jan 26, 2014

benjchristensen commented Jan 26, 2014