Merge pull request #705 from sonatype/proxy-eof

Proxy EOF detection.

nexus-core/src/main/java/org/sonatype/nexus/proxy/LocalStorageEofException.java

@@ -18,22 +18,22 @@
  * during upload, recovery is to have client retry upload).
  * 
  * @author cstamas
- * @since 2.3
+ * @since 2.4
  */
-public class LocalStorageEofException
+public class LocalStorageEOFException
     extends LocalStorageException
 {
-    public LocalStorageEofException( String msg )
+    public LocalStorageEOFException( String msg )
     {
         super( msg );
     }
 
-    public LocalStorageEofException( String msg, Throwable cause )
+    public LocalStorageEOFException( String msg, Throwable cause )
     {
         super( msg, cause );
     }
 
-    public LocalStorageEofException( Throwable cause )
+    public LocalStorageEOFException( Throwable cause )
     {
         super( cause.getMessage(), cause );
     }

nexus-core/src/main/java/org/sonatype/nexus/proxy/RemoteStorageEOFException.java

@@ -0,0 +1,75 @@
+/*
+ * Sonatype Nexus (TM) Open Source Version
+ * Copyright (c) 2007-2012 Sonatype, Inc.
+ * All rights reserved. Includes the third-party code listed at http://links.sonatype.com/products/nexus/oss/attributions.
+ *
+ * This program and the accompanying materials are made available under the terms of the Eclipse Public License Version 1.0,
+ * which accompanies this distribution and is available at http://www.eclipse.org/legal/epl-v10.html.
+ *
+ * Sonatype Nexus (TM) Professional Version is available from Sonatype, Inc. "Sonatype" and "Sonatype Nexus" are trademarks
+ * of Sonatype, Inc. Apache Maven is a trademark of the Apache Software Foundation. M2eclipse is a trademark of the
+ * Eclipse Foundation. All other trademarks are the property of their respective owners.
+ */
+package org.sonatype.nexus.proxy;
+
+import org.sonatype.nexus.proxy.repository.ProxyRepository;
+import com.google.common.base.Preconditions;
+
+/**
+ * Remote storage exception for cases when outbound request is unsuccessful, due to remote peer hang up on us unexpectedly.
+ * Main use of this exception is to "translate" various EOF notification for various Remote Repository Storage implementations
+ * into single exception handled by Nexus Core.
+ *
+ * @author cstamas
+ * @since 2.4
+ */
+public class RemoteStorageEOFException
+    extends RemoteStorageException
+{
+    private final ProxyRepository repository;
+
+    /**
+     * Constructor.
+     *
+     * @param repository
+     * @param message
+     */
+    public RemoteStorageEOFException( final ProxyRepository repository, final String message )
+    {
+        this( repository, message, null );
+    }
+
+    /**
+     * Constructor.
+     *
+     * @param repository
+     * @param message
+     * @param cause
+     */
+    public RemoteStorageEOFException( final ProxyRepository repository, final String message, final Throwable cause )
+    {
+        super( message, cause );
+        this.repository = Preconditions.checkNotNull( repository );
+    }
+
+    /**
+     * Constructor.
+     *
+     * @param repository
+     * @param cause
+     */
+    public RemoteStorageEOFException( final ProxyRepository repository, final Throwable cause )
+    {
+        this( repository, cause.getMessage(), cause );
+    }
+
+    /**
+     * Returns the involved proxy repository. Never returns {@code null}.
+     * 
+     * @return the involved proxy repository.
+     */
+    public ProxyRepository getRepository()
+    {
+        return repository;
+    }
+}

nexus-core/src/main/java/org/sonatype/nexus/proxy/repository/AbstractProxyRepository.java

@@ -33,6 +33,7 @@
 import org.sonatype.nexus.configuration.model.CRepositoryCoreConfiguration;
 import org.sonatype.nexus.proxy.IllegalOperationException;
 import org.sonatype.nexus.proxy.ItemNotFoundException;
+import org.sonatype.nexus.proxy.LocalStorageEOFException;
 import org.sonatype.nexus.proxy.LocalStorageException;
 import org.sonatype.nexus.proxy.RemoteAccessDeniedException;
 import org.sonatype.nexus.proxy.RemoteAccessException;
@@ -1293,7 +1294,7 @@ protected StorageItem doRetrieveItem0( ResourceStoreRequest request, AbstractSto
                                 autoBlockProxying( ex );
                             }
 
-                            if ( ex instanceof RemoteStorageTransportException )
+                            if ( ex instanceof RemoteStorageTransportException || ex instanceof LocalStorageEOFException )
                             {
                                 throw ex;
                             }

nexus-core/src/main/java/org/sonatype/nexus/proxy/storage/local/fs/DefaultFSPeer.java

@@ -31,8 +31,9 @@
 import org.codehaus.plexus.util.IOUtil;
 import org.sonatype.nexus.logging.AbstractLoggingComponent;
 import org.sonatype.nexus.proxy.ItemNotFoundException;
-import org.sonatype.nexus.proxy.LocalStorageEofException;
+import org.sonatype.nexus.proxy.LocalStorageEOFException;
 import org.sonatype.nexus.proxy.LocalStorageException;
+import org.sonatype.nexus.proxy.RemoteStorageEOFException;
 import org.sonatype.nexus.proxy.ResourceStoreRequest;
 import org.sonatype.nexus.proxy.access.Action;
 import org.sonatype.nexus.proxy.item.ContentLocator;
@@ -113,17 +114,28 @@ public void storeItem( final Repository repository, final File repositoryBaseDir
 
                 os.flush();
             }
-            catch ( EOFException e )
+            catch ( EOFException e ) // NXCM-4852: Upload premature end (thrown by Jetty org.eclipse.jetty.io.EofException)
             {
                 if ( hiddenTarget != null )
                 {
                     hiddenTarget.delete();
                 }
 
-                throw new LocalStorageEofException( String.format(
+                throw new LocalStorageEOFException( String.format(
                     "EOF during storing on path \"%s\" (while writing to hiddenTarget: \"%s\")",
                     item.getRepositoryItemUid().toString(), hiddenTarget.getAbsolutePath() ), e );
             }
+            catch ( RemoteStorageEOFException e ) // NXCM-4852: Proxy remote peer response premature end (should be translated by RRS)
+            {
+                if ( hiddenTarget != null )
+                {
+                    hiddenTarget.delete();
+                }
+
+                throw new LocalStorageEOFException( String.format(
+                    "EOF during caching on path \"%s\" (while writing to hiddenTarget: \"%s\")",
+                    item.getRepositoryItemUid().toString(), hiddenTarget.getAbsolutePath() ), e );
+            }
             catch ( IOException e )
             {
                 if ( hiddenTarget != null )

nexus-core/src/main/java/org/sonatype/nexus/proxy/storage/remote/httpclient/Hc4InputStream.java

@@ -0,0 +1,81 @@
+/*
+ * Sonatype Nexus (TM) Open Source Version
+ * Copyright (c) 2007-2012 Sonatype, Inc.
+ * All rights reserved. Includes the third-party code listed at http://links.sonatype.com/products/nexus/oss/attributions.
+ *
+ * This program and the accompanying materials are made available under the terms of the Eclipse Public License Version 1.0,
+ * which accompanies this distribution and is available at http://www.eclipse.org/legal/epl-v10.html.
+ *
+ * Sonatype Nexus (TM) Professional Version is available from Sonatype, Inc. "Sonatype" and "Sonatype Nexus" are trademarks
+ * of Sonatype, Inc. Apache Maven is a trademark of the Apache Software Foundation. M2eclipse is a trademark of the
+ * Eclipse Foundation. All other trademarks are the property of their respective owners.
+ */
+package org.sonatype.nexus.proxy.storage.remote.httpclient;
+
+import java.io.IOException;
+import java.io.InputStream;
+
+import org.apache.http.ConnectionClosedException;
+import org.sonatype.nexus.proxy.RemoteStorageEOFException;
+import org.sonatype.nexus.proxy.repository.ProxyRepository;
+import org.sonatype.nexus.util.WrappingInputStream;
+
+/**
+ * Simple wrapper input stream implementation that translates some HC4 specific exceptions to Nexus Core specific
+ * exceptions, making Core able to properly respond to them.
+ * 
+ * @since 2.4
+ */
+class Hc4InputStream
+    extends WrappingInputStream
+{
+    private final ProxyRepository proxyRepository;
+
+    public Hc4InputStream( final ProxyRepository proxyRepository, final InputStream stream )
+    {
+        super( stream );
+        this.proxyRepository = proxyRepository;
+    }
+
+    @Override
+    public int read()
+        throws IOException
+    {
+        try
+        {
+            return super.read();
+        }
+        catch ( ConnectionClosedException e )
+        {
+            throw new RemoteStorageEOFException( proxyRepository, e );
+        }
+    }
+
+    @Override
+    public int read( byte[] b )
+        throws IOException
+    {
+        try
+        {
+            return super.read( b );
+        }
+        catch ( ConnectionClosedException e )
+        {
+            throw new RemoteStorageEOFException( proxyRepository, e );
+        }
+    }
+
+    @Override
+    public int read( byte b[], int off, int len )
+        throws IOException
+    {
+        try
+        {
+            return super.read( b, off, len );
+        }
+        catch ( ConnectionClosedException e )
+        {
+            throw new RemoteStorageEOFException( proxyRepository, e );
+        }
+    }
+}

nexus-core/src/main/java/org/sonatype/nexus/proxy/storage/remote/httpclient/HttpClientRemoteStorage.java

@@ -177,8 +177,8 @@ public AbstractStorageItem retrieveItem( final ProxyRepository repository, final
             InputStream is;
             try
             {
-                is = new InterruptableInputStream( method, httpResponse.getEntity().getContent() );
-
+                is = new Hc4InputStream( repository, new InterruptableInputStream( method, httpResponse.getEntity().getContent() ) );
+                 
                 String mimeType = ContentType.getOrDefault( httpResponse.getEntity() ).getMimeType();
                 if ( mimeType == null )
                 {

nexus-core/src/main/java/org/sonatype/nexus/proxy/storage/remote/httpclient/InterruptableInputStream.java

@@ -17,24 +17,22 @@
 import java.io.InterruptedIOException;
 
 import org.apache.http.client.methods.AbortableHttpRequest;
+import org.sonatype.nexus.util.WrappingInputStream;
 
 /**
  * Best-effort interruptable InputStream wrapper. The wrapper checks for Thread.isInterrupred before delegating to the
  * actual stream. If the thread is interrupted, the wrapper calls AbortableHttpRequest.abort() and throws
  * InterruptedIOException.
  */
 class InterruptableInputStream
-    extends InputStream
+    extends WrappingInputStream
 {
-
-    private final InputStream stream;
-
     private AbortableHttpRequest request;
 
     public InterruptableInputStream( final AbortableHttpRequest request, final InputStream stream )
     {
+        super(stream);
         this.request = request;
-        this.stream = stream;
     }
 
     public InterruptableInputStream( final InputStream stream )
@@ -60,39 +58,39 @@ public int read()
         throws IOException
     {
         abortIfInterrupted();
-        return stream.read();
+        return super.read();
     }
 
     @Override
     public int read( byte[] b )
         throws IOException
     {
         abortIfInterrupted();
-        return stream.read( b );
+        return super.read( b );
     }
 
     @Override
     public int read( byte b[], int off, int len )
         throws IOException
     {
         abortIfInterrupted();
-        return stream.read( b, off, len );
+        return super.read( b, off, len );
     }
 
     @Override
     public long skip( long n )
         throws IOException
     {
         abortIfInterrupted();
-        return stream.skip( n );
+        return super.skip( n );
     }
 
     @Override
     public int available()
         throws IOException
     {
         abortIfInterrupted();
-        return stream.available();
+        return super.available();
     }
 
     @Override
@@ -101,26 +99,14 @@ public void close()
     {
         // do not throw InterruptedIOException here!
         // this will not close the stream and likely mask original exception!
-        stream.close();
-    }
-
-    @Override
-    public void mark( int readlimit )
-    {
-        stream.mark( readlimit );
+        super.close();
     }
 
     @Override
-    public void reset()
+    public synchronized void reset()
         throws IOException
     {
         abortIfInterrupted();
-        stream.reset();
-    }
-
-    @Override
-    public boolean markSupported()
-    {
-        return stream.markSupported();
+        super.reset();
     }
 }