Use Ref with a single map update

compute hashes in parallel

Author: lolgab

project/src/live.server.scala

@@ -3,7 +3,6 @@ import scala.concurrent.duration.*
 import org.http4s.*
 import org.http4s.HttpApp
 import org.http4s.HttpRoutes
-import org.http4s.client.Client
 import org.http4s.dsl.io.*
 import org.http4s.ember.client.EmberClientBuilder
 import org.http4s.ember.server.EmberServerBuilder
@@ -19,14 +18,12 @@ import fs2.concurrent.Topic
 import scribe.Level
 
 import cats.effect.*
-import cats.effect.std.*
 import cats.implicits.*
 
 import _root_.io.circe.*
 import _root_.io.circe.Encoder
 
 import ProxyConfig.Equilibrium
-import scribe.Scribe
 
 sealed trait FrontendEvent derives Encoder.AsObject
 
@@ -265,27 +262,28 @@ object LiveServer
             .toResource
 
           fileToHashRef <- Ref[IO].of(Map.empty[String, String]).toResource
-          fileToHashMapRef = MapRef.fromSingleImmutableMapRef[IO, String, String](fileToHashRef)
           refreshTopic <- Topic[IO, Unit].toResource
           linkingTopic <- Topic[IO, Unit].toResource
           client <- EmberClientBuilder.default[IO].build
+          outDirPath = fs2.io.file.Path(outDir)
+          baseDirPath = fs2.io.file.Path(baseDir)
 
           proxyRoutes: HttpRoutes[IO] <- makeProxyRoutes(client, pathPrefix, proxyConfig)(logger)
 
           _ <- buildRunner(
             buildTool,
             linkingTopic,
-            fs2.io.file.Path(baseDir),
-            fs2.io.file.Path(outDir),
+            baseDirPath,
+            outDirPath,
             extraBuildArgs,
             millModuleName
           )(logger)
 
-          app <- routes(outDir.toString(), refreshTopic, indexOpts, proxyRoutes, fileToHashRef)(logger)
+          app <- routes(outDir, refreshTopic, indexOpts, proxyRoutes, fileToHashRef)(logger)
 
-          _ <- seedMapOnStart(outDir, fileToHashMapRef)(logger)
+          _ <- updateMapRef(outDirPath, fileToHashRef)(logger).toResource
           // _ <- stylesDir.fold(Resource.unit)(sd => seedMapOnStart(sd, mr))
-          _ <- fileWatcher(fs2.io.file.Path(outDir), fileToHashMapRef, linkingTopic, refreshTopic)(logger)
+          _ <- fileWatcher(outDirPath, fileToHashRef, linkingTopic, refreshTopic)(logger)
           // _ <- stylesDir.fold(Resource.unit[IO])(sd => fileWatcher(fs2.io.file.Path(sd), mr))
           _ <- logger.info(s"Start dev server on http://localhost:$port").toResource
           server <- buildServer(app, port)

project/src/routes.scala

@@ -27,7 +27,6 @@ import cats.effect.*
 import cats.effect.IO
 import cats.effect.kernel.Ref
 import cats.effect.kernel.Resource
-import cats.effect.std.*
 import cats.syntax.all.*
 
 import _root_.io.circe.syntax.EncoderOps
