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);
    }
}

@@ -188,16 +188,56 @@
 perator;
+%0Aimport com.github.davidmoten.rx.RxUtil;
 %0A%0A/**%0A *
@@ -424,27 +424,16 @@
 e final 
-OperationTo
 Operator
@@ -671,67 +671,25 @@
 r = 
-new OperationToOperator%3CObservable%3CInteger%3E, Observable%3CR%3E%3E
+RxUtil.toOperator
 (%0A  

use static construction method