Merge "Making SseEventSink#send asynchronous."

Author: stepanv

core-server/src/main/java/org/glassfish/jersey/server/internal/process/AsyncContext.java renamed to core-server/src/main/java/org/glassfish/jersey/server/AsyncContext.java

@@ -37,7 +37,7 @@
  * only if the new code is made subject to such option by the copyright
  * holder.
  */
-package org.glassfish.jersey.server.internal.process;
+package org.glassfish.jersey.server;
 
 import javax.ws.rs.container.AsyncResponse;
 import javax.ws.rs.core.Response;

core-server/src/main/java/org/glassfish/jersey/server/ChunkedOutput.java

@@ -52,11 +52,11 @@
 import javax.ws.rs.core.GenericType;
 import javax.ws.rs.ext.WriterInterceptor;
 
-import org.glassfish.jersey.internal.util.collection.Value;
+import javax.inject.Provider;
+
 import org.glassfish.jersey.process.internal.RequestContext;
 import org.glassfish.jersey.process.internal.RequestScope;
 import org.glassfish.jersey.server.internal.LocalizationMessages;
-import org.glassfish.jersey.server.internal.process.AsyncContext;
 import org.glassfish.jersey.server.internal.process.MappableException;
 
 /**
@@ -75,14 +75,19 @@ public class ChunkedOutput<T> extends GenericType<T> implements Closeable {
     private final BlockingDeque<T> queue = new LinkedBlockingDeque<>();
     private final byte[] chunkDelimiter;
 
-    private volatile boolean closed = false;
     private boolean flushing = false;
+
+    private volatile boolean closed = false;
+    private volatile boolean resumed = false;
+
+    private volatile AsyncContext asyncContext;
+
     private volatile RequestScope requestScope;
     private volatile RequestContext requestScopeContext;
     private volatile ContainerRequest requestContext;
     private volatile ContainerResponse responseContext;
     private volatile ConnectionCallback connectionCallback;
-    private volatile Value<AsyncContext> asyncContext;
+
 
     /**
      * Create new {@code ChunkedOutput}.
@@ -116,6 +121,23 @@ protected ChunkedOutput(final byte[] chunkDelimiter) {
         }
     }
 
+    /**
+     * Create new {@code ChunkedOutput} with a custom chunk delimiter.
+     *
+     * @param chunkDelimiter custom chunk delimiter bytes. Must not be {code null}.
+     * @since 2.4.1
+     */
+    protected ChunkedOutput(final byte[] chunkDelimiter, Provider<AsyncContext> asyncContextProvider) {
+        if (chunkDelimiter.length > 0) {
+            this.chunkDelimiter = new byte[chunkDelimiter.length];
+            System.arraycopy(chunkDelimiter, 0, this.chunkDelimiter, 0, chunkDelimiter.length);
+        } else {
+            this.chunkDelimiter = ZERO_LENGTH_DELIMITER;
+        }
+
+        this.asyncContext = asyncContextProvider == null ? null : asyncContextProvider.get();
+    }
+
     /**
      * Create new {@code ChunkedOutput} with a custom chunk delimiter.
      *
@@ -181,7 +203,12 @@ public void write(final T chunk) throws IOException {
         flushQueue();
     }
 
-    private void flushQueue() throws IOException {
+    protected void flushQueue() throws IOException {
+        if (!resumed && asyncContext != null) {
+            resumed = true;
+            asyncContext.resume(this);
+        }
+
         if (requestScopeContext == null || requestContext == null || responseContext == null) {
             return;
         }
@@ -245,11 +272,11 @@ public Void call() throws IOException {
                                 responseContext.setEntityStream(writtenStream);
                             }
                         } catch (final IOException ioe) {
-                            connectionCallback.onDisconnect(asyncContext.get());
+                            connectionCallback.onDisconnect(asyncContext);
                             throw ioe;
                         } catch (final MappableException mpe) {
                             if (mpe.getCause() instanceof IOException) {
-                                connectionCallback.onDisconnect(asyncContext.get());
+                                connectionCallback.onDisconnect(asyncContext);
                             }
                             throw mpe;
                         }
@@ -355,21 +382,18 @@ public String toString() {
      * @param requestContext           request context.
      * @param responseContext          response context.
      * @param connectionCallbackRunner connection callback.
-     * @param asyncContext             async context value.
      * @throws IOException when encountered any problem during serializing or writing a chunk.
      */
     void setContext(final RequestScope requestScope,
                     final RequestContext requestScopeContext,
                     final ContainerRequest requestContext,
                     final ContainerResponse responseContext,
-                    final ConnectionCallback connectionCallbackRunner,
-                    final Value<AsyncContext> asyncContext) throws IOException {
+                    final ConnectionCallback connectionCallbackRunner) throws IOException {
         this.requestScope = requestScope;
         this.requestScopeContext = requestScopeContext;
         this.requestContext = requestContext;
         this.responseContext = responseContext;
         this.connectionCallback = connectionCallbackRunner;
-        this.asyncContext = asyncContext;
         flushQueue();
     }
 }

