Fixed testOnErrorViaHasNext in issue #383

[zsxwing]

Hi,

This PR fixed the issue that testOnErrorViaHasNext fails sometimes. It only tries to avoid the failure of unit tests.
However, there is still an issue in the next operator. The hasNext may return true, but the later next throws IllegalStateException("Observable is completed"). An example unit test throws IllegalStateException:

        @Test
        public void test() {
            Subject<String, String> obs = PublishSubject.create();
            Iterator<String> it = next(obs).iterator();
            assertTrue(it.hasNext());
            obs.onCompleted();
            it.next();
        }

I think @abersnaze is right.

hasNext should block until either the onNext or onComplete has been called.

[cloudbees-pull-request-builder]

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

[cloudbees-pull-request-builder]

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

[benjchristensen]

Regarding hasNext being blocking ... that may be the case. Should that be researched further and resolved here, or as a separate pull request?

[zsxwing]

I think it's better to resolve here. I'll work on it.

[benjchristensen]

Thanks.

[petermd]

Thanks for fixing this.

Not sure I understand why hasNext() needs to block. Returning null from takeNext when the completed notification is consumed should work (and pass the example unit-test above)?

[zsxwing]

@petermd I suppose that for an iterator, next should be able to return something if hasNext is true. What's more, null is a valid value in an observable. So we can not simply return a null.

[petermd]

I know it seems a little counter-intuitive but that seems to be what the wiki says

The next() method returns an iterable that on each iteration blocks until the underlying Observable emits another >item, then returns that item (or null if the Observable finishes without emitting another item).

[cloudbees-pull-request-builder]

RxJava-pull-requests #356 FAILURE
Looks like there's a problem with this pull request

[zsxwing]

I updated the next operator.

Now the hasNext method will be blocked until any notification arrives.
In addition, after the observable is completed or fails, the following hasNext always return false and next always return null.

[cloudbees-pull-request-builder]

RxJava-pull-requests #358 FAILURE
Looks like there's a problem with this pull request

[headinthebox]

What is the weirdness?
And why the complexity with schedulers and toenumerable?

Sent from my iPad

On Oct 21, 2013, at 9:45 AM, Shixiong Zhu notifications@github.com wrote:

I found weird behaviors in Rx.Net.

In the following case,

var e = Observable.Create(
o => {
Console.WriteLine("subscribed");
return Disposable.Empty;
}
).SubscribeOn(Scheduler.ThreadPool).Next().GetEnumerator();
Console.WriteLine("before MoveNext");
if (e.MoveNext())
{
Console.WriteLine("Got " + e.Current);
}
else
{
Console.WriteLine("Empty");
}
it will output

subscribed
before MoveNext
In the following another case,

var e = Observable.Empty().SubscribeOn(Scheduler.ThreadPool).Next().GetEnumerator();
Console.WriteLine("before MoveNext");
if (e.MoveNext())
{
Console.WriteLine("Got " + e.Current);
}
else
{
Console.WriteLine("Empty");
}
it will output

before MoveNext
Empty
Does anybody know the difference between these two cases?

—
Reply to this email directly or view it on GitHub.

[zsxwing]

Sorry, I misunderstood about Rx.Net. I removed my related discussion to avoid confusing somebody.

[benjchristensen]

Is there ever a time when someone could call next() in a loop without hasNext()

I don't think it's illegal for someone to do that until it throws a NoSuchElementException thus we need to do the same blocking logic here in case someone doesn't use hasNext()

http://docs.oracle.com/javase/6/docs/api/java/util/Iterator.html#next()

[zsxwing]

I updated the PR to follow the iterator contract. Thanks for your review, @benjchristensen

[cloudbees-pull-request-builder]

RxJava-pull-requests #368 FAILURE
Looks like there's a problem with this pull request

[petermd]

still wondering why this implementation runs contrary to the wiki. that said, i can't find any other implementation of Next operator so can't be 100% sure?

null if the Observable finishes without emitting another item

if the Observable has completed then hasNext() should surely return true and next() return null?

perhaps worth noting that this is meant to be a sampler (it does not buffer) so running

while(it.hasNext()) {
  it.next();
}

is not guaranteed to iterate over the output of the Observable (which is perhaps the intention of the blocking code?) as Observer.onNext() events that happen between hasNext() and next() should be lost.

