Scheduler and also Main file to call server and scheduler

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"

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)

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)

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)
+  }
+}

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()
+  }
+}

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)
 }