// Path: src/main/java/com/github/davidmoten/rx2/Consumers.java
// public final class Consumers {
// 
//     private Consumers() {
//         // prevent instantiation
//     }
// 
//     public static LongConsumer addLongTo(final List<Long> list) {
//         return new LongConsumer() {
// 
//             @Override
//             public void accept(long t) throws Exception {
//                 list.add(t);
//             }
// 
//         };
//     }
// 
//     @SuppressWarnings("unchecked")
//     public static <T extends Closeable> Consumer<T> close() {
//         return (Consumer<T>) CloseHolder.INSTANCE;
//     }
// 
//     private static final class CloseHolder {
//         final static Consumer<Closeable> INSTANCE = new Consumer<Closeable>() {
//             @Override
//             public void accept(Closeable t) throws Exception {
//                 t.close();
//             }
// 
//         };
//     }
// 
//     public static Consumer<Object> increment(final AtomicInteger value) {
//         return new Consumer<Object>() {
//             @Override
//             public void accept(Object t) throws Exception {
//                 value.incrementAndGet();
//             }
//         };
//     }
// 
//     public static Consumer<Throwable> printStackTrace() {
//         // TODO make holder
//         return new Consumer<Throwable>() {
//             @Override
//             public void accept(Throwable e) throws Exception {
//                 e.printStackTrace();
//             }
//         };
//     }
// 
//     @SuppressWarnings("unchecked")
//     public static <T> Consumer<T> doNothing() {
//         return (Consumer<T>) DoNothingHolder.INSTANCE;
//     }
// 
//     private static final class DoNothingHolder {
//         static final Consumer<Object> INSTANCE = new Consumer<Object>() {
// 
//             @Override
//             public void accept(Object t) throws Exception {
//                 // do nothing
//             }
//         };
//     }
// 
//     public static <T> Consumer<T> set(final AtomicReference<T> value) {
//         return new Consumer<T>() {
// 
//             @Override
//             public void accept(T t) throws Exception {
//                 value.set(t);
//             }
//         };
//     }
// 
//     public static Consumer<Integer> set(final AtomicInteger value) {
//         return new Consumer<Integer>() {
//             @Override
//             public void accept(Integer t) throws Exception {
//                 value.set(t);
//             }
//         };
//     }
// 
//     public static Consumer<Object> decrement(final AtomicInteger value) {
//         return new Consumer<Object>() {
// 
//             @Override
//             public void accept(Object t) throws Exception {
//                 value.decrementAndGet();
//             }
// 
//         };
//     }
// 
//     @SuppressWarnings("unchecked")
//     public static <T> Consumer<T> setToTrue(final AtomicBoolean value) {
//         return (Consumer<T>) new Consumer<Object>() {
// 
//             @Override
//             public void accept(Object t) throws Exception {
//                 value.set(true);
//             }
//         };
//     }
// 
//     public static <T> Consumer<T> addTo(final List<T> list) {
//         return new Consumer<T>() {
// 
//             @Override
//             public void accept(T t) throws Exception {
//                 list.add(t);
//             }
//         };
//     }
// 
//     @SuppressWarnings("unchecked")
//     public static <T> Consumer<T> println() {
//         return (Consumer<T>) PrintlnHolder.INSTANCE;
//     }
// 
//     private static final class PrintlnHolder {
//         static final Consumer<Object> INSTANCE = new Consumer<Object>() {
//             @Override
//             public void accept(Object t) throws Exception {
//                 System.out.println(t);
//             }
//         };
//     }
// 
//     public static Consumer<byte[]> assertBytesEquals(final byte[] expected) {
//         // TODO make holder
//         return new Consumer<byte[]>() {
// 
//             @Override
//             public void accept(byte[] array) throws Exception {
//                 if (!Arrays.equals(expected, array)) {
//                     // TODO use custom exception
//                     throw new AssertionException("arrays not equal: expected=" + Arrays.toString(expected) + ",actual="
//                             + Arrays.toString(array));
//                 }
//             }
// 
//         };
//     }
// 
//     public static LongConsumer printLong(final String prefix) {
//         return new LongConsumer() {
//             @Override
//             public void accept(long t) throws Exception {
//                 System.out.println(prefix + t);
//             }
//         };
//     }
// }


