Add synchronously entries from resource collections that do no support concurrent access

src/main/java/org/codehaus/plexus/archiver/ArchiveEntry.java

@@ -60,6 +60,8 @@ public class ArchiveEntry
 
     private PlexusIoResourceAttributes attributes;
 
+    private final boolean addSynchronously;
+
     /**
      * @param name     the filename as it will appear in the archive. This is platform-specific
      *                 normalized with File.separatorChar
@@ -92,6 +94,8 @@ private ArchiveEntry( String name, @Nonnull PlexusIoResource resource, int type,
 
         this.mode = permissions == -1 ? permissions : ( permissions & UnixStat.PERM_MASK ) |
             ( type == FILE ? UnixStat.FILE_FLAG : type == SYMLINK ? UnixStat.LINK_FLAG : UnixStat.DIR_FLAG );
+
+        this.addSynchronously = ( collection != null && !collection.isConcurrentAccessSupported() );
     }
 
     /**
@@ -154,6 +158,17 @@ public int getMode()
             ( type == FILE ? UnixStat.FILE_FLAG : type == SYMLINK ? UnixStat.LINK_FLAG : UnixStat.DIR_FLAG );
     }
 
+    /**
+     * Indicates if this entry should be added to the archive synchronously
+     * before adding the next entry and/or accessing the next entry of {@link ResourceIterator}. 
+     * 
+     * @return {@code true} if this entry should be added synchronously
+     */
+    public boolean shouldAddSynchronously()
+    {
+        return addSynchronously;
+    }
+
     public static ArchiveEntry createFileEntry( String target, PlexusIoResource resource, int permissions,
                                                 PlexusIoResourceCollection collection, int defaultDirectoryPermissions )
         throws ArchiverException

src/main/java/org/codehaus/plexus/archiver/jar/JarArchiver.java

@@ -333,7 +333,7 @@ private void writeManifest( ConcurrentJarCreator zOut, Manifest manifest )
 
         ByteArrayInputStream bais = new ByteArrayInputStream( baos.toByteArray() );
         super.zipFile( createInputStreamSupplier( bais ), zOut, MANIFEST_NAME, System.currentTimeMillis(), null,
-                       DEFAULT_FILE_MODE, null );
+                       DEFAULT_FILE_MODE, null, false );
         super.initZipOutputStream( zOut );
     }
 
@@ -435,15 +435,15 @@ private void createIndexList( ConcurrentJarCreator zOut )
         ByteArrayInputStream bais = new ByteArrayInputStream( baos.toByteArray() );
 
         super.zipFile( createInputStreamSupplier( bais ), zOut, INDEX_NAME, System.currentTimeMillis(), null,
-                       DEFAULT_FILE_MODE, null );
+                       DEFAULT_FILE_MODE, null, true );
     }
 
     /**
      * Overridden from Zip class to deal with manifests and index lists.
      */
     protected void zipFile( InputStreamSupplier is, ConcurrentJarCreator zOut, String vPath,
                             long lastModified, File fromArchive,
-                            int mode, String symlinkDestination )
+                            int mode, String symlinkDestination, boolean addInParallel )
         throws IOException, ArchiverException
     {
         if ( MANIFEST_NAME.equalsIgnoreCase( vPath ) )
@@ -464,7 +464,7 @@ else if ( INDEX_NAME.equalsIgnoreCase( vPath ) && index )
             {
                 rootEntries.addElement( vPath );
             }
-            super.zipFile( is, zOut, vPath, lastModified, fromArchive, mode, symlinkDestination );
+            super.zipFile( is, zOut, vPath, lastModified, fromArchive, mode, symlinkDestination, addInParallel );
         }
     }
 

src/main/java/org/codehaus/plexus/archiver/zip/AbstractZipArchiver.java

@@ -421,11 +421,14 @@ private void addParentDirs(ArchiveEntry archiveEntry, File baseDir, String entry
      * @param lastModified       last modification time for the entry.
      * @param fromArchive        the original archive we are copying this
      * @param symlinkDestination
+     * @param addInParallel      Indicates if the entry should be add in parallel.
+     *                           If set to {@code false} it is added synchronously.
+     *                           If the entry is symbolic link this parameter is ignored.
      */
     @SuppressWarnings( { "JavaDoc" } )
     protected void zipFile( InputStreamSupplier in, ConcurrentJarCreator zOut, String vPath,
                             long lastModified,
-                            File fromArchive, int mode, String symlinkDestination )
+                            File fromArchive, int mode, String symlinkDestination, boolean addInParallel )
         throws IOException, ArchiverException
     {
         getLogger().debug( "adding entry " + vPath );
@@ -446,11 +449,11 @@ protected void zipFile( InputStreamSupplier in, ConcurrentJarCreator zOut, Strin
                 ZipEncoding enc = ZipEncodingHelper.getZipEncoding( getEncoding() );
                 final byte[] bytes = enc.encode( symlinkDestination ).array();
                 payload = new ByteArrayInputStream( bytes );
-                zOut.addArchiveEntry( ze, createInputStreamSupplier( payload ) );
+                zOut.addArchiveEntry( ze, createInputStreamSupplier( payload ), true );
             }
             else
             {
-                zOut.addArchiveEntry( ze, wrappedRecompressor( ze, in ) );
+                zOut.addArchiveEntry( ze, wrappedRecompressor( ze, in ), addInParallel );
             }
         }
     }
