streaming functions for doobie and slick with tests

.github/workflows/build.yml

@@ -51,8 +51,10 @@ jobs:
         with:
           path: |
             ${{ github.workspace }}/core/target/scala-${{ matrix.scalaShort }}/jacoco/report/jacoco.xml
+            ${{ github.workspace }}/streaming/target/scala-${{ matrix.scalaShort }}/jacoco/report/jacoco.xml
             ${{ github.workspace }}/examples/target/scala-${{ matrix.scalaShort }}/jacoco/report/jacoco.xml
             ${{ github.workspace }}/slick/target/scala-${{ matrix.scalaShort }}/jacoco/report/jacoco.xml
+            ${{ github.workspace }}/slick-streaming/target/scala-${{ matrix.scalaShort }}/jacoco/report/jacoco.xml
             ${{ github.workspace }}/doobie/target/scala-${{ matrix.scalaShort }}/jacoco/report/jacoco.xml
           key: ${{ runner.os }}-${{ matrix.scalaShort }}-${{ hashFiles('**/jacoco.xml') }}
 
@@ -79,8 +81,10 @@ jobs:
         with:
           path: |
             ${{ github.workspace }}/core/target/scala-${{ matrix.scalaShort }}/jacoco/report/jacoco.xml
+            ${{ github.workspace }}/streaming/target/scala-${{ matrix.scalaShort }}/jacoco/report/jacoco.xml
             ${{ github.workspace }}/examples/target/scala-${{ matrix.scalaShort }}/jacoco/report/jacoco.xml
             ${{ github.workspace }}/slick/target/scala-${{ matrix.scalaShort }}/jacoco/report/jacoco.xml
+            ${{ github.workspace }}/slick-streaming/target/scala-${{ matrix.scalaShort }}/jacoco/report/jacoco.xml
             ${{ github.workspace }}/doobie/target/scala-${{ matrix.scalaShort }}/jacoco/report/jacoco.xml
           key: ${{ runner.os }}-${{ matrix.scalaShort }}-${{ hashFiles('**/jacoco.xml') }}
       - name: Setup Scala
@@ -93,7 +97,9 @@ jobs:
         with:
           paths: >
             ${{ github.workspace }}/core/target/scala-${{ matrix.scalaShort }}/jacoco/report/jacoco.xml,
+            ${{ github.workspace }}/streaming/target/scala-${{ matrix.scalaShort }}/jacoco/report/jacoco.xml,
             ${{ github.workspace }}/slick/target/scala-${{ matrix.scalaShort }}/jacoco/report/jacoco.xml
+            ${{ github.workspace }}/slick-streaming/target/scala-${{ matrix.scalaShort }}/jacoco/report/jacoco.xml
             ${{ github.workspace }}/doobie/target/scala-${{ matrix.scalaShort }}/jacoco/report/jacoco.xml
           # examples don't need code coverage - at least not now
           token: ${{ secrets.GITHUB_TOKEN }}

README.md

