@@ -0,0 +1,2240 @@
+package com.github.davidmoten.rx.internal.operators;%0A%0Aimport java.io.File;%0Aimport java.util.concurrent.BlockingQueue;%0A%0Aimport org.mapdb.DB;%0Aimport org.mapdb.DBMaker;%0Aimport org.mapdb.Serializer;%0A%0Aimport rx.Observable;%0Aimport rx.Observable.Transformer;%0Aimport rx.functions.Action1;%0Aimport rx.functions.Func0;%0Aimport rx.functions.Func1;%0A%0Apublic final class TransformerOnBackpressureBufferToFile%3CT%3E implements Transformer%3CT, T%3E %7B%0A%0A    private static final String QUEUE_NAME = %22q%22;%0A    private final File file;%0A    private final Serializer%3CT%3E serializer;%0A%0A    public TransformerOnBackpressureBufferToFile(File file, Serializer%3CT%3E serializer) %7B%0A        this.file = file;%0A        this.serializer = serializer;%0A    %7D%0A%0A    @Override%0A    public Observable%3CT%3E call(Observable%3CT%3E o) %7B%0A        return Observable.defer(new Func0%3CObservable%3CT%3E%3E() %7B%0A%0A            @Override%0A            public Observable%3CT%3E call() %7B%0A                Func0%3CDB%3E resourceFactory = new Func0%3CDB%3E() %7B%0A                    @Override%0A                    public DB call() %7B%0A                        return DBMaker.newFileDB(file).make();%0A                    %7D%0A                %7D;%0A                Func1%3CDB, Observable%3CT%3E%3E observableFactory = new Func1%3CDB, Observable%3CT%3E%3E() %7B%0A                    @Override%0A                    public Observable%3CT%3E call(DB db) %7B%0A                        final BlockingQueue%3CT%3E queue;%0A                        if (db.getCatalog().containsKey(QUEUE_NAME)) %7B%0A                            queue = db.getQueue(QUEUE_NAME);%0A                        %7D else %7B%0A                            queue = db.createQueue(QUEUE_NAME, serializer, false);%0A                        %7D%0A                        return Observable.create(new OnSubscribeFromQueue%3CT%3E(queue));%0A                    %7D%0A                %7D;%0A                Action1%3CDB%3E disposeAction = new Action1%3CDB%3E() %7B%0A                    @Override%0A                    public void call(DB db) %7B%0A                        db.close();%0A                    %7D%0A                %7D;%0A                return Observable.using(resourceFactory, observableFactory, disposeAction, true)%0A                        .repeat();%0A            %7D%0A        %7D);%0A    %7D%0A%0A    public static void main(String%5B%5D args) %7B%0A        System.out.println(1 %7C 1);%0A    %7D%0A%0A%7D%0A

add transformer