@@ -147,60 +146,37 @@ def routes(
 
 end routes
 
-def seedMapOnStart(stringPath: String, mr: MapRef[IO, String, Option[String]])(logger: Scribe[IO]) =
-  val asFs2 = fs2.io.file.Path(stringPath)
-  fs2
-    .io
-    .file
-    .Files[IO]
-    .walk(asFs2)
-    .evalMap {
-      f =>
-        Files[IO]
-          .isRegularFile(f)
-          .ifM(
-            // logger.trace(s"hashing $f") >>
-            fileHash(f).flatMap(
-              h =>
-                val key = asFs2.relativize(f)
-                logger.trace(s"hashing $f to put at $key with hash : $h") >>
-                  mr.setKeyValue(key.toString(), h)
-            ),
-            IO.unit
-          )
-    }
+def updateMapRef(stringPath: fs2.io.file.Path, mr: Ref[IO, Map[String, String]])(logger: Scribe[IO]) =
+  Files[IO]
+    .walk(stringPath)
+    .evalFilter(Files[IO].isRegularFile)
+    .parEvalMap(maxConcurrent = 8)(path => fileHash(path).map(path -> _))
     .compile
-    .drain
-    .toResource
-
-end seedMapOnStart
+    .toVector
+    .flatMap(
+      vector =>
+        val newMap = vector
+          .view
+          .map(
+            (path, hash) =>
+              val relativizedPath = stringPath.relativize(path).toString
+              relativizedPath -> hash
+          )
+          .toMap
+        logger.trace(s"Updated hashes $newMap") *> mr.set(newMap)
+    )
 
 private def fileWatcher(
     stringPath: fs2.io.file.Path,
-    mr: MapRef[IO, String, Option[String]],
+    mr: Ref[IO, Map[String, String]],
     linkingTopic: Topic[IO, Unit],
     refreshTopic: Topic[IO, Unit]
 )(logger: Scribe[IO]): ResourceIO[Unit] =
   linkingTopic
     .subscribe(10)
     .evalTap {
       _ =>
-        fs2
-          .io
-          .file
-          .Files[IO]
-          .list(stringPath)
-          .evalTap {
-            path =>
-              fileHash(path).flatMap(
-                h =>
-                  val serveAt = stringPath.relativize(fs2.io.file.Path(path.toString()))
-                  logger.trace(s"$serveAt :: hash -> $h") >>
-                    mr.setKeyValue(serveAt.toString(), h)
-              )
-          }
-          .compile
-          .drain >> refreshTopic.publish1(())
+        updateMapRef(stringPath, mr)(logger) >> refreshTopic.publish1(())
     }
     .compile
     .drain

project/test/src/RoutesSpec.scala

@@ -67,8 +67,7 @@ class RoutesSuite extends CatsEffectSuite:
       for
         logger <- IO(scribe.cats[IO]).toResource
         fileToHashRef <- Ref[IO].of(Map.empty[String, String]).toResource
-        fileToHashMapRef = MapRef.fromSingleImmutableMapRef[IO, String, String](fileToHashRef)
-        _ <- seedMapOnStart(tempDir.toString, fileToHashMapRef)(logger)
+        _ <- updateMapRef(tempDir.toFs2, fileToHashRef)(logger).toResource
       yield fileToHashRef
         .get
         .map {
@@ -86,10 +85,9 @@ class RoutesSuite extends CatsEffectSuite:
         fileToHashRef <- Ref[IO].of(Map.empty[String, String]).toResource
         linkingTopic <- Topic[IO, Unit].toResource
         refreshTopic <- Topic[IO, Unit].toResource
-        fileToHashMapRef = MapRef.fromSingleImmutableMapRef[IO, String, String](fileToHashRef)
-        _ <- fileWatcher(fs2.io.file.Path(tempDir.toString), fileToHashMapRef, linkingTopic, refreshTopic)(logger)
+        _ <- fileWatcher(fs2.io.file.Path(tempDir.toString), fileToHashRef, linkingTopic, refreshTopic)(logger)
         _ <- IO.sleep(100.millis).toResource // wait for watcher to start
-        _ <- seedMapOnStart(tempDir.toString, fileToHashMapRef)(logger)
+        _ <- updateMapRef(tempDir.toFs2, fileToHashRef)(logger).toResource
         _ <- IO.blocking(os.write.over(tempDir / "test.js", "const hi = 'bye, world'")).toResource
         _ <- linkingTopic.publish1(()).toResource
         _ <- refreshTopic.subscribe(1).head.compile.resource.drain
@@ -115,8 +113,7 @@ class RoutesSuite extends CatsEffectSuite:
         val app = for
           logger <- IO(scribe.cats[IO]).toResource
           fileToHashRef <- Ref[IO].of(Map.empty[String, String]).toResource
-          fileToHashMapRef = MapRef.fromSingleImmutableMapRef[IO, String, String](fileToHashRef)
-          _ <- seedMapOnStart(tempDir.toString, fileToHashMapRef)(logger)
+          _ <- updateMapRef(tempDir.toFs2, fileToHashRef)(logger).toResource
           refreshPub <- Topic[IO, Unit].toResource
           theseRoutes <- routes(
             tempDir.toString,

project/test/src/test_extensions.scala

@@ -0,0 +1,3 @@
+extension (path: os.Path)
+  def toFs2: fs2.io.file.Path =
+    fs2.io.file.Path.fromNioPath(path.toNIO)