Build core-common on JDK 11

core-common/src/main/java/org/glassfish/jersey/internal/util/JerseyPublisher.java

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2017, 2018 Oracle and/or its affiliates. All rights reserved.
+ * Copyright (c) 2017, 2019 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
@@ -27,19 +27,21 @@
 import org.glassfish.jersey.internal.LocalizationMessages;
 import org.glassfish.jersey.internal.jsr166.Flow;
 import org.glassfish.jersey.internal.jsr166.SubmissionPublisher;
+import org.glassfish.jersey.internal.jsr166.SubmissionPublisherFactory;
+import org.glassfish.jersey.internal.jsr166.SubmittableFlowPublisher;
 
 
 /**
  * Implementation of {@link Flow.Publisher} corresponding to reactive streams specification.
  * <p>
- * Delegates to {@link SubmissionPublisher} repackaged from jsr166.
+ * Delegates to {@link SubmissionPublisher} repackaged from jsr166 on JDK 8 or to JDK {@code SubmissionPublisher}.
  *
  * @author Adam Lindenthal (adam.lindenthal at oracle.com)
  */
 public class JerseyPublisher<T> implements Flow.Publisher<T> {
 
     private static final int DEFAULT_BUFFER_CAPACITY = 256;
-    private SubmissionPublisher<T> submissionPublisher = new SubmissionPublisher<>();
+    private SubmittableFlowPublisher<T> submissionPublisher = SubmissionPublisherFactory.createSubmissionPublisher();
 
     private final PublisherStrategy strategy;
 
@@ -90,7 +92,7 @@ public JerseyPublisher(final Executor executor) {
      */
     public JerseyPublisher(final Executor executor, final PublisherStrategy strategy) {
         this.strategy = strategy;
-        submissionPublisher = new SubmissionPublisher<>(executor, DEFAULT_BUFFER_CAPACITY);
+        submissionPublisher = SubmissionPublisherFactory.createSubmissionPublisher(executor, DEFAULT_BUFFER_CAPACITY);
     }
 
 
@@ -128,7 +130,7 @@ public JerseyPublisher(final int maxBufferCapacity) {
      */
     public JerseyPublisher(final Executor executor, final int maxBufferCapacity, PublisherStrategy strategy) {
         this.strategy = strategy;
-        submissionPublisher = new SubmissionPublisher<>(executor, maxBufferCapacity);
+        submissionPublisher = SubmissionPublisherFactory.createSubmissionPublisher(executor, maxBufferCapacity);
     }
 
     @Override

core-common/src/main/java/org/glassfish/jersey/internal/jsr166/SubmissionPublisher.java → core-common/src/main/java8/org/glassfish/jersey/internal/jsr166/SubmissionPublisher.java

@@ -134,7 +134,7 @@
  * @author Doug Lea
  * @since 9
  */
-public class SubmissionPublisher<T> implements Flow.Publisher<T>,
+public class SubmissionPublisher<T> implements Flow.Publisher<T>, SubmittableFlowPublisher<T>,
         AutoCloseable {
     /*
      * Most mechanics are handled by BufferedSubscription. This class

core-common/src/main/java8/org/glassfish/jersey/internal/jsr166/SubmissionPublisherFactory.java

@@ -0,0 +1,53 @@
+/*
+ * Copyright (c) 2019 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;
+
+import java.util.concurrent.Executor;
+import java.util.concurrent.ForkJoinPool;
+import java.util.function.BiConsumer;
+
+/**
+ * Factory creating JDK8 compatible SubmissionPublisher (Jdk8SubmissionPublisher) or JDK 9+ SubmissionPublisher
+ */
+public class SubmissionPublisherFactory {
+
+    /**
+     * Creates a new SubmissionPublisher using the {@link
+     * ForkJoinPool#commonPool()} for async delivery to subscribers
+     * (unless it does not support a parallelism level of at least two,
+     * in which case, a new Thread is created to run each task), with
+     * maximum buffer capacity of {@link Flow#defaultBufferSize}, and no
+     * handler for Subscriber exceptions in method {@link
+     * Flow.Subscriber#onNext(Object) onNext}.
+     */
+    public static <T> SubmittableFlowPublisher<T> createSubmissionPublisher() {
+        return new SubmissionPublisher<T>();
+    }
+
+    public static <T> SubmittableFlowPublisher<T> createSubmissionPublisher(Executor executor,
+                                                                            int maxBufferCapacity) {
+        return new SubmissionPublisher<T>(executor, maxBufferCapacity);
+    }
+
+    public static <T> SubmittableFlowPublisher<T> createSubmissionPublisher(Executor executor,
+                                                                            int maxBufferCapacity,
+                                                                            BiConsumer<? super Flow.Subscriber<? super T>,
+                                                                                    ? super Throwable> handler) {
+        return new SubmissionPublisher<T>(executor, maxBufferCapacity, handler);
+    }
+
+}