Treat Multi and Flow as single element schema with SSE media type

core/src/main/java/io/smallrye/openapi/api/constants/KotlinConstants.java

@@ -1,6 +1,7 @@
 package io.smallrye.openapi.api.constants;
 
 import org.jboss.jandex.DotName;
+import org.jboss.jandex.Type;
 
 /**
  * Constants related to the Kotlin language
@@ -24,6 +25,9 @@ public class KotlinConstants {
     public static final DotName FLOW = DotName
             .createSimple("kotlinx.coroutines.flow.Flow");
 
+    public static final Type FLOW_TYPE = Type
+            .create(FLOW, Type.Kind.CLASS);
+
     private KotlinConstants() {
     }
 }

core/src/main/java/io/smallrye/openapi/runtime/io/schema/SchemaFactory.java

@@ -26,6 +26,7 @@
 import org.jboss.jandex.Type.Kind;
 
 import io.smallrye.openapi.api.constants.JDKConstants;
+import io.smallrye.openapi.api.constants.KotlinConstants;
 import io.smallrye.openapi.api.constants.MutinyConstants;
 import io.smallrye.openapi.api.util.MergeUtil;
 import io.smallrye.openapi.internal.models.media.SchemaSupport;
@@ -601,27 +602,13 @@ public static Schema typeToSchema(final AnnotationScannerContext context, Type t
             TypeUtil.applyTypeAttributes(type, schema, schemaAnnotation);
             schema = schemaRegistration(context, type, schema);
         } else if (type.kind() == Type.Kind.ARRAY) {
-            schema = OASFactory.createSchema().addType(SchemaType.ARRAY);
-            ArrayType array = type.asArrayType();
-            int dimensions = array.dimensions();
-            Type componentType = array.constituent();
-            Schema itemSchema;
-
-            if (dimensions > 1) {
-                // Recurse using a new array type with dimensions decremented
-                itemSchema = typeToSchema(context, ArrayType.create(componentType, dimensions - 1), null);
-            } else {
-                // Recurse using the type of the array elements
-                itemSchema = typeToSchema(context, componentType, null);
-                // Maybe dereference
-                itemSchema = lookupRef(context, componentType, itemSchema);
-            }
-
-            schema.setItems(itemSchema);
+            schema = arrayTypeToSchema(context, type.asArrayType());
         } else if (type.kind() == Type.Kind.CLASS) {
             schema = introspectClassToSchema(context, type.asClassType(), true);
         } else if (type.kind() == Type.Kind.PRIMITIVE) {
             schema = OpenApiDataObjectScanner.process(type.asPrimitiveType());
+        } else if (isMultiValuedType(context, type)) {
+            schema = multiValuedTypeToSchema(context, type);
         } else {
             schema = otherTypeToSchema(context, type);
         }
@@ -636,6 +623,26 @@ public static Schema typeToSchema(final AnnotationScannerContext context, Type t
         return schema;
     }
 
+    private static Schema arrayTypeToSchema(final AnnotationScannerContext context, ArrayType array) {
+        Schema schema = OASFactory.createSchema().addType(SchemaType.ARRAY);
+        int dimensions = array.dimensions();
+        Type componentType = array.constituent();
+        Schema itemSchema;
+
+        if (dimensions > 1) {
+            // Recurse using a new array type with dimensions decremented
+            itemSchema = typeToSchema(context, ArrayType.create(componentType, dimensions - 1), null);
+        } else {
+            // Recurse using the type of the array elements
+            itemSchema = typeToSchema(context, componentType, null);
+            // Maybe dereference
+            itemSchema = lookupRef(context, componentType, itemSchema);
+        }
+
+        schema.setItems(itemSchema);
+        return schema;
+    }
+
     /**
      * Convert a Jandex enum class type to a {@link Schema} model. Adds each enum constant
      * name to the list of the given schema's enumeration list.
@@ -805,19 +812,36 @@ private static List<Schema> readClassSchemas(final AnnotationScannerContext cont
                 .collect(Collectors.toList());
     }
 
-    private static Schema otherTypeToSchema(final AnnotationScannerContext context, Type type) {
-        if (TypeUtil.isA(context, type, MutinyConstants.MULTI_TYPE)) {
-            // Treat as an Array
-            Schema schema = OASFactory.createSchema().addType(SchemaType.ARRAY);
-            Type componentType = type.asParameterizedType().arguments().get(0);
+    private static boolean isMultiValuedType(final AnnotationScannerContext context, Type type) {
+        return TypeUtil.isA(context, type, MutinyConstants.MULTI_TYPE)
+                || TypeUtil.isA(context, type, KotlinConstants.FLOW_TYPE);
+    }
 
-            // Recurse using the type of the array elements
-            schema.setItems(typeToSchema(context, componentType));
-            return schema;
+    private static Schema multiValuedTypeToSchema(final AnnotationScannerContext context, Type type) {
+        Type componentType = type.asParameterizedType().arguments().get(0);
+        // Recurse using the type of the parameterize type
+        Schema componentSchema = typeToSchema(context, componentType);
+        Schema typeSchema;
+
+        var mediaTypes = Optional.ofNullable(context.getCurrentProduces())
+                .map(Arrays::asList)
+                .orElseGet(Collections::emptyList);
+
+        if (mediaTypes.contains("text/event-stream")) {
+            // Treat as a single item for SSE
+            typeSchema = componentSchema;
         } else {
-            Type asyncType = resolveAsyncType(context, type);
-            return schemaRegistration(context, asyncType, OpenApiDataObjectScanner.process(context, asyncType));
+            // Treat as an array
+            typeSchema = OASFactory.createSchema().addType(SchemaType.ARRAY);
+            typeSchema.setItems(componentSchema);
         }
+
+        return typeSchema;
+    }
+
+    private static Schema otherTypeToSchema(final AnnotationScannerContext context, Type type) {
+        Type asyncType = resolveAsyncType(context, type);
+        return schemaRegistration(context, asyncType, OpenApiDataObjectScanner.process(context, asyncType));
     }
 
     static Type resolveAsyncType(final AnnotationScannerContext context, Type type) {

core/src/main/java/io/smallrye/openapi/runtime/util/TypeUtil.java

@@ -309,7 +309,6 @@ public class TypeUtil {
         jdkIndex = indexer.complete();
 
         wrapperTypes.addAll(JaxbConstants.JAXB_ELEMENT);
-        wrapperTypes.add(KotlinConstants.FLOW);
         wrapperTypes.add(MutinyConstants.UNI_TYPE.name());
         wrapperTypes.add(JDKConstants.COMPLETION_STAGE_NAME);
         wrapperTypes.add(JDKConstants.COMPLETABLE_FUTURE_NAME);

extension-jaxrs/pom.xml

@@ -81,6 +81,11 @@
             <version>${version.parsson.json}</version>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.jetbrains.kotlinx</groupId>
+            <artifactId>kotlinx-coroutines-core</artifactId>
+            <scope>test</scope>
+        </dependency>
 
         <!-- Depend on core tests and testsuite data classes -->
         <dependency>

extension-jaxrs/src/test/java/io/smallrye/openapi/runtime/scanner/ApiResponseTests.java

@@ -525,4 +525,46 @@ public java.util.stream.Stream<Bean> get() {
 
         assertJsonEquals("responses.stream-return-type.json", Bean.class, StreamAPI.class);
     }
+
+    @Test
+    /*
+     * Test case for Smallrye OpenAPI issue 2535 and Quarkus issue 51109.
+     *
+     * https://github.com/smallrye/smallrye-open-api/issues/2535
+     * https://github.com/quarkusio/quarkus/issues/51109
+     */
+    void testServerSentEvents() throws IOException, JSONException {
+        @jakarta.ws.rs.Path("streams")
+        class ServerSentEventAPI {
+            @jakarta.ws.rs.GET
+            @jakarta.ws.rs.Path("sse-flow")
+            @jakarta.ws.rs.Produces(jakarta.ws.rs.core.MediaType.SERVER_SENT_EVENTS)
+            public kotlinx.coroutines.flow.Flow<String> getSseFlow() {
+                return null;
+            }
+
+            @jakarta.ws.rs.GET
+            @jakarta.ws.rs.Path("json-flow")
+            @jakarta.ws.rs.Produces(jakarta.ws.rs.core.MediaType.APPLICATION_JSON)
+            public kotlinx.coroutines.flow.Flow<String> getJsonFlow() {
+                return null;
+            }
+
+            @jakarta.ws.rs.GET
+            @jakarta.ws.rs.Path("sse-multi")
+            @jakarta.ws.rs.Produces(jakarta.ws.rs.core.MediaType.SERVER_SENT_EVENTS)
+            public io.smallrye.mutiny.Multi<String> getSseMulti() {
+                return null;
+            }
+
+            @jakarta.ws.rs.GET
+            @jakarta.ws.rs.Path("json-multi")
+            @jakarta.ws.rs.Produces(jakarta.ws.rs.core.MediaType.APPLICATION_JSON)
+            public io.smallrye.mutiny.Multi<String> getJsonMulti() {
+                return null;
+            }
+        }
+
+        assertJsonEquals("responses.server-sent-events.json", ServerSentEventAPI.class);
+    }
 }