import java.io.BufferedInputStream;
import java.io.BufferedOutputStream;
import java.io.EOFException;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.io.Serializable;
import java.util.concurrent.Callable;
import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.io.Input;
import com.esotericsoftware.kryo.io.Output;
import com.github.davidmoten.rx2.Consumers;
import io.reactivex.Emitter;
import io.reactivex.Flowable;
import io.reactivex.functions.Consumer;
import io.reactivex.functions.Function;
                    emitter.onError(e);
                }
            }
        });
    }

    /**
     * Returns the deserialized objects from the given {@link File} as a
     * {@link Flowable} stream. Uses buffer of size <code>bufferSize</code>
     * buffer reads from the File.
     * 
     * @param file
     *            the input file
     * @param bufferSize
     *            the buffer size for reading bytes from the file.
     * @param <T>
     *            the generic type of the deserialized objects returned in the
     *            stream
     * @return the stream of deserialized objects from the {@link InputStream}
     *         as a {@link Flowable}.
     */
    public static <T extends Serializable> Flowable<T> read(final File file, final int bufferSize) {
        Callable<ObjectInputStream> resourceFactory = new Callable<ObjectInputStream>() {
            @Override
            public ObjectInputStream call() throws IOException {
                return new ObjectInputStream(new BufferedInputStream(new FileInputStream(file), bufferSize));
            }
        };
        @SuppressWarnings("unchecked")
        Function<ObjectInputStream, Flowable<T>> flowableFactory = (Function<ObjectInputStream, Flowable<T>>) (Function<?, ?>) ObjectInputStreamFlowableFactoryHolder.INSTANCE;
// Path: src/main/java/com/github/davidmoten/rx2/Consumers.java
// public final class Consumers {
// 
//     private Consumers() {
//         // prevent instantiation
//     }
// 
//     public static LongConsumer addLongTo(final List<Long> list) {
//         return new LongConsumer() {
// 
//             @Override
//             public void accept(long t) throws Exception {
//                 list.add(t);
//             }
// 
//         };
//     }
// 
//     @SuppressWarnings("unchecked")
//     public static <T extends Closeable> Consumer<T> close() {
//         return (Consumer<T>) CloseHolder.INSTANCE;
//     }
// 
//     private static final class CloseHolder {
//         final static Consumer<Closeable> INSTANCE = new Consumer<Closeable>() {
//             @Override
//             public void accept(Closeable t) throws Exception {
//                 t.close();
//             }
// 
//         };
//     }
// 
//     public static Consumer<Object> increment(final AtomicInteger value) {
//         return new Consumer<Object>() {
//             @Override
//             public void accept(Object t) throws Exception {
//                 value.incrementAndGet();
//             }
//         };
//     }
// 
//     public static Consumer<Throwable> printStackTrace() {
//         // TODO make holder
//         return new Consumer<Throwable>() {
//             @Override
//             public void accept(Throwable e) throws Exception {
//                 e.printStackTrace();
//             }
//         };
//     }
// 
//     @SuppressWarnings("unchecked")
//     public static <T> Consumer<T> doNothing() {
//         return (Consumer<T>) DoNothingHolder.INSTANCE;
//     }
// 
//     private static final class DoNothingHolder {
//         static final Consumer<Object> INSTANCE = new Consumer<Object>() {
// 
//             @Override
//             public void accept(Object t) throws Exception {
//                 // do nothing
//             }
//         };
//     }
// 
//     public static <T> Consumer<T> set(final AtomicReference<T> value) {
//         return new Consumer<T>() {
// 
//             @Override
//             public void accept(T t) throws Exception {
//                 value.set(t);
//             }
//         };
//     }
// 
//     public static Consumer<Integer> set(final AtomicInteger value) {
//         return new Consumer<Integer>() {
//             @Override
//             public void accept(Integer t) throws Exception {
//                 value.set(t);
//             }
//         };
//     }
// 
//     public static Consumer<Object> decrement(final AtomicInteger value) {
//         return new Consumer<Object>() {
// 
//             @Override
//             public void accept(Object t) throws Exception {
//                 value.decrementAndGet();
//             }
// 
//         };
//     }
// 
//     @SuppressWarnings("unchecked")
//     public static <T> Consumer<T> setToTrue(final AtomicBoolean value) {
//         return (Consumer<T>) new Consumer<Object>() {
// 
//             @Override
//             public void accept(Object t) throws Exception {
//                 value.set(true);
//             }
//         };
//     }
// 
//     public static <T> Consumer<T> addTo(final List<T> list) {
//         return new Consumer<T>() {
// 
//             @Override
//             public void accept(T t) throws Exception {
//                 list.add(t);
//             }
//         };
//     }
// 
//     @SuppressWarnings("unchecked")
//     public static <T> Consumer<T> println() {
//         return (Consumer<T>) PrintlnHolder.INSTANCE;
//     }
// 
//     private static final class PrintlnHolder {
//         static final Consumer<Object> INSTANCE = new Consumer<Object>() {
//             @Override
//             public void accept(Object t) throws Exception {
//                 System.out.println(t);
//             }
//         };
//     }
// 
//     public static Consumer<byte[]> assertBytesEquals(final byte[] expected) {
//         // TODO make holder
//         return new Consumer<byte[]>() {
// 
//             @Override
//             public void accept(byte[] array) throws Exception {
//                 if (!Arrays.equals(expected, array)) {
//                     // TODO use custom exception
//                     throw new AssertionException("arrays not equal: expected=" + Arrays.toString(expected) + ",actual="
//                             + Arrays.toString(array));
//                 }
//             }
// 
//         };
//     }
// 
//     public static LongConsumer printLong(final String prefix) {
//         return new LongConsumer() {
//             @Override
//             public void accept(long t) throws Exception {
//                 System.out.println(prefix + t);
//             }
//         };
//     }
// }


// Path: src/main/java/com/github/davidmoten/rx2/flowable/Serialized.java
import java.io.BufferedInputStream;
import java.io.BufferedOutputStream;
import java.io.EOFException;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.io.Serializable;
import java.util.concurrent.Callable;
import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.io.Input;
import com.esotericsoftware.kryo.io.Output;
import com.github.davidmoten.rx2.Consumers;
import io.reactivex.Emitter;
import io.reactivex.Flowable;
import io.reactivex.functions.Consumer;
import io.reactivex.functions.Function;

                    emitter.onError(e);
                }
            }
        });
    }

    /**
     * Returns the deserialized objects from the given {@link File} as a
     * {@link Flowable} stream. Uses buffer of size <code>bufferSize</code>
     * buffer reads from the File.
     * 
     * @param file
     *            the input file
     * @param bufferSize
     *            the buffer size for reading bytes from the file.
     * @param <T>
     *            the generic type of the deserialized objects returned in the
     *            stream
     * @return the stream of deserialized objects from the {@link InputStream}
     *         as a {@link Flowable}.
     */
    public static <T extends Serializable> Flowable<T> read(final File file, final int bufferSize) {
        Callable<ObjectInputStream> resourceFactory = new Callable<ObjectInputStream>() {
            @Override
            public ObjectInputStream call() throws IOException {
                return new ObjectInputStream(new BufferedInputStream(new FileInputStream(file), bufferSize));
            }
        };
        @SuppressWarnings("unchecked")
        Function<ObjectInputStream, Flowable<T>> flowableFactory = (Function<ObjectInputStream, Flowable<T>>) (Function<?, ?>) ObjectInputStreamFlowableFactoryHolder.INSTANCE;
        Consumer<ObjectInputStream> disposeAction = Consumers.close();