-
Notifications
You must be signed in to change notification settings - Fork 98
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
add endpoints. add retry. stop generating quarter after 8 days
- Loading branch information
1 parent
a6d25b6
commit 631a136
Showing
19 changed files
with
421 additions
and
90 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
60 changes: 60 additions & 0 deletions
60
hmda-data-publisher/src/main/scala/hmda/publisher/api/DataPublisherHttpApi.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,60 @@ | ||
| package hmda.publisher.api | ||
|
|
||
| import akka.actor.ActorRef | ||
| import akka.http.scaladsl.server.Directives._ | ||
| import akka.http.scaladsl.server.Route | ||
| import ch.megard.akka.http.cors.scaladsl.CorsDirectives._ | ||
| import de.heikoseeberger.akkahttpcirce.FailFastCirceSupport._ | ||
| import hmda.publisher.scheduler.AllSchedulers | ||
| import hmda.publisher.scheduler.schedules.{ Schedule, Schedules } | ||
|
|
||
| import scala.concurrent.ExecutionContext | ||
|
|
||
| private class DataPublisherHttpApi( | ||
| schedulers: AllSchedulers | ||
| )(implicit ec: ExecutionContext) { | ||
|
|
||
| //trigger/<schedulername> | ||
| private val triggerScheduler = | ||
| path("trigger" / Segment) { schedulerName => | ||
| post { | ||
| Schedules.withNameOption(schedulerName) match { | ||
| case Some(schedule) => | ||
| triggerSchedule(schedule) | ||
| complete(202 -> s"Schedule ${schedulerName} has been triggered") | ||
| case None => | ||
| complete(404 -> s"Scheduler ${schedulerName} not found. Available: ${Schedules.values.map(_.entryName).mkString(", ")}") | ||
| } | ||
| } | ||
| } | ||
|
|
||
| private def triggerSchedule(msg: Schedule): Unit = { | ||
| import schedulers._ | ||
| val receiver = msg match { | ||
| case Schedules.PanelScheduler2018 => panelScheduler | ||
| case Schedules.PanelScheduler2019 => panelScheduler | ||
| case Schedules.LarPublicScheduler2018 => larPublicScheduler | ||
| case Schedules.LarPublicScheduler2019 => larPublicScheduler | ||
| case Schedules.LarScheduler2018 => larScheduler | ||
| case Schedules.LarScheduler2019 => larScheduler | ||
| case Schedules.LarSchedulerLoanLimit2019 => larScheduler | ||
| case Schedules.LarSchedulerQuarterly2020 => larScheduler | ||
| case Schedules.TsPublicScheduler2018 => tsPublicScheduler | ||
| case Schedules.TsPublicScheduler2019 => tsPublicScheduler | ||
| case Schedules.TsScheduler2018 => tsScheduler | ||
| case Schedules.TsScheduler2019 => tsScheduler | ||
| case Schedules.TsSchedulerQuarterly2020 => tsScheduler | ||
| } | ||
| receiver ! msg | ||
| } | ||
|
|
||
| def routes: Route = | ||
| handleRejections(corsRejectionHandler) { | ||
| cors() { | ||
| encodeResponse { | ||
| triggerScheduler | ||
| } | ||
| } | ||
| } | ||
|
|
||
| } |
32 changes: 32 additions & 0 deletions
32
hmda-data-publisher/src/main/scala/hmda/publisher/api/HmdaDataPublisherApi.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,32 @@ | ||
| package hmda.publisher.api | ||
|
|
||
| import akka.actor.typed.Behavior | ||
| import akka.actor.typed.scaladsl.Behaviors | ||
| import akka.actor.typed.scaladsl.adapter._ | ||
| import akka.actor.{ActorSystem, CoordinatedShutdown} | ||
| import akka.http.scaladsl.server.Directives._ | ||
| import hmda.api.http.directives.HmdaTimeDirectives._ | ||
| import hmda.api.http.routes.BaseHttpApi | ||
| import hmda.publisher.scheduler.AllSchedulers | ||
|
|
||
| import scala.concurrent.ExecutionContext | ||
|
|
||
| // $COVERAGE-OFF$ | ||
| object HmdaDataPublisherApi { | ||
| val name = "hmda-data-publisher-api" | ||
|
|
||
| def apply(allSchedulers: AllSchedulers): Behavior[Nothing] = Behaviors.setup[Nothing] { ctx => | ||
| implicit val ec: ExecutionContext = ctx.executionContext | ||
| implicit val classic: ActorSystem = ctx.system.toClassic | ||
| val shutdown: CoordinatedShutdown = CoordinatedShutdown(ctx.system) | ||
| val config = classic.settings.config | ||
| val host: String = config.getString("hmda.publisher.http.host") | ||
| val port: Int = config.getInt("hmda.publisher.http.port") | ||
|
|
||
| val routes = BaseHttpApi.routes(name) ~ new DataPublisherHttpApi(allSchedulers).routes | ||
| BaseHttpApi.runServer(shutdown, name)(timed(routes), host, port) | ||
|
|
||
| Behaviors.ignore | ||
| } | ||
| } | ||
| // $COVERAGE-ON$ |
34 changes: 34 additions & 0 deletions
34
hmda-data-publisher/src/main/scala/hmda/publisher/helper/QuarterTimeBarrier.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,34 @@ | ||
| package hmda.publisher.helper | ||
|
|
||
| import java.time.{Clock, LocalDate} | ||
|
|
||
| import hmda.publisher.validation.PublishingGuard.Period | ||
| import hmda.util.BankFilterUtils.config | ||
| import hmda.util.Filer | ||
|
|
||
| class QuarterTimeBarrier(clock: Clock) { | ||
|
|
||
| def runIfStillRelevant[T](quarter: Period.Quarter)(thunk: => T): Option[T] = { | ||
| val now = LocalDate.now(clock) | ||
| if (now.isBefore(QuarterTimeBarrier.getEndDateForQuarter(quarter).plusDays(8))) { | ||
| Some(thunk) | ||
| } else { | ||
| None | ||
| } | ||
| } | ||
|
|
||
|
|
||
| } | ||
|
|
||
| object QuarterTimeBarrier { | ||
| private val rulesConfig = Filer.parse(config).fold(error => throw new RuntimeException(s"Failed to parse filing rules in HOCON: $error"), identity) | ||
|
|
||
| def getEndDateForQuarter(quarter: Period.Quarter): LocalDate = { | ||
| quarter match { | ||
| case Period.y2020Q1 => LocalDate.ofYearDay(2020,rulesConfig.qf.q1.endDayOfYear) | ||
| case Period.y2020Q2 => LocalDate.ofYearDay(2020,rulesConfig.qf.q2.endDayOfYear) | ||
| case Period.y2020Q3 => LocalDate.ofYearDay(2020,rulesConfig.qf.q3.endDayOfYear) | ||
| } | ||
| } | ||
|
|
||
| } |
12 changes: 12 additions & 0 deletions
12
hmda-data-publisher/src/main/scala/hmda/publisher/helper/RetryUtils.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,12 @@ | ||
| package hmda.publisher.helper | ||
|
|
||
| import scala.concurrent.{ExecutionContext, Future} | ||
| import scala.concurrent.duration.Duration | ||
|
|
||
| object RetryUtils { | ||
|
|
||
| def retry[T](retries: Int, delay: Duration)(f: () => Future[T])(implicit ec: ExecutionContext): Future[T] = { | ||
| f() recoverWith { case _ if retries > 0 => Future(Thread.sleep(delay.toMillis)).flatMap(_ => retry(retries - 1, delay)(f)) } | ||
| } | ||
|
|
||
| } |
20 changes: 20 additions & 0 deletions
20
hmda-data-publisher/src/main/scala/hmda/publisher/helper/S3Utils.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,20 @@ | ||
| package hmda.publisher.helper | ||
|
|
||
| import akka.NotUsed | ||
| import akka.stream.Materializer | ||
| import akka.stream.alpakka.s3.MultipartUploadResult | ||
| import akka.stream.scaladsl.{ Sink, Source } | ||
| import akka.util.ByteString | ||
|
|
||
| import scala.concurrent.duration.DurationInt | ||
| import scala.concurrent.{ ExecutionContext, Future } | ||
|
|
||
| object S3Utils { | ||
|
|
||
| def uploadWithRetry( | ||
| source: Source[ByteString, NotUsed], | ||
| uploadSink: Sink[ByteString, Future[MultipartUploadResult]] | ||
| )(implicit mat: Materializer, ec: ExecutionContext): Future[MultipartUploadResult] = | ||
| RetryUtils.retry(retries = 3, delay = 1.minute)(() => source.runWith(uploadSink)) | ||
|
|
||
| } |
11 changes: 11 additions & 0 deletions
11
hmda-data-publisher/src/main/scala/hmda/publisher/scheduler/AllSchedulers.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,11 @@ | ||
| package hmda.publisher.scheduler | ||
|
|
||
| import akka.actor.ActorRef | ||
|
|
||
| case class AllSchedulers( | ||
| larPublicScheduler: ActorRef, | ||
| larScheduler: ActorRef, | ||
| panelScheduler: ActorRef, | ||
| tsPublicScheduler: ActorRef, | ||
| tsScheduler: ActorRef | ||
| ) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.