extension-jaxrs/src/test/resources/io/smallrye/openapi/runtime/scanner/responses.server-sent-events.json

@@ -0,0 +1,75 @@
+{
+  "openapi" : "3.1.0",
+  "paths" : {
+    "/streams/json-flow" : {
+      "get" : {
+        "responses" : {
+          "200" : {
+            "description" : "OK",
+            "content" : {
+              "application/json" : {
+                "schema" : {
+                  "type" : "array",
+                  "items" : {
+                    "type" : "string"
+                  }
+                }
+              }
+            }
+          }
+        }
+      }
+    },
+    "/streams/json-multi" : {
+      "get" : {
+        "responses" : {
+          "200" : {
+            "description" : "OK",
+            "content" : {
+              "application/json" : {
+                "schema" : {
+                  "type" : "array",
+                  "items" : {
+                    "type" : "string"
+                  }
+                }
+              }
+            }
+          }
+        }
+      }
+    },
+    "/streams/sse-flow" : {
+      "get" : {
+        "responses" : {
+          "200" : {
+            "description" : "OK",
+            "content" : {
+              "text/event-stream" : {
+                "schema" : {
+                  "type" : "string"
+                }
+              }
+            }
+          }
+        }
+      }
+    },
+    "/streams/sse-multi" : {
+      "get" : {
+        "responses" : {
+          "200" : {
+            "description" : "OK",
+            "content" : {
+              "text/event-stream" : {
+                "schema" : {
+                  "type" : "string"
+                }
+              }
+            }
+          }
+        }
+      }
+    }
+  }
+}

