Add PekkoStream

playframework · Aug 18, 2023 · 8e039d1 · 8e039d1
1 parent 7b2bbea
commit 8e039d1
Show file tree

Hide file tree

Showing 6 changed files with 415 additions and 1 deletion.
diff --git a/build.sbt b/build.sbt
@@ -255,6 +255,8 @@ lazy val akkaVer = Def.setting[String] {
   }
 }
 
+lazy val pekkoVer = Def.setting[String]("1.0.1")
+
 val akkaContribVer = Def.setting[String] {
   if (akkaVer.value.startsWith("2.5")) "0.11+4-91b2f9fa"
   else "0.10"
@@ -300,6 +302,42 @@ lazy val `anorm-akka` = (project in file("akka"))
   )
   .dependsOn(`anorm-core`)
 
+lazy val `anorm-pekko` = (project in file("pekko"))
+  .settings(
+    mimaPreviousArtifacts := {
+      if (scalaBinaryVersion.value == "3") {
+        Set.empty
+      } else {
+        mimaPreviousArtifacts.value
+      }
+    },
+    libraryDependencies ++= Seq("pekko-testkit", "pekko-stream").map { m =>
+      ("org.apache.pekko" %% m % pekkoVer.value % Provided).exclude("org.scala-lang.modules", "*")
+    },
+    libraryDependencies ++= Seq(
+      acolyte,
+      "org.scala-lang.modules" %% "scala-xml" % xmlVer.value % Test
+    ) ++ specs2Test,
+    scalacOptions ++= {
+      if (scalaBinaryVersion.value == "3") {
+        Seq("-Wconf:cat=deprecation&msg=.*(onDownstreamFinish|ActorMaterializer).*:s")
+      } else {
+        Seq("-P:silencer:globalFilters=deprecated")
+      }
+    },
+    Test / unmanagedSourceDirectories ++= {
+      CrossVersion.partialVersion(scalaVersion.value) match {
+        case Some((2, n)) if n < 13 =>
+          Seq((Test / sourceDirectory).value / "scala-2.13-")
+
+        case _ =>
+          Seq((Test / sourceDirectory).value / "scala-2.13+")
+
+      }
+    }
+  )
+  .dependsOn(`anorm-core`)
+
 // ---
 
 lazy val pgVer = sys.env.get("POSTGRES_VERSION").getOrElse("42.6.0")
@@ -356,7 +394,7 @@ lazy val `anorm-enumeratum` = (project in file("enumeratum"))
 
 lazy val `anorm-parent` = (project in file("."))
   .enablePlugins(ScalaUnidocPlugin)
