RxBatchDemo.java

package org.davidwatkins73.rx.playpen;

import rx.Observable;

import java.util.ArrayList;
import java.util.List;

public class RxBatchDemo {

    public static void main(String[] args) {

        List<Integer> batch = new ArrayList<>();

        Observable.range(0, 1000)
                .map(x -> {
                    batch.add(x);
                    return 1;
                })
                .scan(1, (acc, x) -> acc + 1)
                .filter(x -> x % 300 == 0)
                .subscribe(
                        x -> {
                            System.out.println("commit " + x + " == " + batch);
                            batch.clear();
                        },
                        e -> System.err.println(e.getMessage()),
                        () -> System.out.println("Done")
                );


                Observable.range(0, 10)
                        .map(x -> 1)
                        .scan((acc, x) -> acc + x)
                        .map(x -> x / 3)
                        .distinct()
                        .subscribe(System.out::println);
    }


    public static void withSize() {


        List<Integer> batch = new ArrayList<>();

        Observable.range(0, 10)
                .map(x -> {
                    batch.add(x);
                    System.out.println("batch now: "+batch.size());
                    return batch.size();
                })
                .filter(x -> x >= 3)
                .subscribe(
                        x -> {
                            System.out.println("commit " + x + " == " + batch);
                            batch.clear();
                        },
                        e -> System.err.println(e.getMessage()),
                        () -> System.out.println("Done")
                );
    }
}

Playing with ideas for driving jdbc batch via rx.