SSE Flow.Subscriber injectable as event sink

* Java Flow api used only by JerseyEventSink in java 11+ * Reactive streams tck tests for JerseyEventSink * Cancel on close propagation * Remove tests contradicting reactive streams spec https://github.com/reactive-streams/reactive-streams-jvm#2.13 Signed-off-by: Daniel Kec <[email protected]>
eclipse-ee4j · Sep 23, 2020 · 0f65e5d · 0f65e5d
1 parent 56a5634
commit 0f65e5d
Show file tree

Hide file tree

Showing 20 changed files with 1,017 additions and 77 deletions.
diff --git a/...s/helidon-connector/src/test/java/org/glassfish/jersey/helidon/connector/sse/SseTest.java b/...s/helidon-connector/src/test/java/org/glassfish/jersey/helidon/connector/sse/SseTest.java
@@ -110,8 +110,10 @@ public void testSend() throws InterruptedException {
         final StringBuilder sb = new StringBuilder();
         final CountDownLatch latch = new CountDownLatch(10);
         try (SseEventSource source = SseEventSource.target(target().path("simple")).build()) {
-            source.register((event) -> sb.append(event.readData()));
-            source.register((event) -> latch.countDown());
+            source.register((event) -> {
+                sb.append(event.readData());
+                latch.countDown();
+            });
             source.open();
 
             latch.await(WAIT_TIME, TimeUnit.MILLISECONDS);
@@ -158,9 +160,11 @@ private BroadcasterClient(WebTarget target) {
 
         private void register() throws InterruptedException {
             final CountDownLatch latch = new CountDownLatch(1);
-            source.register((event) -> message.append(event.readData()));
-            source.register((event) -> latch.countDown());
-            source.register((event) -> messageLatch.countDown());
+            source.register((event) -> {
+                message.append(event.readData());
+                latch.countDown();
+                messageLatch.countDown();
+            });
             source.open();
 
             latch.await(WAIT_TIME, TimeUnit.MILLISECONDS);

diff --git a/core-common/src/main/java11/org/glassfish/jersey/internal/jsr166/JerseyFlowSubscriber.java b/core-common/src/main/java11/org/glassfish/jersey/internal/jsr166/JerseyFlowSubscriber.java
@@ -0,0 +1,34 @@
+/*
+ * Copyright (c) 2020 Oracle and/or its affiliates. All rights reserved.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License v. 2.0, which is available at
+ * http://www.eclipse.org/legal/epl-2.0.
+ *
+ * This Source Code may also be made available under the following Secondary
+ * Licenses when the conditions for such availability set forth in the
+ * Eclipse Public License v. 2.0 are satisfied: GNU General Public License,
+ * version 2 with the GNU Classpath Exception, which is available at
+ * https://www.gnu.org/software/classpath/license.html.
+ *
+ * SPDX-License-Identifier: EPL-2.0 OR GPL-2.0 WITH Classpath-exception-2.0
+ */
+
+package org.glassfish.jersey.internal.jsr166;
+
+public interface JerseyFlowSubscriber<T> extends Flow.Subscriber<T>, java.util.concurrent.Flow.Subscriber<T> {
+    @Override
+    default void onSubscribe(java.util.concurrent.Flow.Subscription subscription) {
+        this.onSubscribe(new Flow.Subscription() {
+            @Override
+            public void request(final long n) {
+                subscription.request(n);
+            }
+
+            @Override
+            public void cancel() {
+                subscription.cancel();
+            }
+        });
+    }
+}
diff --git a/core-common/src/main/java8/org/glassfish/jersey/internal/jsr166/JerseyFlowSubscriber.java b/core-common/src/main/java8/org/glassfish/jersey/internal/jsr166/JerseyFlowSubscriber.java
@@ -0,0 +1,20 @@
+/*
+ * Copyright (c) 2020 Oracle and/or its affiliates. All rights reserved.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License v. 2.0, which is available at
+ * http://www.eclipse.org/legal/epl-2.0.
+ *
+ * This Source Code may also be made available under the following Secondary
+ * Licenses when the conditions for such availability set forth in the
+ * Eclipse Public License v. 2.0 are satisfied: GNU General Public License,
+ * version 2 with the GNU Classpath Exception, which is available at
+ * https://www.gnu.org/software/classpath/license.html.
+ *
+ * SPDX-License-Identifier: EPL-2.0 OR GPL-2.0 WITH Classpath-exception-2.0
+ */
+
+package org.glassfish.jersey.internal.jsr166;
+
+public interface JerseyFlowSubscriber<T> extends Flow.Subscriber<T> {
+}
diff --git a/core-server/src/main/java/org/glassfish/jersey/server/ChunkedOutput.java b/core-server/src/main/java/org/glassfish/jersey/server/ChunkedOutput.java
@@ -299,6 +299,7 @@ public Void call() throws IOException {
             closed = true;
             // remember the exception (it will get rethrown from finally clause, once it does it's work)
             ex = e;
+            onClose(e);
         } finally {
             if (closed) {
                 try {
@@ -349,6 +350,14 @@ public boolean isClosed() {
         return closed;
     }
 
+    /**
+     * Executed only in case of close being triggered by client.
+     * @param e Exception causing the close
+     */
+    protected void onClose(Exception e){
+
+    }
+
     @SuppressWarnings("EqualsWhichDoesntCheckParameterClass")
     @Override
     public boolean equals(final Object obj) {

diff --git a/core-server/src/main/java/org/glassfish/jersey/server/model/IntrospectionModeller.java b/core-server/src/main/java/org/glassfish/jersey/server/model/IntrospectionModeller.java
@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2010, 2019 Oracle and/or its affiliates. All rights reserved.
+ * Copyright (c) 2010, 2020 Oracle and/or its affiliates. All rights reserved.
  *
  * This program and the accompanying materials are made available under the
  * terms of the Eclipse Public License v. 2.0, which is available at
@@ -40,7 +40,6 @@
 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;
@@ -49,6 +48,7 @@
 import org.glassfish.jersey.server.ManagedAsync;
 import org.glassfish.jersey.server.internal.LocalizationMessages;
 import org.glassfish.jersey.server.model.internal.ModelHelper;
+import org.glassfish.jersey.server.model.internal.SseTypeResolver;
 
 /**
  * Utility class for constructing resource model from JAX-RS annotated POJO.
@@ -298,7 +298,7 @@ private static void introspectAsyncFeatures(AnnotatedMethod am, ResourceMethod.B
         }
 
         for (Class<?> paramType : am.getParameterTypes()) {
-            if (SseEventSink.class.equals(paramType)) {
+            if (SseTypeResolver.isSseSinkParam(paramType)) {
                 resourceMethodBuilder.sse();
             }
         }

diff --git a/core-server/src/main/java/org/glassfish/jersey/server/model/ResourceMethodValidator.java b/core-server/src/main/java/org/glassfish/jersey/server/model/ResourceMethodValidator.java
@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2012, 2018 Oracle and/or its affiliates. All rights reserved.
+ * Copyright (c) 2012, 2020 Oracle and/or its affiliates. All rights reserved.
  *
  * This program and the accompanying materials are made available under the
  * terms of the Eclipse Public License v. 2.0, which is available at
@@ -37,11 +37,11 @@
 import javax.ws.rs.Path;
 import javax.ws.rs.PathParam;
 import javax.ws.rs.QueryParam;
-import javax.ws.rs.sse.SseEventSink;
 
 import org.glassfish.jersey.internal.Errors;
 import org.glassfish.jersey.server.ContainerRequest;
 import org.glassfish.jersey.server.internal.LocalizationMessages;
+import org.glassfish.jersey.server.model.internal.SseTypeResolver;
 import org.glassfish.jersey.server.spi.internal.ParameterValueHelper;
 import org.glassfish.jersey.server.spi.internal.ValueParamProvider;
 
@@ -85,7 +85,7 @@ private void checkMethod(ResourceMethod method) {
         if ("GET".equals(method.getHttpMethod())) {
             final long eventSinkCount = invocable.getParameters()
                     .stream()
-                    .filter(parameter -> SseEventSink.class.equals(parameter.getRawType()))
+                    .filter(parameter -> SseTypeResolver.isSseSinkParam(parameter.getRawType()))
                     .count();
 
             final boolean isSse = eventSinkCount > 0;
@@ -213,7 +213,8 @@ private void checkParameters(ResourceMethod method) {
     }
 
     private boolean isSseInjected(final Invocable invocable) {
-        return invocable.getParameters().stream().anyMatch(parameter -> SseEventSink.class.equals(parameter.getRawType()));
+        return invocable.getParameters().stream()
+                .anyMatch(parameter -> SseTypeResolver.isSseSinkParam(parameter.getRawType()));
     }
 
     private static final Set<Class> PARAM_ANNOTATION_SET = createParamAnnotationSet();

diff --git a/...java/org/glassfish/jersey/server/model/internal/JavaResourceMethodDispatcherProvider.java b/...java/org/glassfish/jersey/server/model/internal/JavaResourceMethodDispatcherProvider.java
@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2011, 2019 Oracle and/or its affiliates. All rights reserved.
+ * Copyright (c) 2011, 2020 Oracle and/or its affiliates. All rights reserved.
  *
  * This program and the accompanying materials are made available under the
  * terms of the Eclipse Public License v. 2.0, which is available at
@@ -76,7 +76,7 @@ public ResourceMethodDispatcher create(final Invocable resourceMethod,
             // return type is void
             int i = 0;
             for (final Parameter parameter : resourceMethod.getParameters()) {
-                if (SseEventSink.class.equals(parameter.getRawType())) {
+                if (SseTypeResolver.isSseSinkParam(parameter.getRawType())) {
                     resourceMethodDispatcher =
                             new SseEventSinkInvoker(resourceMethod, invocationHandler, valueProviders, validator, i);
                     break;

diff --git a/core-server/src/main/java/org/glassfish/jersey/server/model/internal/SseTypeResolver.java b/core-server/src/main/java/org/glassfish/jersey/server/model/internal/SseTypeResolver.java
@@ -0,0 +1,49 @@
+/*
+ * Copyright (c) 2020 Oracle and/or its affiliates. All rights reserved.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License v. 2.0, which is available at
+ * http://www.eclipse.org/legal/epl-2.0.
+ *
+ * This Source Code may also be made available under the following Secondary
+ * Licenses when the conditions for such availability set forth in the
+ * Eclipse Public License v. 2.0 are satisfied: GNU General Public License,
+ * version 2 with the GNU Classpath Exception, which is available at
+ * https://www.gnu.org/software/classpath/license.html.
+ *
+ * SPDX-License-Identifier: EPL-2.0 OR GPL-2.0 WITH Classpath-exception-2.0
+ */
+package org.glassfish.jersey.server.model.internal;
+
+import java.security.AccessController;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Set;
+
+import org.glassfish.jersey.internal.util.ReflectionHelper;
+
+public final class SseTypeResolver {
+
+    private static final Set<Class<?>> SUPPORTED_SSE_SINK_TYPES;
+
+    private SseTypeResolver() {
+    }
+
+    static {
+        Set<Class<?>> set = new HashSet<>(8);
+
+        set.add(org.glassfish.jersey.internal.jsr166.Flow.Subscriber.class);
+        set.add(javax.ws.rs.sse.SseEventSink.class);
+        Class<?> clazz = AccessController
+                .doPrivileged(ReflectionHelper.classForNamePA("java.util.concurrent.Flow$Subscriber", null));
+
+        if (clazz != null) {
+            set.add(clazz);
+        }
+        SUPPORTED_SSE_SINK_TYPES = Collections.unmodifiableSet(set);
+    }
+
+    public static boolean isSseSinkParam(Class<?> type) {
+        return SUPPORTED_SSE_SINK_TYPES.contains(type);
+    }
+}