Сжатие InputStream с помощью gzip

Ответ 1

Чтобы сжать данные, вам понадобится GZIPOutputStream. Но так как вам нужно прочитать данные назад, как если бы из InputStream вам нужно было преобразовать OutputStream в InputStream. Вы можете использовать getBytes() для этого:

GZIPOutputStream gout = new GZIPOutputStream();
//... Code to read from your original uncompressed data and write to gout.

//Convert to InputStream.
new ByteArrayInputStream(gout.getBytes());

Но у этого метода есть ограничение, которое нужно сначала прочитать во всех данных, а это значит, что у вас должно быть достаточно памяти для хранения этого буфера.

Альтернативные подходы, использующие Pipes, упоминаются в этом потоке - Как преобразовать OutputStream в InputStream?

Ответ 2

DeflaterInputStream - это не то, что вы хотите, потому что ему не хватает заголовка/трейлера gzip и использует немного другое сжатие.

Если вы измените с OutputStream (push) на InputStream (pull), вам нужно сделать что-то другое.

Что GzipOutputStream делает:

• записать статический заголовок gzip
• напишите дефлированный поток, используя DeflaterOutputStream. Когда поток написан, контрольная сумма CRC32 построена из несжатых данных, а количество байтов - count
• напишите трейлер, содержащий контрольную сумму CRC32 и количество байтов.

Если вы хотите сделать то же самое с InputStreams, вам нужен поток, который содержит:

• заголовок
• сдутый контент
• трейлер

Лучший способ сделать это - предоставить 3 разных потока и объединить их в один. К счастью, SequenceInputStream выполняет объединение потоков для вас.

Здесь моя реализация плюс простой unit test:

import java.io.ByteArrayInputStream;
import java.io.FileInputStream;
import java.io.FilterInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.SequenceInputStream;
import java.util.Enumeration;
import java.util.zip.CRC32;
import java.util.zip.Deflater;
import java.util.zip.DeflaterInputStream;
import java.util.zip.DeflaterOutputStream;

/**
 * @author mwyraz
 * Wraps an input stream and compresses it contents. Similiar to DeflateInputStream but adds GZIP-header and trailer
 * See GzipOutputStream for details.
 * LICENSE: Free to use. Contains some lines from GzipOutputStream, so oracle license might apply as well!
 */
public class GzipCompressingInputStream extends SequenceInputStream
{
    public GzipCompressingInputStream(InputStream in) throws IOException
    {
        this(in,512);
    }
    public GzipCompressingInputStream(InputStream in, int bufferSize) throws IOException
    {
        super(new StatefullGzipStreamEnumerator(in,bufferSize));
    }

    static enum StreamState
    {
        HEADER,
        CONTENT,
        TRAILER
    }

    protected static class StatefullGzipStreamEnumerator implements Enumeration<InputStream>
    {

        protected final InputStream in;
        protected final int bufferSize;
        protected StreamState state;

        public StatefullGzipStreamEnumerator(InputStream in, int bufferSize)
        {
            this.in=in;
            this.bufferSize=bufferSize;
            state=StreamState.HEADER;
        }

        public boolean hasMoreElements()
        {
            return state!=null;
        }
        public InputStream nextElement()
        {
            switch (state)
            {
                case HEADER:
                    state=StreamState.CONTENT;
                    return createHeaderStream();
                case CONTENT:
                    state=StreamState.TRAILER;
                    return createContentStream();
                case TRAILER:
                    state=null;
                    return createTrailerStream();
            }
            return null;
        }

        static final int GZIP_MAGIC = 0x8b1f;
        static final byte[] GZIP_HEADER=new byte[] {
                (byte) GZIP_MAGIC,        // Magic number (short)
                (byte)(GZIP_MAGIC >> 8),  // Magic number (short)
                Deflater.DEFLATED,        // Compression method (CM)
                0,                        // Flags (FLG)
                0,                        // Modification time MTIME (int)
                0,                        // Modification time MTIME (int)
                0,                        // Modification time MTIME (int)
                0,                        // Modification time MTIME (int)
                0,                        // Extra flags (XFLG)
                0                         // Operating system (OS)
        };
        protected InputStream createHeaderStream()
        {
            return new ByteArrayInputStream(GZIP_HEADER);
        }
        protected InternalGzipCompressingInputStream contentStream;
        protected InputStream createContentStream()
        {
            contentStream=new InternalGzipCompressingInputStream(new CRC32InputStream(in), bufferSize);
            return contentStream;
        }
        protected InputStream createTrailerStream()
        {
            return new ByteArrayInputStream(contentStream.createTrailer());
        }
    }

