Merge pull request #163 from ahjohannessen/wip-tests-for-metastreams

core/tests: convenience methods for metastreams + tests

build.sbt

@@ -1,5 +1,17 @@
 import Dependencies._
 
+def scalaVersionSpecificFolders(srcName: String, srcBaseDir: java.io.File, scalaVersion: String) = {
+
+  def extraDirs(suffix: String): List[File] =
+    List(srcBaseDir / srcName / suffix)
+
+  CrossVersion.partialVersion(scalaVersion) match {
+    case Some((2, 13)) => extraDirs("scala-2.13")
+    case Some((0, _))  => extraDirs("scala-3")
+    case _             => Nil
+  }
+}
+
 lazy val root = project
   .in(file("."))
   .settings(skip in publish := true)
@@ -24,7 +36,9 @@ lazy val `sec-core` = project
   .settings(
     name := "sec-core",
     libraryDependencies ++=
-      compileM(cats, catsEffect, fs2, log4cats, log4catsNoop, scodecBits, circe, circeParser)
+      compileM(cats, catsEffect, fs2, log4cats, log4catsNoop, scodecBits, circe, circeParser),
+    Compile / unmanagedSourceDirectories ++=
+      scalaVersionSpecificFolders("main", sourceDirectory.value, scalaVersion.value)
   )
   .settings(libraryDependencies := libraryDependencies.value.map(_.withDottyCompat(scalaVersion.value)))
   .dependsOn(`sec-protos`)

sec-core/src/main/scala-2.13/sec/syntax/api.scala

