Reimplementation of Concat, improved handling of Observable<Observable<T... #205
Conversation
…e<T>>. The old version required all of the Observable<T>s to be generated and buffered before the concat could begin. If the outer Observable was asynchronous, items could be dropped (test added). The new version passes the test, and does the best job I could (after examining several possible strategies) of achieving clear and consistent semantics in accordance with the principle of least surprise.
|
RxJava-pull-requests #45 ABORTED |
|
There is another issue with this implementation. The countdown latch will always block the calling thread, which is not desirable. Anything that returns an observable should never be blocking. Rx is meant to support composition of operations and allows each operation to run concurrently. |
| @@ -2,5 +2,5 @@ rootProject.name='rxjava' | |||
| include 'rxjava-core', \ | |||
| 'language-adaptors:rxjava-groovy', \ | |||
| 'language-adaptors:rxjava-jruby', \ | |||
| 'language-adaptors:rxjava-clojure', \ | |||
There was a problem hiding this comment.
Why did this file get changed for the pull request?
There was a problem hiding this comment.
Oops, sorry, I think I pushed from the wrong branch. I will fix this (and
the busted unit-test).
On Mar 26, 2013 4:43 PM, "Ben Christensen" notifications@github.com wrote:
In settings.gradle:
@@ -2,5 +2,5 @@ rootProject.name='rxjava'
include 'rxjava-core',
'language-adaptors:rxjava-groovy',
'language-adaptors:rxjava-jruby',
-'language-adaptors:rxjava-clojure', \Why did this file get changed for the pull request?
—
Reply to this email directly or view it on GitHubhttps://github.com//pull/205/files#r3539571
.
|
While reviewing this I started playing with the unit tests and added 2 new ones. I also made them use inOrder.verify so they ensure correct ordering. I've pasted it below so you can merge into your branch instead of me complicating the merge with another branch. The tests pass pass except I haven't thought through enough yet to determine if there is a way to solve that or if the unit test is just testing the wrong thing and thus a bad test. The comment above from @billy about CountDownLatch originated from a discussion between him and I - but while reviewing this it is probably the right approach. I was playing with other approaches that are purely non-blocking while retaining the correct concat sequential behavior but they all seem to require either:
My biggest issue right now is that concat(o1, o2) is a very common usage but that results in List which will be blocking and unsubscribe doesn't work. I'll think through this more but I'd appreciate your thoughts on how to handle public static class UnitTest {
@Test
public void testConcat() {
@SuppressWarnings("unchecked")
Observer<String> observer = mock(Observer.class);
final String[] o = { "1", "3", "5", "7" };
final String[] e = { "2", "4", "6" };
final Observable<String> odds = Observable.toObservable(o);
final Observable<String> even = Observable.toObservable(e);
@SuppressWarnings("unchecked")
Observable<String> concat = Observable.create(concat(odds, even));
concat.subscribe(observer);
verify(observer, times(7)).onNext(anyString());
}
@Test
public void testConcatWithList() {
@SuppressWarnings("unchecked")
Observer<String> observer = mock(Observer.class);
final String[] o = { "1", "3", "5", "7" };
final String[] e = { "2", "4", "6" };
final Observable<String> odds = Observable.toObservable(o);
final Observable<String> even = Observable.toObservable(e);
final List<Observable<String>> list = new ArrayList<Observable<String>>();
list.add(odds);
list.add(even);
Observable<String> concat = Observable.create(concat(list));
concat.subscribe(observer);
verify(observer, times(7)).onNext(anyString());
}
@Test
public void testConcatUnsubscribe() {
final CountDownLatch callOnce = new CountDownLatch(1);
final CountDownLatch okToContinue = new CountDownLatch(1);
final TestObservable<String> w1 = new TestObservable<String>("one", "two", "three");
final TestObservable<String> w2 = new TestObservable<String>(callOnce, okToContinue, "four", "five", "six");
@SuppressWarnings("unchecked")
Observer<String> aObserver = mock(Observer.class);
@SuppressWarnings("unchecked")
Observable<String> concat = Observable.create(concat(w1, w2));
System.out.println("before subscribe");
Subscription s1 = concat.subscribe(aObserver);
System.out.println("after subscribe");
try {
//Block main thread to allow observable "w1" to complete and observable "w2" to call onNext once.
System.out.println("before callOnce.await");
callOnce.await();
System.out.println("after callOnce.await");
s1.unsubscribe();
System.out.println("after s1.unsubscribe");
//Unblock the observable to continue.
okToContinue.countDown();
System.out.println("after countDown");
w1.t.join();
w2.t.join();
} catch (Exception e) {
e.printStackTrace();
fail(e.getMessage());
}
InOrder inOrder = inOrder(aObserver);
inOrder.verify(aObserver, times(1)).onNext("one");
inOrder.verify(aObserver, times(1)).onNext("two");
inOrder.verify(aObserver, times(1)).onNext("three");
inOrder.verify(aObserver, times(1)).onNext("four");
inOrder.verify(aObserver, never()).onNext("five");
inOrder.verify(aObserver, never()).onNext("six");
inOrder.verify(aObserver, times(1)).onCompleted();
}
@Test
public void testMergeObservableOfObservables() {
@SuppressWarnings("unchecked")
Observer<String> observer = mock(Observer.class);
final String[] o = { "1", "3", "5", "7" };
final String[] e = { "2", "4", "6" };
final Observable<String> odds = Observable.toObservable(o);
final Observable<String> even = Observable.toObservable(e);
Observable<Observable<String>> observableOfObservables = Observable.create(new Func1<Observer<Observable<String>>, Subscription>() {
@Override
public Subscription call(Observer<Observable<String>> observer) {
// simulate what would happen in an observable
observer.onNext(odds);
observer.onNext(even);
observer.onCompleted();
return new Subscription() {
@Override
public void unsubscribe() {
// unregister ... will never be called here since we are executing synchronously
}
};
}
});
Observable<String> concat = Observable.create(concat(observableOfObservables));
concat.subscribe(observer);
verify(observer, times(7)).onNext(anyString());
}
/**
* Simple concat of 2 asynchronous observables ensuring it emits in correct order.
*/
@SuppressWarnings("unchecked")
@Test
public void testSimpleAsyncConcat() {
Observer<String> observer = mock(Observer.class);
TestObservable<String> o1 = new TestObservable<String>("one", "two", "three");
TestObservable<String> o2 = new TestObservable<String>("four", "five", "six");
Observable.concat(o1, o2).subscribe(observer);
try {
// wait for async observables to complete
o1.t.join();
o2.t.join();
} catch (Exception e) {
throw new RuntimeException("failed waiting on threads");
}
InOrder inOrder = inOrder(observer);
inOrder.verify(observer, times(1)).onNext("one");
inOrder.verify(observer, times(1)).onNext("two");
inOrder.verify(observer, times(1)).onNext("three");
inOrder.verify(observer, times(1)).onNext("four");
inOrder.verify(observer, times(1)).onNext("five");
inOrder.verify(observer, times(1)).onNext("six");
}
/**
* Test an async Observable that emits more async Observables
*/
@SuppressWarnings("unchecked")
@Test
public void testNestedAsyncConcat() throws Exception {
Observer<String> observer = mock(Observer.class);
final TestObservable<String> o1 = new TestObservable<String>("one", "two", "three");
final TestObservable<String> o2 = new TestObservable<String>("four", "five", "six");
final TestObservable<String> o3 = new TestObservable<String>("seven", "eight", "nine");
final CountDownLatch allowThird = new CountDownLatch(1);
final AtomicReference<Thread> parent = new AtomicReference<Thread>();
Observable<Observable<String>> observableOfObservables = Observable.create(new Func1<Observer<Observable<String>>, Subscription>() {
@Override
public Subscription call(final Observer<Observable<String>> observer) {
final BooleanSubscription s = new BooleanSubscription();
parent.set(new Thread(new Runnable() {
@Override
public void run() {
try {
// emit first
if (!s.isUnsubscribed()) {
System.out.println("Emit o1");
observer.onNext(o1);
}
// emit second
if (!s.isUnsubscribed()) {
System.out.println("Emit o2");
observer.onNext(o2);
}
// wait until sometime later and emit third
try {
allowThird.await();
} catch (InterruptedException e) {
observer.onError(e);
}
if (!s.isUnsubscribed()) {
System.out.println("Emit o3");
observer.onNext(o3);
}
} catch (Exception e) {
observer.onError(e);
} finally {
System.out.println("Done parent Observable");
observer.onCompleted();
}
}
}));
parent.get().start();
return s;
}
});
Observable.create(concat(observableOfObservables)).subscribe(observer);
// wait for parent to start
while (parent.get() == null) {
Thread.sleep(1);
}
try {
// wait for first 2 async observables to complete
while (o1.t == null) {
Thread.sleep(1);
}
System.out.println("Thread1 started ... waiting for it to complete ...");
o1.t.join();
while (o2.t == null) {
Thread.sleep(1);
}
System.out.println("Thread2 started ... waiting for it to complete ...");
o2.t.join();
} catch (Exception e) {
throw new RuntimeException("failed waiting on threads", e);
}
InOrder inOrder = inOrder(observer);
inOrder.verify(observer, times(1)).onNext("one");
inOrder.verify(observer, times(1)).onNext("two");
inOrder.verify(observer, times(1)).onNext("three");
inOrder.verify(observer, times(1)).onNext("four");
inOrder.verify(observer, times(1)).onNext("five");
inOrder.verify(observer, times(1)).onNext("six");
// we shouldn't have the following 3 yet
inOrder.verify(observer, never()).onNext("seven");
inOrder.verify(observer, never()).onNext("eight");
inOrder.verify(observer, never()).onNext("nine");
// we should not be completed yet
verify(observer, never()).onCompleted();
verify(observer, never()).onError(any(Exception.class));
// now allow the third
allowThird.countDown();
try {
while (o3.t == null) {
Thread.sleep(1);
}
// wait for 3rd to complete
o3.t.join();
} catch (Exception e) {
throw new RuntimeException("failed waiting on threads", e);
}
inOrder.verify(observer, times(1)).onNext("seven");
inOrder.verify(observer, times(1)).onNext("eight");
inOrder.verify(observer, times(1)).onNext("nine");
inOrder.verify(observer, times(1)).onCompleted();
verify(observer, never()).onError(any(Exception.class));
}
@SuppressWarnings("unchecked")
@Test
public void testBlockedObservableOfObservables() {
Observer<String> observer = mock(Observer.class);
final String[] o = { "1", "3", "5", "7" };
final String[] e = { "2", "4", "6" };
final Observable<String> odds = Observable.toObservable(o);
final Observable<String> even = Observable.toObservable(e);
final CountDownLatch callOnce = new CountDownLatch(1);
final CountDownLatch okToContinue = new CountDownLatch(1);
TestObservable<Observable<String>> observableOfObservables = new TestObservable<Observable<String>>(callOnce, okToContinue, odds, even);
Func1<Observer<String>, Subscription> concatF = concat(observableOfObservables);
Observable<String> concat = Observable.create(concatF);
concat.subscribe(observer);
try {
//Block main thread to allow observables to serve up o1.
callOnce.await();
} catch (Exception ex) {
ex.printStackTrace();
fail(ex.getMessage());
}
// The concated observable should have served up all of the odds.
verify(observer, times(1)).onNext("1");
verify(observer, times(1)).onNext("3");
verify(observer, times(1)).onNext("5");
verify(observer, times(1)).onNext("7");
try {
// unblock observables so it can serve up o2 and complete
okToContinue.countDown();
observableOfObservables.t.join();
} catch (Exception ex) {
ex.printStackTrace();
fail(ex.getMessage());
}
// The concatenated observable should now have served up all the evens.
verify(observer, times(1)).onNext("2");
verify(observer, times(1)).onNext("4");
verify(observer, times(1)).onNext("6");
}
private static class TestObservable<T> extends Observable<T> {
private final Subscription s = new Subscription() {
@Override
public void unsubscribe() {
subscribed = false;
}
};
private final List<T> values;
private Thread t = null;
private int count = 0;
private boolean subscribed = true;
private final CountDownLatch once;
private final CountDownLatch okToContinue;
public TestObservable(T... values) {
this(null, null, values);
}
public TestObservable(CountDownLatch once, CountDownLatch okToContinue, T... values) {
this.values = Arrays.asList(values);
this.once = once;
this.okToContinue = okToContinue;
}
@Override
public Subscription subscribe(final Observer<T> observer) {
t = new Thread(new Runnable() {
@Override
public void run() {
try {
while (count < values.size() && subscribed) {
observer.onNext(values.get(count));
count++;
//Unblock the main thread to call unsubscribe.
if (null != once)
once.countDown();
//Block until the main thread has called unsubscribe.
if (null != once)
okToContinue.await();
}
if (subscribed)
observer.onCompleted();
} catch (InterruptedException e) {
e.printStackTrace();
fail(e.getMessage());
}
}
});
t.start();
return s;
}
}
} |
Also restore the errant change to settings.gradle.
|
Thanks for the comments and the extra tests. I have merged them and Regarding testConcatUnsubscribe: I'm still not entirely sure of my footing, Am I right that in general, an observer should block onNext() until it has I do find it a little strange that Observable.from(Iterable) is a "shotgun I have updated the test case to call concat.subscribe(aObserver) in a (I had also tried those three non-blocking approaches before uploading this
|
|
RxJava-pull-requests #59 SUCCESS |
|
RxJava-pull-requests #60 SUCCESS |
|
Thanks for update. Let me take a look and I will do some more On Fri, Mar 29, 2013 at 2:22 AM, CloudBees pull request builder plugin <
|
|
RxJava-pull-requests #72 SUCCESS |
|
My original test case was intended to test the unsubscribe inside one of public void testConcatUnsubscribeObservableOfObservable() { "one", "two", "three"); newTestObservable(callOnce, okToContinue, TestObservable<Observable> observableOfObservables = concat(observableOfObservables); and observable "w2" to call onNext exactly once. } On Tue, Apr 2, 2013 at 8:55 PM, CloudBees pull request builder plugin <
|
|
The RX .NET version of Concat uses the ImmediateScheduler to execute the On Wed, Apr 3, 2013 at 10:20 AM, Billy Yuen billyy@gmail.com wrote:
|
Rx does not (or rarely does) add concurrency (see guideline 6.12) thus A All Rx operators must be capable of handling both synchronous and asynchronous sequences, and if it's synchronous that means unsubscribe will not work (which means it's up to the implementor of an Observable to understand this and decided if it's okay to firehose the results). For the case of concat my concern is that if all the sequences being combined are async then we need to retain the async behavior and not block and prevent unsubscription. For example:
I should be able to do this:
This should get the 10 items from A, 20 items from B and first 20 items from C and then unsubscribe and continue. Does this work with the current implementation and correctly unsubscribe from ObservableC? |
|
Good point. The best way to tell is to build an unit test for this case. On Wed, Apr 3, 2013 at 2:29 PM, Ben Christensen notifications@github.comwrote:
|
…te limiter. (ReactiveX#205) * ReactiveX#35 Add Reactor support for circuit breaker, bulkhead and rate limiter. * Fix Codacity warnings * Commit build.gradle file for resilence4j-reactor * Static import assertThat
...>>.
The old version required all of the Observables to be generated and buffered
before the concat could begin. If the outer Observable was asynchronous, items
could be dropped (test added). The new version passes the test, and does the
best job I could (after examining several possible strategies) of achieving
clear and consistent semantics in accordance with the principle of least
surprise.
(My attempt to fix issue #202)