    /**
     * Internal stream without header/trailer  
     */
    protected static class CRC32InputStream extends FilterInputStream
    {
        protected CRC32 crc = new CRC32();
        protected long byteCount;
        public CRC32InputStream(InputStream in)
        {
            super(in);
        }

        @Override
        public int read() throws IOException
        {
            int val=super.read();
            if (val>=0)
            {
                crc.update(val);
                byteCount++;
            }
            return val;
        }
        @Override
        public int read(byte[] b, int off, int len) throws IOException
        {
            len=super.read(b, off, len);
            if (len>=0)
            {
                crc.update(b,off,len);
                byteCount+=len;
            }
            return len;
        }
        public long getCrcValue()
        {
            return crc.getValue();
        }
        public long getByteCount()
        {
            return byteCount;
        }
    }

    /**
     * Internal stream without header/trailer  
     */
    protected static class InternalGzipCompressingInputStream extends DeflaterInputStream
    {
        protected final CRC32InputStream crcIn;
        public InternalGzipCompressingInputStream(CRC32InputStream in, int bufferSize)
        {
            super(in, new Deflater(Deflater.DEFAULT_COMPRESSION, true),bufferSize);
            crcIn=in;
        }
        public void close() throws IOException
        {
            if (in != null)
            {
                try
                {
                    def.end();
                    in.close();
                }
                finally
                {
                    in = null;
                }
            }
        }

        protected final static int TRAILER_SIZE = 8;

        public byte[] createTrailer()
        {
            byte[] trailer= new byte[TRAILER_SIZE];
            writeTrailer(trailer, 0);
            return trailer;
        }

        /*
         * Writes GZIP member trailer to a byte array, starting at a given
         * offset.
         */
        private void writeTrailer(byte[] buf, int offset)
        {
            writeInt((int)crcIn.getCrcValue(), buf, offset); // CRC-32 of uncompr. data
            writeInt((int)crcIn.getByteCount(), buf, offset + 4); // Number of uncompr. bytes
        }

        /*
         * Writes integer in Intel byte order to a byte array, starting at a
         * given offset.
         */
        private void writeInt(int i, byte[] buf, int offset)
        {
            writeShort(i & 0xffff, buf, offset);
            writeShort((i >> 16) & 0xffff, buf, offset + 2);
        }

        /*
         * Writes short integer in Intel byte order to a byte array, starting
         * at a given offset
         */
        private void writeShort(int s, byte[] buf, int offset)
        {
            buf[offset] = (byte)(s & 0xff);
            buf[offset + 1] = (byte)((s >> 8) & 0xff);
        }
    }

}

import static org.junit.Assert.*;

import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.util.Arrays;
import java.util.zip.CRC32;
import java.util.zip.GZIPInputStream;

import org.junit.Test;

public class TestGzipCompressingInputStream
{

    @Test
    public void test() throws Exception
    {
        testCompressor("test1 test2 test3");
        testCompressor("1MB binary data",createTestPattern(1024*1024));
        for (int i=0;i<4096;i++)
        {
            testCompressor(i+" bytes of binary data",createTestPattern(i));
        }
    }

    protected byte[] createTestPattern(int size)
    {
        byte[] data=new byte[size];
        byte pattern=0;
        for (int i=0;i<size;i++)
        {
            data[i]=pattern++;
        }
        return data;
    }

    protected void testCompressor(String data) throws IOException
    {
        testCompressor("String: "+data,data.getBytes());
    }
    protected void testCompressor(String dataInfo, byte[] data) throws IOException
    {
        InputStream uncompressedIn=new ByteArrayInputStream(data);
        InputStream compressedIn=new GzipCompressingInputStream(uncompressedIn);
        InputStream uncompressedOut=new GZIPInputStream(compressedIn);

        byte[] result=StreamHelper.readBinaryStream(uncompressedOut);

        assertTrue("Test failed for: "+dataInfo,Arrays.equals(data,result));

    }

}

Ответ 3

Если вы не хотите загружать контент в большой массив байтов и нуждаетесь в истинном потоковом решении:

package x.y.z;

import org.apache.commons.io.IOUtils;

import java.io.*;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.zip.GZIPInputStream;
import java.util.zip.GZIPOutputStream;
import java.util.zip.ZipOutputStream;

/**
 * Stream Compression Utility
 *
 * @author Thamme Gowda N
 */
public enum CompressionUtil {
    INSTANCE;

    public static final int NUM_THREADS = 5;
    private final ExecutorService pool;

    CompressionUtil(){
        this.pool = Executors.newFixedThreadPool(NUM_THREADS);
    }

    public static CompressionUtil getInstance(){
        return INSTANCE;
    }