@@ -14,6 +14,8 @@ ___
 - [Usage](#usage)
 - [Concepts](#concepts)
 - [Slick module](#slick-module)
+- [Slick streaming module](#slick-streaming-module)
+- [Doobie module](#doobie-module)
 - [Testing](#testing)
 - [How to Release](#how-to-release)
 <!-- tocstop -->
@@ -54,10 +56,12 @@ Currently, the library is developed with Postgres as the target DB. But the appr
 
 #### Sbt
 
-Import one of the two available module at the moment. Slick module works with Scala Futures. Doobie module works with any effect type (typically IO or ZIO) provided cats effect's Async instance is available.
+Slick module works with Scala Futures. Slick-streaming provides also streaming support for Slick with fs2 streams and any effect type provided cats effect's Async instance is available. 
+Doobie module provides both non-streaming and streaming functions and works with any effect type (typically IO or ZIO) provided cats effect's Async instance is available.
 
 ```scala
 libraryDependencies *= "za.co.absa.fa-db" %% "slick"  % "X.Y.Z"
+libraryDependencies *= "za.co.absa.fa-db" %% "slick-streaming"  % "X.Y.Z"
 libraryDependencies *= "za.co.absa.fa-db" %% "doobie"  % "X.Y.Z"
 ```
 
@@ -67,7 +71,9 @@ libraryDependencies *= "za.co.absa.fa-db" %% "doobie"  % "X.Y.Z"
 
 Modules:
 * Core [![Maven Central](https://maven-badges.herokuapp.com/maven-central/za.co.absa.fa-db/core_2.12/badge.svg)](https://maven-badges.herokuapp.com/maven-central/za.co.absa.fa-db/core_2.12)
+* Streaming [![Maven Central](https://maven-badges.herokuapp.com/maven-central/za.co.absa.fa-db/streaming_2.12/badge.svg)](https://maven-badges.herokuapp.com/maven-central/za.co.absa.fa-db/streaming_2.12)
 * Slick [![Maven Central](https://maven-badges.herokuapp.com/maven-central/za.co.absa.fa-db/slick_2.12/badge.svg)](https://maven-badges.herokuapp.com/maven-central/za.co.absa.fa-db/slick_2.12)
+* Slick-streaming [![Maven Central](https://maven-badges.herokuapp.com/maven-central/za.co.absa.fa-db/slick-streaming_2.12/badge.svg)](https://maven-badges.herokuapp.com/maven-central/za.co.absa.fa-db/slick-streaming_2.12)
 * Doobie [![Maven Central](https://maven-badges.herokuapp.com/maven-central/za.co.absa.fa-db/doobie_2.12/badge.svg)](https://maven-badges.herokuapp.com/maven-central/za.co.absa.fa-db/doobie_2.12)
 
 ```xml
@@ -76,6 +82,11 @@ Modules:
     <artifactId>slick_2.12</artifactId>
     <version>${latest_version}</version>
 </dependency>
+<dependency>
+    <groupId>za.co.absa.fa-db</groupId>
+    <artifactId>slick-streaming_2.12</artifactId>
+    <version>${latest_version}</version>
+</dependency>
 <dependency>
     <groupId>za.co.absa.fa-db</groupId>
     <artifactId>doobie_2.12</artifactId>
@@ -86,7 +97,9 @@ Modules:
 ### Scala 2.13
 Modules:
 * Core [![Maven Central](https://maven-badges.herokuapp.com/maven-central/za.co.absa.fa-db/core_2.13/badge.svg)](https://maven-badges.herokuapp.com/maven-central/za.co.absa.fa-db/core_2.13)
+* Streaming [![Maven Central](https://maven-badges.herokuapp.com/maven-central/za.co.absa.fa-db/streaming_2.13/badge.svg)](https://maven-badges.herokuapp.com/maven-central/za.co.absa.fa-db/streaming_2.13)
 * Slick [![Maven Central](https://maven-badges.herokuapp.com/maven-central/za.co.absa.fa-db/slick_2.13/badge.svg)](https://maven-badges.herokuapp.com/maven-central/za.co.absa.fa-db/slick_2.13)
+* Slick-streaming [![Maven Central](https://maven-badges.herokuapp.com/maven-central/za.co.absa.fa-db/slick-streaming_2.13/badge.svg)](https://maven-badges.herokuapp.com/maven-central/za.co.absa.fa-db/slick-streaming_2.13)
 * Doobie [![Maven Central](https://maven-badges.herokuapp.com/maven-central/za.co.absa.fa-db/doobie_2.13/badge.svg)](https://maven-badges.herokuapp.com/maven-central/za.co.absa.fa-db/doobie_2.13)
 
 ```xml
@@ -95,6 +108,11 @@ Modules:
     <artifactId>slick_2.13</artifactId>
     <version>${latest_version}</version>
 </dependency>
+<dependency>
+    <groupId>za.co.absa.fa-db</groupId>
+    <artifactId>slick-streaming_2.13</artifactId>
+    <version>${latest_version}</version>
+</dependency>
 <dependency>
     <groupId>za.co.absa.fa-db</groupId>
     <artifactId>doobie_2.13</artifactId>
@@ -141,18 +159,27 @@ val hStore: Option[Map[String, String]] = pr.nextHStoreOption
 val macAddr: Option[MacAddrString] = pr.nextMacAddrOption
 ```
 
+## Slick streaming module
+
+It additionally brings:
+
+* `class SlickPgStreamingEngine` - implementation of _Streaming_'s `DBStreamingEngine` executing the queries via Slick in a streaming fashion
+* `class SlickStreamingResultFunction` - abstract class for DB functions returning sequence of results (fs2.Stream) in a streaming fashion
+
 ## Doobie module
 
 As the name suggests it runs on [Doobie library](https://tpolecat.github.io/doobie/). The main benefit of the module is that it allows to use any effect type (typically IO or ZIO) therefore is more suitable for functional programming. It also brings in the [Doobie-Postgres library](https://tpolecat.github.io/doobie/docs/14-PostgreSQL.html) for extended Postgres type support.
 
 It brings:
 
 * `class DoobieEngine` - implementation of _Core_'s `DBEngine` executing the queries via Doobie. The class is type parameterized with the effect type.
+* `class DoobieStreamingEngine` - implementation of _Streamings_'s `DBStreamingEngine` executing the queries via Doobie in a streaming fashion. The class is type parameterized with the effect type.
 * `class DoobieSingleResultFunction` - abstract class for DB functions returning single result
 * `class DoobieMultipleResultFunction` - abstract class for DB functions returning sequence of results
 * `class DoobieOptionalResultFunction` - abstract class for DB functions returning optional result
 * `class DoobieSingleResultFunctionWithStatus` - abstract class for DB functions with status handling; it requires an implementation of `StatusHandling` to be mixed-in (`StandardStatusHandling` available out-of-the-box)
-
+* `class DoobieStreamingResultFunction` - abstract class for DB functions returning sequence of results (fs2.Stream) in a streaming fashion
+* 
 Since Doobie also interoperates with ZIO, there is an example of how a database connection can be properly established within a ZIO application. Please see [this file](doobie/zio-setup.md) for more details.
 
 ## Testing

build.sbt

@@ -49,7 +49,7 @@ lazy val commonJacocoExcludes: Seq[String] = Seq(
 )
 
 lazy val parent = (project in file("."))
-  .aggregate(faDbCore, faDBSlick, faDBDoobie, faDBExamples)
+  .aggregate(faDbCore, faDBStreaming, faDBSlick, faDBSlickStreaming, faDBDoobie, faDBExamples)
   .settings(
     name := "root",
     libraryDependencies ++= rootDependencies(scalaVersion.value),
@@ -73,21 +73,51 @@ lazy val faDbCore = (project in file("core"))
     jacocoExcludes := commonJacocoExcludes
   )
 
+lazy val faDBStreaming = (project in file("streaming"))
+  .configs(IntegrationTest)
+  .settings(
+    name := "streaming",
+    libraryDependencies ++= streamingDependencies(scalaVersion.value),
+    javacOptions ++= commonJavacOptions,
+    scalacOptions ++= commonScalacOptions,
+    (Compile / compile) := ((Compile / compile) dependsOn printScalaVersion).value,
+    Defaults.itSettings,
+  ).dependsOn(faDbCore)
+  .settings(
+    jacocoReportSettings := commonJacocoReportSettings.withTitle(s"fa-db:streaming Jacoco Report - scala:${scalaVersion.value}"),
+    jacocoExcludes := commonJacocoExcludes
+  )
+
 lazy val faDBSlick = (project in file("slick"))
   .configs(IntegrationTest)
   .settings(
     name := "slick",
     libraryDependencies ++= slickDependencies(scalaVersion.value),
     javacOptions ++= commonJavacOptions,
     scalacOptions ++= commonScalacOptions,
-    (Compile / compile) := ((Compile / compile) dependsOn printScalaVersion).value, // printScalaVersion is run with compile
+    (Compile / compile) := ((Compile / compile) dependsOn printScalaVersion).value,
     Defaults.itSettings,
   ).dependsOn(faDbCore)
   .settings(
     jacocoReportSettings := commonJacocoReportSettings.withTitle(s"fa-db:slick Jacoco Report - scala:${scalaVersion.value}"),
     jacocoExcludes := commonJacocoExcludes
   )
 
+lazy val faDBSlickStreaming = (project in file("slick-streaming"))
+  .configs(IntegrationTest)
+  .settings(
+    name := "slick-streaming",
+    libraryDependencies ++= slickStreamingDependencies(scalaVersion.value),
+    javacOptions ++= commonJavacOptions,
+    scalacOptions ++= commonScalacOptions,
+    (Compile / compile) := ((Compile / compile) dependsOn printScalaVersion).value,
+    Defaults.itSettings,
+  ).dependsOn(faDbCore, faDBStreaming, faDBSlick)
+  .settings(
+    jacocoReportSettings := commonJacocoReportSettings.withTitle(s"fa-db:slick-streaming Jacoco Report - scala:${scalaVersion.value}"),
+    jacocoExcludes := commonJacocoExcludes
+  )
+
 lazy val faDBDoobie = (project in file("doobie"))
   .configs(IntegrationTest)
   .settings(
@@ -96,7 +126,7 @@ lazy val faDBDoobie = (project in file("doobie"))
     javacOptions ++= commonJavacOptions,
     scalacOptions ++= commonScalacOptions,
     Defaults.itSettings,
-  ).dependsOn(faDbCore)
+  ).dependsOn(faDbCore, faDBStreaming)
   .settings(
     jacocoReportSettings := commonJacocoReportSettings.withTitle(s"fa-db:doobie Jacoco Report - scala:${scalaVersion.value}"),
     jacocoExcludes := commonJacocoExcludes
@@ -108,7 +138,7 @@ lazy val faDBExamples = (project in file("examples"))
     name := "examples",
     libraryDependencies ++= examplesDependencies(scalaVersion.value),
     Test / parallelExecution := false,
-    (Compile / compile) := ((Compile / compile) dependsOn printScalaVersion).value, // printScalaVersion is run with compile
+    (Compile / compile) := ((Compile / compile) dependsOn printScalaVersion).value,
     publish / skip := true
   ).dependsOn(faDbCore, faDBSlick)
   .settings(

core/src/main/scala/za/co/absa/fadb/DBEngine.scala

@@ -23,7 +23,7 @@ import za.co.absa.fadb.exceptions.StatusException
 import scala.language.higherKinds
 
 /**
- *  `DBEngine` is an abstract class that represents a database engine.
+ *  [[DBEngine]] is an abstract class that represents a database engine.
  *  It provides methods to execute queries and fetch results from a database.
  *  @tparam F - The type of the context in which the database queries are executed.
  */

core/src/main/scala/za/co/absa/fadb/DBFunction.scala

@@ -22,7 +22,7 @@ import za.co.absa.fadb.status.handling.StatusHandling
 import scala.language.higherKinds
 
 /**
- *  `DBFunction` is an abstract class that represents a database function.
+ *  [[DBFunction]] is an abstract class that represents a database function.
  *  @param functionNameOverride - Optional parameter to override the class name if it does not match the database function name.
  *  @param schema - The schema the function belongs to.
  *  @param dBEngine - The database engine that is supposed to execute the function (contains connection to the database).
@@ -73,7 +73,7 @@ abstract class DBFunction[I, R, E <: DBEngine[F], F[_]](functionNameOverride: Op
 }
 
 /**
- *  `DBFunctionWithStatus` is an abstract class that represents a database function with a status.
+ *  [[DBFunctionWithStatus]] is an abstract class that represents a database function with a status.
  *  It extends the [[DBFunction]] class and adds handling for the status of the function invocation.
  *  @param functionNameOverride - Optional parameter to override the class name if it does not match the database function name.
  *  @param schema - The schema the function belongs to.
@@ -100,7 +100,7 @@ abstract class DBFunctionWithStatus[I, R, E <: DBEngine[F], F[_]](functionNameOv
 
   /**
    *  Executes the database function and returns multiple results.
-   *  @param values
+   *  @param values the values to pass over to the database function.
    *  @return A sequence of results from the database function.
    */
   def apply(values: I): F[Either[StatusException, R]] = dBEngine.runWithStatus(query(values))
@@ -131,7 +131,7 @@ abstract class DBFunctionWithStatus[I, R, E <: DBEngine[F], F[_]](functionNameOv
 object DBFunction {
 
   /**
-   *  `DBMultipleResultFunction` is an abstract class that represents a database function returning multiple results.
+   *  [[DBMultipleResultFunction]] is an abstract class that represents a database function returning multiple results.
    *  It extends the [[DBFunction]] class and overrides the apply method to return a sequence of results.
    */
   abstract class DBMultipleResultFunction[I, R, E <: DBEngine[F], F[_]](
@@ -155,7 +155,7 @@ object DBFunction {
   }
 
   /**
-   *  `DBSingleResultFunction` is an abstract class that represents a database function returning a single result.
+   *  [[DBSingleResultFunction]] is an abstract class that represents a database function returning a single result.
    *  It extends the [[DBFunction]] class and overrides the apply method to return a single result.
    */
   abstract class DBSingleResultFunction[I, R, E <: DBEngine[F], F[_]](
@@ -178,7 +178,7 @@ object DBFunction {
   }
 
   /**
-   *  `DBOptionalResultFunction` is an abstract class that represents a database function returning an optional result.
+   *  [[DBOptionalResultFunction]] is an abstract class that represents a database function returning an optional result.
    *  It extends the [[DBFunction]] class and overrides the apply method to return an optional result.
    */
   abstract class DBOptionalResultFunction[I, R, E <: DBEngine[F], F[_]](

core/src/main/scala/za/co/absa/fadb/DBSchema.scala

@@ -22,7 +22,7 @@ import za.co.absa.fadb.naming.NamingConvention
  *  An abstract class, an ancestor to represent a database schema
  *  The database name of the schema is derived from the class name based on the provided naming convention
  *  @param schemaNameOverride - in case the class name would not match the database schema name, this gives the
- *  @param namingConvention   - the [[za.co.absa.fadb.naming.NamingConvention NamingConvention]]
+ *  @param namingConvention   - the [[za.co.absa.fadb.naming.NamingConvention]]
  *                           prescribing how to convert a class name into a db object name
  */
 abstract class DBSchema(schemaNameOverride: Option[String] = None)(implicit val namingConvention: NamingConvention) {
@@ -46,7 +46,7 @@ abstract class DBSchema(schemaNameOverride: Option[String] = None)(implicit val
 
   /**
    *  Function to convert a class to the associated DB object name, based on the class' name. For transformation from the
-   *  class name to usual db name the schema's [[za.co.absa.fadb.naming.NamingConvention NamingConvention]] is used.
+   *  class name to usual db name the schema's [[za.co.absa.fadb.naming.NamingConvention]] is used.
    *  @param c  - class which name to use to get the DB object name
    *  @return   - the db object name
    */

core/src/main/scala/za/co/absa/fadb/naming/ExplicitNamingRequired.scala

@@ -19,15 +19,15 @@ package za.co.absa.fadb.naming
 import za.co.absa.fadb.exceptions.NamingException
 
 /**
- *  `ExplicitNamingRequired` is a `NamingConvention` that throws a `NamingConvention` for any string.
+ *  [[ExplicitNamingRequired]] is a [[NamingConvention]] that throws a [[za.co.absa.fadb.exceptions.NamingException]] for any string.
  *  This is used when explicit naming is required and no other naming convention should be applied.
  */
 class ExplicitNamingRequired extends NamingConvention {
 
   /**
-   *  Throws a `NamingConvention` with a message indicating that explicit naming is required.
+   *  Throws a [[za.co.absa.fadb.exceptions.NamingException]] with a message indicating that explicit naming is required.
    *  @param original - The original string.
-   *  @return Nothing, as a `NamingException` is always thrown.
+   *  @return Nothing, as a [[za.co.absa.fadb.exceptions.NamingException]] is always thrown.
    */
   override def stringPerConvention(original: String): String = {
     val message = s"No convention for '$original', explicit naming required."
@@ -36,14 +36,14 @@ class ExplicitNamingRequired extends NamingConvention {
 }
 
 /**
- *  `ExplicitNamingRequired.Implicits` provides an implicit `NamingConvention` instance that
- *  throws a `NamingException` for any string.
+ *  [[ExplicitNamingRequired.Implicits]] provides an implicit [[NamingConvention]] instance that
+ *  throws a [[za.co.absa.fadb.exceptions.NamingException]] for any string.
  */
 object ExplicitNamingRequired {
   object Implicits {
 
     /**
-     *  An implicit `NamingConvention` instance that throws a `NamingException` for any string.
+     *  An implicit [[NamingConvention]] instance that throws a [[za.co.absa.fadb.exceptions.NamingException]] for any string.
      */
     implicit val namingConvention: NamingConvention = new ExplicitNamingRequired()
   }

core/src/main/scala/za/co/absa/fadb/naming/LettersCase.scala

@@ -17,7 +17,7 @@
 package za.co.absa.fadb.naming
 
 /**
- *  `LettersCase` is a sealed trait that represents different cases of letters.
+ *  [[LettersCase]] is a sealed trait that represents different cases of letters.
  *  It provides a method to convert a string to the specific case.
  */
 sealed trait LettersCase {
@@ -33,21 +33,21 @@ sealed trait LettersCase {
 object LettersCase {
 
   /**
-   *  `AsIs` is a [[LettersCase]] that leaves strings as they are.
+   *  [[AsIs]] is a [[LettersCase]] that leaves strings as they are.
    */
   case object AsIs extends LettersCase {
     override def convert(s: String): String = s
   }
 
   /**
-   *  `LowerCase` is a [[LettersCase]] that converts strings to lower case.
+   *  [[LowerCase]] is a [[LettersCase]] that converts strings to lower case.
    */
   case object LowerCase extends LettersCase {
     override def convert(s: String): String = s.toLowerCase
   }
 
   /**
-   *  `UpperCase` is a [[LettersCase]] that converts strings to upper case.
+   *  [[UpperCase]] is a [[LettersCase]] that converts strings to upper case.
    */
   case object UpperCase extends LettersCase {
     override def convert(s: String): String = s.toUpperCase