package au.gov.amsa.streams;

import java.io.IOException;
import java.io.Reader;

import rx.Observable;
import rx.Observer;
import rx.observables.SyncOnSubscribe;

public final class OnSubscribeReader extends SyncOnSubscribe<Reader, String> {

    private final Reader reader;
    private final int size;

    public OnSubscribeReader(Reader reader, int size) {
        this.reader = reader;
        this.size = size;
    }

    @Override
    protected Reader generateState() {
        return reader;
    }

    @Override
    protected Reader next(Reader reader, Observer<? super String> observer) {
        char[] buffer = new char[size];
        try {
            int count = reader.read(buffer);
            if (count == -1)
                observer.onCompleted();
            else
                observer.onNext(String.valueOf(buffer, 0, count));
        } catch (IOException e) {
            observer.onError(e);
        }
        return reader;
    }

    public Observable<String> toObservable() {
        return Observable.create(this);
    }
}

@@ -297,16 +297,54 @@
 nt size;
+%0A    %0A    private final char%5B%5D buffer;
 %0A%0A    pu
@@ -445,24 +445,62 @@
 ize = size;%0A
+        this.buffer = new char%5Bsize%5D;%0A
     %7D%0A%0A    @
@@ -583,32 +583,32 @@
 %0A%0A    @Override%0A
+
     protected Re
@@ -673,48 +673,8 @@
 ) %7B%0A
-        char%5B%5D buffer = new char%5Bsize%5D;%0A
     

increase performance of Strings.from(Reader) by reducing allocations