I'm trying to realise simple server on FS2, which receives json message and decodes it with circe-library.
My code is following:
import cats.effect.{IO, IOApp, Temporal}
import cats.effect.std.Console
import cats.effect.kernel.Concurrent
import com.comcast.ip4s.IpLiteralSyntax
import fs2.io.net.Network
import fs2.{Stream, text}
import io.circe.Decoder
import io.circe.fs2._
import io.circe.generic.semiauto.deriveDecoder
object ServerApp extends IOApp.Simple {
case class Auth(password: String, clientId: String)
implicit val decoder: Decoder[Auth] = deriveDecoder[Auth]
override def run: IO[Unit] = cloudServer[IO]
def cloudServer[F[_]: Concurrent: Console: Network: Temporal]: F[Unit] = {
Network[F].server(port = Some(port"5555")).map { client =>
client.reads
.through(text.utf8.decode)
.through(stringStreamParser)
.through(decoder[F, Auth])
.evalTap { auth =>
Console[F].println(s"Received auth: $auth")
}
.handleErrorWith { error =>
Stream.eval(Console[F].errorln(s"Error processing client: $error"))
}
}.parJoinUnbounded.compile.drain
}
}
But at string ".through(stringStreamParser)" I receive compile error: "No implicit found for evidence"
It's something strange, taking into consideration, that "implicit val decoder: Decoder[Auth] = deriveDecoder[Auth]" is in a scope of object ServerApp
What am I doing wrong?
If we look at the sourcecode of the circe-fs2 package: https://github.com/circe/circe-fs2/blob/master/fs2/src/main/scala/io/circe/fs2/package.scala
We can see that stringStreamParser
requires a Sync[F]
but you only have a Concurrent[F]
here, meaning that if you need both you will need to ask for an Async[F]
there... which is pretty much equal to IO
.
My humble advice, tagless final is a lot of complexity for little value, just use IO
directly. But, if you want to keep following that style, then the best you can do is split that function into smaller ones, so you only require Async
at the top and more fine-grained capabilities in other places.
BTW, rather than doing through(text.utf8.decode)
and through(stringStreamParser)
, you can just use byteStreamParser
.
Thus, overall, I would write that code like this:
(I didn't test the code, so it may have typos)
object ServerApp extends IOApp.Simple {
final case class Auth(password: String, clientId: String)
implicit val decoder: Decoder[Auth] = deriveDecoder[Auth]
override final val run: IO[Unit] =
Network[IO].server(port = Some(port"5555")).map { client =>
client
.reads
.through(byteStreamParser)
.through(decoder[F, Auth])
.evalTap { auth =>
IO.println(s"Received auth: ${auth}")
}.handleErrorWith { error =>
Stream.eval(IO.println(s"Error processing client: ${error}"))
}
}.parJoinUnbounded.compile.drain
}
}