bot/WeatherTool
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Dependency inject logger and replace println

Browse Source
This commit is contained in:
Guntis Smaukstelis
2023-05-21 22:37:33 +03:00
parent 00dd4a4f4a
commit 760ef4e719
10 changed files with 97 additions and 59 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/scala/db/DBService.scala
+2 -2
View File
@@ -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"))
)
}
src/main/scala/db/Main.scala
+8 -5
View File
@@ -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()
src/main/scala/fetch/FetchService.scala
+13 -6
View File
@@ -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
}
src/main/scala/fetch/FileFetchScheduler.scala
+17 -15
View File
@@ -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")
}
}
}
}
}
}
}
}
src/main/scala/fetch/Main.scala
+7 -4
View File
@@ -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)
}
src/main/scala/fetch/Scheduler.scala
+19 -5
View File
@@ -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()
}
}
src/main/scala/parse/Main.scala
+3 -3
View File
@@ -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 ()
}
src/main/scala/server/Main.scala
+5 -4
View File
@@ -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
}
src/main/scala/server/Server.scala
+22 -14
View File
@@ -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")),