@@ -502,7 +505,8 @@ public InputStream get()
         };
         try
         {
-            zipFile( in, zOut, vPath, resource.getLastModified(), null, entry.getMode(), symlinkTarget );
+            zipFile( in, zOut, vPath, resource.getLastModified(), null, entry.getMode(), symlinkTarget,
+                !entry.shouldAddSynchronously() );
         }
         catch ( IOException e )
         {
@@ -583,15 +587,15 @@ protected void zipDir( PlexusIoResource dir, ConcurrentJarCreator zOut, String v
 
             if ( !isSymlink )
             {
-                zOut.addArchiveEntry( ze, createInputStreamSupplier( new ByteArrayInputStream( "".getBytes() ) ) );
+                zOut.addArchiveEntry( ze, createInputStreamSupplier( new ByteArrayInputStream( "".getBytes() ) ), true );
             }
             else
             {
                 String symlinkDestination = ( (SymlinkDestinationSupplier) dir ).getSymlinkDestination();
                 ZipEncoding enc = ZipEncodingHelper.getZipEncoding( encodingToUse );
                 final byte[] bytes = enc.encode( symlinkDestination ).array();
                 ze.setMethod( ZipArchiveEntry.DEFLATED );
-                zOut.addArchiveEntry( ze, createInputStreamSupplier( new ByteArrayInputStream( bytes ) ) );
+                zOut.addArchiveEntry( ze, createInputStreamSupplier( new ByteArrayInputStream( bytes ) ), true );
             }
         }
     }

src/main/java/org/codehaus/plexus/archiver/zip/ConcurrentJarCreator.java

