Remove deps

Author: Baccata

build.sc

@@ -86,7 +86,7 @@ object examples extends mill.define.Module {
  }
 
  object client extends ScalaModule {
- def ivyDeps = Agg(ivy"co.fs2::fs2-io:${versions.fs2}", ivy"eu.monniot::fs2-process:0.4.4")
+ def ivyDeps = Agg(ivy"co.fs2::fs2-io:${versions.fs2}")
  def moduleDeps = Seq(fs2.jvm(versions.scala213))
  def scalaVersion = versions.scala213Version
  def forkEnv: Target[Map[String, String]] = T {

examples/client/src/examples/client/ChildProcess.scala

@@ -0,0 +1,66 @@
+package examples.client
+
+import fs2.Stream
+import cats.effect._
+import cats.syntax.all._
+import scala.jdk.CollectionConverters._
+import java.io.OutputStream
+
+trait ChildProcess[F[_]] {
+ def stdin: fs2.Pipe[F, Byte, Unit]
+ def stdout: Stream[F, Byte]
+ def stderr: Stream[F, Byte]
+}
+
+object ChildProcess {
+
+ def spawn[F[_]: Async](command: String*): Stream[F, ChildProcess[F]] =
+ Stream.bracket(start[F](command))(_._2).map(_._1)
+
+ val readBufferSize = 512
+ private def start[F[_]: Async](command: Seq[String]) = Async[F].interruptible {
+ val p =
+ new java.lang.ProcessBuilder(command.asJava)
+ .start() // .directory(new java.io.File(wd)).start()
+ val done = Async[F].fromCompletableFuture(Sync[F].delay(p.onExit()))
+
+ val terminate: F[Unit] = Sync[F].interruptible(p.destroy())
+
+ import cats._
+ val onGlobal = new (F ~> F) {
+ def apply[A](fa: F[A]): F[A] = Async[F].evalOn(fa, scala.concurrent.ExecutionContext.global)
+ }
+
+ val cp = new ChildProcess[F] {
+ def stdin: fs2.Pipe[F, Byte, Unit] =
+ writeOutputStreamFlushingChunks[F](Sync[F].interruptible(p.getOutputStream()))
+
+ def stdout: fs2.Stream[F, Byte] = fs2.io
+ .readInputStream[F](Sync[F].interruptible(p.getInputStream()), chunkSize = readBufferSize).translate(onGlobal)
+
+ def stderr: fs2.Stream[F, Byte] = fs2.io
+ .readInputStream[F](Sync[F].blocking(p.getErrorStream()), chunkSize = readBufferSize).translate(onGlobal)
+ // Avoids broken pipe - we cut off when the program ends.
+ // Users can decide what to do with the error logs using the exitCode value
+ .interruptWhen(done.void.attempt)
+ }
+ (cp, terminate)
+ }
+
+ /** Adds a flush after each chunk
+ */
+ def writeOutputStreamFlushingChunks[F[_]](
+ fos: F[OutputStream],
+ closeAfterUse: Boolean = true
+ )(implicit F: Sync[F]): fs2.Pipe[F, Byte, Nothing] =
+ s => {
+ def useOs(os: OutputStream): Stream[F, Nothing] =
+ s.chunks.foreach(c => F.interruptible(os.write(c.toArray)) >> F.blocking(os.flush()))
+
+ val os =
+ if (closeAfterUse) Stream.bracket(fos)(os => F.blocking(os.close()))
+ else Stream.eval(fos)
+ os.flatMap(os => useOs(os) ++ Stream.exec(F.blocking(os.flush())))
+ }
+
+}

examples/client/src/examples/client/ClientMain.scala

@@ -11,11 +11,10 @@ import cats.syntax.all._
 import fs2.Stream
 import jsonrpclib.StubTemplate
 import cats.effect.std.Dispatcher
-import scala.sys.process.ProcessIO
 import cats.effect.implicits._
-import scala.sys.process.{Process => SProcess}
 import java.io.OutputStream
 import java.io.InputStream
+import examples.client.ChildProcess
 
 object ClientMain extends IOApp.Simple {
 
@@ -38,69 +37,22 @@ object ClientMain extends IOApp.Simple {
  _ <- log("Starting client")
  serverJar <- sys.env.get("SERVER_JAR").liftTo[IOStream](new Exception("SERVER_JAR env var does not exist"))
  // Starting the server
- (serverStdin, serverStdout, serverStderr) <- Stream.resource(process("java", "-jar", serverJar))
- pipeErrors = serverStderr.through(fs2.io.stderr)
+ rp <- ChildProcess.spawn[IO]("java", "-jar", serverJar)
  // Creating a channel that will be used to communicate to the server
  fs2Channel <- FS2Channel
- .lspCompliant[IO](serverStdout, serverStdin, cancelTemplate = cancelEndpoint.some)
- .concurrently(pipeErrors)
+ .lspCompliant[IO](rp.stdout, rp.stdin, cancelTemplate = cancelEndpoint.some)
+ .concurrently(rp.stderr.through(fs2.io.stderr))
  // Opening the stream to be able to send and receive data
  _ <- fs2Channel.openStream
  // Creating a `IntWrapper => IO[IntWrapper]` stub that can call the server
  increment = fs2Channel.simpleStub[IntWrapper, IntWrapper]("increment")
- result <- Stream.eval(increment(IntWrapper(0)))
- _ <- log(s"Client received $result")
+ result1 <- Stream.eval(increment(IntWrapper(0)))
+ _ <- log(s"Client received $result1")
+ result2 <- Stream.eval(increment(result1))
+ _ <- log(s"Client received $result2")
  _ <- log("Terminating client")
  } yield ()
- run.compile.drain.timeout(2.second)
+ run.compile.drain
  }
 
- /** Wraps the spawning of a subprocess into fs2 friendly semantics
- */
- import scala.concurrent.duration._
- def process(command: String*) = for {
- dispatcher <- Dispatcher[IO]
- stdinPromise <- IO.deferred[fs2.Pipe[IO, Byte, Unit]].toResource
- stdoutPromise <- IO.deferred[fs2.Stream[IO, Byte]].toResource
- stderrPromise <- IO.deferred[fs2.Stream[IO, Byte]].toResource
- makeProcessBuilder = IO(sys.process.stringSeqToProcess(command))
- makeProcessIO = IO(
- new ProcessIO(
- in = { (outputStream: OutputStream) =>
- val pipe = writeOutputStreamFlushingChunks(IO(outputStream))
- val fulfil = stdinPromise.complete(pipe)
- dispatcher.unsafeRunSync(fulfil)
- },
- out = { (inputStream: InputStream) =>
- val stream = fs2.io.readInputStream(IO(inputStream), 512)
- val fulfil = stdoutPromise.complete(stream)
- dispatcher.unsafeRunSync(fulfil)
- },
- err = { (inputStream: InputStream) =>
- val stream = fs2.io.readInputStream(IO(inputStream), 512)
- val fulfil = stderrPromise.complete(stream)
- dispatcher.unsafeRunSync(fulfil)
- }
- )
- )
- makeProcess = (makeProcessBuilder, makeProcessIO).flatMapN { case (b, io) => IO.blocking(b.run(io)) }
- _ <- Resource.make(makeProcess)((runningProcess) => IO.blocking(runningProcess.destroy()))
- pipes <- (stdinPromise.get, stdoutPromise.get, stderrPromise.get).tupled.toResource
- } yield pipes
-
- /** Adds a flush after each chunk
- */
- def writeOutputStreamFlushingChunks[F[_]](
- fos: F[OutputStream],
- closeAfterUse: Boolean = true
- )(implicit F: Sync[F]): fs2.Pipe[F, Byte, Nothing] =
- s => {
- def useOs(os: OutputStream): Stream[F, Nothing] =
- s.chunks.foreach(c => F.interruptible(os.write(c.toArray)) >> F.blocking(os.flush()))
-
- val os =
- if (closeAfterUse) Stream.bracket(fos)(os => F.blocking(os.close()))
- else Stream.eval(fos)
- os.flatMap(os => useOs(os) ++ Stream.exec(F.blocking(os.flush())))
- }
 }