From 15cfbd1801802f2373f5ab82e67b5e3a55c08563 Mon Sep 17 00:00:00 2001 From: Guntis Smaukstelis Date: Fri, 14 Feb 2025 14:46:31 +0200 Subject: [PATCH] Refactor to one scheduler, separated csv/grib under one folder --- build.sbt | 2 +- src/main/scala/{app => }/Main.scala | 20 ++++---- src/main/scala/data/DataService.scala | 2 +- src/main/scala/db/Main.scala | 8 +-- src/main/scala/db/PostgresService.scala | 6 +-- src/main/scala/fetch/FileFetchScheduler.scala | 34 ------------- src/main/scala/fetch/Scheduler.scala | 42 --------------- .../scala/fetch/{ => csv}/FetchService.scala | 4 +- .../fetch/{ => csv}/FileNameService.scala | 4 +- .../dmi}/FetchService.scala | 8 +-- .../dmi}/FetchServiceTest.scala | 2 +- .../{fetchDMI => fetch/dmi}/FileName.scala | 2 +- .../scala/parse/{ => csv}/Aggregate.scala | 3 +- src/main/scala/parse/{ => csv}/Main.scala | 9 +--- src/main/scala/parse/{ => csv}/Parser.scala | 2 +- .../scala/parse/{ => csv}/WeatherData.scala | 2 +- src/main/scala/{ => parse}/grib/Codes.scala | 3 +- src/main/scala/{ => parse}/grib/Grib.scala | 3 +- .../scala/{ => parse}/grib/GribParser.scala | 2 +- .../{ => parse}/grib/GribParserTest.scala | 6 +-- .../{fetchDMI => scheduler}/Scheduler.scala | 6 +-- src/main/scala/server/Server.scala | 18 +++---- src/main/scala/server/ValidateRoutes.scala | 4 +- .../scala/fetch/FileFetchSchedulerSpec.scala | 51 ------------------- .../scala/fetch/FileNameServiceSpec.scala | 2 +- src/test/scala/fetch/SchedulerSpec.scala | 25 --------- src/test/scala/parse/ParserSpec.scala | 2 +- 27 files changed, 56 insertions(+), 216 deletions(-) rename src/main/scala/{app => }/Main.scala (64%) delete mode 100644 src/main/scala/fetch/FileFetchScheduler.scala delete mode 100644 src/main/scala/fetch/Scheduler.scala rename src/main/scala/fetch/{ => csv}/FetchService.scala (99%) rename src/main/scala/fetch/{ => csv}/FileNameService.scala (99%) rename src/main/scala/{fetchDMI => fetch/dmi}/FetchService.scala (99%) rename src/main/scala/{fetchDMI => fetch/dmi}/FetchServiceTest.scala (98%) rename src/main/scala/{fetchDMI => fetch/dmi}/FileName.scala (98%) rename src/main/scala/parse/{ => csv}/Aggregate.scala (99%) rename src/main/scala/parse/{ => csv}/Main.scala (90%) rename src/main/scala/parse/{ => csv}/Parser.scala (98%) rename src/main/scala/parse/{ => csv}/WeatherData.scala (99%) rename src/main/scala/{ => parse}/grib/Codes.scala (99%) rename src/main/scala/{ => parse}/grib/Grib.scala (97%) rename src/main/scala/{ => parse}/grib/GribParser.scala (99%) rename src/main/scala/{ => parse}/grib/GribParserTest.scala (97%) rename src/main/scala/{fetchDMI => scheduler}/Scheduler.scala (86%) delete mode 100644 src/test/scala/fetch/FileFetchSchedulerSpec.scala delete mode 100644 src/test/scala/fetch/SchedulerSpec.scala diff --git a/build.sbt b/build.sbt index 8bbf265..f9d9870 100644 --- a/build.sbt +++ b/build.sbt @@ -5,7 +5,7 @@ ThisBuild / scalaVersion := "2.13.10" lazy val root = (project in file(".")) .settings( name := "WeatherTool", - Compile / mainClass := Some("app.Main") + Compile / mainClass := Some("Main") ) val doobieVersion = "1.0.0-RC1" diff --git a/src/main/scala/app/Main.scala b/src/main/scala/Main.scala similarity index 64% rename from src/main/scala/app/Main.scala rename to src/main/scala/Main.scala index 203b895..c793f43 100644 --- a/src/main/scala/app/Main.scala +++ b/src/main/scala/Main.scala @@ -1,10 +1,10 @@ -package app - import cats.effect._ import cats.implicits.catsSyntaxTuple4Parallel import data.DataService import db.{DBConnection, PostgresService} -import fetch.{FetchService, FileFetchScheduler} +import fetch.csv.{FetchService, FileNameService} +import fetch.dmi +import scheduler.Scheduler import server.Server object Main extends IOApp { @@ -14,19 +14,19 @@ object Main extends IOApp { postgresService <- PostgresService.of(transactor) _ <- postgresService.createWeatherTable // create table if it does not exists - fetch <- FetchService.of - fileFetchScheduler <- FileFetchScheduler.of(postgresService, fetch) - fetchCsvTask = fileFetchScheduler.run.compile.drain - - scheduler <- fetchDMI.Scheduler.of + scheduler <- Scheduler.of dataService <- DataService.of + fetchService <- FetchService.of + + fetchCsvList = new FileNameService().generateCurrentHour.flatMap(fetchService.fetchSingleFile) + fetchCsvTask = scheduler.scheduleTask("Fetch CSV", List(31), fetchCsvList).compile.drain cleanupTask = scheduler.scheduleTask("Cleanup", List(1), dataService.deleteOldForecasts()).compile.drain - fetchGrib <- fetchDMI.FetchService.of(dataService) + fetchGrib <- dmi.FetchService.of(dataService) fetchGribTask = scheduler.scheduleTask("Fetch Grib", List(2), fetchGrib.fetchRecentForecasts()).compile.drain - server <- Server.of(postgresService, dataService, fetch) + server <- Server.of(postgresService, dataService, fetchService) serverTask = server.run exitCode <- (serverTask, fetchCsvTask, cleanupTask, fetchGribTask).parMapN((_, _, _, _) => ExitCode.Success) diff --git a/src/main/scala/data/DataService.scala b/src/main/scala/data/DataService.scala index 85dde00..567fec1 100644 --- a/src/main/scala/data/DataService.scala +++ b/src/main/scala/data/DataService.scala @@ -3,9 +3,9 @@ package data import cats.effect.IO import cats.implicits.toTraverseOps import fs2.io.file.{Files, Path} -import grib.{Grib, GribParser} import org.typelevel.log4cats.Logger import org.typelevel.log4cats.slf4j.Slf4jLogger +import parse.grib.{Grib, GribParser} import java.nio.file.Paths import java.time.{ZoneId, ZoneOffset, ZonedDateTime} diff --git a/src/main/scala/db/Main.scala b/src/main/scala/db/Main.scala index 59a42d9..a2ccab1 100644 --- a/src/main/scala/db/Main.scala +++ b/src/main/scala/db/Main.scala @@ -7,13 +7,13 @@ import doobie._ import doobie.implicits._ import doobie.postgres.implicits._ import io.circe.syntax.EncoderOps -import parse.Aggregate.{AggregateKey, UserQuery} -import parse.Aggregate.AggregateValueImplicits.aggregateValueEncoder -import parse.Aggregate.userQueryEncoder +import parse.csv.Aggregate.{AggregateKey, UserQuery} +import parse.csv.Aggregate.AggregateValueImplicits.aggregateValueEncoder +import parse.csv.Aggregate.userQueryEncoder import io.circe.{Json, Printer} import io.circe.generic.auto._ import io.circe.syntax._ -import parse.Aggregate +import parse.csv.Aggregate import java.time.{LocalDate, LocalDateTime} import java.time.format.DateTimeFormatter diff --git a/src/main/scala/db/PostgresService.scala b/src/main/scala/db/PostgresService.scala index 05238ef..1ac766f 100644 --- a/src/main/scala/db/PostgresService.scala +++ b/src/main/scala/db/PostgresService.scala @@ -11,10 +11,10 @@ import java.time.{LocalDate, LocalDateTime, OffsetDateTime} import doobie.postgres.implicits._ import org.typelevel.log4cats.Logger import org.typelevel.log4cats.slf4j.Slf4jLogger -import parse.Aggregate.{AggregateKey, AggregateValue, DistinctStringList, DoubleValue, TimeDoubleList, UserQuery} -import parse.{Parser, WeatherStationData} -import shapeless.syntax.std.tuple.productTupleOps +import parse.csv.Aggregate.{AggregateKey, AggregateValue, DistinctStringList, DoubleValue, TimeDoubleList, UserQuery} +import parse.csv.{Parser, WeatherStationData} +import shapeless.syntax.std.tuple.productTupleOps import io.circe._ import io.circe.parser._ diff --git a/src/main/scala/fetch/FileFetchScheduler.scala b/src/main/scala/fetch/FileFetchScheduler.scala deleted file mode 100644 index 9218cc3..0000000 --- a/src/main/scala/fetch/FileFetchScheduler.scala +++ /dev/null @@ -1,34 +0,0 @@ -package fetch - -import cats.effect.IO -import db.PostgresService -import fs2.Stream -import org.typelevel.log4cats.Logger -import org.typelevel.log4cats.slf4j.Slf4jLogger - - -object FileFetchScheduler { - def of(postgresService: PostgresService, fetch: FetchService): IO[FileFetchScheduler] = { - Scheduler.of.flatMap { scheduler => - Slf4jLogger.create[IO].map { - new FileFetchScheduler(postgresService, fetch, new FileNameService(), scheduler, _) - } - } - } -} - -class FileFetchScheduler(postgresService: PostgresService, fetch: FetchService, fileNameService: FileNameService, scheduler: Scheduler, log: Logger[IO]) { - def run: Stream[IO, Unit] = { - val fetchTask = fileNameService.generateCurrentHour.flatMap(fetch.fetchSingleFile) - scheduler.scheduleTask(fetchTask) - .evalMap { - case Left(fetchErr) => - log.error(s"Fetch error: $fetchErr") - case Right((name, content)) => - postgresService.save(name, content).attempt.flatMap { - 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/Scheduler.scala b/src/main/scala/fetch/Scheduler.scala deleted file mode 100644 index 1e4f5c8..0000000 --- a/src/main/scala/fetch/Scheduler.scala +++ /dev/null @@ -1,42 +0,0 @@ -package fetch - -import cats.effect._ -import cats.implicits.catsSyntaxApply -import fs2.Stream -import org.typelevel.log4cats.Logger -import org.typelevel.log4cats.slf4j.Slf4jLogger - -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 - - def durationToNextHalfHour(implicit clock: Clock[IO]): IO[FiniteDuration] = { - clock.realTime.map { duration => - val now = LocalTime.ofSecondOfDay((duration.toMillis / 1000) % (24 * 60 * 60)) - val nextHalfHour = if (now.getMinute < downloadMinute) now.withMinute(downloadMinute) - else now.plusHours(1).withMinute(downloadMinute) - val durationToNext = Duration.between(now, nextHalfHour) - FiniteDuration(durationToNext.toMillis, MILLISECONDS) -// FiniteDuration(2000, MILLISECONDS) - } - } - - def scheduleTask(task: IO[Either[Throwable, (String, String)]]): Stream[IO, Either[Throwable, (String, String)]] = { - Stream.eval(durationToNextHalfHour).flatMap { delay => { - Stream.eval(log.info(s"CSV scheduler started with delay: ${delay.toMinutes} min")) *> - (Stream.sleep[IO](delay) ++ Stream.awakeEvery[IO](1.hour)) - .evalMap(_ => task) - }} - } -} \ No newline at end of file diff --git a/src/main/scala/fetch/FetchService.scala b/src/main/scala/fetch/csv/FetchService.scala similarity index 99% rename from src/main/scala/fetch/FetchService.scala rename to src/main/scala/fetch/csv/FetchService.scala index 0293c2a..d3d6755 100644 --- a/src/main/scala/fetch/FetchService.scala +++ b/src/main/scala/fetch/csv/FetchService.scala @@ -1,11 +1,11 @@ -package fetch +package fetch.csv import cats.effect._ import cats.implicits._ import org.http4s._ import org.http4s.client.Client -import org.http4s.headers.Authorization import org.http4s.ember.client.EmberClientBuilder +import org.http4s.headers.Authorization import org.typelevel.log4cats.Logger import org.typelevel.log4cats.slf4j.Slf4jLogger diff --git a/src/main/scala/fetch/FileNameService.scala b/src/main/scala/fetch/csv/FileNameService.scala similarity index 99% rename from src/main/scala/fetch/FileNameService.scala rename to src/main/scala/fetch/csv/FileNameService.scala index af34403..5dbd192 100644 --- a/src/main/scala/fetch/FileNameService.scala +++ b/src/main/scala/fetch/csv/FileNameService.scala @@ -1,9 +1,9 @@ -package fetch +package fetch.csv import cats.effect._ -import java.time.format.DateTimeFormatter import java.time.{Duration, LocalDate, LocalDateTime, ZoneId, ZonedDateTime} +import java.time.format.DateTimeFormatter class FileNameService { diff --git a/src/main/scala/fetchDMI/FetchService.scala b/src/main/scala/fetch/dmi/FetchService.scala similarity index 99% rename from src/main/scala/fetchDMI/FetchService.scala rename to src/main/scala/fetch/dmi/FetchService.scala index 766317c..e3e2c7f 100644 --- a/src/main/scala/fetchDMI/FetchService.scala +++ b/src/main/scala/fetch/dmi/FetchService.scala @@ -1,16 +1,16 @@ -package fetchDMI +package fetch.dmi import cats.effect._ import cats.implicits.toTraverseOps import data.DataService import fs2.io.file.{CopyFlag, CopyFlags, Files, Path} -import grib.GribParser +import io.circe.Json +import io.circe.parser.decode import org.http4s._ import org.http4s.ember.client.EmberClientBuilder import org.typelevel.log4cats.Logger import org.typelevel.log4cats.slf4j.Slf4jLogger -import io.circe.parser.decode -import io.circe.Json +import parse.grib.GribParser import java.time.ZonedDateTime import scala.util.Try diff --git a/src/main/scala/fetchDMI/FetchServiceTest.scala b/src/main/scala/fetch/dmi/FetchServiceTest.scala similarity index 98% rename from src/main/scala/fetchDMI/FetchServiceTest.scala rename to src/main/scala/fetch/dmi/FetchServiceTest.scala index 9182d79..22ab5c4 100644 --- a/src/main/scala/fetchDMI/FetchServiceTest.scala +++ b/src/main/scala/fetch/dmi/FetchServiceTest.scala @@ -1,4 +1,4 @@ -package fetchDMI; +package fetch.dmi import cats.effect.IO import cats.effect.unsafe.implicits.global diff --git a/src/main/scala/fetchDMI/FileName.scala b/src/main/scala/fetch/dmi/FileName.scala similarity index 98% rename from src/main/scala/fetchDMI/FileName.scala rename to src/main/scala/fetch/dmi/FileName.scala index 9629f8b..25ce8b5 100644 --- a/src/main/scala/fetchDMI/FileName.scala +++ b/src/main/scala/fetch/dmi/FileName.scala @@ -1,4 +1,4 @@ -package fetchDMI +package fetch.dmi import java.time.{ZoneOffset, ZonedDateTime} diff --git a/src/main/scala/parse/Aggregate.scala b/src/main/scala/parse/csv/Aggregate.scala similarity index 99% rename from src/main/scala/parse/Aggregate.scala rename to src/main/scala/parse/csv/Aggregate.scala index 30c78d9..d15cfcd 100644 --- a/src/main/scala/parse/Aggregate.scala +++ b/src/main/scala/parse/csv/Aggregate.scala @@ -1,8 +1,7 @@ -package parse +package parse.csv import cats.data.NonEmptyList import cats.implicits.{catsSyntaxOptionId, toFoldableOps} -import io.circe.generic.semiauto.deriveEncoder import io.circe.syntax.EncoderOps import io.circe.{Encoder, Json} diff --git a/src/main/scala/parse/Main.scala b/src/main/scala/parse/csv/Main.scala similarity index 90% rename from src/main/scala/parse/Main.scala rename to src/main/scala/parse/csv/Main.scala index c249035..36ef983 100644 --- a/src/main/scala/parse/Main.scala +++ b/src/main/scala/parse/csv/Main.scala @@ -1,15 +1,8 @@ -package parse - -import cats.data.NonEmptyList -import cats.effect.unsafe.implicits.global +package parse.csv //import db.FileService -import io.circe.syntax.EncoderOps -import parse.Aggregate.{AggregateKey, UserQuery} - import java.time.LocalDateTime import java.time.format.DateTimeFormatter import java.time.temporal.ChronoUnit -import parse.Aggregate.AggregateValueImplicits.aggregateValueEncoder // TODO remake this as a test with different granularities and especially check avg value calculations object Main { diff --git a/src/main/scala/parse/Parser.scala b/src/main/scala/parse/csv/Parser.scala similarity index 98% rename from src/main/scala/parse/Parser.scala rename to src/main/scala/parse/csv/Parser.scala index d0e259f..1c9355e 100644 --- a/src/main/scala/parse/Parser.scala +++ b/src/main/scala/parse/csv/Parser.scala @@ -1,4 +1,4 @@ -package parse +package parse.csv import java.time.LocalDateTime import java.time.format.DateTimeFormatter diff --git a/src/main/scala/parse/WeatherData.scala b/src/main/scala/parse/csv/WeatherData.scala similarity index 99% rename from src/main/scala/parse/WeatherData.scala rename to src/main/scala/parse/csv/WeatherData.scala index 2ae78d6..6564b86 100644 --- a/src/main/scala/parse/WeatherData.scala +++ b/src/main/scala/parse/csv/WeatherData.scala @@ -1,4 +1,4 @@ -package parse +package parse.csv import java.time.LocalDateTime import scala.reflect.runtime.universe._ diff --git a/src/main/scala/grib/Codes.scala b/src/main/scala/parse/grib/Codes.scala similarity index 99% rename from src/main/scala/grib/Codes.scala rename to src/main/scala/parse/grib/Codes.scala index a39d958..0ae4c4d 100644 --- a/src/main/scala/grib/Codes.scala +++ b/src/main/scala/parse/grib/Codes.scala @@ -1,5 +1,4 @@ -package grib - +package parse.grib object Codes { def codesToString(discipline: Int, category: Int, product: Int): String = { diff --git a/src/main/scala/grib/Grib.scala b/src/main/scala/parse/grib/Grib.scala similarity index 97% rename from src/main/scala/grib/Grib.scala rename to src/main/scala/parse/grib/Grib.scala index 2483b1e..c091441 100644 --- a/src/main/scala/grib/Grib.scala +++ b/src/main/scala/parse/grib/Grib.scala @@ -1,4 +1,5 @@ -package grib +package parse.grib + case class Grib( version: Int, length: Long, diff --git a/src/main/scala/grib/GribParser.scala b/src/main/scala/parse/grib/GribParser.scala similarity index 99% rename from src/main/scala/grib/GribParser.scala rename to src/main/scala/parse/grib/GribParser.scala index cbbd945..cf4bc3a 100644 --- a/src/main/scala/grib/GribParser.scala +++ b/src/main/scala/parse/grib/GribParser.scala @@ -1,4 +1,4 @@ -package grib +package parse.grib import cats.effect._ import fs2.io.file.{Files, Path} diff --git a/src/main/scala/grib/GribParserTest.scala b/src/main/scala/parse/grib/GribParserTest.scala similarity index 97% rename from src/main/scala/grib/GribParserTest.scala rename to src/main/scala/parse/grib/GribParserTest.scala index f0f9e93..fd55ce6 100644 --- a/src/main/scala/grib/GribParserTest.scala +++ b/src/main/scala/parse/grib/GribParserTest.scala @@ -1,11 +1,11 @@ -package grib +package parse.grib import cats.effect.IO +import cats.effect.unsafe.implicits.global +import data.DataService import fs2.io.file.Path import io.circe.generic.auto._ import io.circe.syntax._ -import cats.effect.unsafe.implicits.global -import data.DataService object GribParserTest { def main(args: Array[String]): Unit = { diff --git a/src/main/scala/fetchDMI/Scheduler.scala b/src/main/scala/scheduler/Scheduler.scala similarity index 86% rename from src/main/scala/fetchDMI/Scheduler.scala rename to src/main/scala/scheduler/Scheduler.scala index 777dda8..9a9047e 100644 --- a/src/main/scala/fetchDMI/Scheduler.scala +++ b/src/main/scala/scheduler/Scheduler.scala @@ -1,4 +1,4 @@ -package fetchDMI +package scheduler import cats.effect._ import cats.implicits.catsSyntaxApply @@ -28,8 +28,8 @@ class Scheduler(log: Logger[IO]) { } } - def scheduleTask(name: String, minutes: List[Int], task: IO[List[String]]): Stream[IO, List[String]] = { - def streamForMinute(minute: Int): Stream[IO, List[String]] = { + def scheduleTask[A](name: String, minutes: List[Int], task: IO[A]): Stream[IO, A] = { + def streamForMinute(minute: Int): Stream[IO, A] = { Stream.eval(durationToNext(minute)).flatMap { delay => Stream.eval(log.info(s"$name scheduler in: ${delay.toMinutes} min")) *> (Stream.sleep[IO](delay) ++ Stream.awakeEvery[IO](1.hour)) diff --git a/src/main/scala/server/Server.scala b/src/main/scala/server/Server.scala index cd1b0b3..2aacb33 100644 --- a/src/main/scala/server/Server.scala +++ b/src/main/scala/server/Server.scala @@ -1,12 +1,11 @@ package server import cats.effect._ -import cats.implicits.{catsSyntaxApplicativeError, toTraverseOps} +import cats.implicits.toTraverseOps import com.comcast.ip4s.IpLiteralSyntax import data.DataService import db.PostgresService -import fetch.FetchService -import parse.Aggregate +import fetch.csv.FetchService import server.ValidateRoutes.{AggFieldList, AggKey, CityList, DateTimeRange, Granularity, ValidateDate, ValidateDateTime, ValidateInt, ValidateMonths, ValidateZonedDateTime} import io.circe.{Json, Printer} import org.http4s._ @@ -19,16 +18,16 @@ import org.http4s.server.staticcontent.FileService import org.http4s.ember.server.EmberServerBuilder import io.circe.generic.auto._ import io.circe.syntax._ -import org.http4s.circe.CirceEntityCodec.circeEntityEncoder -import parse.Aggregate.AggregateValueImplicits.aggregateValueEncoder -import parse.Aggregate.userQueryEncoder -import parse.Aggregate.{AggregateKey, UserQuery} +import parse.csv.Aggregate.AggregateValueImplicits.aggregateValueEncoder +import parse.csv.Aggregate.userQueryEncoder +import parse.csv.Aggregate.{AggregateKey, UserQuery} import org.http4s.circe.jsonEncoder import org.typelevel.log4cats.Logger import org.typelevel.log4cats.slf4j.Slf4jLogger +import fs2.io.file.Path +import parse.csv.Aggregate import scala.concurrent.duration.DurationInt -import scala.reflect.io.File object Server { @@ -137,8 +136,9 @@ class Server(postgresService: PostgresService, dataService: DataService, fetch: private val apiRoutesCors = CORS(apiRoutes, corsConfig) private val httpApp = Router( - "/api" -> apiRoutesCors, "/" -> staticcontent.fileService[IO](FileService.Config("./web/dist")), + "/api" -> apiRoutesCors, + // TODO rewrite in more generic way "/station" -> staticcontent.fileService[IO](FileService.Config("./web/dist")), "/cities" -> staticcontent.fileService[IO](FileService.Config("./web/dist")), diff --git a/src/main/scala/server/ValidateRoutes.scala b/src/main/scala/server/ValidateRoutes.scala index 98c128e..7c86650 100644 --- a/src/main/scala/server/ValidateRoutes.scala +++ b/src/main/scala/server/ValidateRoutes.scala @@ -1,8 +1,8 @@ package server import cats.data.NonEmptyList -import parse.Aggregate.AggregateKey -import parse.WeatherData +import parse.csv.Aggregate.AggregateKey +import parse.csv.WeatherData import java.time.{LocalDate, LocalDateTime, ZonedDateTime} import java.time.format.DateTimeFormatter diff --git a/src/test/scala/fetch/FileFetchSchedulerSpec.scala b/src/test/scala/fetch/FileFetchSchedulerSpec.scala deleted file mode 100644 index 1509adb..0000000 --- a/src/test/scala/fetch/FileFetchSchedulerSpec.scala +++ /dev/null @@ -1,51 +0,0 @@ -package fetch - -import base.IOSuite -import cats.effect.{Clock, IO} -import cats.effect.kernel.Ref -import cats.implicits.catsSyntaxTuple3Semigroupal -import db.PostgresService -import org.scalatest.matchers.should.Matchers -import org.scalatest.wordspec.AsyncWordSpec -import org.typelevel.log4cats.slf4j.Slf4jLogger -import fs2.Stream -import org.scalamock.scalatest.MockFactory - - -class FileFetchSchedulerSpec extends AsyncWordSpec with Matchers with MockFactory with IOSuite { - "FileFetchScheduler" should { - "save into db" in runIO { - for { - refDb <- Ref.of[IO, Option[String]](None) - refFetch <- Ref.of[IO, Option[(String, String)]](None) - refScheduler <- Ref.of[IO, Option[Either[Throwable, (String, String)]]](None) - log <- Slf4jLogger.create[IO] - mockDatabaseOps = mock[PostgresService] - fileNameService = new FileNameService { - override def generateCurrentHour(implicit clock: Clock[IO]): IO[String] = IO.pure("file_test") - } - fetchService = new FetchService(fileNameService, log) { - override def fetchSingleFile(fileName: String): IO[Either[Throwable, (String, String)]] = { - refFetch.set(Some((fileName, "content"))).as(Right((fileName, "content"))) - } - } - scheduler = new Scheduler(log) { - override def scheduleTask(task: IO[Either[Throwable, (String, String)]]): Stream[IO, Either[Throwable, (String, String)]] = { - Stream.eval(task).flatMap { result => - Stream.eval(refScheduler.set(Some(result))).as(result) - } - } - } - - res = new FileFetchScheduler(mockDatabaseOps, fetchService, fileNameService, scheduler, log) - .run.compile.drain *> - (refDb.get, refFetch.get, refScheduler.get).tupled.map { case (db, fetch, scheduler) => - db shouldBe Some("file_test") - fetch shouldBe Some(("file_test", "content")) - scheduler shouldBe Some(Right(("file_test", "content"))) - } - } yield res - } - } -} - diff --git a/src/test/scala/fetch/FileNameServiceSpec.scala b/src/test/scala/fetch/FileNameServiceSpec.scala index 96417a0..4a81768 100644 --- a/src/test/scala/fetch/FileNameServiceSpec.scala +++ b/src/test/scala/fetch/FileNameServiceSpec.scala @@ -1,8 +1,8 @@ import cats.Applicative import cats.effect.unsafe.implicits.global -import fetch.FileNameService import org.scalatest.funsuite.AnyFunSuite import cats.effect.{Clock, IO} +import fetch.csv.FileNameService import java.util.concurrent.TimeUnit import java.time.LocalDateTime diff --git a/src/test/scala/fetch/SchedulerSpec.scala b/src/test/scala/fetch/SchedulerSpec.scala deleted file mode 100644 index c21e1f6..0000000 --- a/src/test/scala/fetch/SchedulerSpec.scala +++ /dev/null @@ -1,25 +0,0 @@ -package fetch - -import cats.effect.{Clock, IO} -import fs2.Stream - -class SchedulerSpec { -// private def testTask: IO[Unit] = { -// log.info("Running task") -// } -// -// def run(implicit clock: Clock[IO]): Stream[IO, Nothing] = { -// Stream.eval(durationToNextHalfHour).flatMap { delay => -// (Stream.sleep[IO](delay) ++ Stream.awakeEvery[IO](1.hour)).evalMap(_ => testTask).drain -// } -// } -// -// def main(args: Array[String]): Unit = { -// // run.compile.drain.unsafeRunSync() -// for { -// scheduler <- Scheduler.of -// fetch <- FetchService.of -// fetchTask = new FileNameService().generateCurrentHour.flatMap(fetch.fetchSingleFile) -// } yield scheduler.scheduleTask(fetchTask).compile.drain.unsafeRunSync() -// } -} diff --git a/src/test/scala/parse/ParserSpec.scala b/src/test/scala/parse/ParserSpec.scala index 0e97786..7b83045 100644 --- a/src/test/scala/parse/ParserSpec.scala +++ b/src/test/scala/parse/ParserSpec.scala @@ -4,7 +4,7 @@ import cats.effect.unsafe.implicits.global //import db.FileService import org.scalatest.funsuite.AnyFunSuite import org.scalatest.matchers.should.Matchers -import parse.Aggregate.{AggregateKey, DoubleValue, TimeDoubleList, UserQuery} +import parse.csv.Aggregate.{AggregateKey, DoubleValue, TimeDoubleList, UserQuery} import java.time.LocalDateTime import java.time.format.DateTimeFormatter