Merge pull request #1 from alexarchambault/develop

Add client API

Author: alexarchambault

library/src/libdaemonjvm/client/Connect.scala

@@ -0,0 +1,53 @@
+package libdaemonjvm.client
+
+import java.net.Socket
+import java.nio.channels.SocketChannel
+import java.nio.charset.StandardCharsets
+import java.nio.file.Files
+
+import libdaemonjvm.internal.{LockProcess, SocketFile, SocketHandler}
+import libdaemonjvm.LockFiles
+
+object Connect {
+
+  def tryConnect(files: LockFiles): Option[Either[ConnectError, Either[Socket, SocketChannel]]] =
+    tryConnect(files, LockProcess.default)
+
+  def tryConnect(
+    files: LockFiles,
+    proc: LockProcess
+  ): Option[Either[ConnectError, Either[Socket, SocketChannel]]] = {
+
+    def ifProcessRunning(pid: Int): Either[ConnectError, Either[Socket, SocketChannel]] =
+      SocketFile.connect(files.socketPaths) match {
+        case Left(e) =>
+          Left(new ConnectError.ZombieFound(pid, e))
+        case Right(s) =>
+          Right(s)
+      }
+
+    def ifFiles(hasLock: Boolean): Option[Either[ConnectError, Either[Socket, SocketChannel]]] = {
+      val b = Files.readAllBytes(files.pidFile)
+
+      // FIXME Catch malformed content errors here?
+      val s = new String(b, StandardCharsets.UTF_8).trim()
+      val pidOpt =
+        if (s.nonEmpty && s.forall(_.isDigit)) Some(s.toInt)
+        else None
+
+      pidOpt.flatMap { pid =>
+        if (proc.isRunning(pid)) Some(ifProcessRunning(pid))
+        else None
+      }
+    }
+
+    def pidSocketFilesFound(): Boolean =
+      Files.exists(files.pidFile) &&
+      (SocketHandler.usesWindowsPipe || Files.exists(files.socketPaths.path))
+
+    if (pidSocketFilesFound())
+      ifFiles(hasLock = false)
+    else
+      None
+  }
+}

library/src/libdaemonjvm/client/ConnectError.scala

@@ -0,0 +1,11 @@
+package libdaemonjvm.client
+
+sealed abstract class ConnectError(
+  message: String,
+  cause: Throwable = null
+) extends Exception(message, cause)
+
+object ConnectError {
+  final class ZombieFound(val pid: Int, val connectionError: Throwable)
+      extends ConnectError(s"Cannot connect to process $pid", connectionError)
+}

library/src/libdaemonjvm/LockProcess.scala renamed to library/src/libdaemonjvm/internal/LockProcess.scala