    /**
     * Supported compression type names
     */
    public static enum CompressionType {
        GZIP,
        ZIP
    }

    /**
     * Wraps the given stream in a Compressor stream based on given type
     * @param sourceStream : Stream to be wrapped
     * @param type         : Compression type
     * @return source stream wrapped in a compressor stream
     * @throws IOException when some thing bad happens
     */
    public static OutputStream getCompressionWrapper(OutputStream sourceStream,
                                     CompressionType type) throws IOException {

        switch (type) {
            case GZIP:
                return new GZIPOutputStream(sourceStream);
            case ZIP:
                return new ZipOutputStream(sourceStream);
            default:
                throw new IllegalArgumentException("Possible values :"
                        + Arrays.toString(CompressionType.values()));
        }
    }

    /**
     * Gets Compressed Stream for given input Stream
     * @param sourceStream  : Input Stream to be compressed to
     * @param type: Compression types such as GZIP
     * @return  Compressed Stream
     * @throws IOException when some thing bad happens
     */
    public static InputStream getCompressedStream(final InputStream sourceStream,
                                    CompressionType type ) throws IOException {

        if(sourceStream == null) {
            throw new IllegalArgumentException("Source Stream cannot be NULL");
        }

        /**
         *  sourceStream --> zipperOutStream(->intermediateStream -)--> resultStream
         */
        final PipedInputStream resultStream = new PipedInputStream();
        final PipedOutputStream intermediateStream = new PipedOutputStream(resultStream);
        final OutputStream zipperOutStream = getCompressionWrapper(intermediateStream, type);

        Runnable copyTask = new Runnable() {

            @Override
            public void run() {
                try {
                    int c;
                    while((c = sourceStream.read()) >= 0) {
                        zipperOutStream.write(c);
                    }
                    zipperOutStream.flush();
                } catch (IOException e) {
                    IOUtils.closeQuietly(resultStream);  // close it on error case only
                    throw new RuntimeException(e);
                } finally {
                    // close source stream and intermediate streams
                    IOUtils.closeQuietly(sourceStream);
                    IOUtils.closeQuietly(zipperOutStream);
                    IOUtils.closeQuietly(intermediateStream);
                }
            }
        };
        getInstance().pool.submit(copyTask);
        return resultStream;
    }

    public static void main(String[] args) throws IOException {
        String input = "abcdefghij";
        InputStream sourceStream = new ByteArrayInputStream(input.getBytes());
        InputStream compressedStream =
                getCompressedStream(sourceStream, CompressionType.GZIP);

        GZIPInputStream decompressedStream = new GZIPInputStream(compressedStream);
        List<String> lines = IOUtils.readLines(decompressedStream);
        String output = lines.get(0);
        System.out.println("test passed ? " + input.equals(output));

    }
}

Ответ 4

Разве вы не смотрите на GZIPOutputStream в этом случае?

public OutputStream getCompressedStream(InputStream input) {
    OutputStream output = new GZIPOutputStream(new ByteArrayOutputStream()); 
    IOUtils.copy(input, output);
    return output;
}

Ответ 5

Кажется, я опоздал на 3 года, но, возможно, мне будет полезно. Мое решение похоже на решение @Michael Wyraz, только разница в том, что мое решение основано на FilterInputStream

import java.io.ByteArrayInputStream;
import java.io.FilterInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.util.zip.CRC32;
import java.util.zip.Deflater;

public class GZipInputStreamDeflater extends FilterInputStream {

    private static enum Stage {
        HEADER,
        DATA,
        FINALIZATION,
        TRAILER,
        FINISH
    }

    private GZipInputStreamDeflater.Stage stage = Stage.HEADER;

    private final Deflater deflater = new Deflater( Deflater.DEFLATED, true );
    private final CRC32 crc = new CRC32();

    /* GZIP header magic number */
    private final static int GZIP_MAGIC = 0x8b1f;

    private ByteArrayInputStream trailer = null;
    private ByteArrayInputStream header = new ByteArrayInputStream( new byte[] {
        (byte) GZIP_MAGIC, // Magic number (short)
        (byte) ( GZIP_MAGIC >> 8 ), // Magic number (short)
        Deflater.DEFLATED, // Compression method (CM)
        0, // Flags (FLG)
        0, // Modification time MTIME (int)
        0, // Modification time MTIME (int)
        0, // Modification time MTIME (int)
        0, // Modification time MTIME (int)
        0, // Extra flags (XFLG)
        0, // Operating system (OS)
    } );

    public GZipInputStreamDeflater(InputStream in) {
        super( in );
        crc.reset();
    }