@@ -37,6 +37,7 @@ public class ConcurrentJarCreator {
     private final ScatterZipOutputStream directories;
     private final ScatterZipOutputStream metaInfDir;
     private final ScatterZipOutputStream manifest;
+    private final ScatterZipOutputStream synchronousEntries;
 
     private final ParallelScatterZipCreator parallelScatterZipCreator;
     private long zipCloseElapsed;
@@ -61,6 +62,7 @@ public ConcurrentJarCreator(int nThreads) throws IOException {
         directories = createDeferred(defaultSupplier);
         manifest = createDeferred(defaultSupplier);
         metaInfDir = createDeferred( defaultSupplier );
+        synchronousEntries = createDeferred( defaultSupplier );
 
         parallelScatterZipCreator = new ParallelScatterZipCreator(Executors.newFixedThreadPool(nThreads), defaultSupplier);
     }
@@ -72,10 +74,12 @@ public ConcurrentJarCreator(int nThreads) throws IOException {
      *
      * @param zipArchiveEntry The entry to add. Compression method
      * @param source          The source input stream supplier
+     * @param addInParallel   Indicates if the entry should be add in parallel.
+     *                        If set to {@code false} the entry is added synchronously.
      * @throws java.io.IOException
      */
-
-    public void addArchiveEntry(final ZipArchiveEntry zipArchiveEntry, final InputStreamSupplier source) throws IOException {
+    public void addArchiveEntry(final ZipArchiveEntry zipArchiveEntry, final InputStreamSupplier source,
+            final boolean addInParallel) throws IOException {
         final int method = zipArchiveEntry.getMethod();
         if (method == -1) throw new IllegalArgumentException("Method must be set on the supplied zipArchiveEntry");
         if (zipArchiveEntry.isDirectory() && !zipArchiveEntry.isUnixSymlink()) {
@@ -93,8 +97,10 @@ public void addArchiveEntry(final ZipArchiveEntry zipArchiveEntry, final InputSt
             if (zipArchiveEntry.isDirectory()) zipArchiveEntry.setMethod(ZipEntry.STORED);
             manifest.addArchiveEntry(createZipArchiveEntryRequest(zipArchiveEntry, createInputStreamSupplier(payload)));
             payload.close();
-        } else {
+        } else if (addInParallel) {
             parallelScatterZipCreator.addArchiveEntry(zipArchiveEntry, source);
+        } else {
+            synchronousEntries.addArchiveEntry( createZipArchiveEntryRequest( zipArchiveEntry, source ) );
         }
     }
 
@@ -110,12 +116,14 @@ public void writeTo(ZipArchiveOutputStream targetStream) throws IOException, Exe
         metaInfDir.writeTo( targetStream );
         manifest.writeTo(targetStream);
         directories.writeTo(targetStream);
+        synchronousEntries.writeTo( targetStream );
         parallelScatterZipCreator.writeTo( targetStream);
         long startAt = System.currentTimeMillis();
         targetStream.close();
         zipCloseElapsed = System.currentTimeMillis() - startAt;
         manifest.close();
         directories.close();
+        synchronousEntries.close();
     }
 
     /**

src/test/java/org/codehaus/plexus/archiver/zip/ArchiveFileComparator.java

@@ -94,6 +94,22 @@ private static void assertEquals( ArchiveFile file1, TarArchiveEntry entry1,
         is1.close();
         is2.close();
     }
+
+    private static void assertEquals( ArchiveFile file1, TarArchiveEntry entry1,
+                                      ZipFile file2, ZipArchiveEntry entry2 )
+            throws IOException
+    {
+        Assert.assertEquals( entry1.isDirectory(), entry2.isDirectory() );
+
+        final InputStream is1 = file1.getInputStream( entry1 );
+        final InputStream is2 = file2.getInputStream( entry2 );
+        final byte[] bytes1 = IOUtil.toByteArray( is1 );
+        final byte[] bytes2 = IOUtil.toByteArray( is2 );
+        Assert.assertTrue( Arrays.equals( bytes1, bytes2 ) );
+        is1.close();
+        is2.close();
+    }
+
     private static void assertEquals( ZipFile file1, ZipArchiveEntry entry1,
                                       ZipFile file2, ZipArchiveEntry entry2 )
             throws Exception
@@ -136,6 +152,26 @@ public void accept( TarArchiveEntry ze1 )
         Assert.assertTrue( map2.isEmpty() );
     }
 
+
+    public static void assertEquals( final ArchiveFile file1, final ZipFile file2, final String prefix )
+        throws Exception
+    {
+        final Map<String,ZipArchiveEntry> map2 = getFileEntries( file2 );
+        forEachTarArchiveEntry( file1, new TarArchiveEntryConsumer()
+        {
+            public void accept( TarArchiveEntry ze1 )
+                throws IOException
+            {
+                final String name1 = ze1.getName();
+                final String name2 = prefix == null ? name1 : ( prefix + name1 );
+                ZipArchiveEntry ze2 = map2.remove( name2 );
+                Assert.assertNotNull( ze2 );
+                assertEquals( file1, ze1, file2, ze2 );
+            }
+        } );
+        Assert.assertTrue( map2.isEmpty() );
+    }
+
     public static void assertEquals( org.apache.commons.compress.archivers.zip.ZipFile file1, org.apache.commons.compress.archivers.zip.ZipFile file2, String prefix )
             throws Exception
     {

src/test/java/org/codehaus/plexus/archiver/zip/ConcurrentJarCreatorTest.java

@@ -75,7 +75,7 @@ public InputStream get() {
                         throw new RuntimeException(e);
                     }
                 }
-            });
+            }, true);
         }
 
     }

src/test/java/org/codehaus/plexus/archiver/zip/ZipArchiverTest.java

@@ -34,6 +34,8 @@
 import org.codehaus.plexus.archiver.BasePlexusArchiverTest;
 import org.codehaus.plexus.archiver.UnArchiver;
 import org.codehaus.plexus.archiver.UnixStat;
+import org.codehaus.plexus.archiver.tar.TarArchiver;
+import org.codehaus.plexus.archiver.tar.TarFile;
 import org.codehaus.plexus.archiver.util.ArchiveEntryUtils;
 import org.codehaus.plexus.archiver.util.DefaultArchivedFileSet;
 import org.codehaus.plexus.archiver.util.DefaultFileSet;
@@ -675,6 +677,36 @@ public void testCreateResourceCollection()
         cmp2.close();
     }
 
+    public void testZipNonConcurrentResourceCollection()
+        throws Exception
+    {
+        final File tarFile = getTestFile( "target/output/zip-non-concurrent.tar" );
+        TarArchiver tarArchiver = (TarArchiver) lookup( Archiver.ROLE, "tar" );
+        tarArchiver.setDestFile( tarFile );
+        // We're testing concurrency issue so we need large amount of files
+        for (int i = 0; i < 100; i++)
+        {
+            tarArchiver.addFile( getTestFile( "src/test/resources/manifests/manifest1.mf" ),
+                "manifest1.mf" + i );
+            // Directories are added separately so let's test them too
+            tarArchiver.addFile( getTestFile( "src/test/resources/manifests/manifest2.mf" ),
+                "subdir" + i + "/manifest2.mf" );
+        }
+        tarArchiver.createArchive();
+
+        final File zipFile = new File( "target/output/zip-non-concurrent.zip" );
+        ZipArchiver zipArchive = getZipArchiver( zipFile );
+        zipArchive.addArchivedFileSet( tarFile, "prfx/" );
+        zipArchive.setEncoding( "UTF-8" );
+        zipArchive.createArchive();
+
+        final TarFile cmp1 = new TarFile( tarFile );
+        final ZipFile cmp2 = new ZipFile( zipFile );
+        ArchiveFileComparator.assertEquals( cmp1, cmp2, "prfx/" );
+        cmp1.close();
+        cmp2.close();
+    }
+
     public void testDefaultUTF8()
         throws IOException
     {