Merge pull request #47 from neandertech/globs

Revise endpoint and channel constructs to facilitate proxyfication

Author: Baccata

core/src/jsonrpclib/Endpoint.scala

@@ -6,11 +6,19 @@ sealed trait Endpoint[F[_]] {
 
 object Endpoint {
 
-  def apply[F[_]](method: String): PartiallyAppliedEndpoint[F] = new PartiallyAppliedEndpoint[F](method)
+  type MethodPattern = String
+  type Method = String
 
-  class PartiallyAppliedEndpoint[F[_]](method: String) {
+  def apply[F[_]](method: Method): PartiallyAppliedEndpoint[F] = new PartiallyAppliedEndpoint[F](method)
+
+  class PartiallyAppliedEndpoint[F[_]](method: MethodPattern) {
     def apply[In, Err, Out](
         run: In => F[Either[Err, Out]]
+    )(implicit inCodec: Codec[In], errCodec: ErrorCodec[Err], outCodec: Codec[Out]): Endpoint[F] =
+      RequestResponseEndpoint(method, (_: InputMessage, in: In) => run(in), inCodec, errCodec, outCodec)
+
+    def full[In, Err, Out](
+        run: (InputMessage, In) => F[Either[Err, Out]]
     )(implicit inCodec: Codec[In], errCodec: ErrorCodec[Err], outCodec: Codec[Out]): Endpoint[F] =
       RequestResponseEndpoint(method, run, inCodec, errCodec, outCodec)
 
@@ -25,16 +33,22 @@ object Endpoint {
       )
 
     def notification[In](run: In => F[Unit])(implicit inCodec: Codec[In]): Endpoint[F] =
+      NotificationEndpoint(method, (_: InputMessage, in: In) => run(in), inCodec)
+
+    def notificationFull[In](run: (InputMessage, In) => F[Unit])(implicit inCodec: Codec[In]): Endpoint[F] =
       NotificationEndpoint(method, run, inCodec)
 
   }
 
-  final case class NotificationEndpoint[F[_], In](method: String, run: In => F[Unit], inCodec: Codec[In])
-      extends Endpoint[F]
+  final case class NotificationEndpoint[F[_], In](
+      method: MethodPattern,
+      run: (InputMessage, In) => F[Unit],
+      inCodec: Codec[In]
+  ) extends Endpoint[F]
 
   final case class RequestResponseEndpoint[F[_], In, Err, Out](
-      method: String,
-      run: In => F[Either[Err, Out]],
+      method: Method,
+      run: (InputMessage, In) => F[Either[Err, Out]],
       inCodec: Codec[In],
       errCodec: ErrorCodec[Err],
       outCodec: Codec[Out]

core/src/jsonrpclib/internals/Message.scala renamed to core/src/jsonrpclib/Message.scala

@@ -1,34 +1,32 @@
 package jsonrpclib
-package internals
 
 import com.github.plokhotnyuk.jsoniter_scala.core.JsonReader
 import com.github.plokhotnyuk.jsoniter_scala.core.JsonValueCodec
 import com.github.plokhotnyuk.jsoniter_scala.core.JsonWriter
 
 sealed trait Message { def maybeCallId: Option[CallId] }
-private[jsonrpclib] sealed trait InputMessage extends Message { def method: String }
-private[jsonrpclib] sealed trait OutputMessage extends Message {
+sealed trait InputMessage extends Message { def method: String }
+sealed trait OutputMessage extends Message {
   def callId: CallId; final override def maybeCallId: Option[CallId] = Some(callId)
 }
 
-private[jsonrpclib] object InputMessage {
+object InputMessage {
   case class RequestMessage(method: String, callId: CallId, params: Option[Payload]) extends InputMessage {
     def maybeCallId: Option[CallId] = Some(callId)
   }
   case class NotificationMessage(method: String, params: Option[Payload]) extends InputMessage {
     def maybeCallId: Option[CallId] = None
   }
 }
-
-private[jsonrpclib] object OutputMessage {
+object OutputMessage {
   def errorFrom(callId: CallId, protocolError: ProtocolError): OutputMessage =
     ErrorMessage(callId, ErrorPayload(protocolError.code, protocolError.getMessage(), None))
 
   case class ErrorMessage(callId: CallId, payload: ErrorPayload) extends OutputMessage
   case class ResponseMessage(callId: CallId, data: Payload) extends OutputMessage
 }
 
-private[jsonrpclib] object Message {
+object Message {
 
   implicit val messageJsonValueCodecs: JsonValueCodec[Message] = new JsonValueCodec[Message] {
     val rawMessageCodec = implicitly[JsonValueCodec[internals.RawMessage]]

core/src/jsonrpclib/internals/MessageDispatcher.scala

@@ -1,11 +1,10 @@
 package jsonrpclib
 package internals
 
-import jsonrpclib.internals._
 import jsonrpclib.Endpoint.NotificationEndpoint
 import jsonrpclib.Endpoint.RequestResponseEndpoint
-import jsonrpclib.internals.OutputMessage.ErrorMessage
-import jsonrpclib.internals.OutputMessage.ResponseMessage
+import jsonrpclib.OutputMessage.ErrorMessage
+import jsonrpclib.OutputMessage.ResponseMessage
 import scala.util.Try
 
 private[jsonrpclib] abstract class MessageDispatcher[F[_]](implicit F: Monadic[F]) extends Channel.MonadicChannel[F] {
@@ -41,8 +40,8 @@ private[jsonrpclib] abstract class MessageDispatcher[F[_]](implicit F: Monadic[F
       }
   }
 
-  protected[jsonrpclib] def handleReceivedPayload(payload: Payload): F[Unit] = {
-    Codec.decode[Message](Some(payload)).map {
+  protected[jsonrpclib] def handleReceivedMessage(message: Message): F[Unit] = {
+    message match {
       case im: InputMessage =>
         doFlatMap(getEndpoint(im.method)) {
           case Some(ep) => background(im.maybeCallId, executeInputMessage(im, ep))
@@ -61,29 +60,25 @@ private[jsonrpclib] abstract class MessageDispatcher[F[_]](implicit F: Monadic[F
           case Some(pendingCall) => pendingCall(om)
           case None              => doPure(()) // TODO do something
         }
-    } match {
-      case Left(error) =>
-        sendProtocolError(error)
-      case Right(dispatch) => dispatch
     }
   }
 
-  private def sendProtocolError(callId: CallId, pError: ProtocolError): F[Unit] =
+  protected def sendProtocolError(callId: CallId, pError: ProtocolError): F[Unit] =
     sendMessage(OutputMessage.errorFrom(callId, pError))
-  private def sendProtocolError(pError: ProtocolError): F[Unit] =
+  protected def sendProtocolError(pError: ProtocolError): F[Unit] =
     sendProtocolError(CallId.NullId, pError)
 
   private def executeInputMessage(input: InputMessage, endpoint: Endpoint[F]): F[Unit] = {
     (input, endpoint) match {
       case (InputMessage.NotificationMessage(_, params), ep: NotificationEndpoint[F, in]) =>
         ep.inCodec.decode(params) match {
-          case Right(value) => ep.run(value)
+          case Right(value) => ep.run(input, value)
           case Left(value)  => reportError(params, value, ep.method)
         }
       case (InputMessage.RequestMessage(_, callId, params), ep: RequestResponseEndpoint[F, in, err, out]) =>
         ep.inCodec.decode(params) match {
           case Right(value) =>
-            doFlatMap(ep.run(value)) {
+            doFlatMap(ep.run(input, value)) {
               case Right(data) =>
                 val responseData = ep.outCodec.encode(data)
                 sendMessage(OutputMessage.ResponseMessage(callId, responseData))

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

@@ -41,8 +41,8 @@ object ClientMain extends IOApp.Simple {
       // Creating a channel that will be used to communicate to the server
       fs2Channel <- FS2Channel[IO](cancelTemplate = cancelEndpoint.some)
       _ <- Stream(())
-        .concurrently(fs2Channel.output.through(lsp.encodePayloads).through(rp.stdin))
-        .concurrently(rp.stdout.through(lsp.decodePayloads).through(fs2Channel.input))
+        .concurrently(fs2Channel.output.through(lsp.encodeMessages).through(rp.stdin))
+        .concurrently(rp.stdout.through(lsp.decodeMessages).through(fs2Channel.inputOrBounce))
         .concurrently(rp.stderr.through(fs2.io.stderr[IO]))
       // Creating a `IntWrapper => IO[IntWrapper]` stub that can call the server
       increment = fs2Channel.simpleStub[IntWrapper, IntWrapper]("increment")

examples/server/src/examples/server/ServerMain.scala

@@ -34,8 +34,8 @@ object ServerMain extends IOApp.Simple {
         .flatMap(channel =>
           fs2.Stream
             .eval(IO.never) // running the server forever
-            .concurrently(stdin[IO](512).through(lsp.decodePayloads).through(channel.input))
-            .concurrently(channel.output.through(lsp.encodePayloads).through(stdout[IO]))
+            .concurrently(stdin[IO](512).through(lsp.decodeMessages).through(channel.inputOrBounce))
+            .concurrently(channel.output.through(lsp.encodeMessages).through(stdout[IO]))
         )
         .compile
         .drain

fs2/src/jsonrpclib/fs2/FS2Channel.scala

@@ -13,14 +13,15 @@ import cats.effect.std.Supervisor
 import cats.syntax.all._
 import cats.effect.syntax.all._
 import jsonrpclib.internals.MessageDispatcher
-import jsonrpclib.internals._
 
 import scala.util.Try
+import java.util.regex.Pattern
 
 trait FS2Channel[F[_]] extends Channel[F] {
 
-  def input: Pipe[F, Payload, Unit]
-  def output: Stream[F, Payload]
+  def input: Pipe[F, Message, Unit]
+  def inputOrBounce: Pipe[F, Either[ProtocolError, Message], Unit]
+  def output: Stream[F, Message]
 
   def withEndpoint(endpoint: Endpoint[F])(implicit F: Functor[F]): Resource[F, FS2Channel[F]] =
     Resource.make(mountEndpoint(endpoint))(_ => unmountEndpoint(endpoint.method)).map(_ => this)
@@ -52,8 +53,8 @@ object FS2Channel {
   ): Stream[F, FS2Channel[F]] = {
     for {
       supervisor <- Stream.resource(Supervisor[F])
-      ref <- Ref[F].of(State[F](Map.empty, Map.empty, Map.empty, 0)).toStream
-      queue <- cats.effect.std.Queue.bounded[F, Payload](bufferSize).toStream
+      ref <- Ref[F].of(State[F](Map.empty, Map.empty, Map.empty, Vector.empty, 0)).toStream
+      queue <- cats.effect.std.Queue.bounded[F, Message](bufferSize).toStream
       impl = new Impl(queue, ref, supervisor, cancelTemplate)
 
       // Creating a bespoke endpoint to receive cancelation requests
@@ -73,6 +74,7 @@ object FS2Channel {
       runningCalls: Map[CallId, Fiber[F, Throwable, Unit]],
       pendingCalls: Map[CallId, OutputMessage => F[Unit]],
       endpoints: Map[String, Endpoint[F]],
+      globEndpoints: Vector[(Pattern, Endpoint[F])],
       counter: Long
   ) {
     def nextCallId: (State[F], CallId) = (this.copy(counter = counter + 1), CallId.NumberId(counter))
@@ -82,11 +84,27 @@ object FS2Channel {
       val result = pendingCalls.get(callId)
       (this.copy(pendingCalls = pendingCalls.removed(callId)), result)
     }
-    def mountEndpoint(endpoint: Endpoint[F]): Either[ConflictingMethodError, State[F]] =
-      endpoints.get(endpoint.method) match {
-        case None    => Right(this.copy(endpoints = endpoints + (endpoint.method -> endpoint)))
-        case Some(_) => Left(ConflictingMethodError(endpoint.method))
+    def mountEndpoint(endpoint: Endpoint[F]): Either[ConflictingMethodError, State[F]] = {
+      import endpoint.method
+      if (method.contains("*")) {
+        val parts = method
+          .split("\\*", -1)
+          .map { // Don't discard trailing empty string, if any.
+            case ""  => ""
+            case str => Pattern.quote(str)
+          }
+        val glob = Pattern.compile(parts.mkString(".*"))
+        Right(this.copy(globEndpoints = globEndpoints :+ (glob -> endpoint)))
+      } else {
+        endpoints.get(endpoint.method) match {
+          case None    => Right(this.copy(endpoints = endpoints + (endpoint.method -> endpoint)))
+          case Some(_) => Left(ConflictingMethodError(endpoint.method))
+        }
       }
+    }
+    def getEndpoint(method: String): Option[Endpoint[F]] = {
+      endpoints.get(method).orElse(globEndpoints.find(_._1.matcher(method).matches()).map(_._2))
+    }
     def removeEndpoint(method: String): State[F] =
       copy(endpoints = endpoints.removed(method))
 
@@ -98,16 +116,20 @@ object FS2Channel {
   }
 
   private class Impl[F[_]](
-      private val queue: cats.effect.std.Queue[F, Payload],
+      private val queue: cats.effect.std.Queue[F, Message],
       private val state: Ref[F, FS2Channel.State[F]],
       supervisor: Supervisor[F],
       maybeCancelTemplate: Option[CancelTemplate]
   )(implicit F: Concurrent[F])
       extends MessageDispatcher[F]
       with FS2Channel[F] {
 
-    def output: Stream[F, Payload] = Stream.fromQueueUnterminated(queue)
-    def input: Pipe[F, Payload, Unit] = _.evalMap(handleReceivedPayload)
+    def output: Stream[F, Message] = Stream.fromQueueUnterminated(queue)
+    def inputOrBounce: Pipe[F, Either[ProtocolError, Message], Unit] = _.evalMap {
+      case Left(error)    => sendProtocolError(error)
+      case Right(message) => handleReceivedMessage(message)
+    }
+    def input: Pipe[F, Message, Unit] = _.evalMap(handleReceivedMessage)
 
     def mountEndpoint(endpoint: Endpoint[F]): F[Unit] = state
       .modify(s =>
@@ -135,8 +157,8 @@ object FS2Channel {
           }
       }
     protected def reportError(params: Option[Payload], error: ProtocolError, method: String): F[Unit] = ???
-    protected def getEndpoint(method: String): F[Option[Endpoint[F]]] = state.get.map(_.endpoints.get(method))
-    protected def sendMessage(message: Message): F[Unit] = queue.offer(Codec.encode(message))
+    protected def getEndpoint(method: String): F[Option[Endpoint[F]]] = state.get.map(_.getEndpoint(method))
+    protected def sendMessage(message: Message): F[Unit] = queue.offer(message)
 
     protected def nextCallId(): F[CallId] = state.modify(_.nextCallId)
     protected def createPromise[A](callId: CallId): F[(Try[A] => F[Unit], () => F[A])] = Deferred[F, Try[A]].map {

fs2/src/jsonrpclib/fs2/lsp.scala

@@ -6,15 +6,26 @@ import fs2.Chunk
 import fs2.Stream
 import fs2.Pipe
 import jsonrpclib.Payload
+import jsonrpclib.Codec
 
 import java.nio.charset.Charset
 import java.nio.charset.StandardCharsets
+import jsonrpclib.Message
+import jsonrpclib.ProtocolError
 
 object lsp {
 
+  def encodeMessages[F[_]]: Pipe[F, Message, Byte] =
+    (_: Stream[F, Message]).map(Codec.encode(_)).through(encodePayloads)
+
   def encodePayloads[F[_]]: Pipe[F, Payload, Byte] =
     (_: Stream[F, Payload]).map(writeChunk).flatMap(Stream.chunk(_))
 
+  def decodeMessages[F[_]: MonadThrow]: Pipe[F, Byte, Either[ProtocolError, Message]] =
+    (_: Stream[F, Byte]).through(decodePayloads).map { payload =>
+      Codec.decode[Message](Some(payload))
+    }
+
   /** Split a stream of bytes into payloads by extracting each frame based on information contained in the headers.
     *
     * See https://microsoft.github.io/language-server-protocol/specifications/lsp/3.17/specification/#contentPart

fs2/test/src/jsonrpclib/fs2/FS2ChannelSpec.scala

@@ -54,6 +54,32 @@ object FS2ChannelSpec extends SimpleIOSuite {
     }
   }
 
+  testRes("Round trip (glob)") {
+    val endpoint: Endpoint[IO] = Endpoint[IO]("**").simple((int: IntWrapper) => IO(IntWrapper(int.int + 1)))
+
+    for {
+      clientSideChannel <- setup(endpoint)
+      remoteFunction = clientSideChannel.simpleStub[IntWrapper, IntWrapper]("inc/test")
+      result <- remoteFunction(IntWrapper(1)).toStream
+    } yield {
+      expect.same(result, IntWrapper(2))
+    }
+  }
+
+  testRes("Globs have lower priority than strict endpoints") {
+    val endpoint: Endpoint[IO] = Endpoint[IO]("inc").simple((int: IntWrapper) => IO(IntWrapper(int.int + 1)))
+    val globEndpoint: Endpoint[IO] =
+      Endpoint[IO]("**").simple((_: IntWrapper) => IO.raiseError[IntWrapper](new Throwable("Boom")))
+
+    for {
+      clientSideChannel <- setup(globEndpoint, endpoint)
+      remoteFunction = clientSideChannel.simpleStub[IntWrapper, IntWrapper]("inc")
+      result <- remoteFunction(IntWrapper(1)).toStream
+    } yield {
+      expect.same(result, IntWrapper(2))
+    }
+  }
+
   testRes("Endpoint not mounted") {
 
     for {