Generalize serialization and deserialization of proto messages for conformance tests

Author: igor-vovk

conformance/src/main/scala/org/ivovk/connect_rpc_scala/conformance/Http4sServerLauncher.scala

@@ -2,10 +2,14 @@ package org.ivovk.connect_rpc_scala.conformance
 
 import cats.effect.{IO, IOApp}
 import com.comcast.ip4s.{host, port, Port}
-import connectrpc.conformance.v1.{ConformanceServiceFs2GrpcTrailers, ServerCompatResponse}
+import connectrpc.conformance.v1.{
+  ConformanceServiceFs2GrpcTrailers,
+  ServerCompatRequest,
+  ServerCompatResponse,
+}
 import org.http4s.ember.server.EmberServerBuilder
 import org.http4s.server.middleware.Logger
-import org.ivovk.connect_rpc_scala.conformance.util.ServerCompatSerDeser
+import org.ivovk.connect_rpc_scala.conformance.util.ProtoSerDeser
 import org.ivovk.connect_rpc_scala.http4s.Http4sRouteBuilder
 import org.slf4j.LoggerFactory
 
@@ -29,7 +33,7 @@ object Http4sServerLauncher extends IOApp.Simple {
 
   override def run: IO[Unit] = {
     val res = for
-      req <- ServerCompatSerDeser.readRequest[IO](System.in).toResource
+      req <- ProtoSerDeser.read[IO, ServerCompatRequest](System.in).toResource
 
       service <- ConformanceServiceFs2GrpcTrailers.bindServiceResource(
         ConformanceServiceImpl[IO]()
@@ -61,7 +65,7 @@ object Http4sServerLauncher extends IOApp.Simple {
       addr = server.address
       resp = ServerCompatResponse(addr.getHostString, addr.getPort)
 
-      _ <- ServerCompatSerDeser.writeResponse[IO](System.out, resp).toResource
+      _ <- ProtoSerDeser.write[IO](System.out, resp).toResource
 
       _ = System.err.println(s"Server started on $addr...")
     yield ()

conformance/src/main/scala/org/ivovk/connect_rpc_scala/conformance/NettyServerLauncher.scala

@@ -1,8 +1,12 @@
 package org.ivovk.connect_rpc_scala.conformance
 
 import cats.effect.{IO, IOApp}
-import connectrpc.conformance.v1.{ConformanceServiceFs2GrpcTrailers, ServerCompatResponse}
-import org.ivovk.connect_rpc_scala.conformance.util.ServerCompatSerDeser
+import connectrpc.conformance.v1.{
+  ConformanceServiceFs2GrpcTrailers,
+  ServerCompatRequest,
+  ServerCompatResponse,
+}
+import org.ivovk.connect_rpc_scala.conformance.util.ProtoSerDeser
 import org.ivovk.connect_rpc_scala.netty.NettyServerBuilder
 import org.slf4j.LoggerFactory
 
@@ -26,7 +30,7 @@ object NettyServerLauncher extends IOApp.Simple {
 
   override def run: IO[Unit] = {
     val res = for
-      req <- ServerCompatSerDeser.readRequest[IO](System.in).toResource
+      req <- ProtoSerDeser.read[IO, ServerCompatRequest](System.in).toResource
 
       service <- ConformanceServiceFs2GrpcTrailers.bindServiceResource(
         ConformanceServiceImpl[IO]()
@@ -45,7 +49,7 @@ object NettyServerLauncher extends IOApp.Simple {
 
       resp = ServerCompatResponse(server.host, server.port)
 
-      _ <- ServerCompatSerDeser.writeResponse[IO](System.out, resp).toResource
+      _ <- ProtoSerDeser.write[IO](System.out, resp).toResource
 
       _ = System.err.println(s"Server started on ${server.host}:${server.port}...")
       _ = logger.info(s"Netty-server started on ${server.host}:${server.port}...")

conformance/src/main/scala/org/ivovk/connect_rpc_scala/conformance/util/ProtoSerDeser.scala

@@ -0,0 +1,22 @@
+package org.ivovk.connect_rpc_scala.conformance.util
+
+import cats.effect.Sync
+import scalapb.{GeneratedMessage as Message, GeneratedMessageCompanion as Companion}
+
+import java.io.{InputStream, OutputStream}
+
+object ProtoSerDeser {
+  def read[F[_]: Sync, T <: Message](in: InputStream)(using comp: Companion[T]): F[T] =
+    Sync[F].delay {
+      val size = IntSerDeser.read(in)
+      comp.parseFrom(in.readNBytes(size))
+    }
+
+  def write[F[_]: Sync](out: OutputStream, msg: Message): F[Unit] =
+    Sync[F].delay {
+      IntSerDeser.write(out, msg.serializedSize)
+      out.flush()
+      out.write(msg.toByteArray)
+      out.flush()
+    }
+}