Merge pull request #27 from SwissBorg/scala3

Cross compile to Scala 3

.github/workflows/scala.yml

@@ -11,10 +11,11 @@ on:
 jobs:
   test:
     runs-on: ubuntu-latest
-    name: Java ${{matrix.java}}
+    name: Test with Scala ${{ matrix.scala }} and Java ${{ matrix.java }})
     strategy:
       matrix:
         java: ['temurin:1.11', 'temurin:1.17', 'temurin:1.21']
+        scala: [ 2.13.13, 3.3.3 ]
     steps:
     - uses: actions/checkout@v4
     - name: Setup Scala and Java
@@ -24,4 +25,4 @@ jobs:
     - name: Start Postgres
       run: sh ./scripts/launch-postgres.sh
     - name: Run tests
-      run: sbt +test
+      run: sbt ++${{ matrix.scala }} test

.scalafmt.conf

@@ -9,4 +9,4 @@ rewrite.imports.groups =  [["(?!javax?\\.).*", "(?!scala\\.).*"], ["javax?\\..*"
 // Scala 3
 rewrite.scala3.convertToNewSyntax = true
 rewrite.scala3.removeOptionalBraces = false
-runner.dialectOverride.allowUnderscoreAsTypePlaceholder = true
+runner.dialectOverride.allowUnderscoreAsTypePlaceholder = true

core/src/main/scala/org/apache/pekko/persistence/postgres/db/ExtendedPostgresProfile.scala

@@ -13,7 +13,6 @@ trait ExtendedPostgresProfile
     with PgSearchSupport
     with PgNetSupport
     with PgLTreeSupport
-    with PgJsonSupport
     with array.PgArrayJdbcTypes
     with PgCirceJsonSupport {
 

core/src/main/scala/org/apache/pekko/persistence/postgres/journal/dao/JournalTables.scala

@@ -38,7 +38,7 @@ class FlatJournalTable private[dao] (_tableTag: Tag, journalTableCfg: JournalTab
     message,
     tags,
     metadata
-  ) <> (JournalRow.tupled, JournalRow.unapply)
+  ) <> ((JournalRow.apply _).tupled, JournalRow.unapply)
 
   val ordering: Rep[Long] = column[Long](journalTableCfg.columnNames.ordering, O.AutoInc)
   val persistenceId: Rep[String] =
@@ -69,7 +69,7 @@ class PartitionedJournalTable private (_tableTag: Tag, journalTableCfg: JournalT
     message,
     tags,
     metadata
-  ) <> (JournalRow.tupled, JournalRow.unapply)
+  ) <> ((JournalRow.apply _).tupled, JournalRow.unapply)
 
   val ordering: Rep[Long] = column[Long](journalTableCfg.columnNames.ordering)
   val persistenceId: Rep[String] =
@@ -106,7 +106,7 @@ class JournalMetadataTable(_tableTag: Tag, journalMetadataTableCfg: JournalMetad
     maxSequenceNumber,
     minOrdering,
     maxOrdering
-  ) <> (JournalMetadataRow.tupled, JournalMetadataRow.unapply)
+  ) <> ((JournalMetadataRow.apply _).tupled, JournalMetadataRow.unapply)
 
   val id: Rep[Long] = column[Long](journalMetadataTableCfg.columnNames.id)
   val persistenceId: Rep[String] =

core/src/main/scala/org/apache/pekko/persistence/postgres/query/JournalSequenceActor.scala