does storing the unconsumed next value in the hasNext() method not also break this contract?

[zsxwing]

still wondering why this implementation runs contrary to the wiki.

It's possible to implement it following the wiki. But the wiki is against the Iterator contract in Java. I suppose it's better to follow the Iterator contract so that Java users may feel conformable.

Since hasNext and next are always called together, I think it's OK to buffer a value when hasNext is called.

[petermd]

the problem is without buffering the full Observable you cannot implement the generally accepted contract ie hasNext/next can return different results for the same Observable - that seems pretty uncomfortable?

I think the real issue is, there isn't a good example of how you would use this operator. The name "next" suggests you use it to get an Iterator that will return the "next" value from the Observable (or null..) but the implementation does something rather different now (it returns an Iterator that is either empty or has a value from the Observable)

[DavidMGross]

As the main wiki writer, I should note that I'm attempting to be
descriptive rather than prescriptive. If the wiki doesn't match the
implementation, or doesn't match the correct/intuitive implementation --
implement it correctly and then let me know (or correct the wiki yourself).

On Wed, Oct 23, 2013 at 7:50 AM, Peter McDonnell
notifications@github.comwrote:

the problem is without buffering the full Observable you cannot implement
the generally accepted contract ie hasNext/next can return different
results for the same Observable - that seems pretty uncomfortable?

I think the real issue is, there isn't a good example of how you would use
this operator. The name "next" suggests you use it to get an Iteratorthat
will return the "next" value from the Observable (or null..) but the
implementation does something rather different now (it returns an Iterator
that is either empty or has a value from the Observable)

—
Reply to this email directly or view it on GitHubhttps://github.com//pull/433#issuecomment-26911239
.

David M. Gross
PLP Consulting

[zsxwing]

hasNext/next can return different results for the same Observable

I think this behavior is very strange. Actually, hasNext is an indicator that if you will receive an value from next in the Java Iterator contract.

I tried the Next opeartor in Rx.Net and got the similar result.

            IObservable<int> ob =
                    Observable.Create<int>(o =>
                        {
                            Console.WriteLine("Subscribed: Before onNext");
                            Thread.Sleep(2000);
                            o.OnNext(1);
                            Console.WriteLine("Subscribed: After OnNext");
                            o.OnCompleted();
                            return Disposable.Empty;
                        }
                    );
            var iter = ob.SubscribeOn(Scheduler.NewThread).Next().GetEnumerator();
            Console.WriteLine("Before MoveNext");
            while (iter.MoveNext())
            {
                Console.WriteLine("Find a value");
                Thread.Sleep(5000);
                Console.WriteLine("Got " + iter.Current);
            }
            Console.ReadLine();
        }

output:

Before MoveNext
Subscribed: Before onNext
Subscribed: After OnNext
Find a value
Got 1

In this case, Rx.Net also blocks in MoveNext. And after 5 seconds, iter.Current returns the cached value.

So I think my current implementation is same as Rx.Net. And we need to update the wiki.

[petermd]

@DavidMGross - thanks for the clarification.

@zsxwing - is the semantic of Enumerable not slightly different (it consists of a hasNext() and next() in the same call)? I don't have a .NET setup to test easily but would be curious about what the output is if you

1. omit the o.OnNext(1) line (so the Observable just completes)
2. put a Thread.Sleep(3000) before the iter.MoveNext() (so the OnNext has occured before MoveNext is called)

I'll stop bugging you about it then :-)

[zsxwing]

omit the o.OnNext(1) line (so the Observable just completes)

Test case:

IObservable<int> ob =
        Observable.Create<int>(o =>
        {
            Console.WriteLine("Subscribed: Before OnCompleted");
            Thread.Sleep(2000);
            o.OnCompleted();
            Console.WriteLine("Subscribed: After OnCompleted");
            return Disposable.Empty;
        }
        );
var iter = ob.SubscribeOn(Scheduler.NewThread).Next().GetEnumerator();
Console.WriteLine("Before MoveNext");
while (iter.MoveNext())
{
    Console.WriteLine("Find a value");
    Thread.Sleep(5000);
    Console.WriteLine("Got " + iter.Current);
}
Console.WriteLine("After MoveNext");
Console.ReadLine();