@@ -0,0 +1,96 @@
+/*
+ * Copyright 2020 Alex Henning Johannessen
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package sec
+package syntax
+
+import scala.concurrent.duration.FiniteDuration
+import cats.syntax.all._
+import sec.api._
+import sec.api.MetaStreams._
+import sec.core._
+import sec.core.EventNumber.Exact
+import sec.core.StreamId.Id
+
+object api {
+
+  implicit def syntaxForMetaStreams[F[_]: ErrorM](ms: MetaStreams[F]): MetaStreamsSyntax[F] =
+    new MetaStreamsSyntax[F](ms)
+
+//====================================================================================================================
+
+  final class MetaStreamsSyntax[F[_]: ErrorM](val ms: MetaStreams[F]) {
+
+    def getMaxAge(id: Id): F[Option[ReadResult[MaxAge]]] =
+      ms.getMaxAge(id, None)
+
+    def setMaxAge(id: Id, expectedRevision: StreamRevision, age: FiniteDuration): F[WriteResult] =
+      MaxAge.lift[F](age) >>= (ms.setMaxAge(id, expectedRevision, _, None))
+
+    def removeMaxAge(id: Id, expectedRevision: StreamRevision): F[WriteResult] =
+      ms.removeMaxAge(id, expectedRevision, None)
+
+    def getMaxCount(id: Id): F[Option[ReadResult[MaxCount]]] =
+      ms.getMaxCount(id, None)
+
+    def setMaxCount(id: Id, expectedRevision: StreamRevision, count: Int): F[WriteResult] =
+      MaxCount.lift[F](count) >>= (ms.setMaxCount(id, expectedRevision, _, None))
+
+    def removeMaxCount(id: Id, expectedRevision: StreamRevision): F[WriteResult] =
+      ms.removeMaxCount(id, expectedRevision, None)
+
+    def getCacheControl(id: Id): F[Option[ReadResult[CacheControl]]] =
+      ms.getCacheControl(id, None)
+
+    def setCacheControl(id: Id, expectedRevision: StreamRevision, cacheControl: FiniteDuration): F[WriteResult] =
+      CacheControl.lift[F](cacheControl) >>= (ms.setCacheControl(id, expectedRevision, _, None))
+
+    def removeCacheControl(id: Id, expectedRevision: StreamRevision): F[MetaStreams.WriteResult] =
+      ms.removeCacheControl(id, expectedRevision, None)
+
+    def getAcl(id: Id): F[Option[ReadResult[StreamAcl]]] =
+      ms.getAcl(id, None)
+
+    def setAcl(id: Id, expectedRevision: StreamRevision, acl: StreamAcl): F[WriteResult] =
+      ms.setAcl(id, expectedRevision, acl, None)
+
+    def removeAcl(id: Id, expectedRevision: StreamRevision): F[WriteResult] =
+      ms.removeAcl(id, expectedRevision, None)
+
+    def getTruncateBefore(id: Id): F[Option[ReadResult[Exact]]] =
+      ms.getTruncateBefore(id, None)
+
+    def setTruncateBefore(id: Id, expectedRevision: StreamRevision, truncateBefore: Long): F[WriteResult] =
+      EventNumber.Exact.lift[F](truncateBefore) >>= (ms.setTruncateBefore(id, expectedRevision, _, None))
+
+    def removeTruncateBefore(id: Id, expectedRevision: StreamRevision): F[WriteResult] =
+      ms.removeTruncateBefore(id, expectedRevision, None)
+
+    ///
+
+    private[sec] def getMetadata(id: Id): F[Option[MetaResult]] =
+      ms.getMetadata(id, None)
+
+    private[sec] def setMetadata(id: Id, expectedRevision: StreamRevision, data: StreamMetadata): F[WriteResult] =
+      ms.setMetadata(id, expectedRevision, data, None)
+
+    private[sec] def removeMetadata(id: Id, expectedRevision: StreamRevision): F[WriteResult] =
+      ms.removeMetadata(id, expectedRevision, None)
+  }
+
+//====================================================================================================================
+
+}

sec-core/src/main/scala-3/sec/syntax/api.scala

@@ -0,0 +1,91 @@
+/*
+ * Copyright 2020 Alex Henning Johannessen
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package sec
+package syntax
+package api
+
+import scala.concurrent.duration.FiniteDuration
+import cats.syntax.all._
+import sec.ErrorM
+import sec.api._
+import sec.api.MetaStreams._
+import sec.core._
+import sec.core.EventNumber.Exact
+import sec.core.StreamId.Id
+
+//====================================================================================================================
+
+type Creds = Option[UserCredentials]
+
+extension [F[_]: ErrorM](ms: MetaStreams[F]) {
+
+  def getMaxAge(id: Id): F[Option[ReadResult[MaxAge]]] = 
+    ms.getMaxAge(id, None)
+
+  def setMaxAge(id: Id, expectedRevision: StreamRevision, age: FiniteDuration, creds: Creds = None): F[WriteResult] = 
+    MaxAge.lift[F](age) >>= (ms.setMaxAge(id, expectedRevision, _, creds))
+
+  def removeMaxAge(id: Id, expectedRevision: StreamRevision): F[WriteResult] = 
+    ms.removeMaxAge(id, expectedRevision, None)
+
+  def getMaxCount(id: Id): F[Option[ReadResult[MaxCount]]] = 
+    ms.getMaxCount(id, None)
+
+  def setMaxCount(id: Id, expectedRevision: StreamRevision, count: Int, creds: Creds = None): F[WriteResult] = 
+    MaxCount.lift[F](count) >>= (ms.setMaxCount(id, expectedRevision, _, creds))
+
+  def removeMaxCount(id: Id, expectedRevision: StreamRevision): F[WriteResult] = 
+    ms.removeMaxCount(id, expectedRevision, None)
+
+  def getCacheControl(id: Id): F[Option[ReadResult[CacheControl]]] =
+    ms.getCacheControl(id, None)  
+
+  def setCacheControl(id: Id, expectedRevision: StreamRevision, cacheControl: FiniteDuration, creds: Creds = None): F[WriteResult] = 
+    CacheControl.lift[F](cacheControl) >>= (ms.setCacheControl(id, expectedRevision, _, creds))
+
+  def removeCacheControl(id: Id, expectedRevision: StreamRevision): F[MetaStreams.WriteResult] = 
+    ms.removeCacheControl(id, expectedRevision, None)
+
+  def getAcl(id: Id): F[Option[ReadResult[StreamAcl]]] = 
+    ms.getAcl(id, None)
+
+  def setAcl(id: Id, expectedRevision: StreamRevision, acl: StreamAcl): F[WriteResult] = 
+    ms.setAcl(id, expectedRevision, acl, None)
+
+  def removeAcl(id: Id, expectedRevision: StreamRevision): F[WriteResult] = 
+    ms.removeAcl(id, expectedRevision, None)
+
+  def getTruncateBefore(id: Id): F[Option[ReadResult[Exact]]] = 
+    ms.getTruncateBefore(id, None)
+
+  def setTruncateBefore(id: Id, expectedRevision: StreamRevision, truncateBefore: Long, creds: Creds = None): F[WriteResult] = 
+    EventNumber.Exact.lift[F](truncateBefore) >>= (ms.setTruncateBefore(id, expectedRevision, _, creds))
+
+  def removeTruncateBefore(id: Id, expectedRevision: StreamRevision): F[WriteResult] = 
+    ms.removeTruncateBefore(id, expectedRevision, None)
+
+  private[sec] def getMetadata(id: Id): F[Option[MetaResult]] = 
+    ms.getMetadata(id, None)
+
+  private[sec] def setMetadata(id: Id, expectedRevision: StreamRevision, data: StreamMetadata): F[WriteResult] = 
+    ms.setMetadata(id, expectedRevision, data, None)
+
+  private[sec] def removeMetadata(id: Id, expectedRevision: StreamRevision): F[WriteResult] = 
+    ms.removeMetadata(id, expectedRevision, None)
+}
+
+//====================================================================================================================

sec-core/src/main/scala/sec/core/metadata.scala

@@ -18,6 +18,7 @@ package sec
 package core
 
 import scala.concurrent.duration._
+import scala.util.control.NoStackTrace
 import cats.Show
 import cats.Endo
 import cats.syntax.all._
@@ -124,7 +125,7 @@ object MaxAge {
 
   implicit val showForMaxAge: Show[MaxAge] = Show.show(_.value.toString())
 
-  final case class InvalidMaxAge(msg: String) extends RuntimeException(msg)
+  final case class InvalidMaxAge(msg: String) extends RuntimeException(msg) with NoStackTrace
 }
 
 sealed abstract case class MaxCount(value: Int)
@@ -144,7 +145,7 @@ object MaxCount {
     s"${mc.value} event${if (mc.value == 1) "" else "s"}"
   }
 
-  final case class InvalidMaxCount(msg: String) extends RuntimeException(msg)
+  final case class InvalidMaxCount(msg: String) extends RuntimeException(msg) with NoStackTrace
 
 }
 
@@ -163,7 +164,7 @@ object CacheControl {
 
   implicit val showForCacheControl: Show[CacheControl] = Show.show(_.value.toString())
 
-  final case class InvalidCacheControl(msg: String) extends RuntimeException(msg)
+  final case class InvalidCacheControl(msg: String) extends RuntimeException(msg) with NoStackTrace
 }
 
 //======================================================================================================================

sec-core/src/main/scala/sec/core/version.scala

@@ -17,6 +17,7 @@
 package sec
 package core
 
+import scala.util.control.NoStackTrace
 import cats.{Eq, Order, Show}
 import cats.syntax.all._
 
@@ -48,9 +49,18 @@ object EventNumber {
 
   sealed abstract case class Exact(value: Long) extends EventNumber with StreamRevision
   object Exact {
+
     private[EventNumber] def create(value: Long): Exact = new Exact(value) {}
+
     def apply(value: Long): Attempt[Exact] =
       Either.cond(value >= 0L, create(value), s"value must be >= 0, but is $value")
+
+    def lift[F[_]: ErrorA](value: Long): F[Exact] =
+      Exact(value).leftMap(InvalidExact).liftTo[F]
+
+    ///
+
+    final case class InvalidExact(msg: String) extends RuntimeException(msg) with NoStackTrace
   }
 
   case object End extends EventNumber
@@ -59,7 +69,8 @@ object EventNumber {
 
   ///
 
-  def apply(number: Long): EventNumber = if (number < 0) End else exact(number)
+  private[sec] def apply(number: Long): EventNumber =
+    if (number < 0) End else exact(number)
 
   implicit val orderForEventNumber: Order[EventNumber] = Order.from {
     case (x: Exact, y: Exact) => Order[Exact].compare(x, y)

sec-tests/src/cit/scala/sec/api/spec.scala

@@ -40,13 +40,16 @@ object CSpec {
     endpointFrom("SEC_CLUSTER_ES3_ADDRESS", "SEC_CIT_ES3_PORT", "127.0.0.1", 2116)
   )
 
+  final private val maxRetries = 5
+
   ///
 
   def mkClient[F[_]: ConcurrentEffect: Timer](log: Logger[F]): Resource[F, EsClient[F]] = EsClient
     .cluster[F](seed, authority)
     .withChannelShutdownAwait(0.seconds)
     .withCertificate(caPath)
     .withLogger(log)
+    .withOperationsRetryMaxAttempts(maxRetries)
     .resource
 
 }

sec-tests/src/sit/scala/sec/api/gossip.scala

@@ -19,7 +19,7 @@ package api
 
 import sec.api.Gossip._
 
-class GossipITest extends SnSpec {
+class GossipSuite extends SnSpec {
 
   sequential