Merge "Updated fix for JRFCAF-1344."

Author: pavelbucek

core-common/pom.xml

@@ -200,6 +200,11 @@
             <artifactId>osgi-resource-locator</artifactId>
         </dependency>
 
+        <dependency>
+            <!-- Must be declared before JUnit dependency, otherwise not visible to JUnit. -->
+            <groupId>org.jmockit</groupId>
+            <artifactId>jmockit</artifactId>
+        </dependency>
         <dependency>
             <groupId>junit</groupId>
             <artifactId>junit</artifactId>

core-common/src/main/java/org/glassfish/jersey/message/internal/EntityInputStream.java

@@ -42,9 +42,6 @@
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.PushbackInputStream;
-import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.logging.Level;
-import java.util.logging.Logger;
 
 import javax.ws.rs.ProcessingException;
 
@@ -60,14 +57,13 @@
  * @author Marek Potociar (marek.potociar at oracle.com)
  */
 public class EntityInputStream extends InputStream {
-    private static final Logger LOGGER = Logger.getLogger(EntityInputStream.class.getName());
 
-    private volatile InputStream input;
-    private final AtomicBoolean closed = new AtomicBoolean(false);
+    private InputStream input;
+    private boolean closed = false;
 
     /**
      * Create an entity input stream instance wrapping the original input stream.
-     *
+     * <p/>
      * In case the original entity stream is already of type {@code EntityInputStream},
      * the stream is returned without wrapping.
      *
@@ -133,8 +129,7 @@ public boolean markSupported() {
      * a runtime {@link javax.ws.rs.ProcessingException} is thrown.
      * </p>
      *
-     * @throws javax.ws.rs.ProcessingException
-     *          in case the reset operation on the underlying entity input stream failed.
+     * @throws javax.ws.rs.ProcessingException in case the reset operation on the underlying entity input stream failed.
      */
     @Override
     public void reset() {
@@ -158,15 +153,14 @@ public void close() throws ProcessingException {
         if (in == null) {
             return;
         }
-        if (closed.compareAndSet(false, true)) {
-            // Workaround for JRFCAF-1344: Underlying stream close() may be thread-unsafe
-            // and as such the close() may result in an IOException at the socket input stream level,
-            // if the close() gets called at once from multiple threads somehow.
+        if (!closed) {
             try {
                 in.close();
             } catch (IOException ex) {
-                // This means that the underlying socket stream got closed by other thread somehow
-                LOGGER.log(Level.FINE, LocalizationMessages.MESSAGE_CONTENT_INPUT_STREAM_CLOSE_FAILED(), ex);
+                // This e.g. means that the underlying socket stream got closed by other thread somehow...
+                throw new ProcessingException(LocalizationMessages.MESSAGE_CONTENT_INPUT_STREAM_CLOSE_FAILED(), ex);
+            } finally {
+                closed = true;
             }
         }
     }
@@ -230,7 +224,7 @@ public boolean isEmpty() {
      * @throws IllegalStateException in case the entity input stream has been closed.
      */
     public void ensureNotClosed() throws IllegalStateException {
-        if (closed.get()) {
+        if (closed) {
             throw new IllegalStateException(LocalizationMessages.ERROR_ENTITY_STREAM_CLOSED());
         }
     }
@@ -241,7 +235,7 @@ public void ensureNotClosed() throws IllegalStateException {
      * @return {@code true} if the stream has been closed, {@code false} otherwise.
      */
     public boolean isClosed() {
-        return closed.get();
+        return closed;
     }
 
     /**

core-common/src/main/java/org/glassfish/jersey/message/internal/InboundMessageContext.java

@@ -85,6 +85,7 @@
  * @author Marek Potociar (marek.potociar at oracle.com)
  */
 public abstract class InboundMessageContext {
+
     private static final InputStream EMPTY = new InputStream() {
 
         @Override
@@ -121,6 +122,7 @@ public boolean markSupported() {
      * is used to control the execution of interceptors.
      */
     private static class EntityContent extends EntityInputStream {
+
         private boolean buffered;
 
         EntityContent() {
@@ -266,7 +268,7 @@ private static List<String> iterableToList(final Iterable<?> values) {
 
     /**
      * Get a message header as a single string value.
-     *
+     * <p/>
      * Each single header value is converted to String using a
      * {@link javax.ws.rs.ext.RuntimeDelegate.HeaderDelegate} if one is available
      * via {@link javax.ws.rs.ext.RuntimeDelegate#createHeaderDelegate(java.lang.Class)}
@@ -275,10 +277,10 @@ private static List<String> iterableToList(final Iterable<?> values) {
      *
      * @param name the message header.
      * @return the message header value. If the message header is not present then
-     *         {@code null} is returned. If the message header is present but has no
-     *         value then the empty string is returned. If the message header is present
-     *         more than once then the values of joined together and separated by a ','
-     *         character.
+     * {@code null} is returned. If the message header is present but has no
+     * value then the empty string is returned. If the message header is present
+     * more than once then the values of joined together and separated by a ','
+     * character.
      */
     public String getHeaderString(String name) {
         List<String> values = this.headers.get(name);
@@ -400,7 +402,7 @@ public Set<MatchingEntityTag> getIfNoneMatch() {
     /**
      * Get the language of the entity.
      *
-     * @return the language of the entity or {@code null} if not specified
+     * @return the language of the entity or {@code null} if not specified.
      */
     public Locale getLanguage() {
         return singleHeader(HttpHeaders.CONTENT_LANGUAGE, new Function<String, Locale>() {
@@ -418,8 +420,7 @@ public Locale apply(String input) {
     /**
      * Get Content-Length value.
      *
-     * @return Content-Length as integer if present and valid number. In other
-     *         cases returns -1.
+     * @return Content-Length as integer if present and valid number. In other cases returns -1.
      */
     public int getLength() {
         return singleHeader(HttpHeaders.CONTENT_LENGTH, new Function<String, Integer>() {
@@ -438,7 +439,7 @@ public Integer apply(String input) {
      * Get the media type of the entity.
      *
      * @return the media type or {@code null} if not specified (e.g. there's no
-     *         message entity).
+     * message entity).
      */
     public MediaType getMediaType() {
         return singleHeader(HttpHeaders.CONTENT_TYPE, new Function<String, MediaType>() {
@@ -457,7 +458,7 @@ public MediaType apply(String input) {
      * Get a list of media types that are acceptable for a request.
      *
      * @return a read-only list of requested response media types sorted according
-     *         to their q-value, with highest preference first.
+     * to their q-value, with highest preference first.
      */
     public List<AcceptableMediaType> getQualifiedAcceptableMediaTypes() {
         final String value = getHeaderString(HttpHeaders.ACCEPT);
@@ -477,7 +478,7 @@ public List<AcceptableMediaType> getQualifiedAcceptableMediaTypes() {
      * Get a list of languages that are acceptable for the message.
      *
      * @return a read-only list of acceptable languages sorted according
-     *         to their q-value, with highest preference first.
+     * to their q-value, with highest preference first.
      */
     public List<AcceptableLanguageTag> getQualifiedAcceptableLanguages() {
         final String value = getHeaderString(HttpHeaders.ACCEPT_LANGUAGE);
@@ -497,7 +498,7 @@ public List<AcceptableLanguageTag> getQualifiedAcceptableLanguages() {
      * Get the list of language tag from the "Accept-Charset" of an HTTP request.
      *
      * @return The list of AcceptableToken. This list
-     *         is ordered with the highest quality acceptable charset occurring first.
+     * is ordered with the highest quality acceptable charset occurring first.
      */
     public List<AcceptableToken> getQualifiedAcceptCharset() {
         final String acceptCharset = getHeaderString(HttpHeaders.ACCEPT_CHARSET);
@@ -515,7 +516,7 @@ public List<AcceptableToken> getQualifiedAcceptCharset() {
      * Get the list of language tag from the "Accept-Charset" of an HTTP request.
      *
      * @return The list of AcceptableToken. This list
-     *         is ordered with the highest quality acceptable charset occurring first.
+     * is ordered with the highest quality acceptable charset occurring first.
      */
     public List<AcceptableToken> getQualifiedAcceptEncoding() {
         final String acceptEncoding = getHeaderString(HttpHeaders.ACCEPT_ENCODING);
@@ -553,7 +554,7 @@ public Map<String, Cookie> getRequestCookies() {
      * Get the allowed HTTP methods from the Allow HTTP header.
      *
      * @return the allowed HTTP methods, all methods will returned as upper case
-     *         strings.
+     * strings.
      */
     public Set<String> getAllowedMethods() {
         final String allowed = getHeaderString(HttpHeaders.ALLOW);
@@ -642,7 +643,7 @@ public URI apply(String value) {
      * Get the links attached to the message as header.
      *
      * @return links, may return empty {@link java.util.Set} if no links are present. Never
-     *         returns {@code null}.
+     * returns {@code null}.
      */
     public Set<Link> getLinks() {
         List<String> links = this.headers.get(HttpHeaders.LINK);
@@ -666,7 +667,7 @@ public Set<Link> getLinks() {
      *
      * @param relation link relation.
      * @return {@code true} if the for the relation link exists, {@code false}
-     *         otherwise.
+     * otherwise.
      */
     public boolean hasLink(String relation) {
         for (Link link : getLinks()) {
@@ -701,7 +702,7 @@ public Link getLink(String relation) {
      *
      * @param relation link relation.
      * @return the link builder for the relation, otherwise {@code null} if not
-     *         present.
+     * present.
      */
     public Link.Builder getLinkBuilder(String relation) {
         Link link = getLink(relation);
@@ -735,12 +736,12 @@ public void setWorkers(MessageBodyWorkers workers) {
     /**
      * Check if there is a non-empty entity input stream is available in the
      * message.
-     *
+     * <p/>
      * The method returns {@code true} if the entity is present, returns
      * {@code false} otherwise.
      *
      * @return {@code true} if there is an entity present in the message,
-     *         {@code false} otherwise.
+     * {@code false} otherwise.
      */
     public boolean hasEntity() {
         entityContent.ensureNotClosed();
@@ -785,7 +786,6 @@ public <T> T readEntity(Class<T> rawType, PropertiesDelegate propertiesDelegate)
         return readEntity(rawType, rawType, EMPTY_ANNOTATIONS, propertiesDelegate);
     }
 
-
     /**
      * Read entity from a context entity input stream.
      *
@@ -831,14 +831,14 @@ public <T> T readEntity(Class<T> rawType, Type type, Annotation[] annotations, P
 
         entityContent.ensureNotClosed();
 
-// TODO: revise if we need to re-introduce the check for performance reasons or once non-blocking I/O is supported.
-// The code has been commended out because in case of streaming input (e.g. SSE) the call might block until a first
-// byte is available, which would make e.g. the SSE EventSource construction or EventSource.open() method to block
-// until a first event is received, which is undesirable.
-//
-//        if (entityContent.isEmpty()) {
-//            return null;
-//        }
+        // TODO: revise if we need to re-introduce the check for performance reasons or once non-blocking I/O is supported.
+        // The code has been commended out because in case of streaming input (e.g. SSE) the call might block until a first
+        // byte is available, which would make e.g. the SSE EventSource construction or EventSource.open() method to block
+        // until a first event is received, which is undesirable.
+        //
+        //        if (entityContent.isEmpty()) {
+        //            return null;
+        //        }
 
         if (workers == null) {
             return null;
@@ -847,7 +847,6 @@ public <T> T readEntity(Class<T> rawType, Type type, Annotation[] annotations, P
         MediaType mediaType = getMediaType();
         mediaType = mediaType == null ? MediaType.APPLICATION_OCTET_STREAM_TYPE : mediaType;
 
-
         boolean shouldClose = !buffered;
         try {
             T t = (T) workers.readFrom(
@@ -868,7 +867,11 @@ public <T> T readEntity(Class<T> rawType, Type type, Annotation[] annotations, P
             throw new ProcessingException(LocalizationMessages.ERROR_READING_ENTITY_FROM_INPUT_STREAM(), ex);
         } finally {
             if (shouldClose) {
-                entityContent.close();
+                // Workaround for JRFCAF-1344: the underlying stream close() implementation may be thread-unsafe
+                // and as such the close() may result in an IOException at the socket input stream level,
+                // if the close() gets called at once from multiple threads somehow.
+                // We want to ignore these exceptions in the readEntity/bufferEntity operations though.
+                ReaderWriter.safelyClose(entityContent);
             }
         }
     }
@@ -877,8 +880,7 @@ public <T> T readEntity(Class<T> rawType, Type type, Annotation[] annotations, P
      * Buffer the entity stream (if not empty).
      *
      * @return {@code true} if the entity input stream was successfully buffered.
-     * @throws javax.ws.rs.ProcessingException
-     *          in case of an IO error.
+     * @throws javax.ws.rs.ProcessingException in case of an IO error.
      */
     public boolean bufferEntity() throws ProcessingException {
         entityContent.ensureNotClosed();
@@ -893,7 +895,11 @@ public boolean bufferEntity() throws ProcessingException {
             try {
                 ReaderWriter.writeTo(entityStream, baos);
             } finally {
-                entityStream.close();
+                // Workaround for JRFCAF-1344: the underlying stream close() implementation may be thread-unsafe
+                // and as such the close() may result in an IOException at the socket input stream level,
+                // if the close() gets called at once from multiple threads somehow.
+                // We want to ignore these exceptions in the readEntity/bufferEntity operations though.
+                ReaderWriter.safelyClose(entityStream);
             }
 
             entityContent.setContent(new ByteArrayInputStream(baos.toByteArray()), true);

core-common/src/main/java/org/glassfish/jersey/message/internal/MessageBodyFactory.java

@@ -1085,8 +1085,10 @@ public Object readFrom(final Class<?> rawType,
             final Object instance = executor.proceed();
             if (!(instance instanceof Closeable) && !(instance instanceof Source)) {
                 final InputStream stream = executor.getInputStream();
-                if (stream != null) {
-                    stream.close();
+                if (stream != entityStream && stream != null) {
+                    // We only close stream if it differs from the received entity stream,
+                    // otherwise we let the caller close the stream.
+                    ReaderWriter.safelyClose(stream);
                 }
             }
 

core-common/src/main/java/org/glassfish/jersey/message/internal/ReaderInterceptorExecutor.java

@@ -37,7 +37,6 @@
  * only if the new code is made subject to such option by the copyright
  * holder.
  */
-
 package org.glassfish.jersey.message.internal;
 
 import java.io.IOException;
@@ -83,16 +82,16 @@ public final class ReaderInterceptorExecutor extends InterceptorExecutor<ReaderI
 
     private static final Logger LOGGER = Logger.getLogger(ReaderInterceptorExecutor.class.getName());
 
-    private InputStream inputStream;
     private final MultivaluedMap<String, String> headers;
-
     private final Iterator<ReaderInterceptor> interceptors;
-    private int processedCount;
     private final MessageBodyWorkers workers;
     private final boolean translateNce;
 
     private final ServiceLocator serviceLocator;
 
+    private InputStream inputStream;
+    private int processedCount;
+
     /**
      * Constructs a new executor to read given type from provided {@link InputStream entityStream}.
      *