diff --git a/src/main/scala/db/DBService.scala b/src/main/scala/db/DBService.scala index f1be0b9..a0a49bf 100644 --- a/src/main/scala/db/DBService.scala +++ b/src/main/scala/db/DBService.scala @@ -59,10 +59,10 @@ class DBService(log: Logger[IO]) { IO(Files.writeString(path, content)) .redeemWith( error => IO(Left(error)) -// .flatTap(_ => IO.println(s"Write file '$fileName' failed with error: ${error.getMessage}")) +// .flatTap(_ => log.error(s"Write file '$fileName' failed with error: ${error.getMessage}")) , _ => IO(Right(fileName)) -// .flatTap(_ => IO.println(s"write: $fileName")) +// .flatTap(_ => log.info(s"write: $fileName")) ) } diff --git a/src/main/scala/db/Main.scala b/src/main/scala/db/Main.scala index 8a7096b..7559511 100644 --- a/src/main/scala/db/Main.scala +++ b/src/main/scala/db/Main.scala @@ -3,6 +3,7 @@ package db import cats.effect.IO import cats.effect.unsafe.implicits.global import cats.implicits.toTraverseOps +import org.typelevel.log4cats.slf4j.Slf4jLogger import java.time.{LocalDate, LocalDateTime} import java.time.format.DateTimeFormatter @@ -15,30 +16,32 @@ object Main { val to = LocalDateTime.parse("20230501_1230", dateFormatter) for { + log <- Slf4jLogger.create[IO] dbService <- DBService.of lines <- dbService.getInRange(from, to) - _ <- lines.traverse(IO.println) + _ <- lines.traverse(log.info(_)) } yield () } private def testGetDates: IO[Unit] = { for { + log <- Slf4jLogger.create[IO] dbService <- DBService.of dates <- dbService.getDates() - _ <- IO.println(dates) + _ <- log.info(s"$dates") } yield () } private def testGetDate: IO[Unit] = { for { + log <- Slf4jLogger.create[IO] dbService <- DBService.of - dates <- dbService.getDateFileNames(LocalDate.of(2023, 4, 23)) - _ <- IO.println(dates) + dates <- dbService.getDateFileNames(LocalDate.of(2023, 5, 15)) + _ <- log.info(s"$dates") } yield () } def main(args: Array[String]): Unit = { - println("----------------> db main") // testGetInRange.unsafeRunSync() // testGetDates.unsafeRunSync() testGetDate.unsafeRunSync() diff --git a/src/main/scala/fetch/FetchService.scala b/src/main/scala/fetch/FetchService.scala index 20b7a3f..943a65b 100644 --- a/src/main/scala/fetch/FetchService.scala +++ b/src/main/scala/fetch/FetchService.scala @@ -6,7 +6,8 @@ import org.http4s._ import org.http4s.client.Client import org.http4s.headers.Authorization import org.http4s.ember.client.EmberClientBuilder - +import org.typelevel.log4cats.Logger +import org.typelevel.log4cats.slf4j.Slf4jLogger import pureconfig._ import pureconfig.generic.auto._ @@ -20,6 +21,12 @@ final case class WeatherServerConfig( ) object FetchService { + def of: IO[FetchService] = { + Slf4jLogger.create[IO].map(logger => new FetchService(logger)) + } +} + +class FetchService(log: Logger[IO]) { private val weatherServerConfig: WeatherServerConfig = ConfigSource.default.load[WeatherServerConfig] match { case Right(config) => config case Left(errors) => throw new RuntimeException(s"Unable to load config: $errors") @@ -36,10 +43,10 @@ object FetchService { client.expect[String](request).redeemWith( error => IO(Left(error)) - // .flatTap(_ => IO.println(s"Request failed to url: $url with error: ${error.getMessage}")), +// .flatTap(_ => log.error(s"Request failed to url: $url with error: ${error.getMessage}")) , fileContent => IO(Right((fileName, fileContent))) - // .flatTap(_ => IO.println(s"Fetched: $fileName")) +// .flatTap(_ => log.info(s"Fetched: $fileName")) ) } @@ -54,11 +61,11 @@ object FetchService { fetchFiles(List(fileName)).map { results => results.headOption match { case Some(Right(result)) => - IO.println(s"fetched: $fileName").as(Right(result)) + log.info(s"fetched: $fileName").as(Right(result)) case Some(Left(err)) => - IO.println(s"failed fetch: $fileName with error: ${err.getMessage}").as(Left(err)) + log.error(s"failed fetch: $fileName with error: ${err.getMessage}").as(Left(err)) case None => - IO.println(s"failed fetch: $fileName").as(Left(new Exception("No file fetched"))) + log.error(s"failed fetch: $fileName").as(Left(new Exception("No file fetched"))) } }.flatten } diff --git a/src/main/scala/fetch/FileFetchScheduler.scala b/src/main/scala/fetch/FileFetchScheduler.scala index f61fa2a..d95e71a 100644 --- a/src/main/scala/fetch/FileFetchScheduler.scala +++ b/src/main/scala/fetch/FileFetchScheduler.scala @@ -8,28 +8,30 @@ import org.typelevel.log4cats.slf4j.Slf4jLogger object FileFetchScheduler { - def of(dbService: DBService): IO[FileFetchScheduler] = { + def of(dbService: DBService, fetch: FetchService): IO[FileFetchScheduler] = { Slf4jLogger.create[IO].map { - new FileFetchScheduler(dbService, _) + new FileFetchScheduler(dbService, fetch, _) } } } -class FileFetchScheduler(dbService: DBService, log: Logger[IO]) { +class FileFetchScheduler(dbService: DBService, fetch: FetchService, log: Logger[IO]) { def run: Stream[IO, Unit] = { - val fetchTask = FileNameService.generateCurrentHour.flatMap(FetchService.fetchSingleFile) - Scheduler.scheduleTask(fetchTask) - .evalMap { - case Left(fetchErr) => - log.error(s"Fetch error: $fetchErr") - case Right((name, content)) => + val fetchTask = FileNameService.generateCurrentHour.flatMap(fetch.fetchSingleFile) - dbService.save(name, content).attempt.flatMap { - case Left(err) => log.error(s"error: $err") - case Right(saveResult) => saveResult match { + Stream.eval(Scheduler.of).flatMap { scheduler => + scheduler.scheduleTask(fetchTask) + .evalMap { + case Left(fetchErr) => + log.error(s"Fetch error: $fetchErr") + case Right((name, content)) => + dbService.save(name, content).attempt.flatMap { case Left(err) => log.error(s"error: $err") - case Right(savedName) => log.info(s"saved: $savedName") + case Right(saveResult) => saveResult match { + case Left(err) => log.error(s"error: $err") + case Right(savedName) => log.info(s"saved: $savedName") + } } - } - } + } + } } } \ No newline at end of file diff --git a/src/main/scala/fetch/Main.scala b/src/main/scala/fetch/Main.scala index a98a767..4eba071 100644 --- a/src/main/scala/fetch/Main.scala +++ b/src/main/scala/fetch/Main.scala @@ -4,6 +4,7 @@ import cats.effect.IO import cats.effect.unsafe.implicits.global import cats.implicits.toTraverseOps import db.DBService +import org.typelevel.log4cats.slf4j.Slf4jLogger import java.time.LocalDateTime @@ -12,8 +13,10 @@ object Main { val from = LocalDateTime.of(2023, 4, 28, 10, 0) val to = LocalDateTime.of(2023, 4, 28, 13, 30) for { -// fetchResultEither <- FetchService.fetchFromDate(LocalDate.of(2023, 4, 28)).attempt - fetchResultEither <- FetchService.fetchInRange(from, to).attempt + log <- Slf4jLogger.create[IO] + fetch <- FetchService.of +// fetchResultEither <- fetch.fetchFromDate(LocalDate.of(2023, 4, 28)).attempt + fetchResultEither <- fetch.fetchInRange(from, to).attempt fetchServiceError = fetchResultEither.left.toOption.map(e => s"FetchServiceError: ${e.getMessage}").toList fetchResult = fetchResultEither.getOrElse(List.empty) (fetchErrors, successDownloads) = fetchResult.partitionMap(identity) @@ -22,8 +25,8 @@ object Main { (saveErrors, successSaves) = saveResults.partitionMap(identity) successes = successDownloads.map(s => s"fetched: ${s._1}") ++ successSaves.map(s => s"saved: $s") errors = fetchServiceError ++ fetchErrors.map(e => s"FetchError: ${e.getMessage}") ++ saveErrors.map(e => s"SaveError: ${e.getMessage}") - _ <- IO.println(s"errors: $errors") - _ <- IO.println(s"successes: $successes") + _ <- log.info(s"errors: $errors") + _ <- log.info(s"successes: $successes") } yield (successes, errors) } diff --git a/src/main/scala/fetch/Scheduler.scala b/src/main/scala/fetch/Scheduler.scala index 71cdf9d..644b541 100644 --- a/src/main/scala/fetch/Scheduler.scala +++ b/src/main/scala/fetch/Scheduler.scala @@ -4,15 +4,27 @@ import cats.effect._ import cats.effect.unsafe.implicits.global import cats.implicits.catsSyntaxApply import fs2.Stream +import org.typelevel.log4cats.Logger +import org.typelevel.log4cats.slf4j.Slf4jLogger +import server.Server import java.time.{Duration, LocalTime} import scala.concurrent.duration._ + object Scheduler { + def of: IO[Scheduler] = { + Slf4jLogger.create[IO].map { + new Scheduler(_) + } + } +} + +class Scheduler(log: Logger[IO]) { private val downloadMinute = 31 // private def testTask: IO[Unit] = { -// IO(println("Running task")) +// log.info("Running task") // } def durationToNextHalfHour(implicit clock: Clock[IO]): IO[FiniteDuration] = { @@ -34,16 +46,18 @@ object Scheduler { def scheduleTask(task: IO[Either[Throwable, (String, String)]]): Stream[IO, Either[Throwable, (String, String)]] = { Stream.eval(durationToNextHalfHour).flatMap { delay => { - Stream.eval(IO.println(s"Scheduler started with delay: ${delay.toMinutes} min")) *> + Stream.eval(log.info(s"Scheduler started with delay: ${delay.toMinutes} min")) *> (Stream.sleep[IO](delay) ++ Stream.awakeEvery[IO](1.hour)) .evalMap(_ => task) }} } + // This is just for testing def main(args: Array[String]): Unit = { // run.compile.drain.unsafeRunSync() - - val fetchTask = FileNameService.generateCurrentHour.flatMap(FetchService.fetchSingleFile) - scheduleTask(fetchTask).compile.drain.unsafeRunSync() + for { + fetch <- FetchService.of + fetchTask = FileNameService.generateCurrentHour.flatMap(fetch.fetchSingleFile) + } yield scheduleTask(fetchTask).compile.drain.unsafeRunSync() } } diff --git a/src/main/scala/parse/Main.scala b/src/main/scala/parse/Main.scala index 476a539..9d9d195 100644 --- a/src/main/scala/parse/Main.scala +++ b/src/main/scala/parse/Main.scala @@ -4,6 +4,7 @@ import cats.effect.IO import cats.effect.unsafe.implicits.global import db.DBService import io.circe.syntax.EncoderOps +import org.typelevel.log4cats.slf4j.Slf4jLogger import parse.Aggregate.AggregateValueImplicits.aggregateValueEncoder import parse.Aggregate.{AggregateKey, UserQuery} @@ -12,8 +13,6 @@ import java.time.format.DateTimeFormatter object Main { private def run: IO[Unit] = { - println("================ start parser") - val formatter = DateTimeFormatter.ofPattern("yyyyMMdd_HHmm") val from = LocalDateTime.parse("20230515_0905", formatter) val to = LocalDateTime.parse("20230516_0942", formatter) @@ -21,10 +20,11 @@ object Main { // val userQuery = UserQuery(List("Daugavgrīva"), "precipitation", AggregateKey.List) for { + log <- Slf4jLogger.create[IO] dbService <- DBService.of lines <- dbService.getInRange(from, to) parsed <- IO.pure(Parser.queryData(userQuery, lines)) - _ <- IO.println(parsed.asJson) + _ <- log.info(parsed.asJson.toString) } yield () } diff --git a/src/main/scala/server/Main.scala b/src/main/scala/server/Main.scala index ab339a8..c348385 100644 --- a/src/main/scala/server/Main.scala +++ b/src/main/scala/server/Main.scala @@ -1,17 +1,18 @@ package server import cats.effect._ import cats.implicits.catsSyntaxTuple2Parallel - import db.DBService -import fetch.FileFetchScheduler +import fetch.{FetchService, FileFetchScheduler} object Main extends IOApp { def run(args: List[String]): IO[ExitCode] = { for { dbService <- DBService.of - fileFetchScheduler <- FileFetchScheduler.of(dbService) + fetch <- FetchService.of + fileFetchScheduler <- FileFetchScheduler.of(dbService, fetch) schedulerTask = fileFetchScheduler.run.compile.drain - serverTask = Server.run + server <- Server.of(dbService, fetch) + serverTask = server.run exitCode <- (serverTask, schedulerTask).parMapN((_, _) => ExitCode.Success) } yield exitCode } diff --git a/src/main/scala/server/Server.scala b/src/main/scala/server/Server.scala index e7baf3f..33caf64 100644 --- a/src/main/scala/server/Server.scala +++ b/src/main/scala/server/Server.scala @@ -20,11 +20,20 @@ import io.circe.syntax._ import parse.Aggregate.AggregateValueImplicits.aggregateValueEncoder import parse.Aggregate.{AggregateKey, UserQuery} import org.http4s.circe.jsonEncoder - +import org.typelevel.log4cats.Logger +import org.typelevel.log4cats.slf4j.Slf4jLogger import scala.concurrent.duration.DurationInt object Server { + def of(dbService: DBService, fetch: FetchService): IO[Server] = { + Slf4jLogger.create[IO].map { + new Server(dbService, fetch, _) + } + } +} + +class Server(dbService: DBService, fetch: FetchService, log: Logger[IO]) { // Define the extension method `pretty` for Json implicit class JsonPrettyPrinter(json: Json) { @@ -38,25 +47,24 @@ object Server { // http://0.0.0.0:8080/api/query/20230414_2200-20230501_1230/Liepāja,Rēzekne/tempMax/max case GET -> Root / "query" / DateTimeRange(from, to) / CityList(cities) / field / AggKey(key) => - DBService.of.flatMap(_.getInRange(from, to) + dbService.getInRange(from, to) .map(Parser.queryData(UserQuery(cities, field, key), _)) - .flatMap(result => Ok(result.asJson.pretty))) + .flatMap(result => Ok(result.asJson.pretty)) // http://0.0.0.0:8080/api/fetch/date/20230514 case GET -> Root / "fetch" / "date" / ValidDate(date) => val result = for { - fetchResultEither <- FetchService.fetchFromDate(date).attempt + fetchResultEither <- fetch.fetchFromDate(date).attempt fetchServiceError = fetchResultEither.left.toOption.map(e => s"FetchServiceError: ${e.getMessage}").toList fetchResult = fetchResultEither.getOrElse(List.empty) (fetchErrors, successDownloads) = fetchResult.partitionMap(identity) - dbService <- DBService.of saveResults <- successDownloads.traverse { case (name, content) => dbService.save(name, content) } (saveErrors, successSaves) = saveResults.partitionMap(identity) // successes = successDownloads.map(s => s"fetched: ${s._1}") ++ successSaves.map(s => s"saved: $s") successes = successSaves errors = fetchServiceError ++ fetchErrors.map(e => s"FetchError: ${e.getMessage}") ++ saveErrors.map(e => s"SaveError: ${e.getMessage}") - _ <- IO.println(s"errors: $errors") - _ <- IO.println(s"successes: $successes") + _ <- log.error(s"errors: $errors") + _ <- log.info(s"successes: $successes") } yield (successes, errors) result.flatMap { case (successes, errors) => @@ -68,19 +76,19 @@ object Server { // http://0.0.0.0:8080/api/show/all_dates case GET -> Root / "show" / "all_dates" => - DBService.of.flatMap(_.getDates().flatMap(dates => + dbService.getDates().flatMap(dates => Ok(dates.asJson.pretty) - )) + ) // http://0.0.0.0:8080/api/show/date/20230423 case GET -> Root / "show" / "date" / ValidDate(date) => - DBService.of.flatMap(_.getDateFileNames(date).flatMap(fileNames => + dbService.getDateFileNames(date).flatMap(fileNames => Ok(fileNames.asJson.pretty) - )) + ) // http://0.0.0.0:8080/api/show/file/20230423_12:30.csv case GET -> Root / "show" / "file" / (fileName: String) => - DBService.of.flatMap(_.readFile(fileName).flatMap(content => Ok(content.asJson))) + dbService.readFile(fileName).flatMap(content => Ok(content.asJson)) // http://0.0.0.0:8080/api/help case GET -> Root / "help" => { @@ -98,14 +106,14 @@ object Server { } } - val corsConfig = CORSConfig.default + private val corsConfig = CORSConfig.default .withAnyOrigin(true) .withAnyMethod(true) .withAllowedMethods(Some(Set(Method.GET, Method.POST))) .withAllowCredentials(false) .withMaxAge(1.day) - val apiRoutesCors = CORS(apiRoutes, corsConfig) + private val apiRoutesCors = CORS(apiRoutes, corsConfig) private val httpApp = Router( "/" -> staticcontent.fileService[IO](FileService.Config("./web/dist")), diff --git a/web/src/App.tsx b/web/src/App.tsx index 27d98e2..10b293d 100644 --- a/web/src/App.tsx +++ b/web/src/App.tsx @@ -10,7 +10,7 @@ console.log("api host:", import.meta.env.VITE_API_HOST); type Section = "aggregator" | "fileManager"; const App: Component = () => { - const [getSection, setSection] = createSignal
("fileManager"); + const [getSection, setSection] = createSignal
("aggregator"); const section = () => getSection() === "aggregator" ? : ;