@@ -1,6 +1,4 @@
-package libdaemonjvm
-
-import libdaemonjvm.internal.{Pid, Processes}
+package libdaemonjvm.internal
 
 trait LockProcess {
   def pid(): Int

library/src/libdaemonjvm/internal/SocketFile.scala

@@ -9,18 +9,29 @@ import libdaemonjvm.SocketPaths
 
 object SocketFile {
   def canConnect(paths: SocketPaths): Either[Throwable, Unit] = {
+    var s: Either[Throwable, Either[Socket, SocketChannel]] = null
+    try {
+      s = connect(paths)
+      s.map(_ => ())
+    }
+    finally if (s != null)
+      s.toOption.foreach(_.merge.close())
+  }
+  def connect(paths: SocketPaths): Either[Throwable, Either[Socket, SocketChannel]] = {
     var s: Either[Socket, SocketChannel] = null
     try {
       s = SocketHandler.client(paths)
-      Right(())
+      Right(s)
     }
     catch {
       case e: SocketException =>
+        if (s != null)
+          s.merge.close()
         Left(e)
       case e: SocketExceptionLike =>
+        if (s != null)
+          s.merge.close()
         Left(e)
     }
-    finally if (s != null)
-      s.merge.close()
   }
 }

library/src/libdaemonjvm/Lock.scala renamed to library/src/libdaemonjvm/server/Lock.scala

@@ -1,10 +1,11 @@
-package libdaemonjvm
+package libdaemonjvm.server
 
 import java.nio.channels.ServerSocketChannel
 import java.nio.charset.StandardCharsets
 import java.nio.file.{Files, Path}
 
-import libdaemonjvm.internal.{SocketFile, SocketHandler}
+import libdaemonjvm.LockFiles
+import libdaemonjvm.internal.{LockProcess, SocketFile, SocketHandler}
 import java.net.ServerSocket
 
 object Lock {

library/src/libdaemonjvm/LockError.scala renamed to library/src/libdaemonjvm/server/LockError.scala

@@ -1,4 +1,4 @@
-package libdaemonjvm
+package libdaemonjvm.server
 
 import java.nio.file.Path
 

library/src/libdaemonjvm/SigInt.scala renamed to library/src/libdaemonjvm/server/SigInt.scala

@@ -1,4 +1,4 @@
-package libdaemonjvm
+package libdaemonjvm.server
 
 import sun.misc.{Signal, SignalHandler}
 

manual/server/src/libdaemonjvm/TestServer.scala

@@ -6,6 +6,7 @@ import scala.concurrent.duration._
 import java.util.concurrent.atomic.AtomicInteger
 
 import libdaemonjvm.internal.SocketFile
+import libdaemonjvm.server.Lock
 import java.io.IOException
 
 object TestServer {

tests/test/src/libdaemonjvm/tests/LockTests.scala

@@ -2,20 +2,24 @@ package libdaemonjvm.tests
 
 import com.eed3si9n.expecty.Expecty.expect
 import libdaemonjvm._
-import libdaemonjvm.internal.SocketFile
+import libdaemonjvm.internal._
+import libdaemonjvm.server._
 
 import java.nio.file.Files
-import libdaemonjvm.internal.SocketHandler
+import libdaemonjvm.client.Connect
 
 class LockTests extends munit.FunSuite {
 
   test("simple") {
     TestUtil.withTestDir { dir =>
-      TestUtil.tryAcquire(dir) { (files, maybeServerChannel) =>
+      val files = TestUtil.lockFiles(dir)
+      TestUtil.tryAcquire(files) { maybeServerChannel =>
         expect(maybeServerChannel.isRight)
-        val canConnect = SocketFile.canConnect(files.socketPaths)
-        expect(canConnect.isRight)
+        val connectRes = Connect.tryConnect(files)
+        expect(connectRes.exists(_.isRight))
       }
+      val after = Connect.tryConnect(files)
+      expect(after.exists(_.isLeft)) // zombie found (our own process, not listening on the socket)
     }
   }
 

tests/test/src/libdaemonjvm/tests/TestUtil.scala

@@ -5,10 +5,14 @@ import java.util.concurrent.atomic.AtomicInteger
 import java.nio.channels.ServerSocketChannel
 
 import libdaemonjvm._
+import libdaemonjvm.internal._
+import libdaemonjvm.server._
 import java.net.ServerSocket
 import scala.util.Properties
 import java.util.concurrent.CountDownLatch
 import java.net.Socket
+import java.util.concurrent.atomic.AtomicBoolean
+import libdaemonjvm.internal.SocketFile
 
 object TestUtil {
   private lazy val testDirBase = {
@@ -61,12 +65,17 @@ object TestUtil {
     var maybeServerChannel: Either[LockError, Either[ServerSocket, ServerSocketChannel]] = null
     var acceptThreadOpt = Option.empty[Thread]
     val accepting       = new CountDownLatch(1)
+    val shouldStop      = new AtomicBoolean(false)
     try {
       maybeServerChannel = Lock.tryAcquire(files, proc)
       if (Properties.isWin)
         // Windows named pipes seem no to accept clients unless accept is being called on the server socket
         acceptThreadOpt =
-          maybeServerChannel.toOption.flatMap(_.left.toOption.map(acceptAndDiscard(_, accepting)))
+          maybeServerChannel.toOption.flatMap(_.left.toOption.map(acceptAndDiscard(
+            _,
+            accepting,
+            () => shouldStop.get()
+          )))
       for (t <- acceptThreadOpt) {
         t.start()
         accepting.await()
@@ -75,14 +84,19 @@ object TestUtil {
       f(maybeServerChannel)
     }
     finally {
+      shouldStop.set(true)
+      SocketFile.canConnect(files.socketPaths) // unblock the server thread last accept
       for (e <- Option(maybeServerChannel); channel <- e)
         channel.merge.close()
-      acceptThreadOpt.foreach(_.interrupt()) // not sure this has an effect... :|
     }
   }
 
   val acceptAndDiscardCount = new AtomicInteger
-  def acceptAndDiscard(s: ServerSocket, accepting: CountDownLatch): Thread =
+  def acceptAndDiscard(
+    s: ServerSocket,
+    accepting: CountDownLatch,
+    shouldStop: () => Boolean
+  ): Thread =
     new Thread(
       s"libdaemonjvm-tests-accept-and-discard-${acceptAndDiscardCount.incrementAndGet()}"
     ) {
@@ -99,7 +113,7 @@ object TestUtil {
       }
       override def run(): Unit = {
         accepting.countDown()
-        while (true) {
+        while (!shouldStop()) {
           val client = s.accept()
           // closing the client socket in the background, as this call seems to block a few seconds
           closeSocket(client)