Implemented TakeUntil operation

[mairbek]

Fixes issue #86

[benjchristensen]

This is a very elegant solution. I wouldn't have considered using the Pair with map/merge/takeWhile like that.

The only thing I'd like to add is an instance version:

public <E> Observable<T> takeUntil(final Observable<E> other)

This way I can chain off an Observable I already have instead of only statically.

[prabirshrestha]

What about having Tuple2<A, B>, Tuple3<A,B,C> and so on like func and actions instead of using Pair<A, B>?
Even ReactiveCocoa has RACTuple.

[benjchristensen]

@prabirshrestha that's a good thought, I've considered it but been holding off for the following reasons:

1. We haven't had an operator yet need them so didn't want to eagerly add them until actually needed.
2. How will they work with each of the languages? Will they be idiomatic or awkward? Perhaps we can be typed but also implement Iterable and List so they can be accessed dynamically like an array rather than via fields/accessors.
3. Will these duplicate/collide with anything coming in Java8 and if so can be future-proof by matching the signature and then being able to extend from those interface when they arrive?

For example, Groovy and Closure generally prefer arrays/sequence of values without strong typing and typically just return something simple like [a, b].

I probably should have noted in this review that it would be best to not put Pair in rx.util but keep it private (inner class perhaps) until nailing down those questions above.

[benjchristensen]

@mairbek

Do you want to implement the instance method of takeUntil and re-submit, or do you want me to just add that myself?

Based on the previous question/comment as well I think Pair needs to be kept private for now until decisions on Tuple are made.

What do you think about moving the implementation into an OperationTakeUntil class so that Observable doesn't keep growing huge?

Since we lack extensions methods in Java we have to put all the static and instance methods in there, but we can extract implementation details and their testing into Operation* classes to try and keep some sanity.

[mairbek]

@benjchristensen thank you for the review. I'll update the pull request.

[benjchristensen]

You may want to start a new pull request on a new updated branch as this no longer can merge without conflicts.

[mairbek]

Seems like code doesn't handle all cases.

When source observable completes, observer should unsubscribe immediately.

To be precise following test fails

        @Test
        public void testTakeUntilCompleted() {
            Subscription sSource = mock(Subscription.class);
            Subscription sOther = mock(Subscription.class);
            TestObservable source = new TestObservable(sSource);
            TestObservable other = new TestObservable(sOther);

            Observer<String> result = mock(Observer.class);
            Observable<String> stringObservable = takeUntil(source, other);
            stringObservable.subscribe(result);
            source.sendOnNext("one");
            source.sendOnNext("two");
            source.sendOnCompleted();

            verify(result, times(1)).onNext("one");
            verify(result, times(1)).onNext("two");
            verify(sSource, times(1)).unsubscribe();
            verify(sOther, times(1)).unsubscribe();

        }

I'll update the code to handle this use case.

[mairbek]

I modified TakeUntil a bit. It's not as elegant as it was before, but it still uses merge operation to avoid dealing with concurrency. And it works correctly :) (more unit tests passed).

[cloudbees-pull-request-builder]

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

[mairbek]

@teyc you are right, I'll update the code.

Thank you!

[cloudbees-pull-request-builder]

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

[cloudbees-pull-request-builder]

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

[benjchristensen]

Based on my reading of the code, MSDN docs and the unit tests this looks good. Merging ...