diff --git a/src/main/scala/fetch/FetchService.scala b/src/main/scala/fetch/FetchService.scala index 1dddd34..5615250 100644 --- a/src/main/scala/fetch/FetchService.scala +++ b/src/main/scala/fetch/FetchService.scala @@ -20,13 +20,19 @@ final case class WeatherServerConfig( url: String, ) +trait FetchServiceTrait { + def fetchSingleFile(fileName: String): IO[Either[Throwable, (String, String)]] + def fetchInRange(from: LocalDateTime, to: LocalDateTime): IO[List[Either[Throwable, (String, String)]]] + def fetchFromDate(date: LocalDate): IO[List[Either[Throwable, (String, String)]]] +} + object FetchService { def of: IO[FetchService] = { Slf4jLogger.create[IO].map(logger => new FetchService(new FileNameService, logger)) } } -class FetchService(fileNameService: FileNameService, log: Logger[IO]) { +class FetchService(fileNameService: FileNameService, log: Logger[IO]) extends FetchServiceTrait { 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") @@ -38,7 +44,7 @@ class FetchService(fileNameService: FileNameService, log: Logger[IO]) { private val baseUrl: IO[Uri] = IO(Uri.unsafeFromString(weatherServerConfig.url)) private def makeRequest(client: Client[IO], url: Uri): IO[Either[Throwable, (String, String)]] = { - val fileName = url.path.toString() + val fileName = url.path.toString().tail val request = Request[IO](Method.GET, url).withHeaders(Authorization(basicCredentials)) client.expect[String](request).redeemWith( diff --git a/src/main/scala/fetch/FileFetchScheduler.scala b/src/main/scala/fetch/FileFetchScheduler.scala index ed3d4e5..1e338fa 100644 --- a/src/main/scala/fetch/FileFetchScheduler.scala +++ b/src/main/scala/fetch/FileFetchScheduler.scala @@ -8,7 +8,7 @@ import org.typelevel.log4cats.slf4j.Slf4jLogger object FileFetchScheduler { - def of(dbService: DBService, fetch: FetchService): IO[FileFetchScheduler] = { + def of(dbService: DBService, fetch: FetchServiceTrait): IO[FileFetchScheduler] = { Scheduler.of.flatMap { scheduler => Slf4jLogger.create[IO].map { new FileFetchScheduler(dbService, fetch, new FileNameService(), scheduler, _) @@ -17,7 +17,7 @@ object FileFetchScheduler { } } -class FileFetchScheduler(dbService: DBService, fetch: FetchService, fileNameService: FileNameService, scheduler: Scheduler, log: Logger[IO]) { +class FileFetchScheduler(dbService: DBService, fetch: FetchServiceTrait, fileNameService: FileNameService, scheduler: Scheduler, log: Logger[IO]) { def run: Stream[IO, Unit] = { val fetchTask = fileNameService.generateCurrentHour.flatMap(fetch.fetchSingleFile) scheduler.scheduleTask(fetchTask) diff --git a/src/main/scala/fetch/FileNameService.scala b/src/main/scala/fetch/FileNameService.scala index 7c7e71a..af34403 100644 --- a/src/main/scala/fetch/FileNameService.scala +++ b/src/main/scala/fetch/FileNameService.scala @@ -5,11 +5,8 @@ import cats.effect._ import java.time.format.DateTimeFormatter import java.time.{Duration, LocalDate, LocalDateTime, ZoneId, ZonedDateTime} -trait FileNameServiceTrait { - def generateCurrentHour(implicit clock: Clock[IO]): IO[String] -} -class FileNameService extends FileNameServiceTrait { +class FileNameService { private val interval = Duration.ofMinutes(30) private val formatter = DateTimeFormatter.ofPattern("yyyyMMdd_HHmm") @@ -27,17 +24,30 @@ class FileNameService extends FileNameServiceTrait { time.plusMinutes(adjustment) } - override def generateCurrentHour(implicit clock: Clock[IO]): IO[String] = { + def generateCurrentHour(implicit clock: Clock[IO]): IO[String] = { clock.realTime.map { duration => val instant = java.time.Instant.ofEpochMilli(duration.toMillis) val zonedDateTime = ZonedDateTime.ofInstant(instant, ZoneId.of("Europe/Riga")) val now = zonedDateTime.toLocalDateTime.withMinute(30) val csvName = s"${formatter.format(now)}.csv" -// println("println", csvName) csvName } } + def generateLast24Hours(implicit clock: Clock[IO]): IO[List[String]] = { + clock.realTime.map { duration => + val instant = java.time.Instant.ofEpochMilli(duration.toMillis) + val zonedDateTime = ZonedDateTime.ofInstant(instant, ZoneId.of("Europe/Riga")) + val now = zonedDateTime.toLocalDateTime.withMinute(30) + val formatter = DateTimeFormatter.ofPattern("yyyyMMdd_HHmm") + + (0 until 24).toList.map { hour => + val previousHour = now.minusHours(hour) + s"${formatter.format(previousHour)}.csv" + } + } + } + // currently csv files are generated each 30 mins of hour def generate(startTime: LocalDateTime, endTime: LocalDateTime): List[String] = { val roundStartTime = roundToInterval(startTime, true) diff --git a/src/main/scala/fetch/Main.scala b/src/main/scala/fetch/Main.scala index 4eba071..d48c7ce 100644 --- a/src/main/scala/fetch/Main.scala +++ b/src/main/scala/fetch/Main.scala @@ -9,25 +9,40 @@ import org.typelevel.log4cats.slf4j.Slf4jLogger import java.time.LocalDateTime object Main { +// def run: IO[Unit] = { +// val from = LocalDateTime.of(2023, 4, 28, 10, 0) +// val to = LocalDateTime.of(2023, 4, 28, 13, 30) +// for { +// 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) +// 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") +// errors = fetchServiceError ++ fetchErrors.map(e => s"FetchError: ${e.getMessage}") ++ saveErrors.map(e => s"SaveError: ${e.getMessage}") +// _ <- log.info(s"errors: $errors") +// _ <- log.info(s"successes: $successes") +// } yield (successes, errors) +// } + def run: IO[Unit] = { - val from = LocalDateTime.of(2023, 4, 28, 10, 0) - val to = LocalDateTime.of(2023, 4, 28, 13, 30) for { - log <- Slf4jLogger.create[IO] fetch <- FetchService.of -// fetchResultEither <- fetch.fetchFromDate(LocalDate.of(2023, 4, 28)).attempt - fetchResultEither <- fetch.fetchInRange(from, to).attempt + statefulFetch <- StatefulFetchService.of(fetch) + fetchResultEither <- statefulFetch.fetchSingleFile("20230524_0030.csv").attempt + fetchResultEither <- statefulFetch.fetchSingleFile("20230522_0130.csv").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") - errors = fetchServiceError ++ fetchErrors.map(e => s"FetchError: ${e.getMessage}") ++ saveErrors.map(e => s"SaveError: ${e.getMessage}") - _ <- log.info(s"errors: $errors") - _ <- log.info(s"successes: $successes") - } yield (successes, errors) + fetchResult = fetchResultEither.flatMap(res => res.flatMap(aaa => { + println(s"fffffff: ${aaa._1}") + Right(aaa._1) + })) +// _ = println(s"${fetchResult.map()}") + } yield () } def main(args: Array[String]): Unit = { diff --git a/src/main/scala/fetch/StatefulFetchService.scala b/src/main/scala/fetch/StatefulFetchService.scala new file mode 100644 index 0000000..6217b68 --- /dev/null +++ b/src/main/scala/fetch/StatefulFetchService.scala @@ -0,0 +1,59 @@ +package fetch + +import cats.effect.{IO, Ref} +import cats.implicits.toTraverseOps +import org.typelevel.log4cats.Logger +import org.typelevel.log4cats.slf4j.Slf4jLogger + +import java.time.{LocalDate, LocalDateTime} + +object StatefulFetchService { + def of(fetchService: FetchService): IO[StatefulFetchService] = { + Slf4jLogger.create[IO].map(logger => new StatefulFetchService(fetchService, new FileNameService(), logger)) + } +} + +class StatefulFetchService(fetchService: FetchService, fileNameService: FileNameService, log: Logger[IO]) extends FetchServiceTrait { + private val state: Ref[IO, Map[String, String]] = Ref.unsafe(Map.empty) + + private def logState: IO[Unit] = { + state.get.flatMap(currentState => log.info(s"State keys: ${currentState.keys}")) + } + + private def updateState(fileName: String, content: String): IO[Unit] = { + for { + last24Hours <- fileNameService.generateLast24Hours + _ <- state.update(st => (st + (fileName -> content)).filterKeys(last24Hours.contains).toMap) + _ <- logState + } yield () + } + + def fetchSingleFile(fileName: String): IO[Either[Throwable, (String, String)]] = { + fetchService.fetchSingleFile(fileName).flatMap { + case Right((fileName, content)) => + updateState(fileName, content).as(Right((fileName, content))) + case e@Left(_) => IO(e) + } + } + + def fetchInRange(from: LocalDateTime, to: LocalDateTime): IO[List[Either[Throwable, (String, String)]]] = { + fetchService.fetchInRange(from, to).flatMap { results => + val successfulResults = results.collect { case Right(data) => data } + successfulResults.traverse { case (name, content) => updateState(name, content) }.as(results) + } + } + + def fetchFromDate(date: LocalDate): IO[List[Either[Throwable, (String, String)]]] = { + fetchService.fetchFromDate(date).flatMap { results => + val successfulResults = results.collect { case Right(data) => data } + successfulResults.traverse { case (name, content) => updateState(name, content) }.as(results) + } + } +} + + + + + + + diff --git a/src/main/scala/server/Main.scala b/src/main/scala/server/Main.scala index c348385..3e2e145 100644 --- a/src/main/scala/server/Main.scala +++ b/src/main/scala/server/Main.scala @@ -2,16 +2,17 @@ package server import cats.effect._ import cats.implicits.catsSyntaxTuple2Parallel import db.DBService -import fetch.{FetchService, FileFetchScheduler} +import fetch.{FetchService, FileFetchScheduler, StatefulFetchService} object Main extends IOApp { def run(args: List[String]): IO[ExitCode] = { for { dbService <- DBService.of fetch <- FetchService.of - fileFetchScheduler <- FileFetchScheduler.of(dbService, fetch) + statefulFetch <- StatefulFetchService.of(fetch) + fileFetchScheduler <- FileFetchScheduler.of(dbService, statefulFetch) schedulerTask = fileFetchScheduler.run.compile.drain - server <- Server.of(dbService, fetch) + server <- Server.of(dbService, statefulFetch) 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 33caf64..b57298f 100644 --- a/src/main/scala/server/Server.scala +++ b/src/main/scala/server/Server.scala @@ -4,7 +4,7 @@ import cats.effect._ import cats.implicits.toTraverseOps import com.comcast.ip4s.IpLiteralSyntax import db.DBService -import fetch.FetchService +import fetch.FetchServiceTrait import parse.{Parser, WeatherData} import server.ValidateRoutes.{AggKey, CityList, DateTimeRange, ValidDate} import io.circe.{Json, Printer} @@ -26,14 +26,14 @@ import scala.concurrent.duration.DurationInt object Server { - def of(dbService: DBService, fetch: FetchService): IO[Server] = { + def of(dbService: DBService, fetch: FetchServiceTrait): IO[Server] = { Slf4jLogger.create[IO].map { new Server(dbService, fetch, _) } } } -class Server(dbService: DBService, fetch: FetchService, log: Logger[IO]) { +class Server(dbService: DBService, fetch: FetchServiceTrait, log: Logger[IO]) { // Define the extension method `pretty` for Json implicit class JsonPrettyPrinter(json: Json) { diff --git a/src/test/scala/fetch/FileNameServiceSpec.scala b/src/test/scala/fetch/FileNameServiceSpec.scala index ea2730b..96417a0 100644 --- a/src/test/scala/fetch/FileNameServiceSpec.scala +++ b/src/test/scala/fetch/FileNameServiceSpec.scala @@ -1,10 +1,16 @@ +import cats.Applicative +import cats.effect.unsafe.implicits.global import fetch.FileNameService import org.scalatest.funsuite.AnyFunSuite +import cats.effect.{Clock, IO} +import java.util.concurrent.TimeUnit import java.time.LocalDateTime +import scala.concurrent.duration.FiniteDuration +import scala.concurrent.duration._ + class FileNameServiceSpec extends AnyFunSuite { - test("FileNameService.generate returns correct file names") { val startTime = LocalDateTime.of(2023, 5, 17, 0, 0) val endTime = LocalDateTime.of(2023, 5, 17, 4, 0) @@ -25,11 +31,40 @@ class FileNameServiceSpec extends AnyFunSuite { val date = LocalDateTime.of(2023, 5, 17, 0, 0).toLocalDate val expectedFileNames = (0 to 23).toList - .map(hour => if(hour < 10) "0"+hour else ""+hour) + .map(hour => if (hour < 10) "0" + hour else "" + hour) .map(str => s"20230517_${str}30.csv") val actualFileNames = new FileNameService().generateFromDate(date) assert(actualFileNames == expectedFileNames) } + + val fixedClock: Clock[IO] = new Clock[IO] { + val timeInMillis = 1618862447000L + + override def realTime: IO[FiniteDuration] = + IO.pure(Duration.fromNanos(TimeUnit.MILLISECONDS.toNanos(timeInMillis))) + + override def applicative: Applicative[IO] = + Applicative.apply + + override def monotonic: IO[FiniteDuration] = + IO.pure(Duration.fromNanos(TimeUnit.MILLISECONDS.toNanos(timeInMillis))) + + override def timed[A](fa: IO[A]): IO[(FiniteDuration, A)] = + fa.map(a => (Duration.Zero, a)) + } + + test("FileNameService.generateLast24Hours returns correct file names for a fixed time") { + val fileNameService = new FileNameService() + val filenames = fileNameService.generateLast24Hours(fixedClock).unsafeRunSync() + + assert(filenames.size == 24) + + val expectedFileNames = (0 to 23).toList + .map(hour => if (hour < 10) "0" + hour else "" + hour) + .map(str => s"20210419_${str}30.csv") + + assert(filenames.sorted == expectedFileNames.sorted) + } } \ No newline at end of file