Reimplemented the 'reduce' operator

[zsxwing]

Hi,

I reimplemented the 'reduce' operator. The improvements are as follow:

• As I mentioned in Reduce an empty observable #423, 'reduce' should throw an exception when applying on an empty sequence without an initial value.
• Now 'reduce' does not need a 'takeLast' operator, and should be more efficient.

[cloudbees-pull-request-builder]

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

[benjchristensen]

Why does this need to be synchronized? We have a single Observable sequence that will not interleave onNext calls (by contract) so I don't think we need this here.

[benjchristensen]

Now 'reduce' does not need a 'takeLast' operator, and should be more efficient.

What is the overhead of takeLast(1) that limits its usage? (As a side-note, I think takeLast(1) could be special-cased and made far more efficient than takeLast(n) since it doesn't need a queue, just the last value.)

As I mentioned in #423, 'reduce' should throw an exception when applying on an empty sequence without an initial value.

Can we not achieve this check without re-implementing the entire operator? This duplicates logic between scan and reduce for accumulation. If we can't do it cleanly or efficiently then your implementation looks great.

The guiding principle for this decision is:

[zsxwing]

Hi, @benjchristensen

I removed the synchronized. However, I'm still confused about concurrency in RxJava. Could you review my discussion in #417 and help me solve the problems?

I also updated the unit tests.

Finally, I reviewed the current operators and could not find a way to throw an exception when the observable is empty. Could you provide some suggestion?

[cloudbees-pull-request-builder]

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

[cloudbees-pull-request-builder]

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

[cloudbees-pull-request-builder]

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

[benjchristensen]

Finally, I reviewed the current operators and could not find a way to throw an exception when the observable is empty. Could you provide some suggestion?

Does this do what you need?

.flatMap( o -> {
  if(good) {
   return Observable.just(goodValue);
  } else {
   // if bad ... return an error
   return Observable.error(new RuntimeException());
  }

})

[zsxwing]

Sorry that I still have no idea about how to use flatMap to implement it. If the observable is empty, flatMap does nothing. So how to throw an exception in such situation?

[samuelgruetter]

Maybe it would be nice to have an assertNonEmpty operator, since this could also be useful for other use cases. What do you think?

[abersnaze]

Materialize and dematerialize are good tool for implementing operators that have behavior based on onError and onCompleted.

x = from([])

hadValue = false;
x.materialize().map({ n ->
  if (n.kind == Notification.Kind.OnNext) {
    hadValue = true
  }
  else if (n.kind == Notification.Kind.OnCompleted && !hadValue) {
    return new Notification(new Exception())
  }
  return n
}).dematerialize()

[zsxwing]

Thanks, @abersnaze . It works.

@samuelgruetter , I added an private assertNonEmpty operator in the Observable. I think it's not necessary to be public.

[cloudbees-pull-request-builder]

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

[abersnaze]

I just had this idea. replace the takeLast(1) with takeLast(2) and use a second scan to figure out what to do at the end of the sequence.

    public Observable<T> reduce(Func2<T, T, T> accumulator) {
        Func2<Notification<T>, Notification<T>, Notification<T>> func = new Func2<Notification<T>, Notification<T>, Notification<T>>() {
            @Override
            public Notification<T> call(Notification<T> value, Notification<T> end) {
                if (end.isOnError())
                    return end;
                if (value == null)
                    return new Notification<T>(new UnsupportedOperationException("Can not apply on an empty sequence"));
                return value;
            }
        };
        return create(OperationScan.scan(this, accumulator)).materialize().takeLast(2).scan(null, func).dematerialize();
    }

[zsxwing]

Sounds good. But I'm not sure if it's better that using one more scan to eliminate the isEmpty variable. It's better that eliminating variable, but a little harder to understand.

[zsxwing]

I updated this PR to use the new last operator #470 to implement reduce. This can be merged now.

[cloudbees-pull-request-builder]

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

[zsxwing]

It has been fixed in #474