/*
 * Written by Dawid Kurzyniec and released to the public domain, as explained
 * at http://creativecommons.org/licenses/publicdomain
 */

package edu.emory.mathcs.util.io;

import java.io.*;
import java.util.zip.*;

/**
 * Filter output stream that compresses data and features strong
 * flush semantics. Data is written as gzipped packets of variable size.
 * Flushing causes immediate ending of currently written packet and sending all
 * the data off. Therefore, this stream can be used as a transport for RMI or
 * RPC.
 *
 * Note that standard {@link java.util.zip.ZipOutputStream} and
 * {@link java.util.zip.GZIPOutputStream} are useless for this purpose due to
 * their insufficiently strong flushing semantics: they don't guarantee that
 * flush sends out all the data that was written so far, which leads to
 * deadlocks in request-response-based protocols.
 * <p>
 * Compression ratio decreases with decreasing packet size.
 * Hence, the gain will be low or none in RMI applications that exchange
 * small chunks of data. Since the protocol adds two bytes of metadata per each
 * packet, in the extreme case (flushing every single byte) the number of
 * bytes actually sent is tripled. On the other hand, the compression ratio
 * may be significant if large chunks are exchanged, e.g. if large
 * arrays are sent as parameters or received as return values.
 *
 * @see CompressedInputStream
 *
 * @author Dawid Kurzyniec
 * @version 1.0
 */
public class CompressedOutputStream extends FilterOutputStream {

    final static short DEFLATED = (short)0x8000;
    final static short STORED   = (short)0x0000;
    final static int   STORE_TRESHOLD = 32;

    final byte[] buf;
    final byte[] destbuf;
    final DataOutputStream dos;
    int pos;
    final Deflater deflater;

    public CompressedOutputStream(OutputStream out) {
        this(out, 8192);
    }

    public CompressedOutputStream(OutputStream out, int bufSize) {
        super(out);
        if (bufSize <= 0x0010 || bufSize > 0x7FFF) {
            throw new IllegalArgumentException("Invalid buffer size");
        }
        this.buf = new byte[bufSize];
        this.destbuf = new byte[bufSize];
        this.deflater = new Deflater(Deflater.DEFAULT_COMPRESSION, true);
        this.pos = 0;
        this.dos = new DataOutputStream(out);
    }

    public synchronized void write(int b) throws IOException {
        buf[pos++] = (byte)b;
        if (pos == buf.length) {
            writePacket();
        }
    }

    public synchronized void write(byte[] data, int off, int len) throws IOException {
        while (len > 0) {
            int todo = Math.min(buf.length-pos, len);
            System.arraycopy(data, off, buf, pos, todo);
            pos += todo;
            off += todo;
            len -= todo;
            if (pos == buf.length) {
                writePacket();
            }
        }
    }

    public synchronized void flush() throws IOException {
        // make sure we don't write out empty packets
        if (pos != 0) writePacket();
        out.flush();
    }

    private synchronized void writePacket() throws IOException {
        short header;
        if (pos < STORE_TRESHOLD) {
            // store only
            header = (short)((short)pos | STORED);
            dos.writeShort(header);
            out.write(buf, 0, pos);
        }
        else {
            // check whichever gets smaller packet
            deflater.reset();
            deflater.setInput(buf, 0, pos);
            deflater.finish();
            int off = 0;
            while (!deflater.finished() && off < destbuf.length) {
                int wrote = deflater.deflate(destbuf, off, destbuf.length - off);
                assert (wrote > 0);
                off += wrote;
            }
            if (off < destbuf.length - 2) {
                // deflate
                header = (short)((short)off | DEFLATED);
                dos.writeShort(header);
                out.write(destbuf, 0, off);
            }
            else {
                // store only
                header = (short)((short)pos | STORED);
                dos.writeShort(header);
                out.write(buf, 0, pos);
            }
        }
        pos = 0;
    }
}