@@ -85,7 +85,7 @@ class JournalSequenceActor(readJournalDao: ReadJournalDao, config: JournalSequen
     * @param previousDelay
     *   The last used delay (may change in case failures occur)
     */
-  def receive(
+  private def receive(
       currentMaxOrdering: OrderingId,
       missingByCounter: Map[Int, MissingElements],
       moduloCounter: Int,
@@ -130,7 +130,7 @@ class JournalSequenceActor(readJournalDao: ReadJournalDao, config: JournalSequen
 
   /** This method that implements the "find gaps" algo. It's the meat and main purpose of this actor.
     */
-  def findGaps(
+  private def findGaps(
       elements: Seq[OrderingId],
       currentMaxOrdering: OrderingId,
       missingByCounter: Map[Int, MissingElements],
@@ -186,7 +186,7 @@ class JournalSequenceActor(readJournalDao: ReadJournalDao, config: JournalSequen
     }
   }
 
-  def scheduleQuery(delay: FiniteDuration): Unit = {
+  private def scheduleQuery(delay: FiniteDuration): Unit = {
     timers.startSingleTimer(key = QueryOrderingIdsTimerKey, QueryOrderingIds, delay)
   }
 }

core/src/main/scala/org/apache/pekko/persistence/postgres/query/PostgresReadJournalProvider.scala

@@ -11,7 +11,9 @@ import org.apache.pekko.persistence.query.ReadJournalProvider
 
 class PostgresReadJournalProvider(system: ExtendedActorSystem, config: Config, configPath: String)
     extends ReadJournalProvider {
-  override val scaladslReadJournal = new scaladsl.PostgresReadJournal(config, configPath)(system)
+  override def scaladslReadJournal(): scaladsl.PostgresReadJournal =
+    new scaladsl.PostgresReadJournal(config, configPath)(system)
 
-  override val javadslReadJournal = new javadsl.PostgresReadJournal(scaladslReadJournal)
+  override def javadslReadJournal(): javadsl.PostgresReadJournal =
+    new javadsl.PostgresReadJournal(scaladslReadJournal())
 }

core/src/main/scala/org/apache/pekko/persistence/postgres/snapshot/PostgresSnapshotStore.scala

@@ -65,7 +65,6 @@ class PostgresSnapshotStore(config: Config) extends SnapshotStore {
         snapshotDao.snapshotForMaxSequenceNr(persistenceId, maxSequenceNr)
       case SnapshotSelectionCriteria(maxSequenceNr, maxTimestamp, _, _) =>
         snapshotDao.snapshotForMaxSequenceNrAndMaxTimestamp(persistenceId, maxSequenceNr, maxTimestamp)
-      case _ => Future.successful(None)
     }
 
     result.map(_.map(toSelectedSnapshot))
@@ -88,7 +87,6 @@ class PostgresSnapshotStore(config: Config) extends SnapshotStore {
       snapshotDao.deleteUpToMaxSequenceNr(persistenceId, maxSequenceNr)
     case SnapshotSelectionCriteria(maxSequenceNr, maxTimestamp, _, _) =>
       snapshotDao.deleteUpToMaxSequenceNrAndMaxTimestamp(persistenceId, maxSequenceNr, maxTimestamp)
-    case _ => Future.successful(())
   }
 
   override def postStop(): Unit = {

core/src/main/scala/org/apache/pekko/persistence/postgres/snapshot/dao/SnapshotTables.scala

@@ -30,7 +30,13 @@ trait SnapshotTables {
         _schemaName = snapshotTableCfg.schemaName,
         _tableName = snapshotTableCfg.tableName
       ) {
-    def * = (persistenceId, sequenceNumber, created, snapshot, metadata) <> (SnapshotRow.tupled, SnapshotRow.unapply)
+    def * = (
+      persistenceId,
+      sequenceNumber,
+      created,
+      snapshot,
+      metadata
+    ) <> ((SnapshotRow.apply _).tupled, SnapshotRow.unapply)
 
     val persistenceId: Rep[String] =
       column[String](snapshotTableCfg.columnNames.persistenceId, O.Length(255, varying = true))

core/src/main/scala/org/apache/pekko/persistence/postgres/tag/TagTables.scala

@@ -10,7 +10,7 @@ trait TagTables {
 
   class EventTagTableDefinition(_tableTag: Tag)
       extends Table[EventTag](_tableTag, _schemaName = tagsTableCfg.schemaName, _tableName = tagsTableCfg.tableName) {
-    def * = (id, name) <> (EventTag.tupled, EventTag.unapply)
+    def * = (id, name) <> ((EventTag.apply _).tupled, EventTag.unapply)
 
     val id: Rep[Int] = column[Int](tagsTableCfg.columnNames.id, O.AutoInc)
     val name: Rep[String] = column[String](tagsTableCfg.columnNames.name, O.Length(255, varying = true))

core/src/test/scala/org/apache/pekko/persistence/postgres/SingleActorSystemPerTestSpec.scala

@@ -66,13 +66,13 @@ abstract class SingleActorSystemPerTestSpec(val config: Config)
     closeDb()
   }
 
-  def withActorSystem(f: ActorSystem => Unit): Unit = {
+  def withActorSystem[A](f: ActorSystem => A): Unit = {
     implicit val system: ActorSystem = ActorSystem("test", config)
     f(system)
     system.terminate().futureValue
   }
 
-  def withActorSystem(config: Config = config)(f: ActorSystem => Unit): Unit = {
+  def withActorSystem[A](config: Config = config)(f: ActorSystem => A): Unit = {
     implicit val system: ActorSystem = ActorSystem("test", config)
     f(system)
     system.terminate().futureValue

core/src/test/scala/org/apache/pekko/persistence/postgres/query/JournalSequenceActorTest.scala

@@ -12,17 +12,14 @@ import org.apache.pekko.persistence.postgres.{JournalRow, SharedActorSystemTestS
 import org.apache.pekko.persistence.postgres.config.JournalSequenceRetrievalConfig
 import org.apache.pekko.persistence.postgres.db.ExtendedPostgresProfile
 import org.apache.pekko.persistence.postgres.query.JournalSequenceActor.{GetMaxOrderingId, MaxOrderingId}
-import org.apache.pekko.persistence.postgres.query.dao.{
-  FlatReadJournalDao,
-  PartitionedReadJournalDao,
-  TestProbeReadJournalDao
-}
+import org.apache.pekko.persistence.postgres.query.dao.{FlatReadJournalDao, PartitionedReadJournalDao, TestProbeReadJournalDao}
 import org.apache.pekko.persistence.postgres.tag.{CachedTagIdResolver, SimpleTagDao}
 import org.apache.pekko.persistence.postgres.util.Schema.{NestedPartitions, Partitioned, Plain, SchemaType}
 import org.apache.pekko.serialization.SerializationExtension
 import org.apache.pekko.stream.{Materializer, SystemMaterializer}
 import org.apache.pekko.stream.scaladsl.{Sink, Source}
 import org.apache.pekko.testkit.TestProbe
+import org.scalatest.concurrent.PatienceConfiguration.{Interval, Timeout}
 import org.scalatest.time.Span
 import org.slf4j.LoggerFactory
 import slick.jdbc
@@ -88,11 +85,10 @@ abstract class JournalSequenceActorTest(val schemaType: SchemaType) extends Quer
 
           val startTime = System.currentTimeMillis()
           withJournalSequenceActor(db, maxTries = 100) { actor =>
-            val patienceConfig = PatienceConfig(10.seconds, Span(200, org.scalatest.time.Millis))
-            eventually {
+            eventually(Timeout(10.seconds), Interval(Span(200, org.scalatest.time.Millis))) {
               val currentMax = actor.ask(GetMaxOrderingId).mapTo[MaxOrderingId].futureValue.maxOrdering
               currentMax shouldBe elements
-            }(patienceConfig, implicitly, implicitly)
+            }
           }
           val timeTaken = System.currentTimeMillis() - startTime
           log.info(s"Recovered all events in $timeTaken ms")
@@ -124,11 +120,10 @@ abstract class JournalSequenceActorTest(val schemaType: SchemaType) extends Quer
 
           withJournalSequenceActor(db, maxTries = 2) { actor =>
             // Should normally recover after `maxTries` seconds
-            val patienceConfig = PatienceConfig(10.seconds, Span(200, org.scalatest.time.Millis))
-            eventually {
+            eventually(Timeout(10.seconds), Interval(Span(200, org.scalatest.time.Millis))) {
               val currentMax = actor.ask(GetMaxOrderingId).mapTo[MaxOrderingId].futureValue.maxOrdering
               currentMax shouldBe lastElement
-            }(patienceConfig, implicitly, implicitly)
+            }
           }
         }
       }
@@ -159,11 +154,10 @@ abstract class JournalSequenceActorTest(val schemaType: SchemaType) extends Quer
 
           withJournalSequenceActor(db, maxTries = 2) { actor =>
             // The actor should assume the max after 2 seconds
-            val patienceConfig = PatienceConfig(3.seconds)
-            eventually {
+            eventually(Timeout(3.seconds)) {
               val currentMax = actor.ask(GetMaxOrderingId).mapTo[MaxOrderingId].futureValue.maxOrdering
               currentMax shouldBe highestValue
-            }(patienceConfig, implicitly, implicitly)
+            }
           }
         }
       }

project/Dependencies.scala

@@ -2,7 +2,8 @@ import sbt.*
 
 object Dependencies {
   val Scala213 = "2.13.13"
-  val ScalaVersions = Seq(Scala213)
+  val Scala3 = "3.3.3"
+  val ScalaVersions = Seq(Scala213, Scala3)
 
   val PekkoVersion = "1.0.2"
   val ScaffeineVersion = "5.2.1"

project/ProjectAutoPlugin.scala

@@ -38,18 +38,24 @@ object ProjectAutoPlugin extends AutoPlugin {
       "-language:implicitConversions",
       "-language:reflectiveCalls",
       "-language:higherKinds",
-      "-release:11",
-      "-Xsource:3"
-    ),
+      "-release:11"
+    ) ++ (CrossVersion.partialVersion(scalaVersion.value) match {
+      case Some((2, 13)) =>
+        List(
+          "-Xsource:3"
+        )
+      case Some((3, _)) =>
+        List("-source:3.0-migration")
+      case _ =>
+        Nil
+    }),
     Compile / doc / scalacOptions := scalacOptions.value ++ Seq(
       "-doc-title",
       "Pekko Persistence Postgres",
       "-doc-version",
       version.value,
       "-sourcepath",
       (ThisBuild / baseDirectory).value.toString,
-      "-skip-packages",
-      "pekko.pattern", // for some reason Scaladoc creates this
       "-doc-source-url", {
         val branch = if (isSnapshot.value) "main" else s"v${version.value}"
         s"https://github.com/SwissBorg/pekko-persistence-postgres/tree/${branch}€{FILE_PATH_EXT}#L€{FILE_LINE}"