pom.xml

@@ -24,6 +24,7 @@
         <version.io.smallrye.smallrye-config>3.17.2</version.io.smallrye.smallrye-config>
         <version.io.smallrye.smallrye-common-classloader>2.17.1</version.io.smallrye.smallrye-common-classloader>
         <version.eclipse.microprofile.openapi>4.1.1</version.eclipse.microprofile.openapi>
+        <version.kotlinx-coroutines-core>1.10.2</version.kotlinx-coroutines-core>
         <version.kotlinx-metadata-jvm>0.9.0</version.kotlinx-metadata-jvm>
         <version.org.hamcrest>1.3</version.org.hamcrest>
         <version.org.hamcrest.java-hamcrest>2.0.0.0</version.org.hamcrest.java-hamcrest>
@@ -119,11 +120,16 @@
                 <scope>import</scope>
                 <type>pom</type>
             </dependency>
-          <dependency>
-              <groupId>org.jetbrains.kotlinx</groupId>
-              <artifactId>kotlinx-metadata-jvm</artifactId>
-              <version>${version.kotlinx-metadata-jvm}</version>
-          </dependency>
+            <dependency>
+                <groupId>org.jetbrains.kotlinx</groupId>
+                <artifactId>kotlinx-metadata-jvm</artifactId>
+                <version>${version.kotlinx-metadata-jvm}</version>
+            </dependency>
+            <dependency>
+                <groupId>org.jetbrains.kotlinx</groupId>
+                <artifactId>kotlinx-coroutines-core</artifactId>
+                <version>${version.kotlinx-coroutines-core}</version>
+            </dependency>
 
             <!-- SmallRye Projects -->
             <dependency>