File sync (#4)

Author: assayire

.scalafmt.conf

@@ -1,9 +1,21 @@
-version = "3.5.9"
+version = 3.9.5
 runner.dialect = scala3
 
 maxColumn = 120
 project.git = true
+
+indent.main = 2
+indent.callSite = 2
+indent.ctorSite = 2
+indent.ctrlSite = 2
+indent.defnSite = 2
+indent.matchSite = 2
+indent.significant = 2
+
 align.preset = more
+align.stripMargin = true
+docstrings.style = Asterisk
+assumeStandardLibraryStripMargin = true
 
 align.tokens = [
   "=>", "->", "<-", ":=", "//", "%", "%%", "%%%", "+=",
@@ -19,4 +31,5 @@ align.tokens = [
     "code" = "=>"
     "owners" = [{ regex = "(Importee.Rename|Case)" }]
   }
-]
+]
+

chat-app/.scalafmt.conf

@@ -1,4 +1,3 @@
-# DO NOT EDIT THIS FILE WITHOUT CONSENT OF TEAM.
 version = 3.9.5
 runner.dialect = scala3
 
@@ -13,11 +12,10 @@ indent.defnSite = 2
 indent.matchSite = 2
 indent.significant = 2
 
-docstrings.style = keep
-assumeStandardLibraryStripMargin = true
-align.stripMargin = true
-
 align.preset = more