    @Override
    public int read( byte[] b, int off, int len ) throws IOException {
        int read = -1;

        switch( stage ) {
            case FINISH:
                return -1;
            case HEADER:
                read = header.read( b, off, len );
                if( header.available() == 0 ) {
                    stage = Stage.DATA;
                }
                return read;
            case DATA:
                byte[] b2 = new byte[len];
                read = super.read( b2, 0, len );
                if( read <= 0 ) {
                    stage = Stage.FINALIZATION;
                    deflater.finish();
                    return 0;
                }
                else {
                    deflater.setInput( b2, 0, read );
                    crc.update( b2, 0, read );
                    read = 0;
                    while( !deflater.needsInput() && len - read > 0 ) {
                        read += deflater.deflate( b, off + read, len - read, Deflater.NO_FLUSH );
                    }
                    return read;
                }
            case FINALIZATION:
                if( deflater.finished() ) {
                    stage = Stage.TRAILER;

                    int crcVaue = (int) crc.getValue();
                    int totalIn = deflater.getTotalIn();

                    trailer = new ByteArrayInputStream( new byte[] {
                        (byte) ( crcVaue >> 0 ),
                        (byte) ( crcVaue >> 8 ),
                        (byte) ( crcVaue >> 16 ),
                        (byte) ( crcVaue >> 24 ),

                        (byte) ( totalIn >> 0 ),
                        (byte) ( totalIn >> 8 ),
                        (byte) ( totalIn >> 16 ),
                        (byte) ( totalIn >> 24 ),
                    } );

                    return 0;
                }
                else {
                    read = deflater.deflate( b, off, len, Deflater.FULL_FLUSH );
                    return read;
                }
            case TRAILER:
                read = trailer.read( b, off, len );
                if( trailer.available() == 0 ) {
                    stage = Stage.FINISH;
                }
                return read;
        }
        return -1;
    }

    @Override
    public void close( ) throws IOException {
        super.close();
        deflater.end();
        if( trailer != null ) {
            trailer.close();
        }
        header.close();
    }
}

Использование:

AmazonS3Client s3client = new AmazonS3Client( ... );
try ( InputStream in = new GZipInputStreamDeflater( new URL( "http://....../very-big-file.csv" ).openStream() ); ) {
    PutObjectRequest putRequest = new PutObjectRequest( "BUCKET-NAME", "/object/key", in, new ObjectMetadata() );
    s3client.putObject( putRequest );
}

Ответ 6

В JRE нет DeflatingGZIPInputStream. Для дефлирования с помощью формата сжатия "спящий" используйте java.util.zip.DeflaterInputStream и java.util.zip.DeflaterOutputStream:

public InputStream getCompressedStream(InputStream unCompressedStream) {
    return new DeflaterInputStream(unCompressedStream); 
}

Вы можете получить класс из java.util.zip.DeflaterInputStream, который дефлируется в формате GZIP, посмотрев на источник java.util.zip.GZIPOutputStream.

Ответ 7

Рабочий пример сжатого входного потока можно найти в популярном ESB с открытым исходным кодом Mule: GZIPCompressorInputStream.

Он использует DeflaterInputStream, предоставленный JRE для сжатия, добавляет gzip и добавляет трейлер gzip (aka footer).

К сожалению, он находится под Лицензией CPA, которая, похоже, не очень распространена. Кроме того, похоже, что нет unit test.

Ответ 8

Вы можете использовать EasyStream.

try(final InputStreamFromOutputStream<Void> isOs = new InputStreamFromOutputStream<Void>() {
    @Override
    protected void produce(final OutputStream dataSink) throws Exception {
        InputStream in = new GZIPInputStream(unCompressedStream);
        IOUtils.copy(in, dataSink);
    }
}) {        

    //You can use the compressed input stream here

} catch (final IOException e) {
    //Handle exceptions here
} 

Ответ 9

PipedOutputStream позволяет писать в GZIPOutputStream и выставлять эти данные через InputStream. Он имеет фиксированную стоимость памяти, в отличие от других решений, которые буферизуют весь поток данных в массив или файл. Только проблема заключается в том, что вы не можете читать и писать из той же темы, вы должны использовать отдельный.

private InputStream gzipInputStream(InputStream in) throws IOException {
    PipedInputStream zipped = new PipedInputStream();
    PipedOutputStream pipe = new PipedOutputStream(zipped);
    new Thread(
            () -> {
                try(OutputStream zipper = new GZIPOutputStream(pipe)){
                    IOUtils.copy(in, zipper);
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
    ).start();
    return zipped;
}

Ответ 10

Я бы рекомендовал использовать GzipCompressorInputStream из Apache Commons Compress.