package com.github.davidmoten.rx.jdbc;

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

import com.github.davidmoten.rx.OperationToOperator;

/**
 * {@link Operator} corresonding to {@link QueryUpdateOperation}.
 */
class QueryUpdateOperatorFromObservable<R> implements Operator<Observable<Integer>, Observable<R>> {

    private final OperationToOperator<Observable<Integer>, Observable<R>> operator;

    /**
     * Constructor.
     * 
     * @param builder
     * @param operatorType
     */
    QueryUpdateOperatorFromObservable(final QueryUpdate.Builder builder) {
        operator = new OperationToOperator<Observable<Integer>, Observable<R>>(
                new Func1<Observable<Observable<R>>, Observable<Observable<Integer>>>() {

                    @Override
                    public Observable<Observable<Integer>> call(Observable<Observable<R>> observable) {

                        return observable.map(new Func1<Observable<R>, Observable<Integer>>() {
                            @Override
                            public Observable<Integer> call(Observable<R> parameters) {
                                return builder.clearParameters().parameters(parameters).count();
                            }
                        });
                    }
                });
    }

    @Override
    public Subscriber<? super Observable<R>> call(Subscriber<? super Observable<Integer>> subscriber) {
        return operator.call(subscriber);
    }
}

package com.github.davidmoten.rx.jdbc;

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

import com.github.davidmoten.rx.OperationToOperator;
import com.github.davidmoten.rx.RxUtil;

/**
 * {@link Operator} corresonding to {@link QueryUpdateOperation}.
 */
class QueryUpdateOperatorFromObservable<R> implements Operator<Observable<Integer>, Observable<R>> {

    private final Operator<Observable<Integer>, Observable<R>> operator;

    /**
     * Constructor.
     * 
     * @param builder
     * @param operatorType
     */
    QueryUpdateOperatorFromObservable(final QueryUpdate.Builder builder) {
        operator = RxUtil.toOperator(
                new Func1<Observable<Observable<R>>, Observable<Observable<Integer>>>() {

                    @Override
                    public Observable<Observable<Integer>> call(Observable<Observable<R>> observable) {

                        return observable.map(new Func1<Observable<R>, Observable<Integer>>() {
                            @Override
                            public Observable<Integer> call(Observable<R> parameters) {
                                return builder.clearParameters().parameters(parameters).count();
                            }
                        });
                    }
                });
    }

    @Override
    public Subscriber<? super Observable<R>> call(Subscriber<? super Observable<Integer>> subscriber) {
        return operator.call(subscriber);
    }
}

use static construction method