DRAFT: Scala3 cross-compilation;

build.sbt

@@ -7,17 +7,35 @@ lazy val commonSettings = Seq(
   organizationHomepage := Some(url("https://evolution.com")),
   homepage := Some(url("https://github.com/evolution-gaming/kafka-journal")),
   startYear := Some(2018),
-  crossScalaVersions := Seq("2.13.14"),
+  crossScalaVersions := Seq("2.13.14", "3.3.3"),
   scalaVersion := crossScalaVersions.value.head,
-  scalacOptions ++= Seq("-release:17", "-deprecation", "-Xsource:3"),
+  scalacOptions ++= {
+    CrossVersion.partialVersion(scalaVersion.value) match {
+      case Some((3, _))  => Seq("-release:17", "-Ykind-projector")
+      case Some((2, 13)) => Seq("-release:17", "-Xsource:3", "-Ytasty-reader")
+      case _             => Seq("-deprecation")
+    }
+  },
+  ThisBuild / dependencyOverrides ++= Seq(
+    "org.scala-lang.modules" %% "scala-collection-compat" % "2.12.0",
+    "org.scala-lang.modules" %% "scala-java8-compat"      % "1.0.2",
+  ),
+  scalacOptions += "-deprecation",
   Compile / doc / scalacOptions ++= Seq("-groups", "-implicits", "-no-link-warnings"),
   Compile / doc / scalacOptions -= "-Xfatal-warnings",
   publishTo := Some(Resolver.evolutionReleases),
   licenses := Seq(("MIT", url("https://opensource.org/licenses/MIT"))),
   releaseCrossBuild := true,
   // explanation of flags: https://www.scalatest.org/user_guide/using_the_runner (`Configuring reporters` section)
   Test / testOptions ++= Seq(Tests.Argument(TestFrameworks.ScalaTest, "-oUDNCXEHLOPQRM")),
-  libraryDependencies += compilerPlugin(`kind-projector` cross CrossVersion.full),
+  libraryDependencies ++= {
+    CrossVersion.partialVersion(scalaVersion.value) match {
+      case Some((3, _)) => Seq.empty
+      case Some((2, 13)) =>
+        Seq(compilerPlugin(`kind-projector` cross CrossVersion.full))
+      case _ => Seq.empty
+    }
+  },
   libraryDependencySchemes ++= Seq(
     "org.scala-lang.modules" %% "scala-java8-compat" % "always",
     "org.scala-lang.modules" %% "scala-xml"          % "always",
@@ -91,11 +109,9 @@ lazy val core = project
       `play-json`,
       `play-json-jsoniter`,
       scassandra,
-      hostname,
       Cats.core,
       Cats.effect,
-      Scodec.core,
-      Scodec.bits,
+      Scodec.core(scalaVersion.value),
     ),
   )
 
@@ -118,7 +134,6 @@ lazy val journal = project
       `cats-helper`,
       `play-json`,
       `play-json-jsoniter`,
-      hostname,
       `cassandra-driver`,
       scassandra,
       scache,
@@ -130,8 +145,7 @@ lazy val journal = project
       sstream,
       Cats.core,
       Cats.effect,
-      Scodec.core,
-      Scodec.bits,
+      Scodec.core(scalaVersion.value),
       `resource-pool`,
       Logback.core    % Test,
       Logback.classic % Test,
@@ -153,7 +167,6 @@ lazy val persistence = project
       `akka-serialization`,
       `cats-helper`,
       Akka.persistence,
-      `akka-test-actor` % Test,
     ),
   )
 