+align.stripMargin = true
+docstrings.style = Asterisk
+assumeStandardLibraryStripMargin = true
 
 align.tokens = [
   "=>", "->", "<-", ":=", "//", "%", "%%", "%%%", "+=",

filesync/README.md

@@ -0,0 +1,16 @@
+# File Sync
+
+NOTE: Sometimes, this app throws error when running from IntelliJ. This is because the `agent` is bundled as a resource in the `Sync` project, which requires running `sbt assembly` at least once. So run `sbt assembly` in the terminal before running the app from IntelliJ.
+
+Also, when running the app from IntelliJ, configure the run configuration for `sync.Sync` to pass program arguments (source and destination paths).
+
+## Exercises
+
+- Syncing folders/sub-folders
+
+  Track `Rpc.CreateFolder` case class
+
+- Syncing deleted files/folders 
+  
+  Track `Rpc.DeletePath` case class
+

filesync/agent/src/main/scala/sync/Agent.scala

@@ -0,0 +1,33 @@
+package sync
+
+import java.io.{DataInputStream, DataOutputStream}
+
+object Agent {
+  def main(args: Array[String]): Unit = {
+    val input  = new DataInputStream(System.in)
+    val output = new DataOutputStream(System.out)
+
+    while (true) try {
+      val rpc = Shared.receive[Rpc](input)
+      System.err.println("Agent handling: " + rpc)
+
+      rpc match {
+        case Rpc.StatPath(path) =>
+          Shared.send(output, Rpc.StatInfo(path, Shared.hashPath(os.pwd / path)))
+        case Rpc.WriteOver(bytes, path) =>
+          os.remove.all(os.pwd / path)
+          os.write.over(os.pwd / path, bytes, createFolders = true)
+        case Rpc.DeletePath(path) =>
+          os.remove.all(os.pwd / path)
+        case sync.Rpc.CreateFolder(path) =>
+          os.makeDir.all(os.pwd / path)
+      }
+    } catch {
+      case e: java.io.EOFException =>
+        System.err.println("+--------------------------------------------------+")
+        e.printStackTrace(System.err)
+        System.err.println("+--------------------------------------------------+")
+        System.exit(0)
+    }
+  }
+}

filesync/build.sbt

@@ -0,0 +1,36 @@
+import java.nio.file.Paths
+
+ThisBuild / version      := "0.1.0-SNAPSHOT"
+ThisBuild / scalaVersion := "3.7.0"
+
+ThisBuild / libraryDependencies ++=
+  "com.lihaoyi"   %% "upickle"      % "4.1.0" ::
+    "com.lihaoyi" %% "os-lib"       % "0.11.4" ::
+    "com.lihaoyi" %% "os-lib-watch" % "0.11.4" ::
+    "com.lihaoyi" %% "castor"       % "0.3.0" ::
+    Nil
+
+lazy val shared = project
+
+lazy val agent =
+  project
+    .dependsOn(shared)
+    .settings(
+      assembly / mainClass       := Some("sync.Agent"),
+      assembly / assemblyJarName := "agent.jar"
+    )
+
+lazy val sync =
+  project
+    .dependsOn(shared, agent)
+    .settings(
+      Compile / resources := (Compile / resources).value ++ Seq(
+        {
+          val agentJar = (agent / assembly / assemblyOutputPath).value
+          println(s"************************************** $agentJar")
+          /*val destPath = target.value / "agent2.jar"
+          IO.copyFile(agentJar, destPath)*/
+          file(agentJar.getAbsolutePath)
+        }
+      )
+    )

filesync/project/build.properties

@@ -0,0 +1,2 @@
+sbt.version=1.10.7
+build.properties=1.10.11

filesync/project/plugins.sbt

@@ -0,0 +1 @@
+addSbtPlugin("com.eed3si9n" % "sbt-assembly" % "2.3.1")

filesync/shared/src/main/scala/sync/Rpc.scala

@@ -0,0 +1,27 @@
+package sync
+
+import os.SubPath
+import upickle.default.{ReadWriter, macroRW, readwriter}
+
+sealed trait Rpc
+
+object Rpc {
+  implicit val subPathRw: ReadWriter[SubPath] = readwriter[String].bimap[os.SubPath](_.toString, os.SubPath(_))
+
+  case class StatPath(path: os.SubPath) extends Rpc
+  implicit val statPathRw: ReadWriter[StatPath] = macroRW
+
+  case class WriteOver(src: Array[Byte], path: os.SubPath) extends Rpc
+  implicit val writeOverRw: ReadWriter[WriteOver] = macroRW
+
+  case class CreateFolder(path: os.SubPath) extends Rpc
+  implicit val createFolderRw: ReadWriter[CreateFolder] = macroRW
+
+  case class StatInfo(p: os.SubPath, fileHash: Option[Int])
+  implicit val statInfoRw: ReadWriter[StatInfo] = macroRW
+
+  case class DeletePath(path: os.SubPath) extends Rpc
+  implicit val deletePathRw: ReadWriter[DeletePath] = macroRW
+
+  implicit val msgRw: ReadWriter[Rpc] = macroRW
+}

filesync/shared/src/main/scala/sync/Shared.scala

@@ -0,0 +1,22 @@
+package sync
+
+import upickle.default.{Reader, Writer}
+
+import java.io.{DataInputStream, DataOutputStream}
+
+object Shared:
+  def send[T: Writer](out: DataOutputStream, msg: T): Unit =
+    val bytes = upickle.default.writeBinary(msg)
+    out.writeInt(bytes.length)
+    out.write(bytes)
+    out.flush()
+
+  def receive[T: Reader](in: DataInputStream): T =
+    val buf = new Array[Byte](in.readInt())
+    in.readFully(buf)
+    upickle.default.readBinary[T](buf)
+
+  def hashPath(p: os.Path): Option[Int] =
+    if (!os.isFile(p)) None
+    else Some(java.util.Arrays.hashCode(os.read.bytes(p)))
+end Shared

filesync/sync/src/main/scala/sync/Msg.scala

@@ -0,0 +1,7 @@
+package sync
+
+import sync.Rpc.StatInfo
+
+sealed trait Msg
+case class ChangedPath(value: os.SubPath, deleted: Boolean) extends Msg
+case class AgentResponse(value: StatInfo) extends Msg

filesync/sync/src/main/scala/sync/Sync.scala

@@ -0,0 +1,64 @@
+package sync
+
+import castor.Context.Simple.global
+
+object Sync:
+  def main(args: Array[String]): Unit = {
+    if (args.length != 2) {
+      println("Usage: sync <src> <dest>")
+      System.exit(1)
+    }
+
+    val src  = os.Path(args(0))
+    val dest = os.Path(args(1))
+
+    if (!os.exists(src)) {
+      println(s"Source path does not exist: $src")
+      System.exit(2)
+    }
+
+    println(s"Syncing $src to $dest ...")
+
+    val agentExecutable = os.temp(os.read.bytes(os.resource / "agent.jar"))
+    os.perms.set(agentExecutable, "rwxr-xr-x")
+    val agent = os.proc("java", "-jar", agentExecutable).spawn(cwd = dest)
+
+    object SyncActor extends castor.SimpleActor[Msg]:
+      def run(msg: Msg): Unit = {
+        println("SyncActor handling: " + msg)
+        msg match {
+          case ChangedPath(value, false) =>
+            Shared.send(agent.stdin.data, Rpc.StatPath(value))
+          case ChangedPath(value, true) =>
+            Shared.send(agent.stdin.data, Rpc.DeletePath(value))
+          case AgentResponse(Rpc.StatInfo(p, remoteHash)) =>
+            Shared.hashPath(src / p) match {
+              case None => // It is a folder
+                Shared.send(agent.stdin.data, Rpc.CreateFolder(p))
+              case `remoteHash` => // Local and remote hashes are same. Do nothing
+              case Some(_) => // It is a file and content differs
+                Shared.send(agent.stdin.data, Rpc.WriteOver(os.read.bytes(src / p), p))
+            }
+        }
+      }
+
+    val agentReader = new Thread(() => {
+      while (agent.isAlive()) {
+        SyncActor.send(AgentResponse(Shared.receive[Rpc.StatInfo](agent.stdout.data)))
+      }
+    })
+
+    agentReader.start()
+
+    val watcher = os.watch.watch(
+      Seq(src),
+      onEvent = _.foreach { p =>
+        val path = p.subRelativeTo(src)
+        SyncActor.send(ChangedPath(path, !os.exists(src / path)))
+      }
+    )
+
+    Thread.sleep(Long.MaxValue)
+  }
+
+end Sync

src/main/scala/filesync/SimpleFileSync.scala

@@ -0,0 +1,32 @@
+package filesync
+
+object SimpleFileSync {
+
+  /** There are three cases the above code does not support:
+    *   - If the source path is empty and the destination path contains a folder. This is a "delete" which we will
+    *     ignore for now; supporting this is left as an exercise for the reader
+    *   - If the source path is a folder and the destination path is a folder. In this case doing nothing is fine:
+    *     os.walk will enter the source path folder and process all the files within it recursively
+    */
+  def sync(src: os.Path, dest: os.Path): Unit = {
+    os.walk(src).foreach { srcSubPath =>
+      val subPath     = srcSubPath.subRelativeTo(src)
+      val destSubPath = dest / subPath
+
+      (os.isDir(srcSubPath), os.isDir(destSubPath)) match {
+        case (false, true) | (true, false) =>
+          os.copy.over(srcSubPath, destSubPath, createFolders = true)
+        case (false, false)
+            if !os.exists(destSubPath)
+              || !os.read.bytes(srcSubPath).sameElements(os.read.bytes(destSubPath)) =>
+          os.copy.over(srcSubPath, destSubPath, createFolders = true)
+        case _ => // do nothing
+      }
+    }
+  }
+
+  @main
+  def main(): Unit = {
+    sync(os.pwd / "src/main/scala", os.home / "Temp" / "blah")
+  }
+}