From 7df6cef9d5c6699ef58c8a5c8fe89562ab48f1ef Mon Sep 17 00:00:00 2001 From: Todd Burnside Date: Thu, 23 Nov 2023 16:20:15 -0300 Subject: [PATCH] Upgrade gds-fp v0.0.9 - the typelevel dependencies --- .../scala/edu/gemini/aspen/gds/Main.scala | 2 +- .../GDSConfigurationServiceFactory.scala | 27 ++++++++--- .../gds/configuration/GdsConfiguration.scala | 8 ++-- .../aspen/gds/fits/FitsFileProcessor.scala | 48 ++++++++++--------- .../aspen/gds/fits/FitsFileTransferrer.scala | 20 ++++---- .../aspen/gds/keywords/KeywordManager.scala | 3 +- .../gds/observations/ObservationManager.scala | 11 +++-- .../aspen/gds/seqexec/SeqexecServer.scala | 21 ++++---- pom.xml | 5 +- 9 files changed, 80 insertions(+), 65 deletions(-) diff --git a/gds/src/main/scala/edu/gemini/aspen/gds/Main.scala b/gds/src/main/scala/edu/gemini/aspen/gds/Main.scala index a07a7a082..12fb24531 100644 --- a/gds/src/main/scala/edu/gemini/aspen/gds/Main.scala +++ b/gds/src/main/scala/edu/gemini/aspen/gds/Main.scala @@ -38,7 +38,7 @@ object Main { ) obsMgr <- ObservationManager(config.observation, kwMgr, obsStateQ, fitsQ) fitsProcessor = FitsFileProcessor[IO](config.fitsConfig, keywordConfig) - seqexecServer = SeqexecServer(obsStateQ, config.seqexecHost, config.seqexecPort) + seqexecServer = Stream.eval(SeqexecServer(obsStateQ, config.seqexecHost, config.seqexecPort)) obsPurge = Stream .fixedDelay[IO](config.observation.cleanupRate) .foreach(_ => obsStateQ.offer(ObservationStateEvent.PurgeStale)) diff --git a/gds/src/main/scala/edu/gemini/aspen/gds/configuration/GDSConfigurationServiceFactory.scala b/gds/src/main/scala/edu/gemini/aspen/gds/configuration/GDSConfigurationServiceFactory.scala index 26e770444..a23e88839 100644 --- a/gds/src/main/scala/edu/gemini/aspen/gds/configuration/GDSConfigurationServiceFactory.scala +++ b/gds/src/main/scala/edu/gemini/aspen/gds/configuration/GDSConfigurationServiceFactory.scala @@ -3,7 +3,10 @@ package edu.gemini.aspen.gds.configuration import cats.data.{ NonEmptyChain, ValidatedNec } import cats.data.Validated.{ Invalid, Valid } import cats.syntax.all._ +import com.comcast.ip4s.Host +import com.comcast.ip4s.Port import edu.gemini.aspen.gmp.services.PropertyHolder +import fs2.io.file.{ Path => Fs2Path } import java.nio.file._ import java.util.{ Dictionary } import java.util.logging.Logger @@ -61,8 +64,8 @@ class GDSConfigurationServiceFactory( val eventSleep = asDuration(props, "observation.event.sleep") val keywordRetries = asPosInt(props, "keyword.collection.retries") val keywordSleep = asDuration(props, "keyword.collection.sleep") - val seqexecPort = asPosInt(props, "seqexec.server.port") - val seqexecHost = asString(props, "seqexec.server.host") + val seqexecPort = asPort(props, "seqexec.server.port") + val seqexecHost = asHost(props, "seqexec.server.host") val fitsDelOrig = asBool(props, "fits.deleteOriginal") val configValidated = (keywordConfig, @@ -103,6 +106,16 @@ class GDSConfigurationServiceFactory( private def asString(props: Map[String, _], key: String): ValidatedNec[String, String] = props.get(key).map(_.toString).toValidNec(s"Config value missing for `$key`") + private def asHost(props: Map[String, _], key: String): ValidatedNec[String, Host] = + asString(props, key).andThen(s => + Host.fromString(s).toValidNec(s"Invalid host name `$s` for `$key`") + ) + + private def asPort(props: Map[String, _], key: String): ValidatedNec[String, Port] = + asString(props, key).andThen(s => + Port.fromString(s).toValidNec(s"Invalid port value`$s` for `$key`. Must be a positive integer.") + ) + private def asPosInt(props: Map[String, _], key: String): ValidatedNec[String, Int] = asString(props, key).andThen(s => s.toIntOption @@ -127,17 +140,19 @@ class GDSConfigurationServiceFactory( Option.when(duration.isFinite)(duration.asInstanceOf[FiniteDuration]) } - private def fitsSrcDir: ValidatedNec[String, Path] = dirFromPropHolder(fitsSrcPathKey) - private def fitsDestDir: ValidatedNec[String, Path] = dirFromPropHolder(fitsDestPathKey) + private def fitsSrcDir: ValidatedNec[String, Fs2Path] = dirFromPropHolder(fitsSrcPathKey) + private def fitsDestDir: ValidatedNec[String, Fs2Path] = dirFromPropHolder(fitsDestPathKey) // PropertyHolder shouldn't return null, but... private def propHolderValue(key: String): ValidatedNec[String, String] = Option(propertyHolder.getProperty(key)) .toValidNec(s"Value for PropertyHolder sevice key `$key` cannot be null.") - private def dirFromPropHolder(key: String): ValidatedNec[String, Path] = + private def dirFromPropHolder(key: String): ValidatedNec[String, Fs2Path] = propHolderValue(key).andThen { dir => Try { + // Not using FS2 io here, because it would introduce an `F[_]', which would not work well at + // this point in osgi val path = Paths.get(dir) if (path.toFile().isDirectory()) path.validNec else @@ -145,7 +160,7 @@ class GDSConfigurationServiceFactory( } match { case Failure(e) => s"Error creating Path for PropertyHolder service key `$key`, value `$dir`: ${e.getMessage}".invalidNec - case Success(value) => value + case Success(value) => value.map(p => Fs2Path.fromNioPath(p)) } } diff --git a/gds/src/main/scala/edu/gemini/aspen/gds/configuration/GdsConfiguration.scala b/gds/src/main/scala/edu/gemini/aspen/gds/configuration/GdsConfiguration.scala index e1271c581..5ea9a1260 100644 --- a/gds/src/main/scala/edu/gemini/aspen/gds/configuration/GdsConfiguration.scala +++ b/gds/src/main/scala/edu/gemini/aspen/gds/configuration/GdsConfiguration.scala @@ -1,6 +1,8 @@ package edu.gemini.aspen.gds.configuration -import java.nio.file.Path +import fs2.io.file.Path +import com.comcast.ip4s.Host +import com.comcast.ip4s.Port import scala.concurrent.duration._ final case class ObservationConfig( @@ -25,7 +27,7 @@ final case class GdsConfiguration( keywords: List[KeywordConfigurationItem], observation: ObservationConfig, keywordRetries: RetryConfig, - seqexecPort: Int, - seqexecHost: String, + seqexecPort: Port, + seqexecHost: Host, fitsConfig: FitsConfig ) diff --git a/gds/src/main/scala/edu/gemini/aspen/gds/fits/FitsFileProcessor.scala b/gds/src/main/scala/edu/gemini/aspen/gds/fits/FitsFileProcessor.scala index 1947872cf..14749bf61 100644 --- a/gds/src/main/scala/edu/gemini/aspen/gds/fits/FitsFileProcessor.scala +++ b/gds/src/main/scala/edu/gemini/aspen/gds/fits/FitsFileProcessor.scala @@ -16,7 +16,7 @@ import edu.gemini.aspen.gds.syntax.all._ import edu.gemini.aspen.gds.transfer.FitsFileTransferrer import edu.gemini.aspen.giapi.data.DataLabel import fs2.io.file.{ Files => Fs2Files } -import java.nio.file.Path +import fs2.io.file.Path import java.util.logging.Logger import scala.concurrent.duration._ import scala.sys.process._ @@ -28,10 +28,10 @@ sealed trait FitsFileProcessor[F[_]] { object FitsFileProcessor { private val logger = Logger.getLogger(this.getClass.getName) - def apply[F[_]]( + def apply[F[_]: Async: Fs2Files]( fitsConfig: FitsConfig, keywordConfigs: List[KeywordConfigurationItem] - )(implicit F: Async[F]): FitsFileProcessor[F] = + ): FitsFileProcessor[F] = new FitsFileProcessor[F] { val requiredKeywords: Map[Int, List[String]] = keywordConfigs @@ -47,7 +47,7 @@ object FitsFileProcessor { val result = for { _ <- logger.infoF(s"Preparing to transfer FITS file $outFileName") - source <- F.delay(fitsConfig.sourceDir.resolve(inFileName)) + source = fitsConfig.sourceDir.resolve(inFileName) dest <- safeDestinationFile(fitsConfig.destDir, outFileName) cards <- processKeywordsToCards(keywords) t1 <- Clock[F].realTime @@ -66,16 +66,18 @@ object FitsFileProcessor { ) } - def safeDestinationFile(dir: Path, name: String): F[Path] = for { - fullPath <- F.delay(dir.resolve(name)) - exists <- Fs2Files[F].exists(fullPath) - safePath <- - if (exists) - logger.warningF( - s"Output file $fullPath already exists - generating new name." - ) >> safeDestinationFile(dir, newDestinationFileName(name)) - else fullPath.pure[F] - } yield safePath + def safeDestinationFile(dir: Path, name: String): F[Path] = { + val fullPath = dir.resolve(name) + for { + exists <- Fs2Files[F].exists(fullPath) + safePath <- + if (exists) + logger.warningF( + s"Output file $fullPath already exists - generating new name." + ) >> safeDestinationFile(dir, newDestinationFileName(name)) + else fullPath.pure[F] + } yield safePath + } def newDestinationFileName(fullName: String): String = { val nameRegex = """(\w*)-(\d+)""".r @@ -177,9 +179,9 @@ object FitsFileProcessor { for { _ <- logger.infoF(s"Changing ownership of `$dest` to `${c.owner}` with command: $cmd") b <- runCommand(cmd) - _ <- if (b) F.unit else logger.severeF(s"Failed to change ownership of `$dest`") + _ <- if (b) Async[F].unit else logger.severeF(s"Failed to change ownership of `$dest`") } yield () - case None => F.unit + case None => Async[F].unit } def setPermissions(dest: Path, config: Option[SetPermissionsConfig]): F[Unit] = config match { @@ -190,9 +192,9 @@ object FitsFileProcessor { s"Changing permissions of `$dest` to `${c.permissions}` with command: $cmd" ) b <- runCommand(cmd) - _ <- if (b) F.unit else logger.severeF(s"Failed to change permissions of `$dest`") + _ <- if (b) Async[F].unit else logger.severeF(s"Failed to change permissions of `$dest`") } yield () - case None => F.unit + case None => Async[F].unit } def deleteOriginal(file: Path, deleteIt: Boolean): F[Unit] = @@ -200,7 +202,7 @@ object FitsFileProcessor { logger.infoF(s"Deleting original FITS file: $file") >> Fs2Files[F] .delete(file) .handleErrorWith(e => logger.severeF(s"Failed to delete original file: $file", e)) - else F.unit + else Async[F].unit def runCommand(command: String): F[Boolean] = { val plogger = @@ -208,7 +210,7 @@ object FitsFileProcessor { errorLine => logger.severe(s"Command error: $errorLine") ) for { - process <- F.delay(command.run(plogger)) + process <- Async[F].delay(command.run(plogger)) // we need to timeout in case the command hangs. For example, if sudo is used, and a password is required. // Neither of the commands should take long to completed. result <- waitForResult(process, 10) @@ -224,8 +226,8 @@ object FitsFileProcessor { def waitForResult(process: Process, remaining: Int): F[Option[Int]] = if (process.isAlive()) - if (remaining > 0) F.sleep(250.milliseconds) >> waitForResult(process, remaining - 1) - else F.blocking(process.destroy()).as(none) - else F.blocking(process.exitValue().some) + if (remaining > 0) Async[F].sleep(250.milliseconds) >> waitForResult(process, remaining - 1) + else Async[F].blocking(process.destroy()).as(none) + else Async[F].blocking(process.exitValue().some) } } diff --git a/gds/src/main/scala/edu/gemini/aspen/gds/fits/FitsFileTransferrer.scala b/gds/src/main/scala/edu/gemini/aspen/gds/fits/FitsFileTransferrer.scala index c02742e0f..fc887e1bb 100644 --- a/gds/src/main/scala/edu/gemini/aspen/gds/fits/FitsFileTransferrer.scala +++ b/gds/src/main/scala/edu/gemini/aspen/gds/fits/FitsFileTransferrer.scala @@ -7,8 +7,7 @@ import cats.syntax.all._ import edu.gemini.aspen.gds.fits._ import edu.gemini.aspen.gds.syntax.all._ import fs2._ -import fs2.io.file.Files -import java.nio.file.Path +import fs2.io.file.{ Files, Flags, Path } import java.util.logging.Logger object FitsFileTransferrer { @@ -177,13 +176,12 @@ object FitsFileTransferrer { in => go(in, ParserState.empty).stream } - private def deleteIfExists[F[_]: Async](output: Path): F[Unit] = + private def deleteIfExists[F[_]: Async: Files](output: Path): F[Unit] = Files[F].exists(output).flatMap { case true => - logger.warningF(s"Output file $output already exists. It will be deleted.") >> Files[F] - .deleteIfExists(output) - .void - case false => Applicative[F].unit + logger.warningF(s"Output file $output already exists. It will be deleted.") >> + Files[F].deleteIfExists(output).void + case false => Async[F].unit } def stream[F[_]: Async]( @@ -193,7 +191,7 @@ object FitsFileTransferrer { ): Stream[F, Byte] = input.through(fitsPipe(requiredHeaders, additionalHeaders)) - def transfer[F[_]: Async]( + def transfer[F[_]: Async: Files]( input: Path, output: Path, requiredHeaders: Map[Int, List[String]], @@ -203,15 +201,13 @@ object FitsFileTransferrer { _ <- deleteIfExists(output) ref <- Ref.of(0L) _ <- stream( - Files[F] - .readAll(input, chunkSize = RecordLength), + Files[F].readAll(input, chunkSize = RecordLength, flags = Flags.Read), requiredHeaders, additionalHeaders ) .chunks .evalTap(c => ref.update(_ + c.size)) - // if/when fs2 is upgraded, can switch to `unchunks` instead of flatMap - .flatMap(Stream.chunk(_)) + .unchunks .through(Files[F].writeAll(output)) .compile .drain diff --git a/gds/src/main/scala/edu/gemini/aspen/gds/keywords/KeywordManager.scala b/gds/src/main/scala/edu/gemini/aspen/gds/keywords/KeywordManager.scala index e4fdcc802..e69a8f25a 100644 --- a/gds/src/main/scala/edu/gemini/aspen/gds/keywords/KeywordManager.scala +++ b/gds/src/main/scala/edu/gemini/aspen/gds/keywords/KeywordManager.scala @@ -1,13 +1,12 @@ package edu.gemini.aspen.gds.keywords import cats.effect.kernel._ +import cats.effect.std.MapRef import cats.effect.syntax.all._ import cats.syntax.all._ import edu.gemini.aspen.gds.configuration.RetryConfig import edu.gemini.aspen.gds.syntax.all._ import edu.gemini.aspen.giapi.data.{ DataLabel, ObservationEvent } -import io.chrisdavenport.mapref.MapRef -import io.chrisdavenport.mapref.implicits._ import java.util.logging.Logger sealed trait KeywordManager[F[_]] { diff --git a/gds/src/main/scala/edu/gemini/aspen/gds/observations/ObservationManager.scala b/gds/src/main/scala/edu/gemini/aspen/gds/observations/ObservationManager.scala index 511bb90f4..97ab092b3 100644 --- a/gds/src/main/scala/edu/gemini/aspen/gds/observations/ObservationManager.scala +++ b/gds/src/main/scala/edu/gemini/aspen/gds/observations/ObservationManager.scala @@ -1,6 +1,7 @@ package edu.gemini.aspen.gds.observations -import cats.effect.{ Async, Clock } +import cats.effect.{ Async, Clock, Ref } +import cats.effect.std.MapRef import cats.effect.std.QueueSink import cats.syntax.all._ import edu.gemini.aspen.gds.configuration.ObservationConfig @@ -8,8 +9,6 @@ import edu.gemini.aspen.gds.keywords.{ CollectedKeyword, KeywordManager } import edu.gemini.aspen.gds.observations.ObservationStateEvent._ import edu.gemini.aspen.gds.syntax.all._ import edu.gemini.aspen.giapi.data.DataLabel -import io.chrisdavenport.mapref.MapRef -import io.chrisdavenport.mapref.implicits._ import java.util.logging.Logger import scala.concurrent.duration._ @@ -26,7 +25,9 @@ object ObservationManager { obsStateQ: QueueSink[F, ObservationStateEvent], fitsQ: QueueSink[F, (DataLabel, List[CollectedKeyword])] )(implicit F: Async[F]): F[ObservationManager[F]] = - MapRef.ofConcurrentHashMap[F, DataLabel, ObservationItem[F]]().map { mapref => + Ref.of(Map[DataLabel, ObservationItem[F]]()).map { refOfMap => + val mapref = MapRef.fromSingleImmutableMapRef(refOfMap) + def addDataLabel(dataLabel: DataLabel): F[ObservationItem[F]] = for { _ <- logIfExists(dataLabel) @@ -98,7 +99,7 @@ object ObservationManager { case PurgeStale => for { _ <- logger.infoF("Checking for expired observations.") - keys <- mapref.keys + keys <- refOfMap.get.map(_.keys.toList) now <- Clock[F].realTime _ <- keys.traverse(purgeIfNeeded(_, now)) } yield () diff --git a/gds/src/main/scala/edu/gemini/aspen/gds/seqexec/SeqexecServer.scala b/gds/src/main/scala/edu/gemini/aspen/gds/seqexec/SeqexecServer.scala index 197f322af..7ab6459b0 100644 --- a/gds/src/main/scala/edu/gemini/aspen/gds/seqexec/SeqexecServer.scala +++ b/gds/src/main/scala/edu/gemini/aspen/gds/seqexec/SeqexecServer.scala @@ -3,16 +3,17 @@ package edu.gemini.aspen.gds.seqexec import cats.effect._ import cats.effect.std.QueueSink import cats.syntax.all._ +import com.comcast.ip4s.Host +import com.comcast.ip4s.Port import edu.gemini.aspen.gds.keywords.CollectedKeyword import edu.gemini.aspen.gds.model.KeywordSource import edu.gemini.aspen.gds.observations.ObservationStateEvent import edu.gemini.aspen.gds.observations.ObservationStateEvent._ -import fs2.Stream import org.http4s._ import org.http4s.circe._ import org.http4s.dsl.io._ import org.http4s.implicits._ -import org.http4s.blaze.server._ +import org.http4s.ember.server._ import org.http4s.server.Router import Decoders._ @@ -26,9 +27,9 @@ object SeqexecServer { def apply( obsStateQ: QueueSink[IO, ObservationStateEvent], - host: String, - port: Integer - ): Stream[IO, Unit] = { + host: Host, + port: Port + ): IO[Unit] = { def kwv2Collected(kwv: KeywordValue) = CollectedKeyword.Value(kwv.keyword, KeywordSource.SeqExec, none, kwv.value) @@ -69,10 +70,12 @@ object SeqexecServer { val httpApp = Router("gds-seqexec" -> service).orNotFound - BlazeServerBuilder[IO](scala.concurrent.ExecutionContext.Implicits.global) - .bindHttp(port, host) + EmberServerBuilder + .default[IO] + .withHost(host) + .withPort(port) .withHttpApp(httpApp) - .serve - .as(()) + .build + .use(_ => IO.never) } } diff --git a/pom.xml b/pom.xml index d03e0acb1..69f446733 100644 --- a/pom.xml +++ b/pom.xml @@ -56,9 +56,6 @@ 1.7.30 3.0.0-M1 - 3.0.4 - fs2-core_2.13 - fs2-io_2.13 ${project.version} UTF-8 @@ -761,7 +758,7 @@ edu.gemini.gds gds-fp - 0.0.8 + 0.0.9 co.fs2