@@ -173,8 +186,7 @@ lazy val `tests` = project
   .settings(
     libraryDependencies ++= Seq(
       `cats-helper`,
-      Kafka.kafka                % Test,
-      `kafka-launcher`           % Test,
+     `testcontainers-kafka`      % Test,
       `testcontainers-cassandra` % Test,
       scalatest                  % Test,
       Akka.`persistence-tck`     % Test,

cassandra/src/main/scala/com/evolutiongaming/kafka/journal/cassandra/CassandraConsistencyConfig.scala

@@ -1,8 +1,7 @@
 package com.evolutiongaming.kafka.journal.cassandra
 
 import com.datastax.driver.core.ConsistencyLevel
-import pureconfig.ConfigReader
-import pureconfig.generic.semiauto.deriveReader
+import pureconfig.{ConfigCursor, ConfigReader}
 
 final case class CassandraConsistencyConfig(
   read: CassandraConsistencyConfig.Read   = CassandraConsistencyConfig.Read.default,
@@ -11,7 +10,17 @@ final case class CassandraConsistencyConfig(
 
 object CassandraConsistencyConfig {
 
-  implicit val configReaderConsistencyConfig: ConfigReader[CassandraConsistencyConfig] = deriveReader
+  implicit val configReaderConsistencyConfig: ConfigReader[CassandraConsistencyConfig] = new ConfigReader[CassandraConsistencyConfig] {
+    override def from(cur: ConfigCursor): ConfigReader.Result[CassandraConsistencyConfig] = {
+      for {
+        objCur <- cur.asObjectCursor
+        readCur = objCur.atKeyOrUndefined("read")
+        read <- Read.configReaderRead.from(readCur)
+        writeCur = objCur.atKeyOrUndefined("write")
+        write = if (writeCur.isUndefined) CassandraConsistencyConfig.Write.default else (Write.configReaderWrite.from(writeCur))
+      } yield CassandraConsistencyConfig(read, write)
+    }
+  }
 
   val default: CassandraConsistencyConfig = CassandraConsistencyConfig()
 

core/src/main/scala/com/evolutiongaming/kafka/journal/FromConfigReaderResult.scala

@@ -5,9 +5,11 @@ import com.evolutiongaming.catshelper.ApplicativeThrowable
 import pureconfig.ConfigReader
 import pureconfig.error.ConfigReaderException
 
+import scala.reflect.ClassTag
+
 trait FromConfigReaderResult[F[_]] {
 
-  def apply[A](a: ConfigReader.Result[A]): F[A]
+  def apply[A: ClassTag](a: ConfigReader.Result[A]): F[A]
 }
 
 object FromConfigReaderResult {
@@ -16,7 +18,7 @@ object FromConfigReaderResult {
 
   implicit def lift[F[_]: ApplicativeThrowable]: FromConfigReaderResult[F] = {
     new FromConfigReaderResult[F] {
-      def apply[A](a: ConfigReader.Result[A]) = {
+      def apply[A: ClassTag](a: ConfigReader.Result[A]): F[A] = {
         a.fold(a => ConfigReaderException(a).raiseError[F, A], _.pure[F])
       }
     }

core/src/main/scala/com/evolutiongaming/kafka/journal/HostName.scala

@@ -12,10 +12,68 @@ object HostName {
   def of[F[_]: Sync](): F[Option[HostName]] = {
     Sync[F].delay {
       for {
-        a <- com.evolutiongaming.hostname.HostName()
+        a <- EvoHostName()
       } yield {
         HostName(a)
       }
     }
   }
 }
+
+import java.net.InetAddress
+import java.util.concurrent.Executors
+
+import scala.concurrent.duration._
+import scala.concurrent.{Await, ExecutionContext, Future}
+import scala.sys.process.*
+import scala.util.Properties
+import scala.util.control.NonFatal
+
+/** Copied from https://github.com/evolution-gaming/hostname */
+object EvoHostName {
+
+  def apply(): Option[String] = {
+
+    val os = env("os.name").map(_.toLowerCase)
+
+    {
+      if (os contains "win") win()
+      else if ((os contains "nix") || (os contains "nux") || (os contains "mac")) unix()
+      else unix() orElse win()
+    } orElse {
+      inetAddress()
+    }
+  }
+
+  private def inetAddress() = {
+    val service     = Executors.newSingleThreadExecutor()
+    implicit val ec = ExecutionContext.fromExecutor(service)
+    val future = Future {
+      InetAddress.getLocalHost.getHostName
+    }
+    future.onComplete { _ => service.shutdown() }
+    safe { Await.result(future, 1.second) }.filter(str => str != "localhost")
+  }
+
+  private def win() =
+    env("COMPUTERNAME") orElse
+      exec("hostname")
+
+  private def unix() =
+    env("HOSTNAME") orElse
+      exec("hostname") orElse
+      env("gethostname") orElse
+      exec("cat /etc/hostname")
+
+  private def exec(name: String) = safe { name.!! }
+
+  private def env(name: String) = Properties.envOrNone(name)
+
+  private def safe(f: => String) = {
+    try {
+      Option(f.trim).filter(_.nonEmpty)
+    } catch {
+      case NonFatal(_) => None
+    }
+  }
+}

core/src/main/scala/com/evolutiongaming/kafka/journal/LogOfFromAkka.scala

@@ -22,7 +22,7 @@ object LogOfFromAkka {
 
       def apply(source: String) = log(source)
 
-      def apply(source: Class[_]) = log(source)
+      def apply(source: Class[?]) = log(source)(LogSource.fromClass)
     }
   }
 }

core/src/main/scala/com/evolutiongaming/kafka/journal/Payload.scala

@@ -3,9 +3,9 @@ package com.evolutiongaming.kafka.journal
 import com.evolutiongaming.kafka.journal.util.ScodecHelper.*
 import play.api.libs.functional.syntax.*
 import play.api.libs.json.*
-import scodec.bits.ByteVector
+import scodec.bits.{BitVector, ByteVector}
 import scodec.codecs.{bytes, utf8}
-import scodec.{Codec, TransformSyntax}
+import scodec.{Attempt, Codec, DecodeResult, SizeBound}
 
 import scala.util.Try
 
@@ -39,7 +39,13 @@ object Payload {
 
     val empty: Binary = Binary(ByteVector.empty)
 
-    implicit val codecBinary: Codec[Binary] = bytes.as[Binary]
+    implicit val codecBinary: Codec[Binary] = new Codec[Binary] {
+      override def decode(bits: BitVector): Attempt[DecodeResult[Binary]] = bytes.decode(bits).map(_.map(Binary(_)))
+
+      override def encode(value: Binary): Attempt[BitVector] = bytes.encode(value.value)
+
+      override def sizeBound: SizeBound = bytes.sizeBound
+    }
 
   }
 
@@ -53,7 +59,13 @@ object Payload {
 
   object Text {
 
-    implicit val codecText: Codec[Text] = utf8.as[Text]
+    implicit val codecText: Codec[Text] = new Codec[Text] {
+      override def decode(bits: BitVector): Attempt[DecodeResult[Text]] = utf8.decode(bits).map(_.map(Text(_)))
+
+      override def encode(value: Text): Attempt[BitVector] = utf8.encode(value.value)
+
+      override def sizeBound: SizeBound = utf8.sizeBound
+    }
 
   }
 

core/src/main/scala/com/evolutiongaming/kafka/journal/ReleasedError.scala

@@ -2,4 +2,4 @@ package com.evolutiongaming.kafka.journal
 
 import scala.util.control.NoStackTrace
 
-final case object ReleasedError extends RuntimeException("released") with NoStackTrace
+case object ReleasedError extends RuntimeException("released") with NoStackTrace

core/src/main/scala/com/evolutiongaming/kafka/journal/util/GracefulFiber.scala

@@ -23,10 +23,11 @@ object GracefulFiber {
             def join = fiber.join
 
             def cancel = {
-              for {
-                cancel <- cancelRef.getAndSet(true)
-                _      <- if (cancel) ().pure[F] else fiber.joinWithNever
-              } yield {}
+              cancelRef.getAndSet(true)
+                .flatMap(cancel =>
+                  (if (cancel) ().pure[F] else fiber.joinWithNever.void)
+
+                )
             }
           }
         }

core/src/main/scala/com/evolutiongaming/kafka/journal/util/PlayJsonHelper.scala

@@ -31,7 +31,7 @@ object PlayJsonHelper {
 
     def handleErrorWith[A](fa: JsResult[A])(f: JsError => JsResult[A]): JsResult[A] = {
       fa match {
-        case fa: JsSuccess[A] => fa
+        case fa: JsSuccess[?] => fa
         case fa: JsError      => f(fa)
       }
     }
@@ -40,15 +40,15 @@ object PlayJsonHelper {
 
     def flatMap[A, B](fa: JsResult[A])(f: A => JsResult[B]): JsResult[B] = fa.flatMap(f)
 
-    @tailrec
+    @tailrec //tailrec forces to use explicit match instead of fold and type erasure forces to use asInstanceOf
     def tailRecM[A, B](a: A)(f: A => JsResult[Either[A, B]]): JsResult[B] = {
       f(a) match {
-        case b: JsSuccess[Either[A, B]] =>
-          b.value match {
-            case Right(a) => JsSuccess(a)
-            case Left(b)  => tailRecM(b)(f)
+        case success: JsSuccess[?] =>
+          success.value match {
+            case Right(b) => JsSuccess(b.asInstanceOf[B])
+            case Left(a1)  => tailRecM(a1.asInstanceOf[A])(f)
           }
-        case b: JsError => b
+        case error: JsError => error
       }
     }
   }

core/src/main/scala/com/evolutiongaming/kafka/journal/util/PureConfigHelper.scala

@@ -3,10 +3,12 @@ package com.evolutiongaming.kafka.journal.util
 import com.evolutiongaming.kafka.journal.FromConfigReaderResult
 import pureconfig.ConfigReader
 
+import scala.reflect.ClassTag
+
 object PureConfigHelper {
 
   implicit class ConfigReaderResultOpsPureConfigHelper[A](val self: ConfigReader.Result[A]) extends AnyVal {
 
-    def liftTo[F[_]: FromConfigReaderResult]: F[A] = FromConfigReaderResult[F].apply(self)
+    def liftTo[F[_]: FromConfigReaderResult](implicit ct: ClassTag[A]): F[A] = FromConfigReaderResult[F].apply(self)
   }
 }

core/src/main/scala/com/evolutiongaming/kafka/journal/util/ScodecHelper.scala

@@ -24,23 +24,23 @@ object ScodecHelper {
 
       def handleErrorWith[A](fa: Attempt[A])(f: Failure => Attempt[A]): Attempt[A] = {
         fa match {
-          case fa: Successful[A] => fa
           case fa: Failure       => f(fa)
+          case fa => fa
         }
       }
 
       def pure[A](a: A): Attempt[A] = Successful(a)
 
       def flatMap[A, B](fa: Attempt[A])(f: A => Attempt[B]): Attempt[B] = fa.flatMap(f)
 
-      @tailrec
+      @tailrec //tailrec forces to use explicit match instead of fold and type erasure forces to use asInstanceOf
       def tailRecM[A, B](a: A)(f: A => Attempt[Either[A, B]]): Attempt[B] = {
         f(a) match {
-          case b: Failure => b
-          case b: Successful[Either[A, B]] =>
-            b.value match {
-              case Left(b1) => tailRecM(b1)(f)
-              case Right(a) => Successful(a)
+          case failure: Failure => failure
+          case success: Successful[?] =>
+            success.value match {
+              case Left(a1) => tailRecM(a1.asInstanceOf[A])(f)
+              case Right(b) => Successful(b.asInstanceOf[B])
             }
         }
       }

journal/src/main/scala/com/evolutiongaming/kafka/journal/Bounds.scala

@@ -132,15 +132,15 @@ object Bounds {
         *   `true` if this value is less than `x` for an interval `[x, y]`, `false`
         *   otherwise.
         */
-      def <(bounds: Bounds[A])(implicit order: Order[A]): Boolean = self < bounds.min
+      def <(bounds: Bounds[A])(implicit order: Order[A]): Boolean = order.lt(self, bounds.min)
 
       /** Checks if this value is located after an interval `bounds`.
         *
         * @return
         *   `true` if this value is greater than `y` for an interval `[x, y]`, `false`
         *   otherwise.
         */
-      def >(bounds: Bounds[A])(implicit order: Order[A]): Boolean = self > bounds.max
+      def >(bounds: Bounds[A])(implicit order: Order[A]): Boolean = order.gt(self,bounds.max)
     }
   }
 }