Tests for Sse with injected Flow.Subscriber

Signed-off-by: Daniel Kec <[email protected]>

media/sse/pom.xml

@@ -56,6 +56,34 @@
             <type>pom</type>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>io.reactivex.rxjava2</groupId>
+            <artifactId>rxjava</artifactId>
+            <version>${rxjava2.version}</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.reactivestreams</groupId>
+            <artifactId>reactive-streams</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.eclipse</groupId>
+            <artifactId>yasson</artifactId>
+            <version>${yasson.version}</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.glassfish</groupId>
+            <artifactId>jakarta.json</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.glassfish.jersey.media</groupId>
+            <artifactId>jersey-media-json-binding</artifactId>
+            <version>${project.version}</version>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
 
     <build>

media/sse/src/main/java/org/glassfish/jersey/media/sse/internal/JerseyEventSink.java

@@ -136,7 +136,6 @@ public void flush() throws IOException {
     }
 
     public void onComplete() {
-        cancelUpstream();
         close();
     }
 
@@ -163,6 +162,7 @@ private MediaType resolveMediaType(Object item) {
                     || item instanceof Byte
                     || item instanceof Short
                     || item instanceof Integer
+                    || item instanceof Long
                     || item instanceof Double) {
                 implicitMediaType = MediaType.TEXT_PLAIN_TYPE;
             } else {

media/sse/src/test/java/org/glassfish/jersey/media/sse/JerseyFlowAdapters.java

@@ -0,0 +1,64 @@
+/*
+ * 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.media.sse;
+
+import org.glassfish.jersey.internal.jsr166.Flow;
+
+public class JerseyFlowAdapters {
+
+    /**
+     * Adapt {@link org.glassfish.jersey.internal.jsr166.Flow.Subscriber} to
+     * {@link org.reactivestreams.Subscriber}.
+     *
+     * @param jerseySubscriber Jersey's repackaged {@link org.glassfish.jersey.internal.jsr166.Flow.Subscriber}
+     * @param <T>              payload type
+     * @return Reactive Streams's {@link org.reactivestreams.Subscriber}
+     */
+    static <T> org.reactivestreams.Subscriber<T> toSubscriber(Flow.Subscriber<T> jerseySubscriber) {
+        return new org.reactivestreams.Subscriber<T>() {
+            @Override
+            public void onSubscribe(final org.reactivestreams.Subscription subscription) {
+                jerseySubscriber.onSubscribe(new Flow.Subscription() {
+                    @Override
+                    public void request(final long n) {
+                        subscription.request(n);
+                    }
+
+                    @Override
+                    public void cancel() {
+                        subscription.cancel();
+                    }
+                });
+            }
+
+            @Override
+            public void onNext(final T t) {
+                jerseySubscriber.onNext(t);
+            }
+
+            @Override
+            public void onError(final Throwable throwable) {
+                jerseySubscriber.onError(throwable);
+            }
+
+            @Override
+            public void onComplete() {
+                jerseySubscriber.onComplete();
+            }
+        };
+    }
+}