001/*
002 *  Licensed to the Apache Software Foundation (ASF) under one or more
003 *  contributor license agreements.  See the NOTICE file distributed with
004 *  this work for additional information regarding copyright ownership.
005 *  The ASF licenses this file to You under the Apache License, Version 2.0
006 *  (the "License"); you may not use this file except in compliance with
007 *  the License.  You may obtain a copy of the License at
008 *
009 *      http://www.apache.org/licenses/LICENSE-2.0
010 *
011 *  Unless required by applicable law or agreed to in writing, software
012 *  distributed under the License is distributed on an "AS IS" BASIS,
013 *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
014 *  See the License for the specific language governing permissions and
015 *  limitations under the License.
016 */
017package org.apache.commons.compress.archivers.zip;
018
019import java.io.Closeable;
020import java.io.DataOutput;
021import java.io.IOException;
022import java.io.InputStream;
023import java.io.OutputStream;
024import java.nio.ByteBuffer;
025import java.nio.channels.SeekableByteChannel;
026import java.util.zip.CRC32;
027import java.util.zip.Deflater;
028import java.util.zip.ZipEntry;
029
030import org.apache.commons.compress.parallel.ScatterGatherBackingStore;
031
032/**
033 * Encapsulates a {@link Deflater} and crc calculator, handling multiple types of output streams. Currently {@link java.util.zip.ZipEntry#DEFLATED} and
034 * {@link java.util.zip.ZipEntry#STORED} are the only supported compression methods.
035 *
036 * @since 1.10
037 */
038public abstract class StreamCompressor implements Closeable {
039
040    private static final class DataOutputCompressor extends StreamCompressor {
041        private final DataOutput raf;
042
043        DataOutputCompressor(final Deflater deflater, final DataOutput raf) {
044            super(deflater);
045            this.raf = raf;
046        }
047
048        @Override
049        protected void writeOut(final byte[] data, final int offset, final int length) throws IOException {
050            raf.write(data, offset, length);
051        }
052    }
053
054    private static final class OutputStreamCompressor extends StreamCompressor {
055        private final OutputStream os;
056
057        OutputStreamCompressor(final Deflater deflater, final OutputStream os) {
058            super(deflater);
059            this.os = os;
060        }
061
062        @Override
063        protected void writeOut(final byte[] data, final int offset, final int length) throws IOException {
064            os.write(data, offset, length);
065        }
066    }
067
068    private static final class ScatterGatherBackingStoreCompressor extends StreamCompressor {
069        private final ScatterGatherBackingStore bs;
070
071        ScatterGatherBackingStoreCompressor(final Deflater deflater, final ScatterGatherBackingStore bs) {
072            super(deflater);
073            this.bs = bs;
074        }
075
076        @Override
077        protected void writeOut(final byte[] data, final int offset, final int length) throws IOException {
078            bs.writeOut(data, offset, length);
079        }
080    }
081
082    private static final class SeekableByteChannelCompressor extends StreamCompressor {
083        private final SeekableByteChannel channel;
084
085        SeekableByteChannelCompressor(final Deflater deflater, final SeekableByteChannel channel) {
086            super(deflater);
087            this.channel = channel;
088        }
089
090        @Override
091        protected void writeOut(final byte[] data, final int offset, final int length) throws IOException {
092            channel.write(ByteBuffer.wrap(data, offset, length));
093        }
094    }
095
096    /**
097     * Apparently Deflater.setInput gets slowed down a lot on Sun JVMs when it gets handed a huge buffer. See
098     * https://issues.apache.org/bugzilla/show_bug.cgi?id=45396
099     *
100     * Using a buffer size of {@value} bytes proved to be a good compromise
101     */
102    private static final int DEFLATER_BLOCK_SIZE = 8192;
103    private static final int BUFFER_SIZE = 4096;
104
105    /**
106     * Creates a stream compressor with the given compression level.
107     *
108     * @param os       The DataOutput to receive output
109     * @param deflater The deflater to use for the compressor
110     * @return A stream compressor
111     */
112    static StreamCompressor create(final DataOutput os, final Deflater deflater) {
113        return new DataOutputCompressor(deflater, os);
114    }
115
116    /**
117     * Creates a stream compressor with the given compression level.
118     *
119     * @param compressionLevel The {@link Deflater} compression level
120     * @param bs               The ScatterGatherBackingStore to receive output
121     * @return A stream compressor
122     */
123    public static StreamCompressor create(final int compressionLevel, final ScatterGatherBackingStore bs) {
124        final Deflater deflater = new Deflater(compressionLevel, true);
125        return new ScatterGatherBackingStoreCompressor(deflater, bs);
126    }
127
128    /**
129     * Creates a stream compressor with the default compression level.
130     *
131     * @param os The stream to receive output
132     * @return A stream compressor
133     */
134    static StreamCompressor create(final OutputStream os) {
135        return create(os, new Deflater(Deflater.DEFAULT_COMPRESSION, true));
136    }
137
138    /**
139     * Creates a stream compressor with the given compression level.
140     *
141     * @param os       The stream to receive output
142     * @param deflater The deflater to use
143     * @return A stream compressor
144     */
145    static StreamCompressor create(final OutputStream os, final Deflater deflater) {
146        return new OutputStreamCompressor(deflater, os);
147    }
148
149    /**
150     * Creates a stream compressor with the default compression level.
151     *
152     * @param bs The ScatterGatherBackingStore to receive output
153     * @return A stream compressor
154     */
155    public static StreamCompressor create(final ScatterGatherBackingStore bs) {
156        return create(Deflater.DEFAULT_COMPRESSION, bs);
157    }
158
159    /**
160     * Creates a stream compressor with the given compression level.
161     *
162     * @param os       The SeekableByteChannel to receive output
163     * @param deflater The deflater to use for the compressor
164     * @return A stream compressor
165     * @since 1.13
166     */
167    static StreamCompressor create(final SeekableByteChannel os, final Deflater deflater) {
168        return new SeekableByteChannelCompressor(deflater, os);
169    }
170
171    private final Deflater deflater;
172
173    private final CRC32 crc = new CRC32();
174
175    private long writtenToOutputStreamForLastEntry;
176
177    private long sourcePayloadLength;
178
179    private long totalWrittenToOutputStream;
180
181    private final byte[] outputBuffer = new byte[BUFFER_SIZE];
182
183    private final byte[] readerBuf = new byte[BUFFER_SIZE];
184
185    StreamCompressor(final Deflater deflater) {
186        this.deflater = deflater;
187    }
188
189    @Override
190    public void close() throws IOException {
191        deflater.end();
192    }
193
194    void deflate() throws IOException {
195        final int len = deflater.deflate(outputBuffer, 0, outputBuffer.length);
196        if (len > 0) {
197            writeCounted(outputBuffer, 0, len);
198        }
199    }
200
201    /**
202     * Deflates the given source using the supplied compression method
203     *
204     * @param source The source to compress
205     * @param method The #ZipArchiveEntry compression method
206     * @throws IOException When failures happen
207     */
208
209    public void deflate(final InputStream source, final int method) throws IOException {
210        reset();
211        int length;
212
213        while ((length = source.read(readerBuf, 0, readerBuf.length)) >= 0) {
214            write(readerBuf, 0, length, method);
215        }
216        if (method == ZipEntry.DEFLATED) {
217            flushDeflater();
218        }
219    }
220
221    private void deflateUntilInputIsNeeded() throws IOException {
222        while (!deflater.needsInput()) {
223            deflate();
224        }
225    }
226
227    void flushDeflater() throws IOException {
228        deflater.finish();
229        while (!deflater.finished()) {
230            deflate();
231        }
232    }
233
234    /**
235     * Gets the number of bytes read from the source stream
236     *
237     * @return The number of bytes read, never negative
238     */
239    public long getBytesRead() {
240        return sourcePayloadLength;
241    }
242
243    /**
244     * Gets the number of bytes written to the output for the last entry
245     *
246     * @return The number of bytes, never negative
247     */
248    public long getBytesWrittenForLastEntry() {
249        return writtenToOutputStreamForLastEntry;
250    }
251
252    /**
253     * Gets the crc32 of the last deflated file
254     *
255     * @return the crc32
256     */
257
258    public long getCrc32() {
259        return crc.getValue();
260    }
261
262    /**
263     * Gets the total number of bytes written to the output for all files
264     *
265     * @return The number of bytes, never negative
266     */
267    public long getTotalBytesWritten() {
268        return totalWrittenToOutputStream;
269    }
270
271    void reset() {
272        crc.reset();
273        deflater.reset();
274        sourcePayloadLength = 0;
275        writtenToOutputStreamForLastEntry = 0;
276    }
277
278    /**
279     * Writes bytes to ZIP entry.
280     *
281     * @param b      the byte array to write
282     * @param offset the start position to write from
283     * @param length the number of bytes to write
284     * @param method the comrpession method to use
285     * @return the number of bytes written to the stream this time
286     * @throws IOException on error
287     */
288    long write(final byte[] b, final int offset, final int length, final int method) throws IOException {
289        final long current = writtenToOutputStreamForLastEntry;
290        crc.update(b, offset, length);
291        if (method == ZipEntry.DEFLATED) {
292            writeDeflated(b, offset, length);
293        } else {
294            writeCounted(b, offset, length);
295        }
296        sourcePayloadLength += length;
297        return writtenToOutputStreamForLastEntry - current;
298    }
299
300    public void writeCounted(final byte[] data) throws IOException {
301        writeCounted(data, 0, data.length);
302    }
303
304    public void writeCounted(final byte[] data, final int offset, final int length) throws IOException {
305        writeOut(data, offset, length);
306        writtenToOutputStreamForLastEntry += length;
307        totalWrittenToOutputStream += length;
308    }
309
310    private void writeDeflated(final byte[] b, final int offset, final int length) throws IOException {
311        if (length > 0 && !deflater.finished()) {
312            if (length <= DEFLATER_BLOCK_SIZE) {
313                deflater.setInput(b, offset, length);
314                deflateUntilInputIsNeeded();
315            } else {
316                final int fullblocks = length / DEFLATER_BLOCK_SIZE;
317                for (int i = 0; i < fullblocks; i++) {
318                    deflater.setInput(b, offset + i * DEFLATER_BLOCK_SIZE, DEFLATER_BLOCK_SIZE);
319                    deflateUntilInputIsNeeded();
320                }
321                final int done = fullblocks * DEFLATER_BLOCK_SIZE;
322                if (done < length) {
323                    deflater.setInput(b, offset + done, length - done);
324                    deflateUntilInputIsNeeded();
325                }
326            }
327        }
328    }
329
330    protected abstract void writeOut(byte[] data, int offset, int length) throws IOException;
331}