Tweak Lock.tryAcquire API

Author: alexarchambault

library/src/libdaemonjvm/server/Lock.scala

@@ -10,18 +10,20 @@ import java.net.ServerSocket
 
 object Lock {
 
-  def tryAcquire[T](files: LockFiles)
-    : Either[LockError, Either[ServerSocket, ServerSocketChannel]] =
-    tryAcquire(files, LockProcess.default, SocketHandler.server(files.socketPaths))
   def tryAcquire[T](
-    files: LockFiles,
-    proc: LockProcess
-  ): Either[LockError, Either[ServerSocket, ServerSocketChannel]] =
-    tryAcquire(files, proc, SocketHandler.server(files.socketPaths))
+    files: LockFiles
+  )(
+    startListening: Either[ServerSocket, ServerSocketChannel] => T
+  ): Either[LockError, T] =
+    tryAcquire(files, LockProcess.default) {
+      val socket = SocketHandler.server(files.socketPaths)
+      startListening(socket)
+    }
 
   def tryAcquire[T](
     files: LockFiles,
-    proc: LockProcess,
+    proc: LockProcess
+  )(
     setup: => T
   ): Either[LockError, T] = {
 

manual/server/src/libdaemonjvm/TestServer.scala

@@ -1,6 +1,6 @@
 package libdaemonjvm
 
-import java.io.IOException
+import java.io.{Closeable, IOException}
 import java.nio.file.{Files, Paths}
 import java.nio.file.attribute.PosixFilePermission
 import java.util.concurrent.atomic.AtomicInteger
@@ -15,6 +15,26 @@ import libdaemonjvm.server.Lock
 object TestServer {
   val delay          = 2.seconds
   def runTestClients = false
+
+  def runServer(incomingConn: () => Closeable): Unit = {
+    val count = new AtomicInteger
+    while (true) {
+      println("Waiting for clients")
+      val c   = incomingConn()
+      val idx = count.incrementAndGet()
+      val runnable: Runnable = { () =>
+        println(s"New incoming connection $idx, closing it in $delay")
+        Thread.sleep(delay.toMillis)
+        println(s"Closing incoming connection $idx")
+        c.close()
+        println(s"Closed incoming connection $idx")
+      }
+      val t = new Thread(runnable)
+      t.start()
+      Thread.sleep(1000L) // meh, wait for server to be actually listening
+    }
+  }
+
   def main(args: Array[String]): Unit = {
     val path = Paths.get("data-dir")
     if (!Properties.isWin) {
@@ -29,10 +49,9 @@ object TestServer {
       )
     }
     val files = LockFiles.under(path, "libdaemonjvm\\test-server-client\\pipe")
-    val incomingConn = Lock.tryAcquire(files) match {
-      case Left(e)         => throw e
-      case Right(Left(s))  => () => s.accept()
-      case Right(Right(s)) => () => s.accept()
+    Lock.tryAcquire(files)(s => runServer(() => s.fold(_.accept(), _.accept()))) match {
+      case Left(e)   => throw e
+      case Right(()) =>
     }
 
     def clientRunnable(idx: Int): Runnable = { () =>
@@ -59,22 +78,5 @@ object TestServer {
       runClient(3)
       runClient(4)
     }
-
-    val count = new AtomicInteger
-    while (true) {
-      println("Waiting for clients")
-      val c   = incomingConn()
-      val idx = count.incrementAndGet()
-      val runnable: Runnable = { () =>
-        println(s"New incoming connection $idx, closing it in $delay")
-        Thread.sleep(delay.toMillis)
-        println(s"Closing incoming connection $idx")
-        c.close()
-        println(s"Closed incoming connection $idx")
-      }
-      val t = new Thread(runnable)
-      t.setDaemon(true)
-      t.start()
-    }
   }
 }

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

@@ -68,24 +68,30 @@ object TestUtil {
     files: LockFiles,
     proc: LockProcess
   )(f: Either[LockError, Either[ServerSocket, ServerSocketChannel]] => T): T = {
-    var maybeServerChannel: Either[LockError, Either[ServerSocket, ServerSocketChannel]] = null
-    var acceptThreadOpt = Option.empty[Thread]
-    val accepting       = new CountDownLatch(1)
-    val shouldStop      = new AtomicBoolean(false)
+    var serverChannel: 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,
-            () => shouldStop.get()
-          )))
-      for (t <- acceptThreadOpt) {
-        t.start()
-        accepting.await()
-        Thread.sleep(1000L) // waiting so that the accept call below effectively awaits client... :|
+      val maybeServerChannel = Lock.tryAcquire(files, proc) {
+        serverChannel = SocketHandler.server(files.socketPaths)
+        if (Properties.isWin)
+          // Windows named pipes seem no to accept clients unless accept is being called on the server socket
+          acceptThreadOpt =
+            serverChannel.left.toOption.map(acceptAndDiscard(
+              _,
+              accepting,
+              () => shouldStop.get()
+            ))
+        for (t <- acceptThreadOpt) {
+          t.start()
+          accepting.await()
+          // waiting so that the accept call below effectively awaits client... :|
+          Thread.sleep(
+            1000L
+          )
+        }
+        serverChannel
       }
       f(maybeServerChannel)
     }
@@ -96,7 +102,7 @@ object TestUtil {
         case NonFatal(e) =>
           System.err.println(s"Ignoring $e while trying to unblock last accept")
       }
-      for (e <- Option(maybeServerChannel); channel <- e)
+      for (channel <- Option(serverChannel))
         channel.merge.close()
     }
   }