diff --git a/build.sbt b/build.sbt index 961272f..f597f9d 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("server.Server") + Compile / mainClass := Some("server.Main") ) val http4sVersion = "0.23.16" diff --git a/src/main/scala/fetch/FetchService.scala b/src/main/scala/fetch/FetchService.scala index 1876d86..e39ac2a 100644 --- a/src/main/scala/fetch/FetchService.scala +++ b/src/main/scala/fetch/FetchService.scala @@ -48,6 +48,17 @@ object FetchService { } } + def fetchSingleFile(fileName: String): IO[(String, String)] = { + println("start fetching") + fetchFiles(List(fileName)).flatMap { results => + results.headOption match { + case Some(Right(result)) => IO.pure(result) + case Some(Left(err)) => IO.raiseError(err) + case None => IO.raiseError(new Exception("No file fetched")) + } + } + } + def fetchInRange(from: LocalDateTime, to: LocalDateTime): IO[List[Either[Throwable, (String, String)]]] = { val fileNames = FileNameService.generate(from, to) fetchFiles(fileNames) diff --git a/src/main/scala/fetch/FileNameService.scala b/src/main/scala/fetch/FileNameService.scala index b24de6d..cacee0b 100644 --- a/src/main/scala/fetch/FileNameService.scala +++ b/src/main/scala/fetch/FileNameService.scala @@ -1,7 +1,9 @@ package fetch +import cats.effect._ + import java.time.format.DateTimeFormatter -import java.time.{Duration, LocalDate, LocalDateTime} +import java.time.{Duration, LocalDate, LocalDateTime, ZoneId, ZonedDateTime} object FileNameService { private val interval = Duration.ofMinutes(30) @@ -21,7 +23,18 @@ object FileNameService { time.plusMinutes(adjustment) } - // not sure either data are each 10 minutes or each 30 mins of hour + 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 + } + } + + // currently csv files are generated each 30 mins of hour def generate(startTime: LocalDateTime, endTime: LocalDateTime): List[String] = { val roundStartTime = roundToInterval(startTime, true) val roundEndTime = roundToInterval(endTime, false) diff --git a/src/main/scala/server/Main.scala b/src/main/scala/server/Main.scala new file mode 100644 index 0000000..826d7c5 --- /dev/null +++ b/src/main/scala/server/Main.scala @@ -0,0 +1,22 @@ +package server +import cats.effect._ +import db.DBService +import fetch.{FetchService, FileNameService} +import fs2.Stream + +object Main extends IOApp { + def run(args: List[String]): IO[ExitCode] = { + val fetchTask = FileNameService.generateCurrentHour.flatMap(FetchService.fetchSingleFile) + Stream( + Server.run, + Scheduler.scheduleTask(fetchTask) + .evalMap { case (name, content) => + IO(println(s"fetched: $name")) *> + DBService.save(name, content).attempt.flatMap { + case Right(savedName) => IO(println(s"File saved: $savedName")) + case Left(err) => IO(println(s"Error: $err")) + } + } + ).parJoinUnbounded.compile.drain.as(ExitCode.Success) + } +} \ No newline at end of file diff --git a/src/main/scala/server/Scheduler.scala b/src/main/scala/server/Scheduler.scala new file mode 100644 index 0000000..ba47354 --- /dev/null +++ b/src/main/scala/server/Scheduler.scala @@ -0,0 +1,47 @@ +package server + +import cats.effect._ +import cats.effect.unsafe.implicits.global +import fetch.{FetchService, FileNameService} +import fs2.Stream + +import java.time.{Duration, LocalTime} +import scala.concurrent.duration._ + +object Scheduler { + private val downloadMinute = 31 + +// private def testTask: IO[Unit] = { +// IO(println("Running task")) +// } + + 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 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 scheduleTask(task: IO[(String, String)]): Stream[IO, (String, String)] = { + Stream.eval(durationToNextHalfHour).flatMap { delay => + (Stream.sleep[IO](delay) ++ Stream.awakeEvery[IO](1.hour)).evalMap(_ => task) + } + } + + def main(args: Array[String]): Unit = { +// run.compile.drain.unsafeRunSync() + + val fetchTask = FileNameService.generateCurrentHour.flatMap(FetchService.fetchSingleFile) + scheduleTask(fetchTask).compile.drain.unsafeRunSync() + } +} diff --git a/src/main/scala/server/Server.scala b/src/main/scala/server/Server.scala index 1256c32..c29f6c6 100644 --- a/src/main/scala/server/Server.scala +++ b/src/main/scala/server/Server.scala @@ -6,7 +6,7 @@ import db.DBService import fetch.FetchService import parse.{Parser, WeatherData} import server.ValidateRoutes.{AggKey, CityList, DateTimeRange, ValidDate} -import io.circe.{Encoder, Json, Printer} +import io.circe.{Json, Printer} import org.http4s._ import org.http4s.dsl.io._ import org.http4s.server.Router @@ -14,9 +14,10 @@ import org.http4s.server.blaze.BlazeServerBuilder import io.circe.syntax._ import parse.Aggregate.AggregateValueImplicits.aggregateValueEncoder import parse.Aggregate.{AggregateKey, UserQuery} +import fs2.Stream -object Server extends IOApp { +object Server { // Define the extension method `pretty` for Json implicit class JsonPrettyPrinter(json: Json) { @@ -84,12 +85,9 @@ object Server extends IOApp { private val httpApp = Router("/" -> appRoutes).orNotFound - override def run(args: List[String]): IO[ExitCode] = + def run: Stream[IO, ExitCode] = BlazeServerBuilder[IO] .bindHttp(8080, "0.0.0.0") .withHttpApp(httpApp) .serve - .compile - .drain - .as(ExitCode.Success) }