fs2_temperature.scala

package qiita

import java.util.concurrent.{ExecutorService, Executors}

import cats.effect._
import cats.syntax.functor._
import fs2.{Pipe, Stream, io, text}

import scala.concurrent.ExecutionContext
import scala.util.Try

trait StreamApp extends IOApp {
  def runWithEC(s: Blocker => Stream[IO, Unit]): IO[ExitCode] =
    Stream.resource(Blocker[IO]).flatMap(s).compile.drain.as(ExitCode.Success)
}
trait EchoMain extends StreamApp {
  def echo[F[_]: Sync: ContextShift](bl: Blocker): Stream[F, Unit]
  def run(args: List[String]): IO[ExitCode] = runWithEC(echo[IO])
}

trait ConvertMain extends StreamApp {
  def convert[F[_]: Sync: ContextShift](bl: Blocker): Stream[F, Unit]
  def run(args: List[String]): IO[ExitCode] = runWithEC(convert[IO])
}

object ex02_0 extends EchoMain {
  def echo[F[_]: Sync: ContextShift](bl: Blocker): Stream[F, Unit] =
    io.stdin(4096, bl)
      .through(io.stdout(bl)) // Stream[F, Byte]=>Stream[F, Unit]
}

object ex02_1 extends EchoMain {
  def echo[F[_]: Sync: ContextShift](bl: Blocker): Stream[F, Unit] =
    io.stdin(4096, bl)
      .through(text.utf8Decode) // Stream[F, Byte]  => Stream[F, String]
      .through(text.lines)      // Stream[F, String]=> Stream[F, String]
      .map(line => s"$line\n")
      .through(text.utf8Encode) // Stream[F, String]=> Stream[F, Byte]
      .through(io.stdout(bl))   // Stream[F, Byte]  => Stream[F, Unit]
}

object ex02_2 extends EchoMain {
  def echo[F[_]: Sync: ContextShift](bl: Blocker): Stream[F, Unit] =
    io.stdin(4096, bl)
      .through(text.utf8Decode)
      .through(text.lines)
      .takeWhile(s => s != ":q") // <-- ここ
      .map(line => s"$line\n")
      .through(text.utf8Encode)
      .through(io.stdout(bl))
}

object ex02_3 extends ConvertMain {
  def fahrenheitToCelsius(f: Double): Double = (f - 32.0) * (5.0/9.0)

  def convertF2C(line: String): String =
      s"${ Try(line.toDouble).fold(_ => "error", fahrenheitToCelsius(_).toString) }\n"

  def convert[F[_]: Sync: ContextShift](bl: Blocker): Stream[F, Unit] =
    io.stdin(4096, bl)
      .through(text.utf8Decode)
      .through(text.lines)
      .takeWhile(s => s != ":q")
      .through(_ map convertF2C) // Pipe[F, String, String]
      .through(text.utf8Encode)
      .through(io.stdout(bl))
}

object ex02 extends ConvertMain {
  def fahrenheitToCelsius(f: Double): Double = (f - 32.0) * (5.0/9.0)
  def celsiusToFahrenheit(c: Double): Double = c / (5.0/9.0) + 32.0

  sealed trait Mode
  case object F2C extends Mode
  case object C2F extends Mode

  def changeMode[F[_]]: Pipe[F, String, (Mode, String)] =
    _.scan((F2C: Mode, "")) { case ((mode, _), b) =>
        (b match {
          case ":c" => F2C
          case ":f" => C2F
          case _    => mode
        }, b)
      }
      .filter(p => p._2 != ":c" && p._2 != ":f")
      .drop(1)

  def convertTemperature[F[_]]: Pipe[F, (Mode, String), String] =
    _.map { case (mode, line) =>
      val (f, s) = if (mode == F2C) (fahrenheitToCelsius _, "%s°F ⇒ %s°C")
                   else             (celsiusToFahrenheit _, "%s°C ⇒ %s°F")
      s"${ Try(line.trim.toDouble).fold(_ => "error", d => s.format(d, f(d))) }\n"
    }

  def convert[F[_]: Sync: ContextShift](bl: Blocker): Stream[F, Unit] =
    io.stdin(4096, bl)
      .through(text.utf8Decode)
      .through(text.lines)
      .takeWhile(s => s != ":q")
      .through(changeMode andThen convertTemperature) // <-- ここ
      .through(text.utf8Encode)
      .through(io.stdout(bl))
}