Operator: forEach

[benjchristensen]

Issue #45

Related to Pull #131

[benjchristensen]

Updated in another commit to bring inline with the MSDN docs.

Unit tests that show the behavior:

    @Test
    public void testForEach() {
        Observable.create(new AsyncObservable()).forEach({ result -> a.received(result)});
        verify(a, times(1)).received(1);
        verify(a, times(1)).received(2);
        verify(a, times(1)).received(3);
    }

    @Test
    public void testForEachWithError() {
        try {
            Observable.create(new AsyncObservable()).forEach({ result -> throw new RuntimeException('err')});
            fail("we expect an exception to be thrown");
        }catch(Exception e) {
            // do nothing as we expect this
        }
        verify(a, times(0)).received(1);
        verify(a, times(0)).received(2);
        verify(a, times(0)).received(3);
    }

[mairbek]

Verifications could be removed since there are no usages of a in the forEach closure.

[benjchristensen]

Good point. Submitting a commit to remove those now.

[benjchristensen]

Thanks @mairbek for the review.

@dcapwell @jhusain or anyone else have comments on this before I merge?

I'll wait a few hours for feedback before merging and doing a release with this last batch of changes.

[dcapwell]

Looks good to me. Only comment i would make is in the InterruptedException catch. Since not throwing a InterruptedException but Runtime, it might be good to also add Thread.getCurrentThread().interrupt() to set the interrupt flag again.

    try {
        latch.await();  
    } catch (InterruptedException e) {  
        Thread.getCurrentThread().interrupt();
        throw new RuntimeException("Interrupted while waiting for subscription to complete.", e);
 }

[benjchristensen]

Interesting point, didn't even think of that as I figured the RuntimeException would be sufficient to cause the caller to behave appropriately, but they could theoretically swallow RuntimeExceptions and continue in a while loop check for isInterrupted() which could be lost in this scenario.

Making that change now.