    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 *
    017 */
    018package org.apache.commons.compress.archivers.zip;
    019
    020
    021import org.apache.commons.compress.parallel.FileBasedScatterGatherBackingStore;
    022import org.apache.commons.compress.parallel.ScatterGatherBackingStore;
    023import org.apache.commons.compress.utils.BoundedInputStream;
    024
    025import java.io.Closeable;
    026import java.io.File;
    027import java.io.FileNotFoundException;
    028import java.io.IOException;
    029import java.io.InputStream;
    030import java.util.Iterator;
    031import java.util.Queue;
    032import java.util.concurrent.ConcurrentLinkedQueue;
    033import java.util.concurrent.atomic.AtomicBoolean;
    034import java.util.zip.Deflater;
    035
    036/**
    037 * A zip output stream that is optimized for multi-threaded scatter/gather construction of zip files.
    038 * <p>
    039 * The internal data format of the entries used by this class are entirely private to this class
    040 * and are not part of any public api whatsoever.
    041 * </p>
    042 * <p>It is possible to extend this class to support different kinds of backing storage, the default
    043 * implementation only supports file-based backing.
    044 * </p>
    045 * Thread safety: This class supports multiple threads. But the "writeTo" method must be called
    046 * by the thread that originally created the {@link ZipArchiveEntry}.
    047 *
    048 * @since 1.10
    049 */
    050public class ScatterZipOutputStream implements Closeable {
    051    private final Queue<CompressedEntry> items = new ConcurrentLinkedQueue<>();
    052    private final ScatterGatherBackingStore backingStore;
    053    private final StreamCompressor streamCompressor;
    054    private final AtomicBoolean isClosed = new AtomicBoolean();
    055    private ZipEntryWriter zipEntryWriter;
    056
    057    private static class CompressedEntry {
    058        final ZipArchiveEntryRequest zipArchiveEntryRequest;
    059        final long crc;
    060        final long compressedSize;
    061        final long size;
    062
    063        public CompressedEntry(final ZipArchiveEntryRequest zipArchiveEntryRequest, final long crc, final long compressedSize, final long size) {
    064            this.zipArchiveEntryRequest = zipArchiveEntryRequest;
    065            this.crc = crc;
    066            this.compressedSize = compressedSize;
    067            this.size = size;
    068        }
    069
    070        /**
    071         * Update the original {@link ZipArchiveEntry} with sizes/crc
    072         * Do not use this methods from threads that did not create the instance itself !
    073         * @return the zipArchiveEntry that is basis for this request
    074         */
    075
    076        public ZipArchiveEntry transferToArchiveEntry(){
    077            final ZipArchiveEntry entry = zipArchiveEntryRequest.getZipArchiveEntry();
    078            entry.setCompressedSize(compressedSize);
    079            entry.setSize(size);
    080            entry.setCrc(crc);
    081            entry.setMethod(zipArchiveEntryRequest.getMethod());
    082            return entry;
    083        }
    084    }
    085
    086    public ScatterZipOutputStream(final ScatterGatherBackingStore backingStore,
    087                                  final StreamCompressor streamCompressor) {
    088        this.backingStore = backingStore;
    089        this.streamCompressor = streamCompressor;
    090    }
    091
    092    /**
    093     * Add an archive entry to this scatter stream.
    094     *
    095     * @param zipArchiveEntryRequest The entry to write.
    096     * @throws IOException    If writing fails
    097     */
    098    public void addArchiveEntry(final ZipArchiveEntryRequest zipArchiveEntryRequest) throws IOException {
    099        try (final InputStream payloadStream = zipArchiveEntryRequest.getPayloadStream()) {
    100            streamCompressor.deflate(payloadStream, zipArchiveEntryRequest.getMethod());
    101        }
    102        items.add(new CompressedEntry(zipArchiveEntryRequest, streamCompressor.getCrc32(),
    103                                      streamCompressor.getBytesWrittenForLastEntry(), streamCompressor.getBytesRead()));
    104    }
    105
    106    /**
    107     * Write the contents of this scatter stream to a target archive.
    108     *
    109     * @param target The archive to receive the contents of this {@link ScatterZipOutputStream}.
    110     * @throws IOException If writing fails
    111     * @see #zipEntryWriter()
    112     */
    113    public void writeTo(final ZipArchiveOutputStream target) throws IOException {
    114        backingStore.closeForWriting();
    115        try (final InputStream data = backingStore.getInputStream()) {
    116            for (final CompressedEntry compressedEntry : items) {
    117                try (final BoundedInputStream rawStream = new BoundedInputStream(data,
    118                        compressedEntry.compressedSize)) {
    119                    target.addRawArchiveEntry(compressedEntry.transferToArchiveEntry(), rawStream);
    120                }
    121            }
    122        }
    123    }
    124
    125    public static class ZipEntryWriter implements Closeable {
    126        private final Iterator<CompressedEntry> itemsIterator;
    127        private final InputStream itemsIteratorData;
    128
    129        public ZipEntryWriter(final ScatterZipOutputStream scatter) throws IOException {
    130            scatter.backingStore.closeForWriting();
    131            itemsIterator = scatter.items.iterator();
    132            itemsIteratorData = scatter.backingStore.getInputStream();
    133        }
    134
    135        @Override
    136        public void close() throws IOException {
    137            if (itemsIteratorData != null) {
    138                itemsIteratorData.close();
    139            }
    140        }
    141
    142        public void writeNextZipEntry(final ZipArchiveOutputStream target) throws IOException {
    143            final CompressedEntry compressedEntry = itemsIterator.next();
    144            try (final BoundedInputStream rawStream = new BoundedInputStream(itemsIteratorData, compressedEntry.compressedSize)) {
    145                target.addRawArchiveEntry(compressedEntry.transferToArchiveEntry(), rawStream);
    146            }
    147        }
    148    }
    149
    150    /**
    151     * Get a zip entry writer for this scatter stream.
    152     * @throws IOException If getting scatter stream input stream
    153     * @return the ZipEntryWriter created on first call of the method
    154     */
    155    public ZipEntryWriter zipEntryWriter() throws IOException {
    156        if (zipEntryWriter == null) {
    157            zipEntryWriter = new ZipEntryWriter(this);
    158        }
    159        return zipEntryWriter;
    160    }
    161
    162    /**
    163     * Closes this stream, freeing all resources involved in the creation of this stream.
    164     * @throws IOException If closing fails
    165     */
    166    @Override
    167    public void close() throws IOException {
    168        if (!isClosed.compareAndSet(false, true)) {
    169            return;
    170        }
    171        try {
    172            if (zipEntryWriter != null) {
    173                zipEntryWriter.close();
    174            }
    175            backingStore.close();
    176        } finally {
    177            streamCompressor.close();
    178        }
    179    }
    180
    181    /**
    182     * Create a {@link ScatterZipOutputStream} with default compression level that is backed by a file
    183     *
    184     * @param file The file to offload compressed data into.
    185     * @return A ScatterZipOutputStream that is ready for use.
    186     * @throws FileNotFoundException if the file cannot be found
    187     */
    188    public static ScatterZipOutputStream fileBased(final File file) throws FileNotFoundException {
    189        return fileBased(file, Deflater.DEFAULT_COMPRESSION);
    190    }
    191
    192    /**
    193     * Create a {@link ScatterZipOutputStream} that is backed by a file
    194     *
    195     * @param file             The file to offload compressed data into.
    196     * @param compressionLevel The compression level to use, @see #Deflater
    197     * @return A  ScatterZipOutputStream that is ready for use.
    198     * @throws FileNotFoundException if the file cannot be found
    199     */
    200    public static ScatterZipOutputStream fileBased(final File file, final int compressionLevel) throws FileNotFoundException {
    201        final ScatterGatherBackingStore bs = new FileBasedScatterGatherBackingStore(file);
    202        // lifecycle is bound to the ScatterZipOutputStream returned
    203        final StreamCompressor sc = StreamCompressor.create(compressionLevel, bs); //NOSONAR
    204        return new ScatterZipOutputStream(bs, sc);
    205    }
    206}