[yarpl] Reduce operator for Flowable

[phoad]

Adding Reduce operator for Flowable

• http://reactivex.io/documentation/operators/reduce.html

Flowables::range(0, 10)
->reduce([] (int64_t acc, int64_t x) { return acc + x; })
->subscribe(mySubscriber);

and have mySubscriber get sent a single result of 45 and then get completed.

[alexmalyshev]

Like filter(), this version of flowable::ReduceOperator doesn't have request(n) semantics. A way to see this would be to do Flowables::justN<int64_t>({1, 2, 3})->reduce([] (int a, int b) { return a + b; }); If I subscribe to that flowable with a subscriber that does request(1), then my subscriber should get back onNext(6); onComplete();

Right now, your implementation will forward that request(1) to the justN(), which will send through 1. But you will never get the rest of the items. reduce() needs its upstream to finish sending through all of its items before it can send through its single item and then complete.

Question for @benjchristensen, would you expect Flowables::just<int64_t>(5)->reduce(someFn) to call someFn(0, 5)? I think single-item reduce should never call the accumulator function, but I'm not actually sure what the desired behavior here is... (short of renaming to foldLeft and adding an initial argument).

[alexmalyshev]

Comments above

[phoad]

I believe even with one item, if we reduce we should call the reduce function. Think like reduce function is taking the negative of the value, namely subtracting it from the initial value. Then we will return wrong value if we do not call the function at all..

Flowables::just<int64_t>(5)->reduce([] (int a, int b) { return a - b; });

Result should be -5, not 5.

[phoad]

Furthermore, I believe we need a default initial value for the "a" parameter. It might be correct to use the default of the integral values (for +/- operators), but if we think about reducing with (*, /) operators we will always get 0.. We might also end up with a double/float/long number as the result of the reduce, but we can specify this by using these values as type of the parameter "a".

I wanted to ask before doing further..

[alexmalyshev]

Think like reduce function is taking the negative of the value, namely subtracting it from the initial value. Then we will return wrong value if we do not call the function at all..

Flowables::just<int64_t>(5)->reduce([] (int a, int b) { return a - b; });

Result should be -5, not 5.

I don't agree here. reduce() is about combining N items together, pairwise. If you only have one item, then there's nothing to combine it with (unless you have an initial value of course, but that sounds like a different foldLeft() operator to me). I think using integer subtraction is a bad example because it's not the right use case for reduce(), you wouldn't combine items by subtracting them. Operators that are more about "combining elements", are things like addition, multiplication, string concatenation. All things that work just fine if you don't run the argument function on single-item streams.

Furthermore, I believe we need a default initial value for the "a" parameter

It might just be the case that we want a foldLeft() operator that does have an initial value, separate from reduce().

[phoad]

So, you mean, the initial value better to be the first item in the inputs. Nice!
If we have only one value, it is the initial value, which is the result value, that we do not perform any operation on.
If we have multiple values and we apply multiplication, as the initial value for int is 0, as we use the first item in the inputs as the initial value, we will not end up with 0 at the end for the multiplication. Perfect! Huh.

[phoad]

The definition of the reduce is a bit different: http://reactivex.io/documentation/operators/reduce.html

"The Reduce operator applies a function to the first item emitted by the source Observable"

I will update the code according to your explanations. Thanks Alex!
http://stackoverflow.com/questions/7764197/difference-between-foldleft-and-reduceleft-in-scala

[phoad]

I might add a "#if version" check in here so it can build for older gcc versions. Is it an appreciated way in the team for such features that are only available in the new compiler versions.

[phoad]

I thought again and perhaps we should not throw at all and just call complete without calling onNext.
I will do this update along with implementing Observable Reduce Operator unit tests.

[alexmalyshev]

LGTM, but conflicts need fixing :(

[benjchristensen]

Fixed the conflicts, but there is an asan failure that I don't quite understand yet. Doesn't seem related to this change, but need to understand the cause.

[phoad]

Sorry, these updates should not go to the review. I am reverting these lines.

[phoad]

@alexmalyshev, @lehecka can you help in checking the update in?

"Only those with write access to this repository can merge pull requests."