MathObservable

[benjchristensen]

Moving the average/sum/min/max functionality to the MathObservable similar to StringObservable.

Similar to the rxjava-string module this is seeking to achieve the goal of keeping rxjava-core focused on core functionality.

[cloudbees-pull-request-builder]

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

[davidmoten]

Hi, being a fan of method chaining I'd like to see more Operator implementations of for example the math operations and many more. Moreover, I'd like to have an idiom tested and confirmed by you experts to do this easily for my own stuff. To that end do you think it would be of value to include in the library something like the class below? One possibility would be to add the toOperator static method to the Observable class.

import rx.Observable;
import rx.Observable.Operator;
import rx.Subscriber;
import rx.functions.Func1;
import rx.subjects.PublishSubject;

/**
 * Converts an Operation (a function converting one Observable into another)
 * into an {@link Operator}.
 * 
 * @param <R>
 *            to type
 * @param <T>
 *            from type
 */
public class OperatorFromOperation<R, T> implements Operator<R, T> {

    public static <R, T> Operator<R, T> toOperator(Func1<Observable<T>, Observable<R>> operation) {
        return new OperatorFromOperation<R, T>(operation);
    }

    /**
     * The operation to convert.
     */
    private final Func1<Observable<T>, Observable<R>> operation;

    /**
     * Constructor.
     * 
     * @param operation
     *            to be converted into {@link Operator}
     */
    public OperatorFromOperation(Func1<Observable<T>, Observable<R>> operation) {
        this.operation = operation;
    }

    @Override
    public Subscriber<? super T> call(Subscriber<? super R> subscriber) {
        final PublishSubject<T> subject = PublishSubject.create();
        Subscriber<T> result = createSubscriber(subject);
        subscriber.add(result);
        operation.call(subject).subscribe(subscriber);
        return result;
    }

    /**
     * Creates a subscriber that passes all events on to the subject.
     * 
     * @param subject
     *            receives all events.
     * @return
     */
    private static <T> Subscriber<T> createSubscriber(final PublishSubject<T> subject) {
        return new Subscriber<T>() {

            @Override
            public void onCompleted() {
                subject.onCompleted();
            }

            @Override
            public void onError(Throwable e) {
                subject.onError(e);
            }

            @Override
            public void onNext(T t) {
                subject.onNext(t);
            }
        };
    }
}

[abersnaze]

What could you do with operation that couldn't be done with map(Func1<T, R>) or flatMap(Func1<T, Observable<R>>)?

[davidmoten]

An operation acts upon the observable as a whole not on the individual
items like map and flatMap do. For example map and flatMap themselves
could be considered as operations/operators.
On 9 Mar 2014 04:41, "George Campbell" [email protected] wrote:

What could you do with operation that couldn't be done with map(Func1<T,
R>) or flatMap(Func1<T, Observable>)?

Reply to this email directly or view it on GitHubhttps://github.com//pull/942#issuecomment-37103999
.

[benjchristensen]

I don't understand what use cases are trying to be solved by that code, can you provide some examples please?

[davidmoten]

So for instance if I want to use sumInteger in the middle of a chain of
method calls I need to use lift so I need an operator form of it.

I make an operator form of it by implementing a call to sumInteger in a
Func1<Observable,Observable> and then call
toOperator(func) as described in my example to give me an operator. I
would prefer if there were operator forms of all the static observable
methods I suppose but it least this conversion was part of the codebase
that would make life easier.

I should mention that method chained calls are important to me because the
functional attractiveness of rxjava is weakened significantly every time I
have to break the chain with some wrapper call. In communicating the
elegance of rxjava to my colleagues my case is weakened by those breaks in
the chain. I'm a big fan of lift for this reason.

I'm still new to rxjava so let me know if I'm missing something obvious
here.
Cheers
Dave

On 11 March 2014 03:26, Ben Christensen [email protected] wrote:

I don't understand what use cases are trying to be solved by that code,
can you provide some examples please?

Reply to this email directly or view it on GitHubhttps://github.com//pull/942#issuecomment-37202146
.

[davidmoten]

To add an explicit example I've found the method RxUtil.toOperator as
described above useful when I want to use an operation like sumInteger in
an operator form with lift(). To use it I write:

private final Operator<Integer,Integer> SUM_INTEGER =
RxUtil.toOperator(new Func1<Observable,Observable>() {
@OverRide
public Observable call(Observable source) {
return MathObservable.sumInteger(source);
}});
Then I can write:

Observable.from(asList(1,2,3)).lift(SUM_INTEGER);

On 11 March 2014 08:07, Dave Moten [email protected] wrote:

So for instance if I want to use sumInteger in the middle of a chain of
method calls I need to use lift so I need an operator form of it.

I make an operator form of it by implementing a call to sumInteger in a
Func1<Observable,Observable> and then call
toOperator(func) as described in my example to give me an operator. I
would prefer if there were operator forms of all the static observable
methods I suppose but it least this conversion was part of the codebase
that would make life easier.

I should mention that method chained calls are important to me because the
functional attractiveness of rxjava is weakened significantly every time I
have to break the chain with some wrapper call. In communicating the
elegance of rxjava to my colleagues my case is weakened by those breaks in
the chain. I'm a big fan of lift for this reason.

I'm still new to rxjava so let me know if I'm missing something obvious
here.
Cheers
Dave

On 11 March 2014 03:26, Ben Christensen [email protected] wrote:

I don't understand what use cases are trying to be solved by that code,
can you provide some examples please?

Reply to this email directly or view it on GitHubhttps://github.com//pull/942#issuecomment-37202146
.

[davidmoten]

I think perhaps I should simplify my original question. Suppose I have a
static function like

public static <R,T> Observable something(Observable source) {
//anything here of course
return source.map(someTransform).retry().take(100);
}

What is the simplest way to make an Operator out of it so that I can use it
with lift()? If the answer is longer than the toOperator method usage that
I'm proposing then I'd be interested in having toOperator in the library.

On 12 March 2014 11:35, Dave Moten [email protected] wrote:

To add an explicit example I've found the method RxUtil.toOperator as
described above useful when I want to use an operation like sumInteger in
an operator form with lift(). To use it I write:

private final Operator<Integer,Integer> SUM_INTEGER =
RxUtil.toOperator(new
Func1<Observable,Observable>() {
@OverRide
public Observable call(Observable source) {
return MathObservable.sumInteger(source);
}});
Then I can write:

Observable.from(asList(1,2,3)).lift(SUM_INTEGER);

On 11 March 2014 08:07, Dave Moten [email protected] wrote:

So for instance if I want to use sumInteger in the middle of a chain of
method calls I need to use lift so I need an operator form of it.

I make an operator form of it by implementing a call to sumInteger in a
Func1<Observable,Observable> and then call
toOperator(func) as described in my example to give me an operator. I
would prefer if there were operator forms of all the static observable
methods I suppose but it least this conversion was part of the codebase
that would make life easier.

I should mention that method chained calls are important to me because
the functional attractiveness of rxjava is weakened significantly every
time I have to break the chain with some wrapper call. In communicating the
elegance of rxjava to my colleagues my case is weakened by those breaks in
the chain. I'm a big fan of lift for this reason.

I'm still new to rxjava so let me know if I'm missing something obvious
here.
Cheers
Dave

On 11 March 2014 03:26, Ben Christensen [email protected] wrote:

I don't understand what use cases are trying to be solved by that code,
can you provide some examples please?

Reply to this email directly or view it on GitHubhttps://github.com//pull/942#issuecomment-37202146
.

[headinthebox]

We should add an overload for lift that takes an Observable => Observable function. Very much like http://msdn.microsoft.com/en-us/library/hh229147(v=vs.103).aspx. This is on the TODO list, or you can try to build it yourself, and submit a pull request. Note it should allow subscribing more than once in the lambda, hence "publish" in .NET.

[zsxwing]

We should add an overload for lift that takes an Observable => Observable function.

+1

[davidmoten]

I'll make a pull request with the lift method overload as suggested.
Dave

On 13 March 2014 13:41, Shixiong Zhu [email protected] wrote:

We should add an overload for lift that takes an Observable => Observable
function.

+1

Reply to this email directly or view it on GitHubhttps://github.com//pull/942#issuecomment-37494316
.

[akarnokd]

I think we already have that publish() method: Observable.java:5428 which relies on OperationMulticast. If you could reimplement it to OperatorMulticast, we would gain a lot of operators.

[davidmoten]

I'm a bit lost about the reference to publish() method that started with headinthebox's comment. I am not talking about adding a publish method. I am talking about adding a toOperator method or overload on lift method that does the same thing.
Dave

[davidmoten]

@benjchristensen what do you think about the addition of an overload for lift that takes a Func1<Observable,Observable>? Should I go ahead with a pull request?