BodyParserInstances.scala

package controllers
import akka.stream.SinkShape
import akka.stream.scaladsl.{Broadcast, GraphDSL, Sink}
import akka.util.ByteString
import play.api.libs.functional.{Applicative, Functor}
import play.api.libs.streams.Accumulator
import play.api.mvc.BodyParser

import scala.concurrent.ExecutionContext

trait BodyParserInstances {

  implicit def bodyParserApplicative(
    implicit ec: ExecutionContext
  ): Applicative[BodyParser] = new Applicative[BodyParser] {

    override def pure[A](a: A): BodyParser[A] =
      BodyParser(_ => Accumulator.done(Right(a)))

    override def map[A, B](m: BodyParser[A], f: A => B): BodyParser[B] =
      m.map(f)

    override def apply[A, B](mf: BodyParser[A => B],
                             ma: BodyParser[A]): BodyParser[B] =
      BodyParser { rh =>
        val sinkGraph = GraphDSL.create(mf(rh).toSink, ma(rh).toSink) {
          case (ff, fa) =>
            for {
              eitherF <- ff
              eitherA <- fa
            } yield {
              for {
                f <- eitherF
                a <- eitherA
              } yield f(a)
            }
        } { implicit builder => (sinkF, sinkA) =>
          import GraphDSL.Implicits._

          val bcast = builder.add(Broadcast[ByteString](2, eagerCancel = false))
          bcast.out(0) ~> sinkF
          bcast.out(1) ~> sinkA
          SinkShape(bcast.in)
        }

        Accumulator(Sink.fromGraph(sinkGraph))
      }
  }

  implicit def bodyParserFunctor(
    implicit ec: ExecutionContext
  ): Functor[BodyParser] = new Functor[BodyParser] {
    override def fmap[A, B](m: BodyParser[A], f: A => B): BodyParser[B] =
      m.map(f)
  }
}

object BodyParserInstances extends BodyParserInstances