Output:

Before MoveNext
Subscribed: Before OnCompleted
Subscribed: After OnCompleted
After MoveNext

put a Thread.Sleep(3000) before the iter.MoveNext() (so the OnNext has occured before MoveNext is called)

Test case:

IObservable<int> ob =
        Observable.Create<int>(o =>
        {
            Console.WriteLine("Subscribed: Before onNext");
            Thread.Sleep(2000);
            o.OnNext(1);
            Console.WriteLine("Subscribed: After OnNext");
            o.OnCompleted();
            return Disposable.Empty;
        }
        );
var iter = ob.SubscribeOn(Scheduler.NewThread).Next().GetEnumerator();
Thread.Sleep(5000);
Console.WriteLine("Before MoveNext");
while (iter.MoveNext())
{
    Console.WriteLine("Find a value");
    Thread.Sleep(5000);
    Console.WriteLine("Got " + iter.Current);
}
Console.WriteLine("After MoveNext");
Console.ReadLine();

Output:

Subscribed: Before onNext
Subscribed: After OnNext
Before MoveNext
After MoveNext

[zsxwing]

These behaviors are consistent with my thought.

[petermd]

Thanks @zsxwing - so maybe the correct wiki def should be something like?

The next() method returns an iterable that on each iteration blocks in Iterator.hasNext()or Iterator.next() until the underlying Observable emits another item, Iterator.next() then returns that item.

If the Observable emits an error then Iterator.hasNext() will return true and Iterator.next() will re-throw the exception.

If the Observable finishes without emitting another item then Iterator.hasNext call will return false, and Iterator.next() will throw a NoSuchElementException

[zsxwing]

If the Observable emits an error then Iterator.hasNext() will return true and Iterator.next() will re-throw the exception.

In my codes, if the Observable emits an error then Iterator.hasNext() and Iterator.next() will both re-throw the exception. However, you remind me that I should always throw the exception for the following hasNext or next. I'll fix it.

[zsxwing]

I updated the PR to let hasNext and next always throw the error after the observable fails. And rebase it on latest codes.

[cloudbees-pull-request-builder]

RxJava-pull-requests #373 FAILURE
Looks like there's a problem with this pull request

[zsxwing]

Let me conclude the next operator.

• If the user only uses next method, the next method will be blocked until the new value arrives.
• If the user uses hasNext and next methods together, at first the hasNext will be blocked until the new value arrives. After that, however, before the corresponding next method is called, hasNext only returns the same result without being blocked. After the corresponding next method is called (it returns the cached value), the following hasNext will be blocked.
• If the iterator reaches the end of the observable, the following hasNext will always return false, and next will always throw NoSuchElementException.
• If the observable throws any error, the following hasNext or next will always throw the error.

[headinthebox]

That is the protocol for integrators in .net. You have to call movenext, and then current.http://msdn.microsoft.com/en-us/library/vstudio/system.collections.ienumerator.aspx

I always assumed that this was the same in java, but the Scala folks told me that typically has next does not block, but calling next does. Which in Scala means you use or not use parens. Not sure if that is true in general. It does make iterators hard to implement. In PHP the protocol is also a bit different where you start with rewind, which is typically ignored in .net.

Sent from my iPad

On Oct 28, 2013, at 6:22 AM, Shixiong Zhu notifications@github.com wrote:

Let me conclude the next operator.

If the user only uses next method, the next method will be blocked until the new value arrives.
If the user uses hasNext and next methods together, at first the hasNext will be blocked until the new value arrives. After that, however, before the corresponding next method is called, hasNext only returns the same result without being blocked. After the corresponding next method is called (it returns the cached value), the following hasNext will be blocked.
If the iterator reaches the end of the observable, the following hasNext will always return false, and next will always throw NoSuchElementException.
If the observable throws any error, the following hasNext or next will always throw the error.
—
Reply to this email directly or view it on GitHub.

[zsxwing]

@headinthebox javacoc and scaladoc do not say how to block. So I think it should be OK that we block hasNext. For now, I could not find any way to let hasNext follow the contract Returns true if the iteration has more elements. without blocking hasNext.

[benjchristensen]

Your explanation of how it works sounds correct, merging.