8.3.0: stream application/x-ndjson endpoints without making Spring MVC services reactive

README.md

@@ -30,6 +30,7 @@ Maven and Gradle
   - [Usage](#usage)
   - [Calling an API from your service (callMode)](#calling-an-api-from-your-service-callmode)
   - [Object-typed query parameters and multipart bodies](#object-typed-query-parameters-and-multipart-bodies)
+  - [Streaming endpoints (application/x-ndjson)](#streaming-endpoints-applicationx-ndjson)
   - [Camel case Java names (useCamelCaseNames)](#camel-case-java-names-usecamelcasenames)
   - [Unknown enum values (useUnknownEnumValue)](#unknown-enum-values-useunknownenumvalue)
 - [Property Validation](#property-validation)
@@ -1092,6 +1093,66 @@ own: every generated API takes one argument per part (`MultipartFile file,
 String comment`). A multipart body that references a component schema keeps
 taking that model.
 
+### Streaming endpoints (application/x-ndjson)
+
+An operation whose successful response, or whose request body, is
+`application/x-ndjson` (or the older `application/stream+json`) streams one JSON
+document per line. Its schema is the type of each item, or an array of it, and
+an OpenAPI 3.2 `itemSchema` is read the same way:
+
+```yaml
+responses:
+  '200':
+    content:
+      application/x-ndjson:
+        schema:
+          $ref: "#/components/schemas/Item"
+```
+
+The operation is detected from the contract, so the rest of the API keeps its
+usual signatures, and a Spring MVC service does not have to become reactive
+for it.
+
+- **Spring MVC servers** (`reactive` off) split the operation in two. The
+  endpoint Spring maps sends what a plain Java method returns, and that method
+  is the one to implement:
+
+  ```java
+  @RestController
+  class ItemController implements ItemsApi {
+
+    @Override
+    public ResponseEntity<Stream<ItemDTO>> listItems(final String filter) {
+      return ResponseEntity.ok(itemRepository.streamByFilter(filter));
+    }
+  }
+  ```
+
+  Each item is sent as soon as the stream produces it, on Spring MVC's
+  asynchronous executor, with the application's `ObjectMapper`, and the stream
+  is closed once it is sent, also when sending fails. The status and headers of
+  the returned `ResponseEntity` are sent too. A streamed request body arrives as
+  a `Stream<ItemDTO>` read line by line as it is consumed. The generated
+  `NdjsonSupport` class, next to the interfaces, does the writing and reading.
+
+  When the response also declares `application/json`, which must then be the
+  array of the streamed items, a second endpoint sends the same items as a JSON
+  array, and Spring picks one or the other from the `Accept` header (a request
+  that accepts anything, or sends no `Accept`, gets the JSON array).
+
+  The response status is sent with the first item, so an error in the middle
+  of the stream can only end the response early. Configure
+  `spring.mvc.async.request-timeout` for streams that last long.
+- **WebFlux servers** (`reactive` on) take and return a `Flux<ItemDTO>`.
+- **WebClient clients** (`callMode` and `reactive`) return a `Flux<ItemDTO>`,
+  or a `Mono<ResponseEntity<Flux<ItemDTO>>>` from `...WithHttpInfo`, and send a
+  streamed body from a `Flux<ItemDTO>`. **Reactive `@HttpExchange` interfaces**
+  return a `Mono<ResponseEntity<Flux<ItemDTO>>>`.
+- The **blocking clients** (RestClient, and `@HttpExchange` without `reactive`)
+  read a response whole, so they cannot hand the items over as they arrive, and
+  the generation fails for a streaming operation: generate that client with
+  `reactive` on.
+
 ### Camel case Java names (useCamelCaseNames)
 
 By default the generated Java code uses the names the contract gives, whenever

multiapi-engine/pom.xml

@@ -4,7 +4,7 @@
 
   <groupId>com.sngular</groupId>
   <artifactId>multiapi-engine</artifactId>
-  <version>8.2.1</version>
+  <version>8.3.0</version>
   <packaging>jar</packaging>
 
   <properties>

multiapi-engine/src/main/java/com/sngular/api/generator/plugin/openapi/OpenApiGenerator.java

@@ -200,6 +200,7 @@ private void processFile(final SpecFile specFile) {
     // Determine the actual base URI for resolving external references
     final URI specBaseUri = resolveSpecBaseUri(specFile);
     OpenApiUtil.solvePathRefs(openAPI, specBaseUri);
+    OpenApiUtil.promoteItemSchemas(openAPI);
     final String clientPackage = specFile.getClientPackage();
 
     validateClientOptions(specFile);

multiapi-engine/src/main/java/com/sngular/api/generator/plugin/openapi/model/ContentObject.java

@@ -27,4 +27,10 @@ public class ContentObject {
 
   private SchemaObject schemaObject;
 
+  /**
+   * Whether the media type streams a sequence of documents (such as {@code application/x-ndjson}). The data type is then the
+   * type of each streamed item.
+   */
+  private boolean streaming;
+
 }

multiapi-engine/src/main/java/com/sngular/api/generator/plugin/openapi/model/OperationObject.java

@@ -9,6 +9,8 @@
 import java.util.ArrayList;
 import java.util.List;
 
+import com.sngular.api.generator.plugin.common.model.SchemaFieldObjectType;
+
 import lombok.AllArgsConstructor;
 import lombok.Builder;
 import lombok.Data;
@@ -38,6 +40,18 @@ public final class OperationObject {
 
   private List<String> securities;
 
+  /** Whether the successful response streams its items (such as {@code application/x-ndjson}). */
+  private boolean streamingResponse;
+
+  /** The type of each item the successful response streams. */
+  private SchemaFieldObjectType streamingItemType;
+
+  /** The media type the successful response streams its items as. */
+  private String streamingMediaType;
+
+  /** Whether the successful response can also be sent as a JSON array of the streamed items. */
+  private boolean streamingAlsoJson;
+
   public static final class OperationObjectBuilder {
 
     private final List<String> tags = new ArrayList<>();

multiapi-engine/src/main/java/com/sngular/api/generator/plugin/openapi/model/RequestObject.java

@@ -32,6 +32,9 @@ public class RequestObject {
    */
   private boolean inlineMultipart;
 
+  /** Whether the body streams its items (such as {@code application/x-ndjson}); its content's data type is then the item's. */
+  private boolean streaming;
+
   public static final class RequestObjectBuilder {
 
     private final List<ContentObject> contentObjects = new ArrayList<>();

multiapi-engine/src/main/java/com/sngular/api/generator/plugin/openapi/template/ClasspathTemplateLoader.java

@@ -21,7 +21,7 @@ public class ClasspathTemplateLoader extends CommonTemplateLoader {
   private static final List<String> TEMPLATE_FILES = List.of(TemplateIndexConstants.TEMPLATE_INTERFACE_API, TemplateIndexConstants.TEMPLATE_CALL_WEB_API,
                                                              TemplateIndexConstants.TEMPLATE_WEB_CLIENT, TemplateIndexConstants.TEMPLATE_CALL_REST_API,
                                                              TemplateIndexConstants.TEMPLATE_REST_CLIENT, TemplateIndexConstants.TEMPLATE_REACTIVE_API,
-                                                             TemplateIndexConstants.TEMPLATE_CALL_HTTP_EXCHANGE_API);
+                                                             TemplateIndexConstants.TEMPLATE_CALL_HTTP_EXCHANGE_API, TemplateIndexConstants.TEMPLATE_NDJSON_SUPPORT);
 
   private static final List<String> TEMPLATE_AUTH_FILES = List.of(TemplateIndexConstants.TEMPLATE_API_KEY, TemplateIndexConstants.TEMPLATE_AUTHENTICATION,
                                                                   TemplateIndexConstants.TEMPLATE_HTTP_BASIC, TemplateIndexConstants.TEMPLATE_HTTP_BEARER,

multiapi-engine/src/main/java/com/sngular/api/generator/plugin/openapi/template/TemplateFactory.java

@@ -14,6 +14,7 @@
 import com.sngular.api.generator.plugin.common.template.CommonTemplateFactory;
 import com.sngular.api.generator.plugin.openapi.model.AuthObject;
 import com.sngular.api.generator.plugin.openapi.model.PathObject;
+import com.sngular.api.generator.plugin.openapi.model.RequestObject;
 import com.sngular.api.generator.plugin.openapi.parameter.SpecFile;
 import org.apache.commons.lang3.StringUtils;
 
@@ -89,6 +90,18 @@ public final void fillTemplate(
 
     writeTemplateToFile(specFile.isCallMode() ? getTemplateClientApi(specFile) : getTemplateApi(specFile),
                         StringUtils.defaultIfEmpty(specFile.getApiPackage(), DEFAULT_API_PACKAGE), className + "Api");
+
+    // Spring MVC interfaces send and read application/x-ndjson through a helper written next to them.
+    if (!specFile.isCallMode() && !specFile.isReactive() && streams(pathObjects)) {
+      writeTemplateToFile(TemplateIndexConstants.TEMPLATE_NDJSON_SUPPORT, StringUtils.defaultIfEmpty(specFile.getApiPackage(), DEFAULT_API_PACKAGE),
+                          "NdjsonSupport");
+    }
+  }
+
+  private static boolean streams(final List<PathObject> pathObjects) {
+    return pathObjects.stream()
+                      .flatMap(path -> path.getOperationObjects().stream())
+                      .anyMatch(operation -> operation.isStreamingResponse() || operation.getRequestObjects().stream().anyMatch(RequestObject::isStreaming));
   }
 
   private String getTemplateClientApi(final SpecFile specFile) {

multiapi-engine/src/main/java/com/sngular/api/generator/plugin/openapi/template/TemplateIndexConstants.java

@@ -20,6 +20,8 @@ public class TemplateIndexConstants extends CommonTemplateIndexConstants {
 
   public static final String TEMPLATE_CALL_HTTP_EXCHANGE_API = "templateCallHttpExchange.ftlh";
 
+  public static final String TEMPLATE_NDJSON_SUPPORT = "templateNdjsonSupport.ftlh";
+
   public static final String TEMPLATE_REST_CLIENT = "templateRestClient.ftlh";
 
   public static final String TEMPLATE_WEB_CLIENT = "templateWebClient.ftlh";

multiapi-engine/src/main/java/com/sngular/api/generator/plugin/openapi/utils/MapperPathUtil.java

@@ -13,10 +13,12 @@
 import java.util.HashMap;
 import java.util.Iterator;
 import java.util.List;
+import java.util.Locale;
 import java.util.Map;
 import java.util.Map.Entry;
 import java.util.Objects;
 import java.util.Optional;
+import java.util.Set;
 import java.util.function.BiConsumer;
 
 import com.fasterxml.jackson.databind.JsonNode;
@@ -28,6 +30,7 @@
 import com.sngular.api.generator.plugin.common.tools.MapperUtil;
 import com.sngular.api.generator.plugin.common.tools.SchemaUtil;
 import com.sngular.api.generator.plugin.common.tools.StringCaseUtils;
+import com.sngular.api.generator.plugin.openapi.exception.CodeGenerationException;
 import com.sngular.api.generator.plugin.openapi.exception.DuplicatedOperationException;
 import com.sngular.api.generator.plugin.openapi.exception.InvalidOpenAPIException;
 import com.sngular.api.generator.plugin.openapi.model.AuthSchemaObject;
@@ -57,6 +60,14 @@ public class MapperPathUtil {
 
   public static final String SCHEMA = "schema";
 
+  /**
+   * The media types that stream a sequence of JSON documents, one per line, which Spring reads and writes item by item. Their
+   * schema is the type of each item, or an array of it.
+   */
+  private static final Set<String> STREAMING_MEDIA_TYPES = Set.of("application/x-ndjson", "application/stream+json");
+
+  private static final String JSON_MEDIA_TYPE = "application/json";
+
   private MapperPathUtil() {
   }
 
@@ -152,7 +163,7 @@ private static OperationObject createOperation(
       final JsonNode operation, final String operationType, final SpecFile specFile, final GlobalObject globalObject,
       final List<String> operationIdList, final Path baseDir) {
     final JsonNode tagsNode = operation.has("tags") ? operation.get("tags") : null;
-    return OperationObject.builder()
+    final OperationObject operationObject = OperationObject.builder()
                           .operationId(mapOperationId(getOperationId(operation), operationIdList))
                           .operationType(operationType)
                           .summary(ApiTool.getNodeAsString(operation, "summary"))
@@ -165,6 +176,58 @@ private static OperationObject createOperation(
                           .consumes(getRequestList(operation.at("/requestBody")))
                           .produces(getResponseList(operation.at("/responses")))
                           .build();
+    applyStreaming(operationObject, specFile);
+    return operationObject;
+  }
+
+  public static boolean isStreamingMediaType(final String mediaType) {
+    return Objects.nonNull(mediaType) && STREAMING_MEDIA_TYPES.contains(mediaType.toLowerCase(Locale.ROOT));
+  }
+
+  /**
+   * Marks the operation whose successful response streams its items, and checks that the contract can be generated as a
+   * stream: a JSON alternative of the same response must be the array of the streamed items, and a streamed body its only
+   * content.
+   */
+  private static void applyStreaming(final OperationObject operation, final SpecFile specFile) {
+    final Optional<ResponseObject> streamedResponse = operation.getResponseObjects().stream()
+                                                              .filter(response -> StringUtils.startsWith(response.getResponseName(), "2"))
+                                                              .filter(response -> response.getContentObjects().stream().anyMatch(ContentObject::isStreaming))
+                                                              .findFirst();
+    streamedResponse.ifPresent(response -> {
+      final ContentObject content = response.getContentObjects().stream().filter(ContentObject::isStreaming).findFirst().orElseThrow();
+      operation.setStreamingResponse(true);
+      operation.setStreamingItemType(content.getDataType());
+      operation.setStreamingMediaType(content.getName());
+      operation.setStreamingAlsoJson(hasJsonAlternative(operation.getOperationId(), response, content));
+    });
+    for (final RequestObject request : operation.getRequestObjects()) {
+      if (request.isStreaming() && request.getContentObjects().size() > 1) {
+        throw new CodeGenerationException("Operation '" + operation.getOperationId() + "' streams its request body, which must then declare only the "
+                                          + "streamed media type, but it declares " + request.getContentObjects().stream().map(ContentObject::getName).toList());
+      }
+    }
+    final boolean streams = operation.isStreamingResponse() || operation.getRequestObjects().stream().anyMatch(RequestObject::isStreaming);
+    if (streams && specFile.isCallMode() && !specFile.isReactive()) {
+      throw new CodeGenerationException("Operation '" + operation.getOperationId() + "' streams application/x-ndjson, which the blocking clients cannot "
+                                        + "consume: generate its client with reactive=true (a WebClient client or a reactive @HttpExchange interface)");
+    }
+  }
+
+  /** Whether the streamed response also declares JSON, which then has to be the array of the streamed items. */
+  private static boolean hasJsonAlternative(final String operationId, final ResponseObject response, final ContentObject streamed) {
+    final Optional<ContentObject> json = response.getContentObjects().stream()
+                                                 .filter(content -> JSON_MEDIA_TYPE.equalsIgnoreCase(content.getName()))
+                                                 .findFirst();
+    json.ifPresent(content -> {
+      final SchemaFieldObjectType jsonType = content.getDataType();
+      if (Objects.isNull(jsonType) || !TypeConstants.ARRAY.equals(jsonType.getBaseType()) || Objects.isNull(jsonType.getInnerType())
+          || !jsonType.getInnerType().toString().equals(String.valueOf(streamed.getDataType()))) {
+        throw new CodeGenerationException("Operation '" + operationId + "' streams " + streamed.getDataType() + " items as "
+                                          + streamed.getName() + ", so its application/json response must be the array of them, but it is " + jsonType);
+      }
+    });
+    return json.isPresent();
   }
 
   private static String getOperationId(final JsonNode operation) {
@@ -235,6 +298,10 @@ private static boolean isInlineMultipart(final JsonNode content) {
     return !multipartSchema.isMissingNode() && !ApiTool.hasRef(multipartSchema);
   }
 
+  private static boolean isStreamingContent(final JsonNode content) {
+    return Objects.nonNull(content) && IteratorUtils.toList(content.fieldNames()).stream().anyMatch(MapperPathUtil::isStreamingMediaType);
+  }
+
   private static List<RequestObject> mapRequestObject(
       final SpecFile specFile, final JsonNode operation,
       final GlobalObject globalObject, final Path baseDir) {
@@ -251,6 +318,7 @@ private static List<RequestObject> mapRequestObject(
                                         .required(isRequiredBody(requestBody))
                                         .isFormData(ApiTool.getNode(requestBody, CONTENT).has("multipart/form-data"))
                                         .inlineMultipart(isInlineMultipart(ApiTool.getNode(requestBody, CONTENT)))
+                                        .streaming(isStreamingContent(ApiTool.getNode(requestBody, CONTENT)))
                                         .contentObjects(mapContentObject(specFile, ApiTool.getNode(requestBody, CONTENT),
                                                                          "InlineObject" + operationIdWithCap, globalObject, baseDir))
                                         .build());
@@ -264,6 +332,7 @@ private static List<RequestObject> mapRequestObject(
                                         .required(isRequiredBody(actualRequestBody))
                                         .isFormData(ApiTool.getNode(actualRequestBody, CONTENT).has("multipart/form-data"))
                                         .inlineMultipart(isInlineMultipart(ApiTool.getNode(actualRequestBody, CONTENT)))
+                                        .streaming(isStreamingContent(ApiTool.getNode(actualRequestBody, CONTENT)))
                                         .contentObjects(mapContentObject(specFile, ApiTool.getNode(actualRequestBody, CONTENT),
                                                                          operationIdWithCap, globalObject, baseDir))
                                         .build());
@@ -500,7 +569,8 @@ private static List<ContentObject> mapContentObject(
     if (Objects.nonNull(content)) {
       for (final Iterator<String> it = content.fieldNames(); it.hasNext(); ) {
         final String mediaType = it.next();
-        final var schema = ApiTool.getNode(ApiTool.getNode(content, mediaType), SCHEMA);
+        final boolean streaming = isStreamingMediaType(mediaType);
+        final var schema = streamedItemSchema(ApiTool.getNode(ApiTool.getNode(content, mediaType), SCHEMA), streaming);
         final String pojoName = preparePojoName(inlineObject, schema, specFile);
         final SchemaFieldObjectType dataType = getSchemaType(schema, pojoName, specFile, globalObject, baseDir);
         final String importName = getImportFromType(dataType);
@@ -517,12 +587,18 @@ private static List<ContentObject> mapContentObject(
                                         .name(mediaType)
                                         .importName(importName)
                                         .schemaObject(schemaObject)
+                                        .streaming(streaming)
                                         .build());
       }
     }
     return contentObjects;
   }
 
+  /** The schema of each item a streamed media type sends: its schema, or the items of it when it is declared as an array. */
+  private static JsonNode streamedItemSchema(final JsonNode schema, final boolean streaming) {
+    return streaming && Objects.nonNull(schema) && ApiTool.isArray(schema) && ApiTool.hasItems(schema) ? ApiTool.getItems(schema) : schema;
+  }
+
   private static String preparePojoName(final String inlineObject, final JsonNode schema, final SpecFile specFile) {
     // Use unified ResponseWrapperHandler for wrapper decisions (v7.0)
     // This ensures OpenApiUtil and MapperPathUtil stay in sync