diff --git a/.scalafmt.conf b/.scalafmt.conf index fd7c2805..93f48c01 100644 --- a/.scalafmt.conf +++ b/.scalafmt.conf @@ -1,4 +1,4 @@ -version = "3.1.2" +version = "3.11.1" runner.dialect = scala3 maxColumn = 120 diff --git a/benchmarks/src/main/scala/com/devsisters/shardcake/Server.scala b/benchmarks/src/main/scala/com/devsisters/shardcake/Server.scala index 78ec9d0d..26f2738f 100644 --- a/benchmarks/src/main/scala/com/devsisters/shardcake/Server.scala +++ b/benchmarks/src/main/scala/com/devsisters/shardcake/Server.scala @@ -1,5 +1,6 @@ package com.devsisters.shardcake +import com.devsisters.shardcake.kryo.given import com.devsisters.shardcake.interfaces.Storage import zio.{ Config => _, _ } import zio.stream.ZStream @@ -58,7 +59,6 @@ object Server { val sharding: ZLayer[Config, Throwable, Sharding with GrpcConfig] = ZLayer.makeSome[Config, Sharding with GrpcConfig]( - KryoSerialization.live, memory, grpcConfig, shardManagerClient, diff --git a/build.sbt b/build.sbt index 89ac02e0..1bc5cc9f 100644 --- a/build.sbt +++ b/build.sbt @@ -14,7 +14,6 @@ val redis4catsVersion = "2.0.1" val redissonVersion = "3.45.1" val scalaKryoVersion = "1.4.0" val testContainersVersion = "0.44.1" -val scalaCompatVersion = "2.13.0" inThisBuild( List( @@ -54,6 +53,7 @@ lazy val root = project storageRedis, storageRedisson, serializationKryo, + serializationProteus, grpcProtocol, examples, benchmarks @@ -66,10 +66,9 @@ lazy val core = project .settings( libraryDependencies ++= Seq( - "dev.zio" %% "zio" % zioVersion, - "dev.zio" %% "zio-streams" % zioVersion, - "dev.zio" %% "zio-json" % zioJsonVersion, - "org.scala-lang.modules" %% "scala-collection-compat" % scalaCompatVersion + "dev.zio" %% "zio" % zioVersion, + "dev.zio" %% "zio-streams" % zioVersion, + "dev.zio" %% "zio-json" % zioJsonVersion ) ) @@ -143,7 +142,7 @@ lazy val serializationKryo = project .in(file("serialization-kryo")) .settings(name := "shardcake-serialization-kryo") .settings(commonSettings) - .dependsOn(core) + .dependsOn(entities) .settings( libraryDependencies ++= Seq( @@ -154,6 +153,18 @@ lazy val serializationKryo = project lazy val generateProto = taskKey[Unit]("Regenerate sharding.proto from the Scala protocol definition.") lazy val checkProto = taskKey[Unit]("Fail if sharding.proto is out of sync with the Scala protocol definition.") +lazy val serializationProteus = project + .in(file("serialization-proteus")) + .settings(name := "shardcake-serialization-proteus") + .settings(commonSettings) + .dependsOn(entities) + .settings( + libraryDependencies ++= + Seq( + "com.github.ghostdogpr" %% "proteus-core" % proteusVersion + ) + ) + lazy val grpcProtocol = project .in(file("protocol-grpc")) .settings(name := "shardcake-protocol-grpc") diff --git a/core/src/main/scala/com/devsisters/shardcake/PodAddress.scala b/core/src/main/scala/com/devsisters/shardcake/PodAddress.scala index 724cf34c..e7545a01 100644 --- a/core/src/main/scala/com/devsisters/shardcake/PodAddress.scala +++ b/core/src/main/scala/com/devsisters/shardcake/PodAddress.scala @@ -1,7 +1,5 @@ package com.devsisters.shardcake -import scala.collection.compat._ - import zio.json._ case class PodAddress(host: String, port: Int) { diff --git a/core/src/main/scala/com/devsisters/shardcake/interfaces/MessageCodec.scala b/core/src/main/scala/com/devsisters/shardcake/interfaces/MessageCodec.scala new file mode 100644 index 00000000..4cd6cb97 --- /dev/null +++ b/core/src/main/scala/com/devsisters/shardcake/interfaces/MessageCodec.scala @@ -0,0 +1,51 @@ +package com.devsisters.shardcake.interfaces + +import com.devsisters.shardcake.interfaces.MessageCodec.{ Decoder, Encoder } + +/** + * Per-message-type codec used by Shardcake to serialise entity / topic messages and + * their replies on the wire. + * + * Carried by `RecipientType[Msg]` via an implicit `MessageCodec[Msg]` parameter, so each + * `EntityType` / `TopicType` declaration picks up the codec from the surrounding scope. + * + * Implementations decide how `Msg` is encoded and — for reply-bearing variants — which + * encoder/decoder pair to use for the reply type. + * For uniform backends (e.g. Kryo) the reply codec is the same regardless of which + * variant the message is; for variant-aware backends (e.g. Proteus) it depends on the + * concrete variant. + */ +trait MessageCodec[Msg] { + + /** + * Encode a message to bytes for transport. + */ + def encodeMessage(message: Msg): Array[Byte] + + /** + * Decode a message from bytes. + */ + def decodeMessage(bytes: Array[Byte]): Msg + + /** + * Gets an encoder for the reply of a specific message on the receiver side. + * The returned function will be invoked when the entity calls `replier.reply(value)` + * or `streamReplier.replyStream(...)`. + */ + def replyEncoder(decoded: Msg): Encoder + + /** + * Returns a decoder for the reply of a specific message. + */ + def replyDecoder[Res](sample: Msg): Decoder[Res] + + /** + * Same as `replyDecoder`, but for stream-reply slots. + */ + def streamReplyDecoder[Res](sample: Msg): Decoder[Res] +} + +object MessageCodec { + type Encoder = Any => Array[Byte] + type Decoder[A] = Array[Byte] => A +} diff --git a/core/src/main/scala/com/devsisters/shardcake/interfaces/Pods.scala b/core/src/main/scala/com/devsisters/shardcake/interfaces/Pods.scala index dc288eb9..652dc8b7 100644 --- a/core/src/main/scala/com/devsisters/shardcake/interfaces/Pods.scala +++ b/core/src/main/scala/com/devsisters/shardcake/interfaces/Pods.scala @@ -72,14 +72,14 @@ object Pods { pod: PodAddress, entityId: String, messages: ZStream[Any, Throwable, BinaryMessage] - ): Task[Option[Array[Byte]]] = ZIO.none + ): Task[Option[Array[Byte]]] = ZIO.none def sendMessageAndReceiveStream(pod: PodAddress, message: BinaryMessage): ZStream[Any, Throwable, Array[Byte]] = ZStream.empty def sendStreamAndReceiveStream( pod: PodAddress, entityId: String, messages: ZStream[Any, Throwable, BinaryMessage] - ): ZStream[Any, Throwable, Array[Byte]] = ZStream.empty + ): ZStream[Any, Throwable, Array[Byte]] = ZStream.empty }) case class BinaryMessage(entityId: String, entityType: String, body: Array[Byte], replyId: Option[String]) diff --git a/core/src/main/scala/com/devsisters/shardcake/interfaces/Serialization.scala b/core/src/main/scala/com/devsisters/shardcake/interfaces/Serialization.scala deleted file mode 100644 index b5bd8f0b..00000000 --- a/core/src/main/scala/com/devsisters/shardcake/interfaces/Serialization.scala +++ /dev/null @@ -1,59 +0,0 @@ -package com.devsisters.shardcake.interfaces - -import zio.{ Chunk, Task, ULayer, ZIO, ZLayer } - -import java.io.{ ByteArrayInputStream, ByteArrayOutputStream, ObjectInputStream, ObjectOutputStream } - -/** - * An interface to serialize user messages that will be sent between pods. - */ -trait Serialization { - - /** - * Transforms the given message into binary - */ - def encode(message: Any): Task[Array[Byte]] - - /** - * Transform binary back into the given type - */ - def decode[A](bytes: Array[Byte]): Task[A] - - /** - * Transforms a chunk of messages into binary - */ - def encodeChunk(messages: Chunk[Any]): Task[Chunk[Array[Byte]]] = - ZIO.foreach(messages)(encode) - - /** - * Transforms a chunk of binary back into the given type - */ - def decodeChunk[A](bytes: Chunk[Array[Byte]]): Task[Chunk[A]] = - ZIO.foreach(bytes)(decode[A]) -} - -object Serialization { - - /** - * A layer that uses Java serialization for encoding and decoding messages. - * This is useful for testing and not recommended to use in production. - */ - val javaSerialization: ULayer[Serialization] = - ZLayer.succeed(new Serialization { - def encode(message: Any): Task[Array[Byte]] = - ZIO.scoped { - val stream = new ByteArrayOutputStream() - ZIO - .fromAutoCloseable(ZIO.attempt(new ObjectOutputStream(stream))) - .flatMap(oos => ZIO.attempt(oos.writeObject(message))) - .as(stream.toByteArray) - } - - def decode[A](bytes: Array[Byte]): Task[A] = - ZIO.scoped { - ZIO - .fromAutoCloseable(ZIO.attempt(new ObjectInputStream(new ByteArrayInputStream(bytes)))) - .flatMap(ois => ZIO.attempt(ois.readObject.asInstanceOf[A])) - } - }) -} diff --git a/core/src/test/scala/com/devsisters/shardcake/JavaSerializationSpec.scala b/core/src/test/scala/com/devsisters/shardcake/JavaSerializationSpec.scala deleted file mode 100644 index fd009c40..00000000 --- a/core/src/test/scala/com/devsisters/shardcake/JavaSerializationSpec.scala +++ /dev/null @@ -1,19 +0,0 @@ -package com.devsisters.shardcake - -import com.devsisters.shardcake.interfaces.Serialization -import zio.{ Scope, ZIO } -import zio.test._ - -object JavaSerializationSpec extends ZIOSpecDefault { - def spec: Spec[TestEnvironment with Scope, Any] = - suite("JavaSerializationSpec")( - test("serialize back and forth") { - case class Test(a: Int, b: String) - val expected = Test(2, "test") - for { - bytes <- ZIO.serviceWithZIO[Serialization](_.encode(expected)) - actual <- ZIO.serviceWithZIO[Serialization](_.decode[Test](bytes)) - } yield assertTrue(expected == actual) - } - ).provideShared(Serialization.javaSerialization) -} diff --git a/entities/src/main/scala/com/devsisters/shardcake/LocalSharding.scala b/entities/src/main/scala/com/devsisters/shardcake/LocalSharding.scala index 4655ea59..bcaf2c82 100644 --- a/entities/src/main/scala/com/devsisters/shardcake/LocalSharding.scala +++ b/entities/src/main/scala/com/devsisters/shardcake/LocalSharding.scala @@ -1,6 +1,6 @@ package com.devsisters.shardcake -import com.devsisters.shardcake.interfaces.{ Pods, Serialization, Storage } +import com.devsisters.shardcake.interfaces.{ Pods, Storage } import com.devsisters.shardcake.interfaces.Pods.BinaryMessage import zio.{ Promise, Queue, RLayer, Task, ULayer, URLayer, ZIO, ZLayer } import zio.stream.ZStream @@ -111,8 +111,8 @@ object LocalSharding { * A special layer meant for testing that uses a local queue rather than an external transport. * This layer will only work in a single JVM and is not suitable for production use. */ - val live: RLayer[ShardManagerClient with Storage with Serialization with Config, Sharding] = - ZLayer.makeSome[ShardManagerClient with Storage with Serialization with Config, Sharding]( + val live: RLayer[ShardManagerClient with Storage with Config, Sharding] = + ZLayer.makeSome[ShardManagerClient with Storage with Config, Sharding]( localQueue, localPods, localServer, diff --git a/entities/src/main/scala/com/devsisters/shardcake/Messenger.scala b/entities/src/main/scala/com/devsisters/shardcake/Messenger.scala index 5d89ef8f..ae674e77 100644 --- a/entities/src/main/scala/com/devsisters/shardcake/Messenger.scala +++ b/entities/src/main/scala/com/devsisters/shardcake/Messenger.scala @@ -36,10 +36,13 @@ trait Messenger[-Msg] { /** * Send a stream of messages and receive a stream of responses of type `Res`. + * + * The first message carrying the `StreamReplier` is constructed via `request`; any + * additional messages flow through `rest`. */ - def sendStreamAndReceiveStream[Res](entityId: String)( - messages: StreamReplier[Res] => ZStream[Any, Throwable, Msg] - ): Task[ZStream[Any, Throwable, Res]] + def sendStreamAndReceiveStream[Res]( + entityId: String + )(request: StreamReplier[Res] => Msg, rest: ZStream[Any, Throwable, Msg]): Task[ZStream[Any, Throwable, Res]] /** * Send a message and receive a stream of responses of type `Res` while restarting the stream when the remote entity @@ -80,12 +83,13 @@ trait Messenger[-Msg] { * cursor according to what we've seen in the previous stream of responses. */ def sendStreamAndReceiveStreamAutoRestart[Cursor, Res](entityId: String, cursor: Cursor)( - msg: (Cursor, StreamReplier[Res]) => ZStream[Any, Throwable, Msg] + request: (Cursor, StreamReplier[Res]) => Msg, + rest: Cursor => ZStream[Any, Throwable, Msg] )( updateCursor: (Cursor, Res) => Cursor ): ZStream[Any, Throwable, Res] = ZStream - .unwrap(sendStreamAndReceiveStream[Res](entityId)(msg(cursor, _))) + .unwrap(sendStreamAndReceiveStream[Res](entityId)(request(cursor, _), rest(cursor))) .either .mapAccum(cursor) { case (c, Right(res)) => updateCursor(c, res) -> Right(res) @@ -95,7 +99,7 @@ trait Messenger[-Msg] { case Right(res) => ZStream.succeed(res) case Left((lastSeenCursor, StreamCancelled)) => ZStream.execute(ZIO.sleep(200.millis)) ++ - sendStreamAndReceiveStreamAutoRestart(entityId, lastSeenCursor)(msg)(updateCursor) + sendStreamAndReceiveStreamAutoRestart(entityId, lastSeenCursor)(request, rest)(updateCursor) case Left((_, err)) => ZStream.fail(err) } } diff --git a/entities/src/main/scala/com/devsisters/shardcake/RecipientType.scala b/entities/src/main/scala/com/devsisters/shardcake/RecipientType.scala index e2faebb1..41d378d8 100644 --- a/entities/src/main/scala/com/devsisters/shardcake/RecipientType.scala +++ b/entities/src/main/scala/com/devsisters/shardcake/RecipientType.scala @@ -1,13 +1,20 @@ package com.devsisters.shardcake +import com.devsisters.shardcake.interfaces.MessageCodec + /** - * An abstract type to extend for each type of entity or topic + * An abstract type to extend for each type of entity or topic. + * + * The `MessageCodec[Msg]` carried via the implicit parameter owns serialisation for this + * recipient type. Users provide a codec by importing one of the backend `given`s in scope + * at the declaration site, e.g. `import com.devsisters.shardcake.kryo.given`. + * * @param name a unique string that identifies this entity or topic type * @tparam Msg the type of message that can be sent to this entity or topic type */ -sealed abstract class RecipientType[+Msg](val name: String) { +sealed abstract class RecipientType[Msg](val name: String)(using private[shardcake] val codec: MessageCodec[Msg]) { def getShardId(entityId: String, numberOfShards: Int): ShardId = math.abs(entityId.hashCode % numberOfShards) + 1 } -abstract class EntityType[+Msg](name: String) extends RecipientType[Msg](name) -abstract class TopicType[+Msg](name: String) extends RecipientType[Msg](name) +abstract class EntityType[Msg](name: String)(using MessageCodec[Msg]) extends RecipientType[Msg](name) +abstract class TopicType[Msg](name: String)(using MessageCodec[Msg]) extends RecipientType[Msg](name) diff --git a/entities/src/main/scala/com/devsisters/shardcake/Sharding.scala b/entities/src/main/scala/com/devsisters/shardcake/Sharding.scala index cf7f5bdd..1f312c6b 100644 --- a/entities/src/main/scala/com/devsisters/shardcake/Sharding.scala +++ b/entities/src/main/scala/com/devsisters/shardcake/Sharding.scala @@ -4,8 +4,8 @@ import com.devsisters.shardcake.Messenger.MessengerTimeout import com.devsisters.shardcake.Sharding.{ EntityState, ShardingRegistrationEvent } import com.devsisters.shardcake.errors._ import com.devsisters.shardcake.interfaces.Pods.BinaryMessage -import com.devsisters.shardcake.interfaces.{ Pods, Serialization, Storage } -import com.devsisters.shardcake.internal.{ EntityManager, ReplyChannel, SendChannel } +import com.devsisters.shardcake.interfaces.{ MessageCodec, Pods, Storage } +import com.devsisters.shardcake.internal.{ EntityManager, PendingReply, ReplyChannel, SendChannel } import zio.{ Config => _, _ } import zio.stream.ZStream @@ -22,16 +22,15 @@ class Sharding private ( shardAssignments: Ref[Map[ShardId, PodAddress]], entityStates: Ref[Map[String, EntityState]], singletons: Ref.Synchronized[List[(String, UIO[Nothing], Option[Fiber[Nothing, Nothing]])]], - replyChannels: Ref[Map[String, ReplyChannel[Nothing]]], // channel for each pending reply, + pendingReplies: Ref[Map[String, PendingReply]], lastUnhealthyNodeReported: Ref[OffsetDateTime], isShuttingDownRef: Ref[Boolean], shardManager: ShardManagerClient, pods: Pods, storage: Storage, - serialization: Serialization, eventsHub: Hub[ShardingRegistrationEvent] ) { self => - private[shardcake] def getShardId(recipientType: RecipientType[_], entityId: String): ShardId = + private[shardcake] def getShardId(recipientType: RecipientType[?], entityId: String): ShardId = recipientType.getShardId(entityId, config.numberOfShards) val register: Task[Unit] = @@ -120,14 +119,14 @@ class Sharding private ( ZIO.logDebug(s"Unassigned shards: ${renderShardIds(shards)}") ) - def getPodAddress(recipientType: RecipientType[_], entityId: String): UIO[Option[PodAddress]] = + def getPodAddress(recipientType: RecipientType[?], entityId: String): UIO[Option[PodAddress]] = for { shards <- shardAssignments.get shardId = getShardId(recipientType, entityId) pod = shards.get(shardId) } yield pod - def isEntityOnLocalShards(recipientType: RecipientType[_], entityId: String): UIO[Boolean] = + def isEntityOnLocalShards(recipientType: RecipientType[?], entityId: String): UIO[Boolean] = getPodAddress(recipientType, entityId).map(pod => pod.contains(address)) val getAssignments: UIO[Map[ShardId, PodAddress]] = @@ -187,28 +186,26 @@ class Sharding private ( def sendToLocalEntity(msg: BinaryMessage): Task[Option[Array[Byte]]] = for { - replyChannel <- ReplyChannel.single[Any] + replyChannel <- ReplyChannel.single[Array[Byte]] _ <- sendToLocalEntity(msg, replyChannel) res <- replyChannel.output - bytes <- ZIO.foreach(res)(serialization.encode) - } yield bytes + } yield res def sendToLocalEntityAndReceiveStream(msg: BinaryMessage): ZStream[Any, Throwable, Array[Byte]] = ZStream.unwrap { for { - replyChannel <- ReplyChannel.stream[Any] + replyChannel <- ReplyChannel.stream[Array[Byte]] _ <- sendToLocalEntity(msg, replyChannel) - } yield replyChannel.output.mapChunksZIO(serialization.encodeChunk) + } yield replyChannel.output } def sendStreamToLocalEntity(messages: ZStream[Any, Throwable, BinaryMessage]): Task[Option[Array[Byte]]] = ZIO.scoped { for { - replyChannel <- ReplyChannel.single[Any] + replyChannel <- ReplyChannel.single[Array[Byte]] _ <- messages.runForeach(sendToLocalEntity(_, replyChannel)).onError(replyChannel.fail).forkScoped res <- replyChannel.output - bytes <- ZIO.foreach(res)(serialization.encode) - } yield bytes + } yield res } def sendStreamToLocalEntityAndReceiveStream( @@ -216,34 +213,59 @@ class Sharding private ( ): ZStream[Any, Throwable, Array[Byte]] = ZStream.unwrapScoped { for { - replyChannel <- ReplyChannel.stream[Any] + replyChannel <- ReplyChannel.stream[Array[Byte]] _ <- messages.runForeach(sendToLocalEntity(_, replyChannel)).onError(replyChannel.fail).forkScoped - } yield replyChannel.output.mapChunksZIO(serialization.encodeChunk) + } yield replyChannel.output } - private def sendToLocalEntity(msg: BinaryMessage, replyChannel: ReplyChannel[Nothing]): Task[Unit] = + private def sendToLocalEntity(msg: BinaryMessage, replyChannel: ReplyChannel[Array[Byte]]): Task[Unit] = entityStates.get.flatMap(_.get(msg.entityType) match { case Some(state) => state.processBinary(msg, replyChannel).unit case None => ZIO.fail(new Exception(s"Entity type ${msg.entityType} was not registered.")) }) - private[shardcake] def initReply(id: String, replyChannel: ReplyChannel[Nothing]): UIO[Unit] = - replyChannels - .getAndUpdate(_.updated(id, replyChannel)) - .flatMap(beforeReplyChannels => - replyChannel.await.ensuring(replyChannels.update(_ - id)).forkDaemon.unless(beforeReplyChannels.contains(id)) - ) - .unit + private[shardcake] def isReplyRegistered(id: String): UIO[Boolean] = + pendingReplies.get.map(_.contains(id)) + + private[shardcake] def initReply(id: String, pendingReply: PendingReply): UIO[Unit] = + // Cheap read avoids the CAS + forkDaemon when a prior message in the same stream + // already registered this slot. Also guards against re-init on `offerToQueue` retry. + isReplyRegistered(id).flatMap { + case true => ZIO.unit + case false => + pendingReplies + .modify(map => if (map.contains(id)) (true, map) else (false, map.updated(id, pendingReply))) + .flatMap(alreadyPresent => + pendingReply.await.ensuring(pendingReplies.update(_ - id)).forkDaemon.unless(alreadyPresent) + ) + .unit + } def reply[Reply](reply: Reply, replier: Replier[Reply]): UIO[Unit] = - replyChannels + pendingReplies .modify(repliers => (repliers.get(replier.id), repliers - replier.id)) - .flatMap(ZIO.foreachDiscard(_)(_.asInstanceOf[ReplyChannel[Reply]].replySingle(reply))) + .flatMap { + case None => ZIO.unit + case Some(PendingReply.Value(channel)) => + channel.asInstanceOf[ReplyChannel[Reply]].replySingle(reply) + case Some(PendingReply.Bytes(channel, encode)) => + ZIO + .attempt(encode(reply)) + .foldCauseZIO(channel.fail, channel.replySingle) + } def replyStream[Reply](replies: ZStream[Any, Nothing, Reply], replier: StreamReplier[Reply]): UIO[Unit] = - replyChannels + pendingReplies .modify(repliers => (repliers.get(replier.id), repliers - replier.id)) - .flatMap(ZIO.foreachDiscard(_)(_.asInstanceOf[ReplyChannel[Reply]].replyStream(replies))) + .flatMap { + case None => ZIO.unit + case Some(PendingReply.Value(channel)) => + channel.asInstanceOf[ReplyChannel[Reply]].replyStream(replies) + case Some(PendingReply.Bytes(channel, encode)) => + channel.replyStream( + replies.mapZIO(r => ZIO.attempt(encode(r))) + ) + } private def handleError(ex: Throwable): ZIO[Any, Nothing, Unit] = ZIO @@ -270,7 +292,16 @@ class Sharding private ( entityStates.get.flatMap( _.get(recipientTypeName) match { case Some(state) => - state.entityManager.asInstanceOf[EntityManager[Msg]].send(entityId, msg, replyId, replyChannel) + val em = state.entityManager.asInstanceOf[EntityManager[Msg]] + replyId match { + case Some(rid) => + isReplyRegistered(rid).flatMap { + case true => em.send(entityId, msg, replyId, None) + case false => em.send(entityId, msg, replyId, Some(PendingReply.Value(replyChannel))) + } + case None => + em.send(entityId, msg, None, Some(PendingReply.Value(replyChannel))) + } case None => ZIO.fail(new Exception(s"Entity type $recipientTypeName was not registered.")) } @@ -280,6 +311,8 @@ class Sharding private ( recipientTypeName: String, entityId: String, pod: PodAddress, + codec: MessageCodec[Msg], + replyDecoder: Array[Byte] => Res, sendChannel: SendChannel[Msg], replyChannel: ReplyChannel[Res], replyId: Option[String] @@ -287,25 +320,25 @@ class Sharding private ( if (pod == address && !config.simulateRemotePods) { val run = sendChannel.foreach(sendToSelf(recipientTypeName, entityId, _, replyId, replyChannel)) sendChannel match { - case _: SendChannel.Single[_] => run - case _: SendChannel.Stream[_] => if (replyId.isDefined) (run race replyChannel.await).fork.unit else run + case _: SendChannel.Single[?] => run + case _: SendChannel.Stream[?] => if (replyId.isDefined) (run race replyChannel.await).fork.unit else run } } else { replyChannel match { - case _: ReplyChannel.FromPromise[_] => + case _: ReplyChannel.FromPromise[?] => sendChannel - .send(pods, serialization, pod, entityId, recipientTypeName, replyId) + .send(pods, codec, pod, entityId, recipientTypeName, replyId) .tapError(handleError) .flatMap { - case Some(bytes) => serialization.decode[Res](bytes).flatMap(replyChannel.replySingle) + case Some(bytes) => ZIO.attempt(replyDecoder(bytes)).flatMap(replyChannel.replySingle) case None => replyChannel.end } - case _: ReplyChannel.FromQueue[_] => + case _: ReplyChannel.FromQueue[?] => replyChannel.replyStream( sendChannel - .sendAndReceiveStream(pods, serialization, pod, entityId, recipientTypeName, replyId) + .sendAndReceiveStream(pods, codec, pod, entityId, recipientTypeName, replyId) .tapError(handleError) - .mapChunksZIO(serialization.decodeChunk[Res]) + .mapChunksZIO(chunk => ZIO.attempt(chunk.map(replyDecoder))) ) } } @@ -315,6 +348,7 @@ class Sharding private ( sendTimeout: MessengerTimeout = MessengerTimeout.InheritConfigTimeout ): Messenger[Msg] = new Messenger[Msg] { + private val codec = entityType.codec private val timeout = sendTimeout match { case MessengerTimeout.NoTimeout => None case MessengerTimeout.InheritConfigTimeout => Some(config.sendTimeout) @@ -322,14 +356,15 @@ class Sharding private ( } def sendDiscard(entityId: String)(msg: Msg): Task[Unit] = { - val send = sendMessage(entityId, msg, None) + val send = sendMessage(entityId, msg, None, _ => ()) timeout.fold(send.unit)(t => send.timeout(t).unit) } - def send[Res](entityId: String)(msg: Replier[Res] => Msg): Task[Res] = + def send[Res](entityId: String)(build: Replier[Res] => Msg): Task[Res] = Random.nextUUID.flatMap { uuid => - val body = msg(Replier(uuid.toString)) - val send = sendMessage[Res](entityId, body, Some(uuid.toString)).flatMap { + val body = build(Replier(uuid.toString)) + val decoder = codec.replyDecoder[Res](body) + val send = sendMessage[Res](entityId, body, Some(uuid.toString), decoder).flatMap { case Some(value) => ZIO.succeed(value) case None => ZIO.fail(new Exception(s"Send returned nothing, entityId=$entityId, body=$body")) } @@ -338,112 +373,107 @@ class Sharding private ( def sendAndReceiveStream[Res]( entityId: String - )(msg: StreamReplier[Res] => Msg): Task[ZStream[Any, Throwable, Res]] = + )(build: StreamReplier[Res] => Msg): Task[ZStream[Any, Throwable, Res]] = Random.nextUUID.flatMap { uuid => - sendMessageAndReceiveStream[Res](entityId, msg(StreamReplier(uuid.toString)), Some(uuid.toString)) + val body = build(StreamReplier(uuid.toString)) + val decoder = codec.streamReplyDecoder[Res](body) + sendMessageAndReceiveStream[Res](entityId, body, Some(uuid.toString), decoder) } def sendStream(entityId: String)(messages: ZStream[Any, Throwable, Msg]): Task[Unit] = { - val send = - ReplyChannel.single[Unit].flatMap[Any, Throwable, Unit](sendStreamGeneric(entityId, messages, None, _)) + val send = ReplyChannel.single[Unit].flatMap(sendStreamGeneric(entityId, messages, None, _, _ => ())) timeout.fold(send)(t => send.timeout(t).unit) } - def sendStreamAndReceiveStream[Res](entityId: String)( - messages: StreamReplier[Res] => ZStream[Any, Throwable, Msg] - ): Task[ZStream[Any, Throwable, Res]] = + def sendStreamAndReceiveStream[Res]( + entityId: String + )(request: StreamReplier[Res] => Msg, rest: ZStream[Any, Throwable, Msg]): Task[ZStream[Any, Throwable, Res]] = Random.nextUUID.flatMap { uuid => - sendStreamAndReceiveStream[Res](entityId, messages(StreamReplier(uuid.toString)), Some(uuid.toString)) + val head = request(StreamReplier(uuid.toString)) + val decoder = codec.streamReplyDecoder[Res](head) + val fullStream = ZStream(head) ++ rest + sendStreamAndReceiveStream[Res](entityId, fullStream, Some(uuid.toString), decoder) } - private def sendMessage[Res](entityId: String, msg: Msg, replyId: Option[String]): Task[Option[Res]] = + private def sendMessage[Res]( + entityId: String, + msg: Msg, + replyId: Option[String], + replyDecoder: Array[Byte] => Res + ): Task[Option[Res]] = for { replyChannel <- ReplyChannel.single[Res] - _ <- sendMessageGeneric(entityId, msg, replyId, replyChannel) + _ <- sendMessageGeneric(entityId, msg, replyId, replyChannel, replyDecoder) res <- replyChannel.output } yield res private def sendMessageAndReceiveStream[Res]( entityId: String, msg: Msg, - replyId: Option[String] + replyId: Option[String], + replyDecoder: Array[Byte] => Res ): Task[ZStream[Any, Throwable, Res]] = for { replyChannel <- ReplyChannel.stream[Res] - _ <- sendMessageGeneric(entityId, msg, replyId, replyChannel) + _ <- sendMessageGeneric(entityId, msg, replyId, replyChannel, replyDecoder) } yield replyChannel.output private def sendStreamAndReceiveStream[Res]( entityId: String, messages: ZStream[Any, Throwable, Msg], - replyId: Option[String] + replyId: Option[String], + replyDecoder: Array[Byte] => Res ): Task[ZStream[Any, Throwable, Res]] = for { replyChannel <- ReplyChannel.stream[Res] - _ <- sendStreamGeneric(entityId, messages, replyId, replyChannel) + _ <- sendStreamGeneric(entityId, messages, replyId, replyChannel, replyDecoder) } yield replyChannel.output private def sendMessageGeneric[Res]( entityId: String, msg: Msg, replyId: Option[String], - replyChannel: ReplyChannel[Res] - ): Task[Unit] = { - val shardId = getShardId(entityType, entityId) - def trySend: Task[Unit] = - for { - shards <- shardAssignments.get - pod = shards.get(shardId) - _ <- pod match { - case Some(pod) => - sendToPod[Msg, Res]( - entityType.name, - entityId, - pod, - SendChannel.single(msg), - replyChannel, - replyId - ).catchSome { case _: EntityNotManagedByThisPod | _: PodUnavailable => - Clock.sleep(200.millis) *> trySend - }.onError(replyChannel.fail) - case None => - // no shard assignment, retry - Clock.sleep(100.millis) *> trySend - } - } yield () - - if (shardId >= 1 && shardId <= config.numberOfShards) trySend - else ZIO.fail(InvalidShardId(entityId, shardId)) - } + replyChannel: ReplyChannel[Res], + replyDecoder: Array[Byte] => Res + ): Task[Unit] = + sendGeneric(entityId, SendChannel.single(msg), replyId, replyChannel, replyDecoder) private def sendStreamGeneric[Res]( entityId: String, messages: ZStream[Any, Throwable, Msg], replyId: Option[String], - replyChannel: ReplyChannel[Res] + replyChannel: ReplyChannel[Res], + replyDecoder: Array[Byte] => Res + ): Task[Unit] = + sendGeneric(entityId, SendChannel.stream(messages), replyId, replyChannel, replyDecoder) + + private def sendGeneric[Res]( + entityId: String, + sendChannel: SendChannel[Msg], + replyId: Option[String], + replyChannel: ReplyChannel[Res], + replyDecoder: Array[Byte] => Res ): Task[Unit] = { val shardId = getShardId(entityType, entityId) def trySend: Task[Unit] = - for { - shards <- shardAssignments.get - pod = shards.get(shardId) - _ <- pod match { - case Some(pod) => - sendToPod[Msg, Res]( - entityType.name, - entityId, - pod, - SendChannel.stream(messages), - replyChannel, - replyId - ).catchSome { case _: EntityNotManagedByThisPod | _: PodUnavailable => - Clock.sleep(200.millis) *> trySend - }.onError(replyChannel.fail) - case None => - // no shard assignment, retry - Clock.sleep(100.millis) *> trySend - } - } yield () + shardAssignments.get.map(_.get(shardId)).flatMap { + case Some(pod) => + sendToPod[Msg, Res]( + entityType.name, + entityId, + pod, + codec, + replyDecoder, + sendChannel, + replyChannel, + replyId + ).catchSome { case _: EntityNotManagedByThisPod | _: PodUnavailable => + Clock.sleep(200.millis) *> trySend + }.onError(replyChannel.fail) + case None => + // no shard assignment, retry + Clock.sleep(100.millis) *> trySend + } if (shardId >= 1 && shardId <= config.numberOfShards) trySend else ZIO.fail(InvalidShardId(entityId, shardId)) @@ -455,6 +485,7 @@ class Sharding private ( sendTimeout: MessengerTimeout = MessengerTimeout.InheritConfigTimeout ): Broadcaster[Msg] = new Broadcaster[Msg] { + private val codec = topicType.codec private val timeout = sendTimeout match { case MessengerTimeout.NoTimeout => None case MessengerTimeout.InheritConfigTimeout => Some(config.sendTimeout) @@ -462,15 +493,21 @@ class Sharding private ( } def broadcastDiscard(topic: String)(msg: Msg): UIO[Unit] = - sendMessage(topic, msg, None).unit + sendMessage(topic, msg, None, _ => ()).unit - def broadcast[Res](topic: String)(msg: Replier[Res] => Msg): UIO[Map[PodAddress, Try[Res]]] = + def broadcast[Res](topic: String)(build: Replier[Res] => Msg): UIO[Map[PodAddress, Try[Res]]] = Random.nextUUID.flatMap { uuid => - val body = msg(Replier(uuid.toString)) - sendMessage[Res](topic, body, Some(uuid.toString)).interruptible + val body = build(Replier(uuid.toString)) + val decoder = codec.replyDecoder[Res](body) + sendMessage[Res](topic, body, Some(uuid.toString), decoder).interruptible } - private def sendMessage[Res](topic: String, msg: Msg, replyId: Option[String]): UIO[Map[PodAddress, Try[Res]]] = + private def sendMessage[Res]( + topic: String, + msg: Msg, + replyId: Option[String], + replyDecoder: Array[Byte] => Res + ): UIO[Map[PodAddress, Try[Res]]] = for { pods <- getPods res <- ZIO @@ -478,10 +515,12 @@ class Sharding private ( def trySend: Task[Option[Res]] = for { replyChannel <- ReplyChannel.single[Res] - _ <- sendToPod( + _ <- sendToPod[Msg, Res]( topicType.name, topic, pod, + codec, + replyDecoder, SendChannel.single(msg), replyChannel, replyId @@ -543,15 +582,38 @@ class Sharding private ( entityMaxIdleTime, loadEntity ) - processBinary = (msg: BinaryMessage, replyChannel: ReplyChannel[Nothing]) => - serialization - .decode[Req](msg.body) - .flatMap(entityManager.send(msg.entityId, _, msg.replyId, replyChannel)) + processBinary = (msg: BinaryMessage, replyChannel: ReplyChannel[Array[Byte]]) => + ZIO + .attempt(recipientType.codec.decodeMessage(msg.body)) + .flatMap { decoded => + msg.replyId match { + case None => + entityManager.send( + msg.entityId, + decoded, + None, + Some(PendingReply.Bytes(replyChannel, Sharding.noReplyEncoder)) + ) + case Some(replyId) => + isReplyRegistered(replyId).flatMap { + case true => + entityManager.send(msg.entityId, decoded, msg.replyId, None) + case false => + val encoder = recipientType.codec.replyEncoder(decoded) + entityManager.send( + msg.entityId, + decoded, + msg.replyId, + Some(PendingReply.Bytes(replyChannel, encoder)) + ) + } + } + } .catchAllCause(replyChannel.fail) _ <- entityStates.update(_.updated(recipientType.name, EntityState(entityManager, processBinary))) } yield () - def terminateLocalEntity(entityType: EntityType[_], entityId: String): UIO[Unit] = + def terminateLocalEntity(entityType: EntityType[?], entityId: String): UIO[Unit] = entityStates.get.flatMap(_.get(entityType.name) match { case Some(state) => state.entityManager.terminateEntity(entityId) case None => ZIO.unit @@ -563,33 +625,34 @@ object Sharding { sealed trait ShardingRegistrationEvent object ShardingRegistrationEvent { - case class EntityRegistered(entityType: EntityType[_]) extends ShardingRegistrationEvent { + case class EntityRegistered(entityType: EntityType[?]) extends ShardingRegistrationEvent { override def toString: String = s"Registered entity ${entityType.name}" } case class SingletonRegistered(name: String) extends ShardingRegistrationEvent { override def toString: String = s"Registered singleton $name" } - case class TopicRegistered(topicType: TopicType[_]) extends ShardingRegistrationEvent { + case class TopicRegistered(topicType: TopicType[?]) extends ShardingRegistrationEvent { override def toString: String = s"Registered topic ${topicType.name}" } } private[shardcake] case class EntityState( entityManager: EntityManager[Nothing], - processBinary: (BinaryMessage, ReplyChannel[Nothing]) => UIO[Unit] + processBinary: (BinaryMessage, ReplyChannel[Array[Byte]]) => UIO[Unit] ) + private val noReplyEncoder: Any => Array[Byte] = _ => Array.emptyByteArray + /** * A layer that sets up sharding communication between pods. */ - val live: ZLayer[Pods with ShardManagerClient with Storage with Serialization with Config, Throwable, Sharding] = + val live: ZLayer[Pods with ShardManagerClient with Storage with Config, Throwable, Sharding] = ZLayer.scoped { for { config <- ZIO.service[Config] pods <- ZIO.service[Pods] shardManager <- ZIO.service[ShardManagerClient] storage <- ZIO.service[Storage] - serialization <- ZIO.service[Serialization] shardsCache <- Ref.make(Map.empty[ShardId, PodAddress]) entityStates <- Ref.make[Map[String, EntityState]](Map()) singletons <- Ref.Synchronized @@ -602,7 +665,7 @@ object Sharding { } ) ) - replyChannels <- Ref.make[Map[String, ReplyChannel[Nothing]]](Map()) + pendingReplies <- Ref.make[Map[String, PendingReply]](Map()) cdt <- Clock.currentDateTime lastUnhealthyNodeReported <- Ref.make(cdt) shuttingDown <- Ref.make(false) @@ -613,13 +676,12 @@ object Sharding { shardsCache, entityStates, singletons, - replyChannels, + pendingReplies, lastUnhealthyNodeReported, shuttingDown, shardManager, pods, storage, - serialization, eventsHub ) _ <- sharding.getShardingRegistrationEvents.mapZIO(event => ZIO.logInfo(event.toString)).runDrain.forkDaemon @@ -739,6 +801,6 @@ object Sharding { * This method can only be used if the entity is hosted on the current pod (otherwise it will do nothing). * Typically, you would use this method from inside the entity behavior to stop itself. */ - def terminateLocalEntity(entityType: EntityType[_], entityId: String): URIO[Sharding, Unit] = + def terminateLocalEntity(entityType: EntityType[?], entityId: String): URIO[Sharding, Unit] = ZIO.serviceWithZIO[Sharding](_.terminateLocalEntity(entityType, entityId)) } diff --git a/entities/src/main/scala/com/devsisters/shardcake/internal/EntityManager.scala b/entities/src/main/scala/com/devsisters/shardcake/internal/EntityManager.scala index 52efeab4..a88b2dcf 100644 --- a/entities/src/main/scala/com/devsisters/shardcake/internal/EntityManager.scala +++ b/entities/src/main/scala/com/devsisters/shardcake/internal/EntityManager.scala @@ -11,7 +11,7 @@ private[shardcake] trait EntityManager[-Req] { entityId: String, req: Req, replyId: Option[String], - replyChannel: ReplyChannel[Nothing] + pendingReply: Option[PendingReply] ): IO[EntityNotManagedByThisPod, Unit] def terminateEntity(entityId: String): UIO[Unit] def terminateEntitiesOnShards(shards: Set[ShardId]): UIO[Unit] @@ -104,7 +104,7 @@ private[shardcake] object EntityManager { entityId: String, req: Req, replyId: Option[String], - replyChannel: ReplyChannel[Nothing] + pendingReply: Option[PendingReply] ): IO[EntityNotManagedByThisPod, Unit] = for { // first, verify that this entity should be handled by this pod @@ -119,16 +119,16 @@ private[shardcake] object EntityManager { map <- entities.get _ <- map.get(entityId) match { case Some(Left(queue)) => - offerToQueue(entityId, queue, req, replyId, replyChannel) + offerToQueue(entityId, queue, req, replyId, pendingReply) case None if !loadEntity(req) => - replyChannel.end + pendingReply.fold[UIO[Unit]](ZIO.unit)(_.end) case _ => getOrCreateQueue(entityId).flatMap { case Right(_) => // the queue is shutting down, try again a little later - Clock.sleep(100 millis) *> send(entityId, req, replyId, replyChannel) + Clock.sleep(100 millis) *> send(entityId, req, replyId, pendingReply) case Left(queue) => - offerToQueue(entityId, queue, req, replyId, replyChannel) + offerToQueue(entityId, queue, req, replyId, pendingReply) } } } yield () @@ -138,14 +138,16 @@ private[shardcake] object EntityManager { queue: Queue[Req], req: Req, replyId: Option[String], - replyChannel: ReplyChannel[Nothing] + pendingReply: Option[PendingReply] ): IO[EntityNotManagedByThisPod, Unit] = currentTimeInMilliseconds.flatMap(cdt => entitiesLastReceivedAt.update(_ + (entityId -> cdt))) *> // add the message to the queue and setup the reply channel if needed - (replyId match { - case Some(replyId) => sharding.initReply(replyId, replyChannel) <* queue.offer(req) - case None => queue.offer(req) *> replyChannel.end - }).catchAllCause(_ => Clock.sleep(100 millis) *> send(entityId, req, replyId, replyChannel)) + ((replyId, pendingReply) match { + case (Some(replyId), Some(pr)) => sharding.initReply(replyId, pr) <* queue.offer(req) + case (None, Some(pr)) => queue.offer(req) *> pr.end + // stream continuation: reply is already set up on a prior message — just enqueue + case (_, None) => queue.offer(req).unit + }).catchAllCause(_ => Clock.sleep(100 millis) *> send(entityId, req, replyId, pendingReply)) private def getOrCreateQueue(entityId: String): IO[EntityNotManagedByThisPod, Either[Queue[Req], Signal]] = entities.modifyZIO(map => diff --git a/entities/src/main/scala/com/devsisters/shardcake/internal/GraphQLClient.scala b/entities/src/main/scala/com/devsisters/shardcake/internal/GraphQLClient.scala index 4508b409..8e519e0e 100644 --- a/entities/src/main/scala/com/devsisters/shardcake/internal/GraphQLClient.scala +++ b/entities/src/main/scala/com/devsisters/shardcake/internal/GraphQLClient.scala @@ -101,7 +101,7 @@ private[shardcake] object GraphQLClient { encoder0: ArgEncoder[PodAddressInput], encoder1: ArgEncoder[String], encoder2: ArgEncoder[RoleInput] - ): SelectionBuilder[_root_.caliban.client.Operations.RootMutation, scala.Option[Unit]] = + ): SelectionBuilder[_root_.caliban.client.Operations.RootMutation, scala.Option[Unit]] = _root_.caliban.client.SelectionBuilder.Field( "register", OptionOf(Scalar()), @@ -113,7 +113,7 @@ private[shardcake] object GraphQLClient { ) def unregister(podAddress: PodAddressInput)(implicit encoder0: ArgEncoder[PodAddressInput] - ): SelectionBuilder[_root_.caliban.client.Operations.RootMutation, scala.Option[Unit]] = + ): SelectionBuilder[_root_.caliban.client.Operations.RootMutation, scala.Option[Unit]] = _root_.caliban.client.SelectionBuilder.Field( "unregister", OptionOf(Scalar()), @@ -121,7 +121,7 @@ private[shardcake] object GraphQLClient { ) def notifyUnhealthyPod(podAddress: PodAddressInput)(implicit encoder0: ArgEncoder[PodAddressInput] - ): SelectionBuilder[_root_.caliban.client.Operations.RootMutation, Unit] = + ): SelectionBuilder[_root_.caliban.client.Operations.RootMutation, Unit] = _root_.caliban.client.SelectionBuilder.Field( "notifyUnhealthyPod", Scalar(), diff --git a/entities/src/main/scala/com/devsisters/shardcake/internal/PendingReply.scala b/entities/src/main/scala/com/devsisters/shardcake/internal/PendingReply.scala new file mode 100644 index 00000000..b6ba692a --- /dev/null +++ b/entities/src/main/scala/com/devsisters/shardcake/internal/PendingReply.scala @@ -0,0 +1,21 @@ +package com.devsisters.shardcake.internal + +import zio.{ Cause, UIO } + +/** + * A reply we're waiting on. Wraps the [[ReplyChannel]] that the incoming response will + * be pushed into, and records whether to push it as a raw typed value (local sends, no + * serialisation) or as pre-encoded bytes (remote sends, where the encoder supplied by + * the codec is applied). + */ +private[shardcake] sealed trait PendingReply { + protected def channel: ReplyChannel[?] + def fail(cause: Cause[Throwable]): UIO[Unit] = channel.fail(cause) + val await: UIO[Unit] = channel.await + val end: UIO[Unit] = channel.end +} + +private[shardcake] object PendingReply { + final case class Value(channel: ReplyChannel[Nothing]) extends PendingReply + final case class Bytes(channel: ReplyChannel[Array[Byte]], encode: Any => Array[Byte]) extends PendingReply +} diff --git a/entities/src/main/scala/com/devsisters/shardcake/internal/SendChannel.scala b/entities/src/main/scala/com/devsisters/shardcake/internal/SendChannel.scala index d358fa97..2cb49db9 100644 --- a/entities/src/main/scala/com/devsisters/shardcake/internal/SendChannel.scala +++ b/entities/src/main/scala/com/devsisters/shardcake/internal/SendChannel.scala @@ -1,16 +1,16 @@ package com.devsisters.shardcake.internal import com.devsisters.shardcake.PodAddress +import com.devsisters.shardcake.interfaces.{ MessageCodec, Pods } import com.devsisters.shardcake.interfaces.Pods.BinaryMessage -import com.devsisters.shardcake.interfaces.{ Pods, Serialization } -import zio.Task +import zio.{ Chunk, Task, ZIO } import zio.stream.ZStream -private[shardcake] sealed trait SendChannel[+A] { self => +private[shardcake] sealed trait SendChannel[A] { self => def foreach(f: A => Task[Unit]): Task[Unit] def send( pods: Pods, - serialization: Serialization, + codec: MessageCodec[A], pod: PodAddress, entityId: String, recipientTypeName: String, @@ -18,7 +18,7 @@ private[shardcake] sealed trait SendChannel[+A] { self => ): Task[Option[Array[Byte]]] def sendAndReceiveStream( pods: Pods, - serialization: Serialization, + codec: MessageCodec[A], pod: PodAddress, entityId: String, recipientTypeName: String, @@ -27,70 +27,70 @@ private[shardcake] sealed trait SendChannel[+A] { self => } private[shardcake] object SendChannel { - case class Single[A](msg: A) extends SendChannel[A] { + final case class Single[A](msg: A) extends SendChannel[A] { def foreach(f: A => Task[Unit]): Task[Unit] = f(msg) def send( pods: Pods, - serialization: Serialization, + codec: MessageCodec[A], pod: PodAddress, entityId: String, recipientTypeName: String, replyId: Option[String] - ): Task[Option[Array[Byte]]] = - serialization - .encode(msg) + ): Task[Option[Array[Byte]]] = + ZIO + .attempt(codec.encodeMessage(msg)) .flatMap(bytes => pods.sendMessage(pod, BinaryMessage(entityId, recipientTypeName, bytes, replyId))) def sendAndReceiveStream( pods: Pods, - serialization: Serialization, + codec: MessageCodec[A], pod: PodAddress, entityId: String, recipientTypeName: String, replyId: Option[String] - ): ZStream[Any, Throwable, Array[Byte]] = + ): ZStream[Any, Throwable, Array[Byte]] = ZStream.unwrap( - serialization - .encode(msg) + ZIO + .attempt(codec.encodeMessage(msg)) .map { bytes => val binaryMessage = BinaryMessage(entityId, recipientTypeName, bytes, replyId) pods.sendMessageAndReceiveStream(pod, binaryMessage) } ) } - case class Stream[A](messages: ZStream[Any, Throwable, A]) extends SendChannel[A] { + + final case class Stream[A](messages: ZStream[Any, Throwable, A]) extends SendChannel[A] { def foreach(f: A => Task[Unit]): Task[Unit] = messages.runForeach(f) def send( pods: Pods, - serialization: Serialization, + codec: MessageCodec[A], pod: PodAddress, entityId: String, recipientTypeName: String, replyId: Option[String] - ): Task[Option[Array[Byte]]] = - pods.sendStream( - pod, - entityId, - messages.mapChunksZIO(messages => - serialization - .encodeChunk(messages) - .map(_.map(bytes => BinaryMessage(entityId, recipientTypeName, bytes, replyId))) - ) - ) + ): Task[Option[Array[Byte]]] = { + val requestStream = messages.mapChunksZIO(encodeChunk(codec, _, entityId, recipientTypeName, replyId)) + pods.sendStream(pod, entityId, requestStream) + } def sendAndReceiveStream( pods: Pods, - serialization: Serialization, + codec: MessageCodec[A], pod: PodAddress, entityId: String, recipientTypeName: String, replyId: Option[String] - ): ZStream[Any, Throwable, Array[Byte]] = { - val requestStream = messages.mapChunksZIO(messages => - serialization - .encodeChunk(messages) - .map(_.map(bytes => BinaryMessage(entityId, recipientTypeName, bytes, replyId))) - ) + ): ZStream[Any, Throwable, Array[Byte]] = { + val requestStream = messages.mapChunksZIO(encodeChunk(codec, _, entityId, recipientTypeName, replyId)) pods.sendStreamAndReceiveStream(pod, entityId, requestStream) } + + private def encodeChunk( + codec: MessageCodec[A], + chunk: Chunk[A], + entityId: String, + recipientTypeName: String, + replyId: Option[String] + ): Task[Chunk[BinaryMessage]] = + ZIO.attempt(chunk.map(msg => BinaryMessage(entityId, recipientTypeName, codec.encodeMessage(msg), replyId))) } def single[A](msg: A): SendChannel[A] = diff --git a/entities/src/main/scala/com/devsisters/shardcake/javaSerialization/package.scala b/entities/src/main/scala/com/devsisters/shardcake/javaSerialization/package.scala new file mode 100644 index 00000000..284ed060 --- /dev/null +++ b/entities/src/main/scala/com/devsisters/shardcake/javaSerialization/package.scala @@ -0,0 +1,44 @@ +package com.devsisters.shardcake + +import java.io.{ ByteArrayInputStream, ByteArrayOutputStream, ObjectInputStream, ObjectOutputStream } + +import com.devsisters.shardcake.interfaces.MessageCodec + +/** + * Java-serialization backend given. + * + * Import with `import com.devsisters.shardcake.javaSerialization.given` to enable + * Java-serialization-backed transport for all `EntityType` / `TopicType` declarations + * in the surrounding scope. + * + * Mostly useful for tests and examples. For production deployments prefer + * `com.devsisters.shardcake.kryo` or a macro-derived backend. + */ +package object javaSerialization { + given derive[Msg]: MessageCodec[Msg] = new MessageCodec[Msg] { + private val universalEncoder: Any => Array[Byte] = (v: Any) => writeBytes(v) + + def encodeMessage(message: Msg): Array[Byte] = writeBytes(message) + + def decodeMessage(bytes: Array[Byte]): Msg = readBytes(bytes) + + def replyEncoder(decoded: Msg): Any => Array[Byte] = universalEncoder + + def replyDecoder[Res](sample: Msg): Array[Byte] => Res = bytes => readBytes[Res](bytes) + def streamReplyDecoder[Res](sample: Msg): Array[Byte] => Res = bytes => readBytes[Res](bytes) + + private def writeBytes(value: Any): Array[Byte] = { + val baos = new ByteArrayOutputStream + val oos = new ObjectOutputStream(baos) + try oos.writeObject(value) + finally oos.close() + baos.toByteArray + } + + private def readBytes[A](bytes: Array[Byte]): A = { + val ois = new ObjectInputStream(new ByteArrayInputStream(bytes)) + try ois.readObject().asInstanceOf[A] + finally ois.close() + } + } +} diff --git a/entities/src/test/scala/com/devsisters/shardcake/BroadcastingSpec.scala b/entities/src/test/scala/com/devsisters/shardcake/BroadcastingSpec.scala index b12a0fda..ade7564d 100644 --- a/entities/src/test/scala/com/devsisters/shardcake/BroadcastingSpec.scala +++ b/entities/src/test/scala/com/devsisters/shardcake/BroadcastingSpec.scala @@ -1,6 +1,7 @@ package com.devsisters.shardcake -import com.devsisters.shardcake.interfaces.{ Serialization, Storage } +import com.devsisters.shardcake.interfaces.Storage +import com.devsisters.shardcake.javaSerialization.given import zio.test.TestAspect.{ sequential, withLiveClock } import zio.test._ import zio.{ Config => _, _ } @@ -28,7 +29,6 @@ object BroadcastingSpec extends ZIOSpecDefault { } } ).provideShared( - Serialization.javaSerialization, LocalSharding.live, ShardManagerClient.local, Storage.memory, diff --git a/entities/src/test/scala/com/devsisters/shardcake/ShardingSpec.scala b/entities/src/test/scala/com/devsisters/shardcake/ShardingSpec.scala index d41abf1b..dcee08e3 100644 --- a/entities/src/test/scala/com/devsisters/shardcake/ShardingSpec.scala +++ b/entities/src/test/scala/com/devsisters/shardcake/ShardingSpec.scala @@ -2,7 +2,8 @@ package com.devsisters.shardcake import com.devsisters.shardcake.CounterActor.CounterMessage._ import com.devsisters.shardcake.CounterActor._ -import com.devsisters.shardcake.interfaces.{ Serialization, Storage } +import com.devsisters.shardcake.interfaces.Storage +import com.devsisters.shardcake.javaSerialization.given import zio.stream.{ SubscriptionRef, ZStream } import zio.test.TestAspect.{ sequential, withLiveClock } import zio.test._ @@ -71,7 +72,7 @@ object ShardingSpec extends ZIOSpecDefault { _ <- counter.sendDiscard("c3")(IncrementCounter) c0 <- counter.send("c3")(GetCounter.apply) _ <- Clock.sleep(3 seconds) - c1 <- counter.send("c3")(GetCounter.apply) // counter should be restarted + c1 <- counter.send("c3")(GetCounter.apply) // counter should be restarted } yield assertTrue(c0 == 1, c1 == 0) } }, @@ -84,7 +85,7 @@ object ShardingSpec extends ZIOSpecDefault { _ <- counter.sendDiscard("c3")(IncrementCounter) c0 <- counter.send("c3")(GetCounter.apply) _ <- counter.send("c3")(TriggerTerminate.apply) - c1 <- counter.send("c3")(GetCounter.apply) // counter should be restarted + c1 <- counter.send("c3")(GetCounter.apply) // counter should be restarted } yield assertTrue(c0 == 1, c1 == 0) } }, @@ -100,7 +101,7 @@ object ShardingSpec extends ZIOSpecDefault { _ <- counter.sendDiscard("c3")(IncrementCounter) c1 <- counter.send("c3")(GetCounter.apply) _ <- Clock.sleep(4 seconds) - c2 <- counter.send("c3")(GetCounter.apply) // counter should be restarted + c2 <- counter.send("c3")(GetCounter.apply) // counter should be restarted } yield assertTrue(c0 == 1, c1 == 2, c2 == 0) } }, @@ -131,8 +132,9 @@ object ShardingSpec extends ZIOSpecDefault { _ <- Sharding.registerEntity(Counter, behavior) _ <- Sharding.registerScoped counter <- Sharding.messenger(Counter) - stream <- counter.sendStreamAndReceiveStream[Int]("c1")(replier => - ZStream.succeed(StreamingChanges(replier)) ++ ZStream.fromIterable(1 to 5).as(IncrementCounter) + stream <- counter.sendStreamAndReceiveStream[Int]("c1")( + StreamingChanges.apply, + ZStream.fromIterable(1 to 5).as(IncrementCounter) ) latch <- Promise.make[Nothing, Unit] fiber <- stream.take(5).tap(_ => latch.succeed(())).runCollect.fork @@ -144,7 +146,6 @@ object ShardingSpec extends ZIOSpecDefault { } } ).provideShared( - Serialization.javaSerialization, LocalSharding.live, ShardManagerClient.local, Storage.memory, diff --git a/examples/src/main/scala/example/complex/GuildApp.scala b/examples/src/main/scala/example/complex/GuildApp.scala index c0278acf..d81a945a 100644 --- a/examples/src/main/scala/example/complex/GuildApp.scala +++ b/examples/src/main/scala/example/complex/GuildApp.scala @@ -1,14 +1,11 @@ package example.complex import com.devsisters.shardcake._ -import com.devsisters.shardcake.interfaces.Serialization import dev.profunktor.redis4cats.RedisCommands import example.complex.GuildBehavior.GuildMessage.{ Join, Terminate } import example.complex.GuildBehavior._ import zio.{ Config => _, _ } -import scala.collection.compat._ - object GuildApp extends ZIOAppDefault { val config: ZLayer[Any, SecurityException, Config] = ZLayer( @@ -17,7 +14,7 @@ object GuildApp extends ZIOAppDefault { .map(_.flatMap(_.toIntOption).fold(Config.default)(port => Config.default.copy(shardingPort = port))) ) - val program: ZIO[Sharding with Scope with Serialization with RedisCommands[Task, String, String], Throwable, Unit] = + val program: ZIO[Sharding with Scope with RedisCommands[Task, String, String], Throwable, Unit] = for { _ <- Sharding.registerEntity(Guild, behavior, p => Some(Terminate(p))) _ <- Sharding.registerScoped @@ -40,7 +37,6 @@ object GuildApp extends ZIOAppDefault { ZLayer.succeed(RedisConfig.default), redis, StorageRedis.live, - KryoSerialization.live, ShardManagerClient.liveWithSttp, GrpcPods.live, Sharding.live, diff --git a/examples/src/main/scala/example/complex/GuildBehavior.scala b/examples/src/main/scala/example/complex/GuildBehavior.scala index 1c0ad936..bf10cf68 100644 --- a/examples/src/main/scala/example/complex/GuildBehavior.scala +++ b/examples/src/main/scala/example/complex/GuildBehavior.scala @@ -1,6 +1,7 @@ package example.complex import com.devsisters.shardcake.{ EntityType, Replier, Sharding } +import com.devsisters.shardcake.kryo.given import dev.profunktor.redis4cats.RedisCommands import zio.{ Dequeue, Promise, RIO, Task, ZIO } diff --git a/examples/src/main/scala/example/simple/GuildApp.scala b/examples/src/main/scala/example/simple/GuildApp.scala index e6d407d5..e314d42e 100644 --- a/examples/src/main/scala/example/simple/GuildApp.scala +++ b/examples/src/main/scala/example/simple/GuildApp.scala @@ -26,7 +26,6 @@ object GuildApp extends ZIOAppDefault { .provide( ZLayer.succeed(Config.default), ZLayer.succeed(GrpcConfig.default), - Serialization.javaSerialization, Storage.memory, ShardManagerClient.liveWithSttp, GrpcPods.live, diff --git a/examples/src/main/scala/example/simple/GuildBehavior.scala b/examples/src/main/scala/example/simple/GuildBehavior.scala index 4b1ec15b..b275f1ab 100644 --- a/examples/src/main/scala/example/simple/GuildBehavior.scala +++ b/examples/src/main/scala/example/simple/GuildBehavior.scala @@ -1,6 +1,7 @@ package example.simple import com.devsisters.shardcake.{ EntityType, Replier, Sharding, StreamReplier } +import com.devsisters.shardcake.javaSerialization.given import zio.stream.ZStream import zio.{ Dequeue, RIO, Ref, ZIO } diff --git a/examples/src/test/scala/example/EndToEndSpec.scala b/examples/src/test/scala/example/EndToEndSpec.scala index 3f909e53..a8893a35 100644 --- a/examples/src/test/scala/example/EndToEndSpec.scala +++ b/examples/src/test/scala/example/EndToEndSpec.scala @@ -96,7 +96,6 @@ object EndToEndSpec extends ZIOSpecDefault { } ).provideShared( Sharding.live, - KryoSerialization.live, GrpcPods.live, ShardManagerClient.liveWithSttp, StorageRedis.live, diff --git a/examples/src/test/scala/example/GrpcAuthExampleSpec.scala b/examples/src/test/scala/example/GrpcAuthExampleSpec.scala index 9d25520d..8d242704 100644 --- a/examples/src/test/scala/example/GrpcAuthExampleSpec.scala +++ b/examples/src/test/scala/example/GrpcAuthExampleSpec.scala @@ -54,7 +54,6 @@ object GrpcAuthExampleSpec extends ZIOSpecDefault { config, grpcConfigLayer(validAuthenticationKey), Sharding.live, - KryoSerialization.live, GrpcPods.live, GrpcShardingService.live ) diff --git a/manager/src/main/scala/com/devsisters/shardcake/ShardManager.scala b/manager/src/main/scala/com/devsisters/shardcake/ShardManager.scala index 0a36957a..6b73e0d3 100644 --- a/manager/src/main/scala/com/devsisters/shardcake/ShardManager.scala +++ b/manager/src/main/scala/com/devsisters/shardcake/ShardManager.scala @@ -9,7 +9,6 @@ import zio._ import zio.stream.ZStream import scala.annotation.tailrec -import scala.collection.compat._ /** * A component in charge of assigning and unassigning shards to/from pods @@ -297,7 +296,7 @@ object ShardManager { ZIO.whenDiscard(failedAssignments.nonEmpty)( ZIO.logWarning( s"Ignoring assignments for pods that are no longer alive for role ${role.name}: ${failedAssignments - .mkString("[", ", ", "]")}" + .mkString("[", ", ", "]")}" ) ) _ <- @@ -491,7 +490,7 @@ object ShardManager { // don't assign too many shards to the same pods, unless we need rebalance immediately .filter { case (podAddress, _) => rebalanceImmediately || - assignments.count { case (_, p) => p == podAddress } < state.shards.size * rebalanceRate + assignments.count { case (_, p) => p == podAddress } < state.shards.size * rebalanceRate } // don't assign to a pod that was unassigned in the same rebalance .filterNot { case (podAddress, _) => unassignedPods.contains(podAddress) } diff --git a/manager/src/test/scala/com/devsisters/shardcake/ShardManagerSpec.scala b/manager/src/test/scala/com/devsisters/shardcake/ShardManagerSpec.scala index 941b5a06..433bef10 100644 --- a/manager/src/test/scala/com/devsisters/shardcake/ShardManagerSpec.scala +++ b/manager/src/test/scala/com/devsisters/shardcake/ShardManagerSpec.scala @@ -222,8 +222,8 @@ object ShardManagerSpec extends ZIOSpecDefault { shutdownAssignments <- ZIO.serviceWithZIO[Storage](_.getAssignments(role)) shutdownPods <- ZIO.serviceWithZIO[Storage](_.getPods) } yield - // manager should have saved its state to storage when it shut down - assertTrue(shutdownAssignments.nonEmpty && shutdownPods.nonEmpty) + // manager should have saved its state to storage when it shut down + assertTrue(shutdownAssignments.nonEmpty && shutdownPods.nonEmpty) setup *> test }.provide(Storage.memory) diff --git a/project/build.properties b/project/build.properties index 3093542b..c93d181b 100644 --- a/project/build.properties +++ b/project/build.properties @@ -1 +1 @@ -sbt.version = 1.11.1 +sbt.version = 1.12.11 diff --git a/project/plugins.sbt b/project/plugins.sbt index eb982250..fdd6a6ca 100644 --- a/project/plugins.sbt +++ b/project/plugins.sbt @@ -1,3 +1,3 @@ -addSbtPlugin("org.scalameta" % "sbt-scalafmt" % "2.4.6") +addSbtPlugin("org.scalameta" % "sbt-scalafmt" % "2.6.1") addSbtPlugin("com.github.sbt" % "sbt-ci-release" % "1.11.1") addSbtPlugin("pl.project13.scala" % "sbt-jmh" % "0.4.7") diff --git a/protocol-grpc/src/main/protobuf/sharding.proto b/protocol-grpc/src/main/protobuf/sharding.proto index bb991567..2d5a04df 100644 --- a/protocol-grpc/src/main/protobuf/sharding.proto +++ b/protocol-grpc/src/main/protobuf/sharding.proto @@ -30,7 +30,7 @@ message SendRequest { } message SendResponse { - bytes body = 1; + optional bytes body = 1; } message PingShardsRequest {} diff --git a/protocol-grpc/src/main/scala/com/devsisters/shardcake/GrpcPods.scala b/protocol-grpc/src/main/scala/com/devsisters/shardcake/GrpcPods.scala index ebdb0c58..0b608f3b 100644 --- a/protocol-grpc/src/main/scala/com/devsisters/shardcake/GrpcPods.scala +++ b/protocol-grpc/src/main/scala/com/devsisters/shardcake/GrpcPods.scala @@ -26,7 +26,7 @@ class GrpcPods( map.get(pod) match { case Some((client, _)) => ZIO.succeed((client, map)) case None => - val builder = { + val builder = config.executor match { case Some(executor) => ManagedChannelBuilder @@ -40,7 +40,6 @@ class GrpcPods( .maxInboundMessageSize(config.maxInboundMessageSize) .usePlaintext() } - } val acquireChannel: RIO[Scope, ManagedChannel] = ZIO.acquireRelease(ZIO.attempt(builder.build())) { channel => @@ -106,7 +105,7 @@ class GrpcPods( .send(toSendRequest(message)) .mapBoth( mapClientError(pod, message.entityId, isStream = false), - res => if (res.body.isEmpty) None else Some(res.body) + _.body ) } @@ -120,7 +119,7 @@ class GrpcPods( .sendStream(messages.mapBoth(Status.INTERNAL.withCause(_).asException(), toSendRequest)) .mapBoth( mapClientError(pod, entityId, isStream = true), - res => if (res.body.isEmpty) None else Some(res.body) + _.body ) } @@ -129,7 +128,7 @@ class GrpcPods( .fromZIO(getConnection(pod)) .flatMap( _.sendAndReceiveStream(toSendRequest(message)) - .mapBoth(mapClientError(pod, message.entityId, isStream = true), _.body) + .mapBoth(mapClientError(pod, message.entityId, isStream = true), _.body.getOrElse(Array.emptyByteArray)) ) def sendStreamAndReceiveStream( @@ -142,7 +141,7 @@ class GrpcPods( .flatMap( _.sendStreamAndReceiveStream( messages.mapBoth(Status.INTERNAL.withCause(_).asException(), toSendRequest) - ).mapBoth(mapClientError(pod, entityId, isStream = true), _.body) + ).mapBoth(mapClientError(pod, entityId, isStream = true), _.body.getOrElse(Array.emptyByteArray)) ) } diff --git a/protocol-grpc/src/main/scala/com/devsisters/shardcake/GrpcShardingService.scala b/protocol-grpc/src/main/scala/com/devsisters/shardcake/GrpcShardingService.scala index 9f8724f9..9eda3368 100644 --- a/protocol-grpc/src/main/scala/com/devsisters/shardcake/GrpcShardingService.scala +++ b/protocol-grpc/src/main/scala/com/devsisters/shardcake/GrpcShardingService.scala @@ -22,7 +22,7 @@ abstract class GrpcShardingService(sharding: Sharding, timeout: Duration) { def send(request: SendRequest): ZIO[Any, StatusException, SendResponse] = sharding .sendToLocalEntity(GrpcShardingService.toBinary(request)) - .map(GrpcShardingService.toSendResponse) + .map(SendResponse(_)) .mapError(GrpcShardingService.mapErrorToStatus) .timeoutFail(GrpcShardingService.timeoutException)(timeout) @@ -31,13 +31,13 @@ abstract class GrpcShardingService(sharding: Sharding, timeout: Duration) { ): ZIO[Any, StatusException, SendResponse] = sharding .sendStreamToLocalEntity(requests.map(GrpcShardingService.toBinary)) - .map(GrpcShardingService.toSendResponse) + .map(SendResponse(_)) .mapError(GrpcShardingService.mapErrorToStatus) def sendAndReceiveStream(request: SendRequest): ZStream[Any, StatusException, SendResponse] = sharding .sendToLocalEntityAndReceiveStream(GrpcShardingService.toBinary(request)) - .map(SendResponse(_)) + .map(bytes => SendResponse(Some(bytes))) .mapError(GrpcShardingService.mapErrorToStatus) def sendStreamAndReceiveStream( @@ -45,7 +45,7 @@ abstract class GrpcShardingService(sharding: Sharding, timeout: Duration) { ): ZStream[Any, StatusException, SendResponse] = sharding .sendStreamToLocalEntityAndReceiveStream(requests.map(GrpcShardingService.toBinary)) - .map(SendResponse(_)) + .map(bytes => SendResponse(Some(bytes))) .mapError(GrpcShardingService.mapErrorToStatus) def pingShards(request: PingShardsRequest): ZIO[Any, StatusException, PingShardsResponse] = @@ -57,14 +57,9 @@ object GrpcShardingService { private[shardcake] val timeoutException: StatusException = Status.ABORTED.withDescription("Timeout while handling sharding send grpc").asException() - private val emptySendResponse: SendResponse = SendResponse(Array.emptyByteArray) - private[shardcake] def toBinary(req: SendRequest): BinaryMessage = BinaryMessage(req.entityId, req.entityType, req.body, req.replyId) - private[shardcake] def toSendResponse(body: Option[Array[Byte]]): SendResponse = - body.fold(emptySendResponse)(SendResponse(_)) - private[shardcake] val mapErrorToStatus: Throwable => StatusException = { case e: StatusException => e case e: StatusRuntimeException => e.getStatus.asException() diff --git a/protocol-grpc/src/main/scala/com/devsisters/shardcake/ShardingServerInterceptor.scala b/protocol-grpc/src/main/scala/com/devsisters/shardcake/ShardingServerInterceptor.scala index b7f02d36..052a9b5a 100644 --- a/protocol-grpc/src/main/scala/com/devsisters/shardcake/ShardingServerInterceptor.scala +++ b/protocol-grpc/src/main/scala/com/devsisters/shardcake/ShardingServerInterceptor.scala @@ -72,10 +72,8 @@ object ShardingServerInterceptor { io: GrpcContext => UnaryEffect[Resp] ): Req => GrpcContext => UnaryEffect[Resp] = interceptors.foldRight(((_: Req) => io): Req => GrpcContext => UnaryEffect[Resp]) { (i, acc) => req => - { - val applied: Req => GrpcContext => UnaryEffect[Resp] = i.unary[Req, Resp](acc(req)) - applied(req) - } + val applied: Req => GrpcContext => UnaryEffect[Resp] = i.unary[Req, Resp](acc(req)) + applied(req) } def clientStreaming[Req: ProtobufCodec, Resp: ProtobufCodec]( @@ -90,10 +88,8 @@ object ShardingServerInterceptor { io: GrpcContext => StreamEffect[Resp] ): Req => GrpcContext => StreamEffect[Resp] = interceptors.foldRight(((_: Req) => io): Req => GrpcContext => StreamEffect[Resp]) { (i, acc) => req => - { - val applied: Req => GrpcContext => StreamEffect[Resp] = i.serverStreaming[Req, Resp](acc(req)) - applied(req) - } + val applied: Req => GrpcContext => StreamEffect[Resp] = i.serverStreaming[Req, Resp](acc(req)) + applied(req) } def bidiStreaming[Req: ProtobufCodec, Resp: ProtobufCodec]( diff --git a/protocol-grpc/src/main/scala/com/devsisters/shardcake/protocol/Sharding.scala b/protocol-grpc/src/main/scala/com/devsisters/shardcake/protocol/Sharding.scala index 152e21a0..775b0ddd 100644 --- a/protocol-grpc/src/main/scala/com/devsisters/shardcake/protocol/Sharding.scala +++ b/protocol-grpc/src/main/scala/com/devsisters/shardcake/protocol/Sharding.scala @@ -13,7 +13,7 @@ object Sharding { case class SendRequest(entityId: String, entityType: String, body: Array[Byte], replyId: Option[String]) derives ProtobufCodec - case class SendResponse(body: Array[Byte]) derives ProtobufCodec + case class SendResponse(body: Option[Array[Byte]]) derives ProtobufCodec case class PingShardsRequest() derives ProtobufCodec case class PingShardsResponse() derives ProtobufCodec diff --git a/serialization-kryo/src/main/scala/com/devsisters/shardcake/KryoSerialization.scala b/serialization-kryo/src/main/scala/com/devsisters/shardcake/KryoSerialization.scala deleted file mode 100644 index bf25a634..00000000 --- a/serialization-kryo/src/main/scala/com/devsisters/shardcake/KryoSerialization.scala +++ /dev/null @@ -1,36 +0,0 @@ -package com.devsisters.shardcake - -import com.devsisters.shardcake.interfaces.Serialization -import com.typesafe.config.{ Config, ConfigFactory } -import io.altoo.serialization.kryo.scala.ScalaKryoSerializer -import zio.{ Chunk, Task, ZIO, ZLayer } - -object KryoSerialization { - - /** - * A layer that returns a serialization implementation using the Kryo library. - */ - val live: ZLayer[Any, Throwable, Serialization] = - ZLayer(ZIO.attempt(ConfigFactory.defaultReference()).flatMap(make)) - - /** - * A layer that returns a serialization implementation using the Kryo library, taking a Config object. - * See https://github.com/altoo-ag/scala-kryo-serialization for more details about configuration. - */ - def liveWithConfig(config: Config): ZLayer[Any, Throwable, Serialization] = - ZLayer(make(config)) - - private def make(config: Config): Task[Serialization] = - ZIO.attempt { - new ScalaKryoSerializer(config, getClass.getClassLoader) - }.map(serializer => - new Serialization { - def encode(message: Any): Task[Array[Byte]] = ZIO.fromTry(serializer.serialize(message)) - def decode[A](bytes: Array[Byte]): Task[A] = ZIO.fromTry(serializer.deserialize[A](bytes)) - override def encodeChunk(messages: Chunk[Any]): Task[Chunk[Array[Byte]]] = - ZIO.attempt(messages.map(serializer.serialize(_).get)) - override def decodeChunk[A](bytes: Chunk[Array[Byte]]): Task[Chunk[A]] = - ZIO.attempt(bytes.map(serializer.deserialize[A](_).get)) - } - ) -} diff --git a/serialization-kryo/src/main/scala/com/devsisters/shardcake/kryo/KryoMessageCodec.scala b/serialization-kryo/src/main/scala/com/devsisters/shardcake/kryo/KryoMessageCodec.scala new file mode 100644 index 00000000..e4273746 --- /dev/null +++ b/serialization-kryo/src/main/scala/com/devsisters/shardcake/kryo/KryoMessageCodec.scala @@ -0,0 +1,59 @@ +package com.devsisters.shardcake.kryo + +import com.devsisters.shardcake.interfaces.MessageCodec +import com.typesafe.config.{ Config, ConfigFactory } +import io.altoo.serialization.kryo.scala.ScalaKryoSerializer + +/** + * A universal `MessageCodec[Msg]` backed by Kryo via altoo's `ScalaKryoSerializer`. + * + * Because Kryo is reflective and uniform across types, one generic codec serves every + * message type — there is no per-type derivation. + * Users typically import the package-level given via `import com.devsisters.shardcake.kryo.given` + * and define their `EntityType` / `TopicType` without further ceremony. + * + * To use a custom Kryo configuration, shadow the default `given` with + * `given [Msg]: MessageCodec[Msg] = KryoMessageCodec.fromConfig[Msg](myConfig)`. + */ +final class KryoMessageCodec[Msg] private[kryo] (serializer: ScalaKryoSerializer) extends MessageCodec[Msg] { + private val universalEncoder: Any => Array[Byte] = + (value: Any) => serializer.serialize(value).get + + def encodeMessage(message: Msg): Array[Byte] = + serializer.serialize(message).get + + def decodeMessage(bytes: Array[Byte]): Msg = + serializer.deserialize[Any](bytes).get.asInstanceOf[Msg] + + def replyEncoder(decoded: Msg): Any => Array[Byte] = + universalEncoder + + def replyDecoder[Res](sample: Msg): Array[Byte] => Res = + bytes => serializer.deserialize[Any](bytes).get.asInstanceOf[Res] + + def streamReplyDecoder[Res](sample: Msg): Array[Byte] => Res = + bytes => serializer.deserialize[Any](bytes).get.asInstanceOf[Res] +} + +object KryoMessageCodec { + + /** + * Shared `ScalaKryoSerializer` built from Typesafe Config's reference configuration. + * The serializer manages a Kryo pool internally and is safe to share across the whole pod. + */ + lazy val defaultSerializer: ScalaKryoSerializer = + new ScalaKryoSerializer(ConfigFactory.defaultReference(), getClass.getClassLoader) + + /** + * A `KryoMessageCodec[Msg]` backed by [[defaultSerializer]]. + */ + def default[Msg]: KryoMessageCodec[Msg] = new KryoMessageCodec[Msg](defaultSerializer) + + /** + * A `KryoMessageCodec[Msg]` backed by a fresh `ScalaKryoSerializer` built from the + * given Typesafe Config. See https://github.com/altoo-ag/scala-kryo-serialization for + * the available options. + */ + def fromConfig[Msg](config: Config): KryoMessageCodec[Msg] = + new KryoMessageCodec[Msg](new ScalaKryoSerializer(config, getClass.getClassLoader)) +} diff --git a/serialization-kryo/src/main/scala/com/devsisters/shardcake/kryo/package.scala b/serialization-kryo/src/main/scala/com/devsisters/shardcake/kryo/package.scala new file mode 100644 index 00000000..4694fc34 --- /dev/null +++ b/serialization-kryo/src/main/scala/com/devsisters/shardcake/kryo/package.scala @@ -0,0 +1,17 @@ +package com.devsisters.shardcake + +import com.devsisters.shardcake.interfaces.MessageCodec + +/** + * Kryo backend given. Import with `import com.devsisters.shardcake.kryo.given` to enable + * Kryo-backed serialisation for all `EntityType` / `TopicType` declarations in the + * surrounding scope. + * + * To use a custom Kryo configuration, shadow this given with your own: + * {{{ + * given [Msg]: MessageCodec[Msg] = KryoMessageCodec.fromConfig[Msg](myConfig) + * }}} + */ +package object kryo { + given derive[Msg]: MessageCodec[Msg] = KryoMessageCodec.default[Msg] +} diff --git a/serialization-kryo/src/test/scala/com/devsisters/shardcake/KryoMessageCodecSpec.scala b/serialization-kryo/src/test/scala/com/devsisters/shardcake/KryoMessageCodecSpec.scala new file mode 100644 index 00000000..6a60d673 --- /dev/null +++ b/serialization-kryo/src/test/scala/com/devsisters/shardcake/KryoMessageCodecSpec.scala @@ -0,0 +1,46 @@ +package com.devsisters.shardcake + +import com.devsisters.shardcake.interfaces.MessageCodec +import com.devsisters.shardcake.kryo.given +import zio.Scope +import zio.test._ + +object KryoMessageCodecSpec extends ZIOSpecDefault { + + sealed trait Msg + object Msg { + final case class Plain(name: String, count: Int) extends Msg + final case class WithReplier(id: Int, replier: Replier[String]) extends Msg + final case class WithStreamReplier(replier: StreamReplier[Int]) extends Msg + } + + private val codec = summon[MessageCodec[Msg]] + + def spec: Spec[TestEnvironment with Scope, Any] = + suite("KryoMessageCodecSpec")( + test("roundtrips a message with no Replier") { + val msg = Msg.Plain("hello", 42) + val bytes = codec.encodeMessage(msg) + val decoded = codec.decodeMessage(bytes) + assertTrue(decoded == msg) + }, + test("roundtrips a message containing a Replier and produces a working receiver encoder") { + val msg = Msg.WithReplier(7, Replier("rid-1")) + val bytes = codec.encodeMessage(msg) + val decoded = codec.decodeMessage(bytes).asInstanceOf[Msg.WithReplier] + val encoder = codec.replyEncoder(decoded) + val replyBytes = encoder("ok") + val decodedStr = codec.replyDecoder[String](msg)(replyBytes) + assertTrue(decoded.id == 7, decoded.replier.id == "rid-1", decodedStr == "ok") + }, + test("roundtrips a message containing a StreamReplier and produces a working receiver encoder") { + val msg = Msg.WithStreamReplier(StreamReplier("srid-1")) + val bytes = codec.encodeMessage(msg) + val decoded = codec.decodeMessage(bytes).asInstanceOf[Msg.WithStreamReplier] + val encoder = codec.replyEncoder(decoded) + val replyBytes = encoder(99) + val decodedInt = codec.streamReplyDecoder[Int](msg)(replyBytes) + assertTrue(decoded.replier.id == "srid-1", decodedInt == 99) + } + ) +} diff --git a/serialization-kryo/src/test/scala/com/devsisters/shardcake/KryoSerializationSpec.scala b/serialization-kryo/src/test/scala/com/devsisters/shardcake/KryoSerializationSpec.scala deleted file mode 100644 index c684fdda..00000000 --- a/serialization-kryo/src/test/scala/com/devsisters/shardcake/KryoSerializationSpec.scala +++ /dev/null @@ -1,19 +0,0 @@ -package com.devsisters.shardcake - -import com.devsisters.shardcake.interfaces.Serialization -import zio.{ Scope, ZIO } -import zio.test._ - -object KryoSerializationSpec extends ZIOSpecDefault { - def spec: Spec[TestEnvironment with Scope, Any] = - suite("KryoSerializationSpec")( - test("serialize back and forth") { - case class Test(a: Int, b: String) - val expected = Test(2, "test") - for { - bytes <- ZIO.serviceWithZIO[Serialization](_.encode(expected)) - actual <- ZIO.serviceWithZIO[Serialization](_.decode[Test](bytes)) - } yield assertTrue(expected == actual) - } - ).provideShared(KryoSerialization.live) -} diff --git a/serialization-proteus/src/main/scala/com/devsisters/shardcake/proteus/ProteusMessageCodec.scala b/serialization-proteus/src/main/scala/com/devsisters/shardcake/proteus/ProteusMessageCodec.scala new file mode 100644 index 00000000..bd3e22fb --- /dev/null +++ b/serialization-proteus/src/main/scala/com/devsisters/shardcake/proteus/ProteusMessageCodec.scala @@ -0,0 +1,216 @@ +package com.devsisters.shardcake.proteus + +import _root_.proteus.{ ProtobufCodec, ProtobufDeriver } +import com.devsisters.shardcake.{ Replier, StreamReplier } +import com.devsisters.shardcake.interfaces.MessageCodec + +import scala.compiletime.{ erasedValue, summonFrom, summonInline } +import scala.deriving.Mirror + +/** + * An inline-derived `MessageCodec[Msg]` backed by Proteus. + * + * Walks `Msg` at compile time: + * - For the whole message it uses `ProtobufCodec.derived[Msg]` for encode/decode. + * - For each message shape containing a `Replier[R]` or `StreamReplier[R]` field, directly + * or inside nested case classes / sealed-trait variants, it derives `ProtobufCodec[R]`. + * + * Message shapes without a reply field do not participate in reply dispatch (they are + * fire-and-forget messages). Product types (a single case class as `Msg`) are also supported. + * + * Recursive message graphs are not supported by the reply-selector derivation. Keep + * recursive structures behind non-product containers or provide a custom codec if they need + * reply discovery. + */ +object ProteusMessageCodec { + + private type VariantEntry = (Any => Array[Byte], Array[Byte] => Any) + + private enum ReplyEntrySummary { + case None + case One(entry: VariantEntry) + case Many + + def combine(that: ReplyEntrySummary): ReplyEntrySummary = + (this, that) match { + case (ReplyEntrySummary.None, other) => other + case (one @ ReplyEntrySummary.One(_), ReplyEntrySummary.None) => one + case _ => ReplyEntrySummary.Many + } + + def soleEntry: Option[VariantEntry] = + this match { + case ReplyEntrySummary.One(entry) => Some(entry) + case ReplyEntrySummary.None | ReplyEntrySummary.Many => scala.None + } + } + + private final case class ReplySelector[-A](select: A => Option[VariantEntry], summary: ReplyEntrySummary) + + /** + * Derive a `MessageCodec[Msg]` using Proteus. Proteus requires `Msg` to be a message + * (case class) or an enum / sealed trait at the root — primitives like `Int` / `String` + * are not supported as message types directly; wrap them in a case class. + * + * For sealed traits and case classes, direct or nested `Replier[R]` / `StreamReplier[R]` + * fields have their per-slot reply codec materialised at compile time. + */ + inline def derived[Msg](using m: Mirror.Of[Msg], deriver: ProtobufDeriver): MessageCodec[Msg] = + build(summonOrDeriveProtobufCodec[Msg](using deriver), deriveSelector[Msg](using m, deriver)) + + private inline def summonOrDeriveProtobufCodec[A](using deriver: ProtobufDeriver): ProtobufCodec[A] = + summonFrom { + case codec: ProtobufCodec[A] => codec + case _ => ProtobufCodec.derived[A](using deriver) + } + + private def build[Msg]( + msgCodec: ProtobufCodec[Msg], + replySelector: ReplySelector[Msg] + ): MessageCodec[Msg] = new Impl[Msg](msgCodec, replySelector) + + private final class Impl[Msg]( + msgCodec: ProtobufCodec[Msg], + replySelector: ReplySelector[Msg] + ) extends MessageCodec[Msg] { + + private val fallbackEncode: Any => Array[Byte] = _ => Array.emptyByteArray + + // With exactly one Replier-bearing variant there's only one possible reply type, so + // the same encoder applies regardless of which variant the incoming request was. + // Caching it also prevents `fallbackEncode` from ever being handed out for a request + // that didn't carry the Replier (e.g. stream continuations). + private val singleEncoder: Option[Any => Array[Byte]] = + replySelector.summary.soleEntry.map(_._1) + + def encodeMessage(message: Msg): Array[Byte] = msgCodec.encode(message) + def decodeMessage(bytes: Array[Byte]): Msg = msgCodec.decode(bytes) + + def replyEncoder(decoded: Msg): Any => Array[Byte] = + singleEncoder.getOrElse(replySelector.select(decoded).fold(fallbackEncode)(_._1)) + + def replyDecoder[Res](sample: Msg): Array[Byte] => Res = lookupDecoder(sample, "reply") + def streamReplyDecoder[Res](sample: Msg): Array[Byte] => Res = lookupDecoder(sample, "stream reply") + + private def lookupDecoder[Res](sample: Msg, kind: String): Array[Byte] => Res = + replySelector.select(sample) match { + case Some((_, dec)) => bytes => dec(bytes).asInstanceOf[Res] + case None => + val message = s"No $kind codec registered for variant ${sample.getClass.getName}" + (_: Array[Byte]) => sys.error(message) + } + } + + private inline def deriveSelector[A](using m: Mirror.Of[A], deriver: ProtobufDeriver): ReplySelector[A] = + inline m match { + case s: Mirror.SumOf[A] => + val selectors = collectVariantSelectors[s.MirroredElemTypes] + ReplySelector( + select = value => selectors(s.ordinal(value)).select(value), + summary = selectors.foldLeft(ReplyEntrySummary.None)(_ combine _.summary) + ) + case p: Mirror.ProductOf[A] => + selectorForProduct[A, p.MirroredElemTypes] + } + + private inline def collectVariantSelectors[Variants <: Tuple](using + deriver: ProtobufDeriver + ): Vector[ReplySelector[Any]] = + inline erasedValue[Variants] match { + case _: EmptyTuple => Vector.empty + case _: (head *: tail) => + val current = summonFrom { + case m: Mirror.Of[`head`] => deriveSelector[head](using m, deriver).asInstanceOf[ReplySelector[Any]] + case _ => emptySelector[Any] + } + current +: collectVariantSelectors[tail] + } + + private inline def selectorForProduct[A, Fields <: Tuple](using deriver: ProtobufDeriver): ReplySelector[A] = + inline if (hasDirectReplier[Fields]) { + val entry = directReplyEntry[Fields] + ReplySelector(_ => Some(entry), ReplyEntrySummary.One(entry)) + } else { + val nested = nestedFieldSelectors[Fields](0) + ReplySelector( + select = value => { + val product = value.asInstanceOf[Product] + var remaining = nested + var selected: Option[VariantEntry] = scala.None + while (selected.isEmpty && remaining.nonEmpty) { + val (index, selector) = remaining.head + selected = selector.select(product.productElement(index)) + remaining = remaining.tail + } + selected + }, + summary = nested.foldLeft(ReplyEntrySummary.None)(_ combine _._2.summary) + ) + } + + private inline def hasDirectReplier[Fields <: Tuple]: Boolean = + inline erasedValue[Fields] match { + case _: EmptyTuple => false + case _: (Replier[?] *: tail) => true + case _: (StreamReplier[?] *: tail) => true + case _: (_ *: tail) => hasDirectReplier[tail] + } + + private inline def directReplyEntry[Fields <: Tuple](using deriver: ProtobufDeriver): VariantEntry = + inline erasedValue[Fields] match { + case _: (Replier[r] *: tail) => buildReplyEntry[r] + case _: (StreamReplier[r] *: tail) => buildReplyEntry[r] + case _: (_ *: tail) => directReplyEntry[tail] + } + + private inline def nestedFieldSelectors[Fields <: Tuple](index: Int)(using + deriver: ProtobufDeriver + ): List[(Int, ReplySelector[Any])] = + inline erasedValue[Fields] match { + case _: EmptyTuple => Nil + case _: (Replier[?] *: tail) => nestedFieldSelectors[tail](index + 1) + case _: (StreamReplier[?] *: tail) => nestedFieldSelectors[tail](index + 1) + case _: (head *: tail) => + val rest = nestedFieldSelectors[tail](index + 1) + summonFrom { + case m: Mirror.Of[`head`] => + val selector = deriveSelector[head](using m, deriver).asInstanceOf[ReplySelector[Any]] + selector.summary match { + case ReplyEntrySummary.None => rest + case ReplyEntrySummary.One(_) | ReplyEntrySummary.Many => (index -> selector) :: rest + } + case _ => + rest + } + } + + private def emptySelector[A]: ReplySelector[A] = + ReplySelector(_ => None, ReplyEntrySummary.None) + + /** + * Materialise the per-reply-type encoder/decoder pair for one Replier/StreamReplier slot. + * Prefers an in-scope `ProtobufCodec[R]`; otherwise derives one from a Mirror. Emits a + * clear compile error when neither is available — Proteus needs a case class / sealed + * trait / enum at the root and can't encode bare primitives. + */ + private inline def buildReplyEntry[R](using deriver: ProtobufDeriver): VariantEntry = + summonFrom { + case codec: ProtobufCodec[R] => + ( + (v: Any) => codec.encode(v.asInstanceOf[R]), + (b: Array[Byte]) => codec.decode(b) + ) + case _: Mirror.Of[R] => + val codec = ProtobufCodec.derived[R](using deriver) + ( + (v: Any) => codec.encode(v.asInstanceOf[R]), + (b: Array[Byte]) => codec.decode(b) + ) + case _ => + scala.compiletime.error( + "Cannot derive a Proteus MessageCodec: reply type has no Mirror or in-scope ProtobufCodec. " + + "Proteus requires a case class, sealed trait, or enum at the root — " + + "wrap primitive reply types (Int, String, etc.) in a case class." + ) + } +} diff --git a/serialization-proteus/src/main/scala/com/devsisters/shardcake/proteus/package.scala b/serialization-proteus/src/main/scala/com/devsisters/shardcake/proteus/package.scala new file mode 100644 index 00000000..4fa3394b --- /dev/null +++ b/serialization-proteus/src/main/scala/com/devsisters/shardcake/proteus/package.scala @@ -0,0 +1,24 @@ +package com.devsisters.shardcake + +import _root_.proteus.ProtobufDeriver +import com.devsisters.shardcake.interfaces.MessageCodec + +import scala.deriving.Mirror +import scala.util.NotGiven + +/** + * Proteus backend givens. Import everything in one shot with + * `import com.devsisters.shardcake.proteus.given` to enable Proteus-backed serialisation + * for all `EntityType` / `TopicType` declarations in the surrounding scope. + * + * Bring a custom `given ProtobufDeriver = ...` into scope (e.g. via Proteus's builder API + * to register custom codec instances or modifiers) to override the default deriver — the + * `defaultDeriver` exposed here only materialises when no other `ProtobufDeriver` is in + * scope, so a user-provided one wins without an ambiguity error. + */ +package object proteus { + given defaultDeriver(using NotGiven[ProtobufDeriver]): ProtobufDeriver = ProtobufDeriver + + inline given derive[Msg](using m: Mirror.Of[Msg], deriver: ProtobufDeriver): MessageCodec[Msg] = + ProteusMessageCodec.derived[Msg](using m, deriver) +} diff --git a/serialization-proteus/src/test/scala/com/devsisters/shardcake/ProteusMessageCodecSpec.scala b/serialization-proteus/src/test/scala/com/devsisters/shardcake/ProteusMessageCodecSpec.scala new file mode 100644 index 00000000..fbc3de5e --- /dev/null +++ b/serialization-proteus/src/test/scala/com/devsisters/shardcake/ProteusMessageCodecSpec.scala @@ -0,0 +1,138 @@ +package com.devsisters.shardcake + +import com.devsisters.shardcake.interfaces.MessageCodec +import com.devsisters.shardcake.proteus.given +import zio.Scope +import zio.test._ + +object ProteusMessageCodecSpec extends ZIOSpecDefault { + + sealed trait Msg derives _root_.proteus.ProtobufCodec + object Msg { + final case class Plain(name: String, count: Int) extends Msg + final case class WithReplier(id: Int, replier: Replier[Reply]) extends Msg + final case class WithStreamReplier(replier: StreamReplier[StreamReply]) extends Msg + final case class FireAndForget(payload: String) extends Msg + } + + final case class Reply(text: String, ok: Boolean) derives _root_.proteus.ProtobufCodec + final case class StreamReply(value: Int) derives _root_.proteus.ProtobufCodec + + sealed trait NestedMsg derives _root_.proteus.ProtobufCodec + object NestedMsg { + final case class ToProtocol(request: Protocol[Command, PEvent]) extends NestedMsg + final case class Plain(payload: String) extends NestedMsg + } + + sealed trait Protocol[Command, PEvent] + object Protocol { + final case class Run[Command, PEvent](command: Command) extends Protocol[Command, PEvent] + final case class Prepare[Command, PEvent](transactionId: String, command: Command) extends Protocol[Command, PEvent] + final case class ReadyToPersist[Command, PEvent]( + transactionId: String, + replier: Replier[PersistReadied[PEvent]] + ) extends Protocol[Command, PEvent] + final case class Commit[Command, PEvent](transactionId: String) extends Protocol[Command, PEvent] + + given _root_.proteus.ProtobufCodec[Protocol[ProteusMessageCodecSpec.Command, ProteusMessageCodecSpec.PEvent]] = + _root_.proteus.ProtobufCodec.derived + } + + final case class Command(context: String, function: Function, replier: Replier[CommandReply]) + derives _root_.proteus.ProtobufCodec + + sealed trait Function derives _root_.proteus.ProtobufCodec + object Function { + final case class Add(amount: Int) extends Function + final case class Get() extends Function + } + + sealed trait CommandReply derives _root_.proteus.ProtobufCodec + object CommandReply { + final case class Added(total: Int) extends CommandReply + final case class Got(total: Int) extends CommandReply + } + + sealed trait PEvent derives _root_.proteus.ProtobufCodec + object PEvent { + final case class Changed(delta: Int) extends PEvent + } + + final case class PersistReadied[PEvent](persistenceId: String, events: List[PEvent], fromSequenceNr: Long) + + object PersistReadied { + given _root_.proteus.ProtobufCodec[PersistReadied[ProteusMessageCodecSpec.PEvent]] = + _root_.proteus.ProtobufCodec.derived + } + + private val codec = summon[MessageCodec[Msg]] + private val nestedCodec = summon[MessageCodec[NestedMsg]] + + def spec: Spec[TestEnvironment with Scope, Any] = + suite("ProteusMessageCodecSpec")( + test("roundtrips a message with no Replier") { + val msg = Msg.Plain("hello", 42) + val bytes = codec.encodeMessage(msg) + val decoded = codec.decodeMessage(bytes) + assertTrue(decoded == msg) + }, + test("roundtrips a fire-and-forget variant") { + val msg = Msg.FireAndForget("x") + val bytes = codec.encodeMessage(msg) + val decoded = codec.decodeMessage(bytes) + assertTrue(decoded == msg) + }, + test("dispatches replyEncoder by variant and round-trips the reply") { + val msg = Msg.WithReplier(7, Replier("rid-1")) + val bytes = codec.encodeMessage(msg) + val decoded = codec.decodeMessage(bytes).asInstanceOf[Msg.WithReplier] + val encoder = codec.replyEncoder(decoded) + val reply = Reply("ok", true) + val replyBytes = encoder(reply) + val decoded2 = codec.replyDecoder[Reply](msg)(replyBytes) + assertTrue(decoded.id == 7, decoded.replier.id == "rid-1", decoded2 == reply) + }, + test("dispatches stream reply encoder by variant and round-trips the reply") { + val msg = Msg.WithStreamReplier(StreamReplier("srid-1")) + val bytes = codec.encodeMessage(msg) + val decoded = codec.decodeMessage(bytes).asInstanceOf[Msg.WithStreamReplier] + val encoder = codec.replyEncoder(decoded) + val reply = StreamReply(99) + val replyBytes = encoder(reply) + val decoded2 = codec.streamReplyDecoder[StreamReply](msg)(replyBytes) + assertTrue(decoded.replier.id == "srid-1", decoded2 == reply) + }, + test("finds a Replier nested inside products and sealed-trait variants") { + val msg = NestedMsg.ToProtocol( + Protocol.Run(Command("ctx", Function.Add(1), Replier("nested-rid"))) + ) + val bytes = nestedCodec.encodeMessage(msg) + val decoded = nestedCodec.decodeMessage(bytes).asInstanceOf[NestedMsg.ToProtocol] + val encoder = nestedCodec.replyEncoder(decoded) + val reply = CommandReply.Added(2) + val replyBytes = encoder(reply) + val decoded2 = nestedCodec.replyDecoder[CommandReply](msg)(replyBytes) + + assertTrue( + decoded.request.asInstanceOf[Protocol.Run[Command, PEvent]].command.replier.id == "nested-rid", + decoded2 == reply + ) + }, + test("finds concrete generic reply types in nested protocol variants") { + val msg = NestedMsg.ToProtocol( + Protocol.ReadyToPersist[Command, PEvent]("tid", Replier("ready-rid")) + ) + val bytes = nestedCodec.encodeMessage(msg) + val decoded = nestedCodec.decodeMessage(bytes).asInstanceOf[NestedMsg.ToProtocol] + val encoder = nestedCodec.replyEncoder(decoded) + val reply = PersistReadied("pid", List(PEvent.Changed(3)), 4L) + val replyBytes = encoder(reply) + val decoded2 = nestedCodec.replyDecoder[PersistReadied[PEvent]](msg)(replyBytes) + + assertTrue( + decoded.request.asInstanceOf[Protocol.ReadyToPersist[Command, PEvent]].replier.id == "ready-rid", + decoded2 == reply + ) + } + ) +} diff --git a/serialization-proteus/src/test/scala/com/devsisters/shardcake/ProteusShardingSpec.scala b/serialization-proteus/src/test/scala/com/devsisters/shardcake/ProteusShardingSpec.scala new file mode 100644 index 00000000..05bdbd05 --- /dev/null +++ b/serialization-proteus/src/test/scala/com/devsisters/shardcake/ProteusShardingSpec.scala @@ -0,0 +1,171 @@ +package com.devsisters.shardcake + +import com.devsisters.shardcake.interfaces.Storage +import com.devsisters.shardcake.proteus.given +import zio.stream.{ SubscriptionRef, ZStream } +import zio.test.TestAspect.{ sequential, withLiveClock } +import zio.test._ +import zio.{ Config => _, _ } + +/** + * End-to-end test that exercises the full Sharding + LocalSharding flow using the Proteus + * backend. Reply types are case classes (Proteus only supports messages / enums at the + * root, not primitives), so we wrap `Int` in a small `Count` case class. + * + * The local-pod shortcut means most of the action goes through the typed reply channel + * path; setting `simulateRemotePods = true` forces the byte-channel path through Proteus. + */ +object ProteusShardingSpec extends ZIOSpecDefault { + import ProteusShardingSpec.CounterActor._ + import ProteusShardingSpec.CounterActor.CounterMessage._ + + def spec: Spec[TestEnvironment with Scope, Any] = + suite("ProteusShardingSpec")( + suite("local-pod shortcut")(tests*).provideShared( + LocalSharding.live, + ShardManagerClient.local, + Storage.memory, + ZLayer.succeed(Config.default) + ), + suite("simulate remote pods (forces proteus serialization on every hop)")(tests*).provideShared( + LocalSharding.live, + ShardManagerClient.local, + Storage.memory, + ZLayer.succeed(Config.default.copy(simulateRemotePods = true)) + ) + ) @@ sequential @@ withLiveClock + + private val tests: List[Spec[Sharding, Any]] = List( + test("Send fire-and-forget then ask for the count") { + ZIO.scoped { + for { + _ <- Sharding.registerEntity(Counter, behavior) + _ <- Sharding.registerScoped + counter <- Sharding.messenger(Counter) + _ <- counter.sendDiscard("c1")(IncrementCounter) + _ <- counter.sendDiscard("c1")(IncrementCounter) + _ <- counter.sendDiscard("c1")(DecrementCounter) + _ <- counter.sendDiscard("c2")(IncrementCounter) + _ <- Clock.sleep(500.millis) + c1 <- counter.send("c1")(GetCounter.apply) + c2 <- counter.send("c2")(GetCounter.apply) + } yield assertTrue(c1.value == 1, c2.value == 1) + } + }, + test("Response stream over Proteus") { + ZIO.scoped { + for { + _ <- Sharding.registerEntity(Counter, behavior) + _ <- Sharding.registerScoped + counter <- Sharding.messenger(Counter) + stream <- counter.sendAndReceiveStream("c1")(StreamingChanges.apply) + latch <- Promise.make[Nothing, Unit] + fiber <- stream.take(4).tap(_ => latch.succeed(())).runCollect.fork + _ <- latch.await + _ <- counter.sendDiscard("c1")(IncrementCounter) + _ <- counter.sendDiscard("c1")(IncrementCounter) + _ <- counter.sendDiscard("c1")(DecrementCounter) + items <- fiber.join + } yield assertTrue(items.map(_.value) == Chunk(0, 1, 2, 1)) + } + }, + // The Counter protocol has two reply-bearing variants (GetCounter and StreamingChanges). + // Proteus must resolve the right reply codec for the variant carried by `request`. + test("Multi-variant sendStreamAndReceiveStream over Proteus") { + ZIO.scoped { + for { + _ <- Sharding.registerEntity(Counter, behavior) + _ <- Sharding.registerScoped + counter <- Sharding.messenger(Counter) + latch <- Promise.make[Nothing, Unit] + stream <- counter.sendStreamAndReceiveStream[Count]("c1")( + StreamingChanges.apply, + ZStream.fromZIO(latch.await).drain ++ + ZStream.fromIterable(List(IncrementCounter, IncrementCounter, DecrementCounter)) + ) + items <- stream.tap(_ => latch.succeed(())).take(4).runCollect + } yield assertTrue(items.map(_.value) == Chunk(0, 1, 2, 1)) + } + }, + // Regression test for shardcake's pendingReplies overwrite bug with Proteus's + // per-variant reply encoder. The first message carries the StreamReplier; the + // follow-ups (IncrementCounter / DecrementCounter — no Replier field) are emitted + // eagerly, ahead of any reply being received. Without the fix, the receiver's + // `processBinary` would re-register the pendingReply slot with `fallbackEncode` + // (empty bytes) before the actor's `replyStream` captures the real encoder, and the + // sender would then fail to decode replies with "OneOf field absent". + // + // Uses `Reply` (sealed trait, required OneOf field) so an empty payload from + // `fallbackEncode` is a hard decode failure rather than a silent default value. + test("Eager non-Replier follow-ups don't clobber the reply encoder") { + ZIO.scoped { + for { + _ <- Sharding.registerEntity(Counter, behavior) + _ <- Sharding.registerScoped + counter <- Sharding.messenger(Counter) + stream <- counter.sendStreamAndReceiveStream[Reply]("c1")( + StreamReply.apply, + ZStream.fromIterable(List(IncrementCounter, IncrementCounter, DecrementCounter)) + ) + // Take only the initial ack — `state.changes` subscription timing is unreliable + // once eager follow-ups race ahead. The bug shows up as a decode failure on the + // very first reply, so one element is enough to assert correctness. + first <- stream.take(1).runCollect.timeoutFail("timeout")(10.seconds) + } yield assertTrue(first.size == 1 && first.head == Reply.Ack()) + } + } + ) + + // Reply payload wrapped in a case class — Proteus needs a message at the root, not a primitive. + final case class Count(value: Int) + + // Sealed-trait reply used by the regression test. A required OneOf field means an + // empty payload (what `fallbackEncode` would produce) is a hard decode failure. + sealed trait Reply + object Reply { + final case class Ack() extends Reply + final case class Snapshot(v: Int) extends Reply + } + + object CounterActor { + sealed trait CounterMessage + object CounterMessage { + final case class GetCounter(replier: Replier[Count]) extends CounterMessage + case object IncrementCounter extends CounterMessage + case object DecrementCounter extends CounterMessage + final case class StreamingChanges(replier: StreamReplier[Count]) extends CounterMessage + // Pushes a deterministic `Ack` as the first reply, then a `Snapshot` per + // Increment/Decrement processed by this actor while the stream is open. + final case class StreamReply(replier: StreamReplier[Reply]) extends CounterMessage + } + + object Counter extends EntityType[CounterMessage]("counter") + + def behavior(entityId: String, messages: Dequeue[CounterMessage]): RIO[Sharding, Nothing] = + ZIO.logInfo(s"Started entity $entityId") *> + (SubscriptionRef.make(0) <*> Ref.make[Option[Queue[Reply]]](None)).flatMap { case (state, replyQueueRef) => + messages.take.flatMap { + case CounterMessage.GetCounter(replier) => state.get.flatMap(v => replier.reply(Count(v))) + case CounterMessage.IncrementCounter => + state + .updateAndGet(_ + 1) + .flatMap(v => replyQueueRef.get.flatMap(ZIO.foreachDiscard(_)(_.offer(Reply.Snapshot(v))))) + case CounterMessage.DecrementCounter => + state + .updateAndGet(_ - 1) + .flatMap(v => replyQueueRef.get.flatMap(ZIO.foreachDiscard(_)(_.offer(Reply.Snapshot(v))))) + case CounterMessage.StreamingChanges(replier) => + replier.replyStream(state.changes.ensuring(state.set(-1)).map(Count(_))) + case CounterMessage.StreamReply(replier) => + // Forces eager follow-ups to reach `processBinary` before the actor subscribes. + ZIO.sleep(200.millis) *> + (for { + q <- Queue.unbounded[Reply] + _ <- replyQueueRef.set(Some(q)) + _ <- replier.replyStream(ZStream.fromQueueWithShutdown(q)) + _ <- q.offer(Reply.Ack()) + } yield ()) + }.forever + } + } +} diff --git a/storage-redis/src/main/scala/com/devsisters/shardcake/StorageRedis.scala b/storage-redis/src/main/scala/com/devsisters/shardcake/StorageRedis.scala index 5e9ec7c6..eed67799 100644 --- a/storage-redis/src/main/scala/com/devsisters/shardcake/StorageRedis.scala +++ b/storage-redis/src/main/scala/com/devsisters/shardcake/StorageRedis.scala @@ -9,8 +9,6 @@ import zio.stream.ZStream import zio.stream.interop.fs2z._ import zio.{ Task, ZIO, ZLayer } -import scala.collection.compat._ - object StorageRedis { type fs2Stream[A] = fs2.Stream[Task, A] type Redis = RedisCommands[Task, String, String] with PubSubCommands[Task, fs2Stream, String, String] diff --git a/storage-redisson/src/main/scala/com/devsisters/shardcake/StorageRedis.scala b/storage-redisson/src/main/scala/com/devsisters/shardcake/StorageRedis.scala index 6f230c53..cb38ab30 100644 --- a/storage-redisson/src/main/scala/com/devsisters/shardcake/StorageRedis.scala +++ b/storage-redisson/src/main/scala/com/devsisters/shardcake/StorageRedis.scala @@ -1,6 +1,5 @@ package com.devsisters.shardcake -import scala.collection.compat._ import scala.jdk.CollectionConverters._ import com.devsisters.shardcake.interfaces.Storage diff --git a/vuepress/docs/docs/README.md b/vuepress/docs/docs/README.md index 6b5d9a4f..bd3b3399 100644 --- a/vuepress/docs/docs/README.md +++ b/vuepress/docs/docs/README.md @@ -66,7 +66,7 @@ Shardcake only takes care of starting entities on the right pods as well as the There are 4 pluggable parts that can be implemented with the technology of your choice. - The `Storage` trait defines where shard assignments will be stored. Shardcake provides an implementation using **Redis**. - The `Pods` trait defines how to communicate with remote pods. Shardcake provides an implementation using **gRPC** as the protocol. -- The `Serialization` defines how to encode and decode messages. Shardcake provides an implementation using **Kryo**. +- The `MessageCodec[Msg]` type class defines how to encode and decode messages of a given type. Shardcake provides backends using **Kryo** (reflective, no derivation) and **Proteus** (macro-derived, protobuf wire format). - The `PodsHealth` trait defines how to check if a pod is healthy or not. Shardcake provides an implementation using the **k8s API**. ![architecture diagram](/shardcake/arch.png) @@ -124,9 +124,13 @@ object GuildMessage { } ``` We also need to define an **Entity Type**. This is done by extending `EntityType` with the message type as well as a unique `String` identifier for this type. +The `EntityType` constructor requires a given `MessageCodec[GuildMessage]` in scope, which decides how messages are serialized on the wire. Here we pull in the Kryo backend's universal given by importing it next to the declaration: ```scala +import com.devsisters.shardcake.kryo.given + object Guild extends EntityType[GuildMessage]("guild") ``` +See the [Customization](customization.md#message-codec) section for the other available backends. The behavior itself is a function with the following signature: ```scala @@ -194,7 +198,6 @@ def run: Task[Unit] = ZIO.scoped(program).provide( ZLayer.succeed(Config.default), ZLayer.succeed(GrpcConfig.default), - Serialization.javaSerialization, // use java serialization for messages Storage.memory, // store data in memory ShardManagerClient.liveWithSttp, // client to communicate with the Shard Manager GrpcPods.live, // use gRPC protocol @@ -236,7 +239,7 @@ as well as [a more complex example](https://github.com/devsisters/shardcake/tree To understand how Sharding works under the hood, have a look at the [Architecture](architecture.md) section. The [Configuration](config.md) section explains how to configure the sharding system. -Finally, the [Customization](customization.md) section describes how you can use your own storage, serialization or messaging protocol, as well as the options provided by Shardcake. +Finally, the [Customization](customization.md) section describes how you can use your own storage, message codec or messaging protocol, as well as the options provided by Shardcake. ::: tip Differences with Akka Cluster Sharding ? [Akka Cluster Sharding](https://doc.akka.io/docs/akka/current/typed/cluster-sharding.html) is the main alternative for sharding in Scala. diff --git a/vuepress/docs/docs/customization.md b/vuepress/docs/docs/customization.md index 8a69f758..c5794ce5 100644 --- a/vuepress/docs/docs/customization.md +++ b/vuepress/docs/docs/customization.md @@ -102,31 +102,95 @@ You can then simply use the `GrpcPods.live` layer. On pods, you also need expose the gRPC API. This is done by adding the `GrpcShardingService.live` layer to your environment. You don't need this one on the Shard Manager. -## Serialization +## Message Codec -The `Serialization` trait defines how to serialize user messages that will be sent between pods. -It contains 2 methods `encode` and `decode` that define how to transform a give type from and to bytes. +The `MessageCodec[Msg]` type class defines how a specific message type is encoded for transport between pods and how its replies are encoded and decoded. ```scala -trait Serialization { - def encode(message: Any): Task[Array[Byte]] - def decode[A](bytes: Array[Byte]): Task[A] +trait MessageCodec[Msg] { + def encodeMessage(message: Msg): Array[Byte] + def decodeMessage(bytes: Array[Byte]): Msg + def replyEncoder(decoded: Msg): Any => Array[Byte] + def replyDecoder[Res](sample: Msg): Array[Byte] => Res + def streamReplyDecoder[Res](sample: Msg): Array[Byte] => Res } ``` -For testing, you can use the `Serialization.javaSerialization` layer that uses Java Serialization (not recommended in production). -Shardcake provides an implementation of `Serialization` using the [Kryo](https://github.com/EsotericSoftware/kryo) binary serialization library. To use it, add the following dependency: +`RecipientType[Msg]` (the parent of `EntityType[Msg]` and `TopicType[Msg]`) requires a given `MessageCodec[Msg]` at the declaration site. + +The expected pattern is to import a backend's package-level given right where the type is declared: + +```scala +import com.devsisters.shardcake.kryo.given + +object Guild extends EntityType[GuildMessage]("guild") +``` + +Different entity / topic types can use different backends side by side — just import the right given in each declaration site. + +### Java serialization (tests only) + +Bundled with `shardcake-entities`, intended for tests and examples. Not recommended for production. + +```scala +import com.devsisters.shardcake.javaSerialization.given + +object Guild extends EntityType[GuildMessage]("guild") +``` + +### Kryo + +Uses the [Kryo](https://github.com/EsotericSoftware/kryo) binary serialization library via altoo's [`scala-kryo-serialization`](https://github.com/altoo-ag/scala-kryo-serialization). +Reflective — no derivation, no compile-time constraints on the message type. The given covers every type uniformly. + ```scala libraryDependencies += "com.devsisters" %% "shardcake-serialization-kryo" % "2.7.1" ``` -You can then simply use the `KryoSerialization.live` layer. + +```scala +import com.devsisters.shardcake.EntityType +import com.devsisters.shardcake.kryo.given + +object Guild extends EntityType[GuildMessage]("guild") +``` + +For a custom Kryo configuration (extra registered serializers, references off, etc.), define your own given instead of importing `com.devsisters.shardcake.kryo.given`. + +```scala +import com.devsisters.shardcake.EntityType +import com.devsisters.shardcake.kryo.KryoMessageCodec +import com.devsisters.shardcake.interfaces.MessageCodec +import com.typesafe.config.ConfigFactory + +given [Msg]: MessageCodec[Msg] = KryoMessageCodec.fromConfig[Msg](ConfigFactory.load("my-kryo.conf")) + +object Guild extends EntityType[GuildMessage]("guild") +``` + +### Proteus + +Uses [Proteus](https://github.com/ghostdogpr/proteus) (macro-derived, protobuf-compatible wire format). +The codec is derived per message type at compile time. Message types must be a case class, sealed trait, or enum at the root — bare primitives (`Int`, `String`, …) need to be wrapped in a case class. + +```scala +libraryDependencies += "com.devsisters" %% "shardcake-serialization-proteus" % "2.7.1" +``` + +```scala +import com.devsisters.shardcake.proteus.given + +object Guild extends EntityType[GuildMessage]("guild") +``` + +The Proteus given will pick up any in-scope `ProtobufCodec[T]` you've declared (e.g. via `derives proteus.ProtobufCodec` on your types) before falling back to deriving one — useful if you want to register custom field codecs. + +Proteus also offers protobuf's forward/backward compatibility rules, which makes it a good choice when you need to do rolling updates that change the message format. ::: tip Server updates and message versioning -- Messages are not persisted, which means that if you stop and restart the whole system, you can change anything in the messages format. -- On the other hand, if you wish to do rolling updates (update servers progressively without downtime), you need to be careful with changes in the messages format. -- What you can do largely depends on your serialization mechanism, some solutions allow changes while some others are very restrictive. - [Kryo](https://github.com/EsotericSoftware/kryo) by default is pretty strict and won't support most changes, but there are settings to support more (at the cost of some performance or message size). -- When you can't modify existing messages, an option is to create new messages that won't be used until the rolling update is finished (so you won't have cases where old nodes receive new messages). +- Messages are not persisted, which means that if you stop and restart the whole system, you can change anything in the message format. +- On the other hand, if you wish to do rolling updates (update servers progressively without downtime), you need to be careful with changes in the message format. +- What you can do largely depends on the backend you choose. [Kryo](https://github.com/EsotericSoftware/kryo) is reflective and pretty strict by default — most changes will break the wire format. Proteus uses protobuf's evolution rules, so adding optional fields or new variants is safe. +- When you can't modify an existing message, the usual fallback is to introduce a new message type that won't be used until the rolling update is finished (so old nodes never receive new messages). ::: ## Health @@ -147,5 +211,5 @@ libraryDependencies += "com.devsisters" %% "shardcake-health-k8s" % "2.7.1" You can then simply use the `K8sPodsHealth.live` layer. This is requiring a `Pods` layer that comes from [zio-k8s](https://coralogix.github.io/zio-k8s/docs/overview/overview_gettingstarted). ::: tip Examples -Check the [examples](https://github.com/devsisters/shardcake/tree/series/3.x/examples/src/main/scala/example/complex) folder that contains a full example using Redis, gRPC and Kryo seralization. -::: \ No newline at end of file +Check the [examples](https://github.com/devsisters/shardcake/tree/series/3.x/examples/src/main/scala/example/complex) folder that contains a full example using Redis, gRPC and the Kryo message codec. +:::