codegen: add streaming support for application/octet-stream contents

generated-doc/out/generator/sbt-openapi-codegen.md

@@ -44,6 +44,7 @@ openapiJsonSerdeLib                   circe                                The j
 openapiValidateNonDiscriminatedOneOfs true                                 Whether to fail if variants of a oneOf without a discriminator cannot be disambiguated.
 openapiMaxSchemasPerFile              400                                  Maximum number of schemas to generate in a single file (tweak if hitting javac class size limits).
 openapiAdditionalPackages             Nil                                  Additional packageName/swaggerFile pairs for generating from multiple schemas 
+openapiStreamingImplementation        fs2                                  Backend capability to assume for streaming content. Supports akka, fs2, pekko and zio.  
 ===================================== ==================================== ==================================================================================================
 ```
 

openapi-codegen/cli/src/main/scala/sttp/tapir/codegen/GenScala.scala

@@ -62,6 +62,9 @@ object GenScala {
   private val jsonLibOpt: Opts[Option[String]] =
     Opts.option[String]("jsonLib", "Json library to use for serdes", "j").orNone
 
+  private val streamingImplementationOpt: Opts[Option[String]] =
+    Opts.option[String]("streamingImplementation", "Capability to use for binary streams", "s").orNone
+
   private val destDirOpt: Opts[File] =
     Opts
       .option[String]("destdir", "Destination directory", "d")
@@ -84,7 +87,8 @@ object GenScala {
       headTagForNamesOpt,
       jsonLibOpt,
       validateNonDiscriminatedOneOfsOpt,
-      maxSchemasPerFileOpt
+      maxSchemasPerFileOpt,
+      streamingImplementationOpt
     )
       .mapN {
         case (
@@ -96,7 +100,8 @@ object GenScala {
               headTagForNames,
               jsonLib,
               validateNonDiscriminatedOneOfs,
-              maxSchemasPerFile
+              maxSchemasPerFile,
+              streamingImplementation
             ) =>
           val objectName = maybeObjectName.getOrElse(DefaultObjectName)
 
@@ -109,6 +114,7 @@ object GenScala {
                 targetScala3,
                 headTagForNames,
                 jsonLib.getOrElse("circe"),
+                streamingImplementation.getOrElse("fs2"),
                 validateNonDiscriminatedOneOfs,
                 maxSchemasPerFile.getOrElse(400)
               )

openapi-codegen/core/src/main/scala/sttp/tapir/codegen/BasicGenerator.scala

@@ -21,6 +21,10 @@ object JsonSerdeLib extends Enumeration {
   val Circe, Jsoniter, Zio = Value
   type JsonSerdeLib = Value
 }
+object StreamingImplementation extends Enumeration {
+  val Akka, FS2, Pekko, Zio = Value
+  type StreamingImplementation = Value
+}
 
 object BasicGenerator {
 
@@ -34,6 +38,7 @@ object BasicGenerator {
       targetScala3: Boolean,
       useHeadTagForObjectNames: Boolean,
       jsonSerdeLib: String,
+      streamingImplementation: String,
       validateNonDiscriminatedOneOfs: Boolean,
       maxSchemasPerFile: Int
   ): Map[String, String] = {
@@ -47,9 +52,20 @@ object BasicGenerator {
         )
         JsonSerdeLib.Circe
     }
+    val normalisedStreamingImplementation = streamingImplementation.toLowerCase match {
+      case "akka"  => StreamingImplementation.Akka
+      case "fs2"   => StreamingImplementation.FS2
+      case "pekko" => StreamingImplementation.Pekko
+      case "zio"   => StreamingImplementation.Zio
+      case _ =>
+        System.err.println(
+          s"!!! Unrecognised value $streamingImplementation for streaming impl -- should be one of akka, fs2, pekko or zio. Defaulting to fs2 !!!"
+        )
+        StreamingImplementation.FS2
+    }
 
     val EndpointDefs(endpointsByTag, queryOrPathParamRefs, jsonParamRefs, enumsDefinedOnEndpointParams) =
-      endpointGenerator.endpointDefs(doc, useHeadTagForObjectNames, targetScala3, normalisedJsonLib)
+      endpointGenerator.endpointDefs(doc, useHeadTagForObjectNames, targetScala3, normalisedJsonLib, normalisedStreamingImplementation)
     val GeneratedClassDefinitions(classDefns, jsonSerdes, schemas) =
       classGenerator
         .classDefs(

openapi-codegen/core/src/main/scala/sttp/tapir/codegen/EndpointGenerator.scala

@@ -2,6 +2,8 @@ package sttp.tapir.codegen
 import io.circe.Json
 import sttp.tapir.codegen.BasicGenerator.{indent, mapSchemaSimpleTypeToType, strippedToCamelCase}
 import sttp.tapir.codegen.JsonSerdeLib.JsonSerdeLib
+import sttp.tapir.codegen.StreamingImplementation
+import sttp.tapir.codegen.StreamingImplementation.StreamingImplementation
 import sttp.tapir.codegen.openapi.models.OpenapiModels.{OpenapiDocument, OpenapiParameter, OpenapiPath, OpenapiRequestBody, OpenapiResponse}
 import sttp.tapir.codegen.openapi.models.OpenapiSchemaType.{
   OpenapiSchemaAny,
@@ -10,7 +12,8 @@ import sttp.tapir.codegen.openapi.models.OpenapiSchemaType.{
   OpenapiSchemaEnum,
   OpenapiSchemaMap,
   OpenapiSchemaRef,
-  OpenapiSchemaSimpleType
+  OpenapiSchemaSimpleType,
+  OpenapiSchemaString
 }
 import sttp.tapir.codegen.openapi.models.{OpenapiComponent, OpenapiSchemaType, OpenapiSecuritySchemeType, SpecificationExtensionRenderer}
 import sttp.tapir.codegen.util.JavaEscape
@@ -55,12 +58,13 @@ class EndpointGenerator {
       doc: OpenapiDocument,
       useHeadTagForObjectNames: Boolean,
       targetScala3: Boolean,
-      jsonSerdeLib: JsonSerdeLib
+      jsonSerdeLib: JsonSerdeLib,
+      streamingImplementation: StreamingImplementation
   ): EndpointDefs = {
     val components = Option(doc.components).flatten
     val GeneratedEndpoints(endpointsByFile, queryOrPathParamRefs, jsonParamRefs, definesEnumQueryParam) =
       doc.paths
-        .map(generatedEndpoints(components, useHeadTagForObjectNames, targetScala3, jsonSerdeLib))
+        .map(generatedEndpoints(components, useHeadTagForObjectNames, targetScala3, jsonSerdeLib, streamingImplementation))
         .foldLeft(GeneratedEndpoints(Nil, Set.empty, Set.empty, false))(_ merge _)
     val endpointDecls = endpointsByFile.map { case GeneratedEndpointsForFile(k, ge) =>
       val definitions = ge
@@ -84,7 +88,8 @@ class EndpointGenerator {
       components: Option[OpenapiComponent],
       useHeadTagForObjectNames: Boolean,
       targetScala3: Boolean,
-      jsonSerdeLib: JsonSerdeLib
+      jsonSerdeLib: JsonSerdeLib,
+      streamingImplementation: StreamingImplementation
   )(p: OpenapiPath): GeneratedEndpoints = {
     val parameters = components.map(_.parameters).getOrElse(Map.empty)
     val securitySchemes = components.map(_.securitySchemes).getOrElse(Map.empty)
@@ -106,14 +111,15 @@ class EndpointGenerator {
         }
 
         val name = strippedToCamelCase(m.operationId.getOrElse(m.methodType + p.url.capitalize))
-        val (inParams, maybeLocalEnums) = ins(m.resolvedParameters, m.requestBody, name, targetScala3, jsonSerdeLib)
+        val (inParams, maybeLocalEnums) =
+          ins(m.resolvedParameters, m.requestBody, name, targetScala3, jsonSerdeLib, streamingImplementation)
         val definition =
           s"""|endpoint
               |  .${m.methodType}
               |  ${urlMapper(p.url, m.resolvedParameters)}
               |${indent(2)(security(securitySchemes, m.security))}
               |${indent(2)(inParams)}
-              |${indent(2)(outs(m.responses))}
+              |${indent(2)(outs(m.responses, streamingImplementation))}
               |${indent(2)(tags(m.tags))}
               |$attributeString
               |""".stripMargin.linesIterator.filterNot(_.trim.isEmpty).mkString("\n")
@@ -211,7 +217,8 @@ class EndpointGenerator {
       requestBody: Option[OpenapiRequestBody],
       endpointName: String,
       targetScala3: Boolean,
-      jsonSerdeLib: JsonSerdeLib
+      jsonSerdeLib: JsonSerdeLib,
+      streamingImplementation: StreamingImplementation
   )(implicit location: Location): (String, Option[String]) = {
     def getEnumParamDefn(param: OpenapiParameter, e: OpenapiSchemaEnum, isArray: Boolean) = {
       val enumName = endpointName.capitalize + strippedToCamelCase(param.name).capitalize
@@ -267,7 +274,7 @@ class EndpointGenerator {
     val rqBody = requestBody.flatMap { b =>
       if (b.content.isEmpty) None
       else if (b.content.size != 1) bail(s"We can handle only one requestBody content! Saw ${b.content.map(_.contentType)}")
-      else Some(s".in(${contentTypeMapper(b.content.head.contentType, b.content.head.schema, b.required)})")
+      else Some(s".in(${contentTypeMapper(b.content.head.contentType, b.content.head.schema, streamingImplementation, b.required)})")
     }
 
     (params ++ rqBody).mkString("\n") -> maybeEnumDefns.foldLeft(Option.empty[String]) {
@@ -298,7 +305,7 @@ class EndpointGenerator {
   // treats redirects as ok
   private val okStatus = """([23]\d\d)""".r
   private val errorStatus = """([45]\d\d)""".r
-  private def outs(responses: Seq[OpenapiResponse])(implicit location: Location) = {
+  private def outs(responses: Seq[OpenapiResponse], streamingImplementation: StreamingImplementation)(implicit location: Location) = {
     // .errorOut(stringBody)
     // .out(jsonBody[List[Book]])
 
@@ -315,13 +322,13 @@ class EndpointGenerator {
           case content +: Nil =>
             resp.code match {
               case "200" =>
-                s".out(${contentTypeMapper(content.contentType, content.schema)}$d)"
+                s".out(${contentTypeMapper(content.contentType, content.schema, streamingImplementation)}$d)"
               case okStatus(s) =>
-                s".out(${contentTypeMapper(content.contentType, content.schema)}$d.and(statusCode(sttp.model.StatusCode($s))))"
+                s".out(${contentTypeMapper(content.contentType, content.schema, streamingImplementation)}$d.and(statusCode(sttp.model.StatusCode($s))))"
               case "default" =>
-                s".errorOut(${contentTypeMapper(content.contentType, content.schema)}$d)"
+                s".errorOut(${contentTypeMapper(content.contentType, content.schema, streamingImplementation)}$d)"
               case errorStatus(s) =>
-                s".errorOut(${contentTypeMapper(content.contentType, content.schema)}$d.and(statusCode(sttp.model.StatusCode($s))))"
+                s".errorOut(${contentTypeMapper(content.contentType, content.schema, streamingImplementation)}$d.and(statusCode(sttp.model.StatusCode($s))))"
               case x =>
                 bail(s"Statuscode mapping is incomplete! Cannot handle $x")
             }
@@ -333,7 +340,12 @@ class EndpointGenerator {
       .mkString("\n")
   }
 
-  private def contentTypeMapper(contentType: String, schema: OpenapiSchemaType, required: Boolean = true)(implicit location: Location) = {
+  private def contentTypeMapper(
+      contentType: String,
+      schema: OpenapiSchemaType,
+      streamingImplementation: StreamingImplementation,
+      required: Boolean = true
+  )(implicit location: Location) = {
     contentType match {
       case "text/plain" =>
         "stringBody"
@@ -362,6 +374,31 @@ class EndpointGenerator {
             s"multipartBody[$t]"
           case x => bail(s"$contentType only supports schema ref or binary. Found $x")
         }
+      case "application/octet-stream" =>
+        val capability = streamingImplementation match {
+          case StreamingImplementation.Akka  => "sttp.capabilities.akka.AkkaStreams"
+          case StreamingImplementation.FS2   => "sttp.capabilities.fs2.Fs2Streams[cats.effect.IO]"
+          case StreamingImplementation.Pekko => "sttp.capabilities.pekko.PekkoStreams"
+          case StreamingImplementation.Zio   => "sttp.capabilities.zio.ZioStreams"
+        }
+        schema match {
+          case _: OpenapiSchemaString =>
+            s"streamTextBody($capability)(CodecFormat.OctetStream())"
+          case schema =>
+            val outT = schema match {
+              case st: OpenapiSchemaSimpleType =>
+                val (t, _) = mapSchemaSimpleTypeToType(st)
+                t
+              case OpenapiSchemaArray(st: OpenapiSchemaSimpleType, _) =>
+                val (t, _) = mapSchemaSimpleTypeToType(st)
+                s"List[$t]"
+              case OpenapiSchemaMap(st: OpenapiSchemaSimpleType, _) =>
+                val (t, _) = mapSchemaSimpleTypeToType(st)
+                s"Map[String, $t]"
+              case x => bail(s"Can't create this param as output (found $x)")
+            }
+            s"streamBody($capability)(Schema.binary[$outT], CodecFormat.OctetStream())"
+        }
 
       case x => bail(s"Not all content types supported! Found $x")
     }

openapi-codegen/core/src/test/scala/sttp/tapir/codegen/BasicGeneratorSpec.scala

@@ -17,7 +17,8 @@ class BasicGeneratorSpec extends CompileCheckTestBase {
       useHeadTagForObjectNames = useHeadTagForObjectNames,
       jsonSerdeLib = jsonSerdeLib,
       validateNonDiscriminatedOneOfs = true,
-      maxSchemasPerFile = 400
+      maxSchemasPerFile = 400,
+      streamingImplementation = "fs2"
     )
   }
   def gen(

openapi-codegen/core/src/test/scala/sttp/tapir/codegen/ClassDefinitionGeneratorSpec.scala

@@ -391,7 +391,13 @@ class ClassDefinitionGeneratorSpec extends CompileCheckTestBase {
       case Left(value) => throw new Exception(value)
       case Right(doc) =>
         new EndpointGenerator()
-          .endpointDefs(doc, useHeadTagForObjectNames = false, targetScala3 = false, jsonSerdeLib = JsonSerdeLib.Circe)
+          .endpointDefs(
+            doc,
+            useHeadTagForObjectNames = false,
+            targetScala3 = false,
+            jsonSerdeLib = JsonSerdeLib.Circe,
+            streamingImplementation = StreamingImplementation.FS2
+          )
           .endpointDecls(None)
     }
 

openapi-codegen/core/src/test/scala/sttp/tapir/codegen/EndpointGeneratorSpec.scala

@@ -63,7 +63,13 @@ class EndpointGeneratorSpec extends CompileCheckTestBase {
     )
     val generatedCode = BasicGenerator.imports(JsonSerdeLib.Circe) ++
       new EndpointGenerator()
-        .endpointDefs(doc, useHeadTagForObjectNames = false, targetScala3 = false, jsonSerdeLib = JsonSerdeLib.Circe)
+        .endpointDefs(
+          doc,
+          useHeadTagForObjectNames = false,
+          targetScala3 = false,
+          jsonSerdeLib = JsonSerdeLib.Circe,
+          streamingImplementation = StreamingImplementation.FS2
+        )
         .endpointDecls(None)
     generatedCode should include("val getTestAsdId =")
     generatedCode should include(""".in(query[Option[String]]("fgh-id"))""")
@@ -142,7 +148,13 @@ class EndpointGeneratorSpec extends CompileCheckTestBase {
     )
     BasicGenerator.imports(JsonSerdeLib.Circe) ++
       new EndpointGenerator()
-        .endpointDefs(doc, useHeadTagForObjectNames = false, targetScala3 = false, jsonSerdeLib = JsonSerdeLib.Circe)
+        .endpointDefs(
+          doc,
+          useHeadTagForObjectNames = false,
+          targetScala3 = false,
+          jsonSerdeLib = JsonSerdeLib.Circe,
+          streamingImplementation = StreamingImplementation.FS2
+        )
         .endpointDecls(None) shouldCompile ()
   }
 
@@ -188,7 +200,13 @@ class EndpointGeneratorSpec extends CompileCheckTestBase {
     )
     val generatedCode = BasicGenerator.imports(JsonSerdeLib.Circe) ++
       new EndpointGenerator()
-        .endpointDefs(doc, useHeadTagForObjectNames = false, targetScala3 = false, jsonSerdeLib = JsonSerdeLib.Circe)
+        .endpointDefs(
+          doc,
+          useHeadTagForObjectNames = false,
+          targetScala3 = false,
+          jsonSerdeLib = JsonSerdeLib.Circe,
+          streamingImplementation = StreamingImplementation.FS2
+        )
         .endpointDecls(None)
     generatedCode should include(
       """.out(stringBody.description("Processing").and(statusCode(sttp.model.StatusCode(202))))"""
@@ -253,7 +271,8 @@ class EndpointGeneratorSpec extends CompileCheckTestBase {
       useHeadTagForObjectNames = false,
       jsonSerdeLib = "circe",
       validateNonDiscriminatedOneOfs = true,
-      maxSchemasPerFile = 400
+      maxSchemasPerFile = 400,
+      streamingImplementation = "fs2"
     )("TapirGeneratedEndpoints")
     generatedCode should include(
       """file: sttp.model.Part[java.io.File]"""
@@ -274,7 +293,8 @@ class EndpointGeneratorSpec extends CompileCheckTestBase {
       useHeadTagForObjectNames = false,
       jsonSerdeLib = "circe",
       validateNonDiscriminatedOneOfs = true,
-      maxSchemasPerFile = 400
+      maxSchemasPerFile = 400,
+      streamingImplementation = "fs2"
     )("TapirGeneratedEndpoints")
     generatedCode shouldCompile ()
     val expectedAttrDecls = Seq(

openapi-codegen/sbt-plugin/src/main/scala/sttp/tapir/sbt/OpenapiCodegenKeys.scala

@@ -2,6 +2,18 @@ package sttp.tapir.sbt
 
 import sbt._
 
+case class OpenApiConfiguration(
+    swaggerFile: File,
+    packageName: String,
+    objectName: String,
+    useHeadTagForObjectName: Boolean,
+    jsonSerdeLib: String,
+    streamingImplementation: String,
+    validateNonDiscriminatedOneOfs: Boolean,
+    maxSchemasPerFile: Int,
+    additionalPackages: List[(String, File)]
+)
+
 trait OpenapiCodegenKeys {
   lazy val openapiSwaggerFile = settingKey[File]("The swagger file with the api definitions.")
   lazy val openapiPackage = settingKey[String]("The name for the generated package.")
@@ -13,7 +25,10 @@ trait OpenapiCodegenKeys {
   lazy val openapiValidateNonDiscriminatedOneOfs =
     settingKey[Boolean]("Whether to fail if variants of a oneOf without a discriminator cannot be disambiguated..")
   lazy val openapiMaxSchemasPerFile = settingKey[Int]("Maximum number of schemas to generate for a single file")
-  lazy val openapiAdditionalPackages = taskKey[List[(String, File)]]("Addition package -> spec mappings to generate.")
+  lazy val openapiAdditionalPackages = settingKey[List[(String, File)]]("Addition package -> spec mappings to generate.")
+  lazy val openapiStreamingImplementation = settingKey[String]("Implementation for streamTextBody. Supports: akka, fs2, pekko, zio.")
+  lazy val openapiOpenApiConfiguration =
+    settingKey[OpenApiConfiguration]("Aggregation of other settings. Manually set value will be disregarded.")
 
   lazy val generateTapirDefinitions = taskKey[Unit]("The task that generates tapir definitions based on the input swagger file.")
 }