Mirrored Source for OKIO.

MirroredSource.java

import java.io.IOException;
import java.util.concurrent.atomic.AtomicBoolean;
import okio.Buffer;
import okio.Source;
import okio.Timeout;
import timber.log.Timber;

public class MirroredSource {

    private final Buffer buffer = new Buffer();
    private final Source source;
    private final AtomicBoolean sourceExhausted = new AtomicBoolean();

    public MirroredSource(final Source source) {
        this.source = source;
    }

    public Source original() {
        return new okio.Source() {

            @Override public long read(final Buffer sink, final long byteCount) throws IOException {
                final long bytesRead = source.read(sink, byteCount);
                if (bytesRead > 0) {
                    synchronized (buffer) {
                        sink.copyTo(buffer, sink.size() - bytesRead, bytesRead);
                        // Notfiy the mirror to continue
                        buffer.notify();
                    }
                } else {
                    sourceExhausted.set(true);
                }
                return bytesRead;
            }

            @Override public Timeout timeout() {
                return source.timeout();
            }

            @Override public void close() throws IOException {
                source.close();
                sourceExhausted.set(true);
                synchronized (buffer) {
                    buffer.notify();
                }
            }
        };
    }

    public Source mirror() {
        return new okio.Source() {

            @Override public long read(final Buffer sink, final long byteCount) throws IOException {
               synchronized (buffer) {
                while (!sourceExhausted.get()) {
                    // only need to synchronise on reads when the source is not exhausted.
                    
                        if (buffer.request(byteCount)) {
                            return buffer.read(sink, byteCount);
                        } else {
                            try {
                                buffer.wait();
                            } catch (final InterruptedException e) {
                                Timber.e(e, "Error waiting");
                            }
                        }
                    }
                }
                return buffer.read(sink, byteCount);
            }
            }

            @Override public Timeout timeout() {
                return new Timeout();
            }

            @Override public void close() throws IOException { /* not used */ }
        };
    }
}