core-server/src/main/java/org/glassfish/jersey/server/ServerRuntime.java

@@ -101,7 +101,6 @@
 import org.glassfish.jersey.server.internal.monitoring.EmptyRequestEventBuilder;
 import org.glassfish.jersey.server.internal.monitoring.RequestEventBuilder;
 import org.glassfish.jersey.server.internal.monitoring.RequestEventImpl;
-import org.glassfish.jersey.server.internal.process.AsyncContext;
 import org.glassfish.jersey.server.internal.process.Endpoint;
 import org.glassfish.jersey.server.internal.process.MappableException;
 import org.glassfish.jersey.server.internal.process.RequestProcessingContext;
@@ -115,10 +114,10 @@
 import org.glassfish.jersey.server.spi.ResponseErrorMapper;
 import org.glassfish.jersey.spi.ExceptionMappers;
 
-import static org.glassfish.jersey.server.internal.process.AsyncContext.State.COMPLETED;
-import static org.glassfish.jersey.server.internal.process.AsyncContext.State.RESUMED;
-import static org.glassfish.jersey.server.internal.process.AsyncContext.State.RUNNING;
-import static org.glassfish.jersey.server.internal.process.AsyncContext.State.SUSPENDED;
+import static org.glassfish.jersey.server.AsyncContext.State.COMPLETED;
+import static org.glassfish.jersey.server.AsyncContext.State.RESUMED;
+import static org.glassfish.jersey.server.AsyncContext.State.RUNNING;
+import static org.glassfish.jersey.server.AsyncContext.State.SUSPENDED;
 
 /**
  * Server-side request processing runtime.
@@ -714,8 +713,7 @@ public OutputStream getOutputStream(final int contentLength) throws IOException
                                     runtime.requestScope.referenceCurrent(),
                                     request,
                                     response,
-                                    connectionCallbackRunner,
-                                    processingContext.asyncContextValue());
+                                    connectionCallbackRunner);
                         } catch (final IOException ex) {
                             LOGGER.log(Level.SEVERE, LocalizationMessages.ERROR_WRITING_RESPONSE_ENTITY_CHUNK(), ex);
                             close = true;

core-server/src/main/java/org/glassfish/jersey/server/internal/inject/AsyncResponseValueParamProvider.java

@@ -47,7 +47,7 @@
 import javax.inject.Provider;
 
 import org.glassfish.jersey.server.ContainerRequest;
-import org.glassfish.jersey.server.internal.process.AsyncContext;
+import org.glassfish.jersey.server.AsyncContext;
 import org.glassfish.jersey.server.model.Parameter;
 import org.glassfish.jersey.server.spi.internal.ValueParamProvider;
 

core-server/src/main/java/org/glassfish/jersey/server/internal/inject/ValueParamProviderConfigurator.java

@@ -71,7 +71,7 @@
 import org.glassfish.jersey.server.ContainerRequest;
 import org.glassfish.jersey.server.ServerBootstrapBag;
 import org.glassfish.jersey.server.Uri;
-import org.glassfish.jersey.server.internal.process.AsyncContext;
+import org.glassfish.jersey.server.AsyncContext;
 import org.glassfish.jersey.server.internal.process.RequestProcessingContextReference;
 import org.glassfish.jersey.server.spi.internal.ValueParamProvider;
 

core-server/src/main/java/org/glassfish/jersey/server/internal/process/RequestProcessingConfigurator.java

@@ -57,6 +57,7 @@
 import org.glassfish.jersey.internal.inject.AbstractBinder;
 import org.glassfish.jersey.internal.inject.InjectionManager;
 import org.glassfish.jersey.process.internal.RequestScoped;
+import org.glassfish.jersey.server.AsyncContext;
 import org.glassfish.jersey.server.CloseableService;
 import org.glassfish.jersey.server.ContainerRequest;
 import org.glassfish.jersey.server.ExtendedUriInfo;

core-server/src/main/java/org/glassfish/jersey/server/internal/process/RequestProcessingContext.java

@@ -49,6 +49,7 @@
 import org.glassfish.jersey.internal.util.collection.Values;
 import org.glassfish.jersey.process.internal.ChainableStage;
 import org.glassfish.jersey.process.internal.Stage;
+import org.glassfish.jersey.server.AsyncContext;
 import org.glassfish.jersey.server.CloseableService;
 import org.glassfish.jersey.server.ContainerRequest;
 import org.glassfish.jersey.server.ContainerResponse;
@@ -155,7 +156,7 @@ public CloseableService closeableService() {
 
 
     /**
-     * Lazily initialize {@link org.glassfish.jersey.server.internal.process.AsyncContext} for this
+     * Lazily initialize {@link AsyncContext} for this
      * request processing context.
      * <p>
      * The {@code lazyContextValue} will be only invoked once during the first call to {@link #asyncContext()}.

core-server/src/main/java/org/glassfish/jersey/server/model/IntrospectionModeller.java

@@ -1,7 +1,7 @@
 /*
  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS HEADER.
  *
- * Copyright (c) 2010-2015 Oracle and/or its affiliates. All rights reserved.
+ * Copyright (c) 2010-2017 Oracle and/or its affiliates. All rights reserved.
  *
  * The contents of this file are subject to the terms of either the GNU
  * General Public License Version 2 only ("GPL") or the Common Development
@@ -63,6 +63,7 @@
 import javax.ws.rs.container.AsyncResponse;
 import javax.ws.rs.container.Suspended;
 import javax.ws.rs.core.MediaType;
+import javax.ws.rs.sse.SseEventSink;
 
 import org.glassfish.jersey.internal.Errors;
 import org.glassfish.jersey.internal.util.Producer;
@@ -310,10 +311,15 @@ private static void introspectAsyncFeatures(AnnotatedMethod am, ResourceMethod.B
             for (Annotation annotation : annotations) {
                 if (annotation.annotationType() == Suspended.class) {
                     resourceMethodBuilder.suspended(AsyncResponse.NO_TIMEOUT, TimeUnit.MILLISECONDS);
-                    return;
                 }
             }
         }
+
+        for (Class<?> paramType : am.getParameterTypes()) {
+            if (SseEventSink.class.equals(paramType)) {
+                resourceMethodBuilder.sse();
+            }
+        }
     }
 
     private void addResourceMethods(

core-server/src/main/java/org/glassfish/jersey/server/model/ResourceMethod.java

@@ -148,6 +148,7 @@ public static final class Builder {
         private final Set<MediaType> producedTypes;
         // Suspendable
         private boolean managedAsync;
+        private boolean sse;
         private boolean suspended;
         private long suspendTimeout;
         private TimeUnit suspendTimeoutUnit;
@@ -367,6 +368,17 @@ public Builder suspended(long timeout, TimeUnit unit) {
             return this;
         }
 
+        /**
+         * Set the SSE flag on the method model to {@code true}.
+         *
+         * @return updated builder object.
+         */
+        public Builder sse() {
+            sse = true;
+
+            return this;
+        }
+
         /**
          * Set the managed async required flag on the method model to {@code true}.
          *
@@ -527,6 +539,7 @@ public ResourceMethod build() {
                     producedTypes,
                     managedAsync,
                     suspended,
+                    sse,
                     suspendTimeout,
                     suspendTimeoutUnit,
                     createInvocable(),
@@ -567,6 +580,7 @@ private Invocable createInvocable() {
         // SuspendableComponent
         private final boolean managedAsync;
         private final boolean suspended;
+        private final boolean sse;
         private final long suspendTimeout;
         private final TimeUnit suspendTimeoutUnit;
         // Invocable
@@ -579,7 +593,7 @@ private Invocable createInvocable() {
         private Data(final String httpMethod,
                      final Collection<MediaType> consumedTypes,
                      final Collection<MediaType> producedTypes,
-                     boolean managedAsync, final boolean suspended,
+                     boolean managedAsync, final boolean suspended, boolean sse,
                      final long suspendTimeout,
                      final TimeUnit suspendTimeoutUnit,
                      final Invocable invocable,
@@ -594,6 +608,7 @@ private Data(final String httpMethod,
             this.producedTypes = Collections.unmodifiableList(new ArrayList<>(producedTypes));
             this.invocable = invocable;
             this.suspended = suspended;
+            this.sse = sse;
             this.suspendTimeout = suspendTimeout;
             this.suspendTimeoutUnit = suspendTimeoutUnit;
 
@@ -660,6 +675,15 @@ private Data(final String httpMethod,
             return suspended;
         }
 
+        /**
+         * Flag indicating whether the method requires injection of Sse Event Sink.
+         *
+         * @return {@code true} if the method requires injection of Sse Event Sink, {@code false} otherwise.
+         */
+        /* package */ boolean isSse() {
+            return sse;
+        }
+
         /**
          * Get the suspended timeout value for the method.
          *
@@ -850,6 +874,15 @@ public boolean isSuspendDeclared() {
         return data.isSuspended();
     }
 
+    /**
+     * Check whether the resource method will be producing Server-sent event stream.
+     *
+     * @return {@code true} if the resource method produces Server-sent event stream, {@code false} otherwise.
+     */
+    public boolean isSse() {
+        return data.isSse();
+    }
+
     @Override
     public boolean isManagedAsyncDeclared() {
         return data.isManagedAsync();

core-server/src/main/java/org/glassfish/jersey/server/model/ResourceMethodInvoker.java

@@ -394,7 +394,7 @@ public ContainerResponse apply(final RequestProcessingContext processingContext)
         final ContainerRequest request = processingContext.request();
         final Object resource = processingContext.routingContext().peekMatchedResource();
 
-        if (method.isSuspendDeclared() || method.isManagedAsyncDeclared()) {
+        if (method.isSuspendDeclared() || method.isManagedAsyncDeclared() || method.isSse()) {
             if (!processingContext.asyncContext().suspend()) {
                 throw new ProcessingException(LocalizationMessages.ERROR_SUSPENDING_ASYNC_REQUEST());
             }
@@ -414,6 +414,11 @@ public ContainerResponse apply(final RequestProcessingContext processingContext)
             // TODO replace with processing context factory method.
             Response response = invoke(processingContext, resource);
 
+            // we don't care about the response when SseEventSink is injected - it will be sent asynchronously.
+            if (method.isSse()) {
+                return null;
+            }
+
             if (response.hasEntity()) {
                 Object entityFuture = response.getEntity();
                 if (entityFuture instanceof CompletionStage) {