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 comcast.ip4s.IpLiteralSyntax
import fs2.io.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"))
            }
     }.parJoinUnboundedpile.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?

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 comcast.ip4s.IpLiteralSyntax
import fs2.io.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"))
            }
     }.parJoinUnboundedpile.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?

• scala
• implicit
• circe
• fs2

1 Answer 1

If we look at the sourcecode of the circe-fs2 package: https://github/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}"))
          }
     }.parJoinUnboundedpile.drain
   }
}