-  .aggregate(`anorm-tokenizer`, `anorm-core`, `anorm-iteratee`, `anorm-akka`, `anorm-postgres`, `anorm-enumeratum`)
+  .aggregate(`anorm-tokenizer`, `anorm-core`, `anorm-iteratee`, `anorm-akka`, `anorm-pekko`, `anorm-postgres`, `anorm-enumeratum`)
   .settings(
     mimaPreviousArtifacts := Set.empty,
     (Compile / headerSources) ++=

diff --git a/pekko/src/main/scala/anorm/PekkoStream.scala b/pekko/src/main/scala/anorm/PekkoStream.scala
@@ -0,0 +1,195 @@
+/*
+ * Copyright (C) from 2022 The Play Framework Contributors <https://github.com/playframework>, 2011-2021 Lightbend Inc. <https://www.lightbend.com>
+ */
+
+package anorm
+
+import java.sql.Connection
+
+import scala.util.control.NonFatal
+
+import scala.concurrent.{ Future, Promise }
+
+import org.apache.pekko.stream.Materializer
+import org.apache.pekko.stream.scaladsl.{ Source }
+
+/**
+ * Anorm companion for the [[http://doc.akka.io/docs/akka/2.4.4/scala/stream/]].
+ *
+ * @define materialization It materializes a [[scala.concurrent.Future]] of [[scala.Int]] containing the number of rows read from the source upon completion, and a possible exception if row parsing failed.
+ * @define sqlParam the SQL query
+ * @define materializerParam the stream materializer
+ * @define connectionParam the JDBC connection, which must not be closed until the source is materialized.
+ * @define columnAliaserParam the column aliaser
+ */
+object PekkoStream {
+
+  /**
+   * Returns the rows parsed from the `sql` query as a reactive source.
+   *
+   * $materialization
+   *
+   * @tparam T the type of the result elements
+   * @param sql $sqlParam
+   * @param parser the result (row) parser
+   * @param as $columnAliaserParam
+   * @param m $materializerParam
+   * @param connection $connectionParam
+   *
+   * {{{
+   * import java.sql.Connection
+   *
+   * import scala.concurrent.Future
+   *
+   * import org.apache.pekko.stream.Materializer
+   * import org.apache.pekko.stream.scaladsl.Source
+   *
+   * import anorm._
+   *
+   * def resultSource(implicit m: Materializer, con: Connection): Source[String, Future[Int]] = PekkoStream.source(SQL"SELECT * FROM Test", SqlParser.scalar[String], ColumnAliaser.empty)
+   * }}}
+   */
+  @SuppressWarnings(Array("UnusedMethodParameter"))
+  def source[T](sql: => Sql, parser: RowParser[T], as: ColumnAliaser)(implicit
+      @deprecated("No longer required (will be removed)", "2.5.4") m: Materializer,
+      con: Connection
+  ): Source[T, Future[Int]] = Source.fromGraph(new ResultSource[T](con, sql, as, parser))
+
+  /**
+   * Returns the rows parsed from the `sql` query as a reactive source.
+   *
+   * $materialization
+   *
+   * @tparam T the type of the result elements
+   * @param sql $sqlParam
+   * @param parser the result (row) parser
+   * @param m $materializerParam
+   * @param connection $connectionParam
+   */
+  @SuppressWarnings(Array("UnusedMethodParameter"))
+  def source[T](sql: => Sql, parser: RowParser[T])(implicit
+      @deprecated("No longer required (will be removed)", "2.5.4") m: Materializer,
+      con: Connection
+  ): Source[T, Future[Int]] = source[T](sql, parser, ColumnAliaser.empty)
+
+  /**
+   * Returns the result rows from the `sql` query as an enumerator.
+   * This is equivalent to `source[Row](sql, RowParser.successful, as)`.
+   *
+   * $materialization
+   *
+   * @param sql $sqlParam
+   * @param as $columnAliaserParam
+   * @param m $materializerParam
+   * @param connection $connectionParam
+   */
+  def source(sql: => Sql, as: ColumnAliaser)(implicit
+      m: Materializer,
+      connnection: Connection
+  ): Source[Row, Future[Int]] = source(sql, RowParser.successful, as)
+
+  /**
+   * Returns the result rows from the `sql` query as an enumerator.
+   * This is equivalent to
+   * `source[Row](sql, RowParser.successful, ColumnAliaser.empty)`.
+   *
+   * $materialization
+   *
+   * @param sql $sqlParam
+   * @param m $materializerParam
+   * @param connection $connectionParam
+   */
+  def source(sql: => Sql)(implicit m: Materializer, connnection: Connection): Source[Row, Future[Int]] =
+    source(sql, RowParser.successful, ColumnAliaser.empty)
+
+  // Internal stages
+
+  import scala.util.{ Failure, Success }
+  import java.sql.ResultSet
+  import org.apache.pekko.stream.{ Attributes, Outlet, SourceShape }
+  import org.apache.pekko.stream.stage.{ GraphStageWithMaterializedValue, GraphStageLogic, OutHandler }
+
+  private[anorm] class ResultSource[T](connection: Connection, sql: Sql, as: ColumnAliaser, parser: RowParser[T])
+      extends GraphStageWithMaterializedValue[SourceShape[T], Future[Int]] {
+
+    private[anorm] var resultSet: ResultSet = _
+
+    override val toString     = "AnormQueryResult"
+    val out: Outlet[T]        = Outlet(s"${toString}.out")
+    val shape: SourceShape[T] = SourceShape(out)
+
+    override def createLogicAndMaterializedValue(inheritedAttributes: Attributes): (GraphStageLogic, Future[Int]) = {
+      val result = Promise[Int]()
+
+      val logic = new GraphStageLogic(shape) with OutHandler {
+        private var cursor: Option[Cursor] = None
+        private var counter: Int           = 0
+
+        private def failWith(cause: Throwable): Unit = {
+          result.failure(cause)
+          fail(out, cause)
+          ()
+        }
+
+        override def preStart(): Unit = {
+          try {
+            resultSet = sql.unsafeResultSet(connection)
+            nextCursor()
+          } catch {
+            case NonFatal(cause) => failWith(cause)
+          }
+        }
+
+        override def postStop() = release()
+
+        private def release(): Unit = {
+          val stmt: Option[java.sql.Statement] = {
+            if (resultSet != null && !resultSet.isClosed) {
+              val s = resultSet.getStatement
+              resultSet.close()
+              Option(s)
+            } else None
+          }
+
+          stmt.foreach { s =>
+            if (!s.isClosed) s.close()
+          }
+        }
+
+        private def nextCursor(): Unit = {
+          cursor = Sql.unsafeCursor(resultSet, sql.resultSetOnFirstRow, as)
+        }
+
+        def onPull(): Unit = cursor match {
+          case Some(c) =>
+            c.row.as(parser) match {
+              case Success(parsed) => {
+                counter += 1
+                push(out, parsed)
+                nextCursor()
+              }
+
+              case Failure(cause) =>
+                failWith(cause)
+            }
+
+          case _ => {
+            result.success(counter)
+            complete(out)
+          }
+        }
+
+        override def onDownstreamFinish() = {
+          result.tryFailure(new InterruptedException("Downstream finished"))
+          release()
+          super.onDownstreamFinish()
+        }
+
+        setHandler(out, this)
+      }
+
+      logic -> result.future
+    }
+  }
+
+}
diff --git a/pekko/src/test/resources/reference.conf b/pekko/src/test/resources/reference.conf
@@ -0,0 +1,3 @@
+akka {
+  loglevel = "OFF"
+}
diff --git a/pekko/src/test/scala-2.13+/PekkoCompat.scala b/pekko/src/test/scala-2.13+/PekkoCompat.scala
@@ -0,0 +1,9 @@
+/*
+ * Copyright (C) from 2022 The Play Framework Contributors <https://github.com/playframework>, 2011-2021 Lightbend Inc. <https://www.lightbend.com>
+ */
+
+package anorm
+
+private[anorm] object AkkaCompat {
+  type Seq[T] = _root_.scala.collection.immutable.Seq[T]
+}
diff --git a/pekko/src/test/scala-2.13-/PekkoCompat.scala b/pekko/src/test/scala-2.13-/PekkoCompat.scala
@@ -0,0 +1,9 @@
+/*
+ * Copyright (C) from 2022 The Play Framework Contributors <https://github.com/playframework>, 2011-2021 Lightbend Inc. <https://www.lightbend.com>
+ */
+
+package anorm
+
+private[anorm] object AkkaCompat {
+  type Seq[T] = _root_.scala.collection.Seq[T]
+}