Framework-less, generalized, annotation-based RPC

build.sbt

@@ -52,7 +52,7 @@ val commonSettings = Seq(
     s"-Xlint:-missing-interpolator,-adapted-args,${if (scalaBinaryVersion.value == "2.12") "-unused," else ""}_",
   ),
   // some Java 8 related tests use Java interface static methods, Scala 2.11.12 requires JDK8 target for that
-  scalacOptions in Test ++= (if(scalaBinaryVersion.value == "2.11") Seq("-target:jvm-1.8") else Seq()),
+  scalacOptions in Test ++= (if (scalaBinaryVersion.value == "2.11") Seq("-target:jvm-1.8") else Seq()),
   apiURL := Some(url("http://avsystem.github.io/scala-commons/api")),
   autoAPIMappings := true,
 
@@ -132,6 +132,7 @@ lazy val commons = project.in(file("."))
   .enablePlugins(ScalaUnidocPlugin)
   .aggregate(
     `commons-annotations`,
+    `commons-annotations-js`,
     `commons-macros`,
     `commons-core`,
     `commons-core-js`,
@@ -151,6 +152,7 @@ lazy val commons = project.in(file("."))
     scalacOptions in ScalaUnidoc in unidoc += "-Ymacro-expand:none",
     unidocProjectFilter in ScalaUnidoc in unidoc :=
       inAnyProject -- inProjects(
+        `commons-annotations-js`,
         `commons-macros`,
         `commons-analyzer`,
         `commons-core-js`,
@@ -159,16 +161,6 @@ lazy val commons = project.in(file("."))
       ),
   )
 
-lazy val `commons-annotations` = project
-  .settings(jvmCommonSettings)
-
-lazy val `commons-macros` = project
-  .dependsOn(`commons-annotations`)
-  .settings(
-    jvmCommonSettings,
-    libraryDependencies += "org.scala-lang" % "scala-reflect" % scalaVersion.value,
-  )
-
 def mkSourceDirs(base: File, scalaBinary: String, conf: String): Seq[File] = Seq(
   base / "src" / conf / "scala",
   base / "src" / conf / s"scala-$scalaBinary",
@@ -182,8 +174,25 @@ def sourceDirsSettings(baseMapper: File => File) = Seq(
     mkSourceDirs(baseMapper(baseDirectory.value), scalaBinaryVersion.value, "test"),
 )
 
+lazy val `commons-annotations` = project
+  .settings(jvmCommonSettings)
+
+lazy val `commons-annotations-js` = project.in(`commons-annotations`.base / "js")
+  .enablePlugins(ScalaJSPlugin)
+  .configure(p => if (forIdeaImport) p.dependsOn(`commons-annotations`) else p)
+  .settings(
+    jsCommonSettings,
+    name := (name in `commons-annotations`).value,
+    sourceDirsSettings(_.getParentFile),
+  )
+
+lazy val `commons-macros` = project.settings(
+  jvmCommonSettings,
+  libraryDependencies += "org.scala-lang" % "scala-reflect" % scalaVersion.value,
+)
+
 lazy val `commons-core` = project
-  .dependsOn(`commons-macros`)
+  .dependsOn(`commons-macros`, `commons-annotations`)
   .settings(
     jvmCommonSettings,
     sourceDirsSettings(_ / "jvm"),
@@ -197,7 +206,7 @@ lazy val `commons-core` = project
 lazy val `commons-core-js` = project.in(`commons-core`.base / "js")
   .enablePlugins(ScalaJSPlugin)
   .configure(p => if (forIdeaImport) p.dependsOn(`commons-core`) else p)
-  .dependsOn(`commons-macros`)
+  .dependsOn(`commons-macros`, `commons-annotations-js`)
   .settings(
     jsCommonSettings,
     name := (name in `commons-core`).value,

commons-akka/src/main/scala/com/avsystem/commons/rpc/akka/AkkaRPCFramework.scala

@@ -7,8 +7,9 @@ import akka.actor.{ActorRef, ActorSystem}
 import akka.util.ByteString
 import com.avsystem.commons.rpc.akka.client.ClientRawRPC
 import com.avsystem.commons.rpc.akka.server.ServerActor
-import com.avsystem.commons.rpc.{FunctionRPCFramework, GetterRPCFramework, ProcedureRPCFramework}
+import com.avsystem.commons.rpc.{FunctionRPCFramework, GetterRPCFramework, MetadataAnnotation, ProcedureRPCFramework, RpcMetadataCompanion, TypedMetadata, infer, multi, reifyAnnot, reifyName, verbatim}
 import com.avsystem.commons.serialization.{GenCodec, StreamInput, StreamOutput}
+import monix.reactive.Observable
 
 /**
   * RPC Framework implemented with Akka as transportation layer.
@@ -17,6 +18,7 @@ import com.avsystem.commons.serialization.{GenCodec, StreamInput, StreamOutput}
   */
 object AkkaRPCFramework extends GetterRPCFramework with ProcedureRPCFramework with FunctionRPCFramework with MonixRPCFramework {
   trait RawRPC extends GetterRawRPC with ProcedureRawRPC with FunctionRawRPC with MonixRawRPC
+  object RawRPC extends BaseRawRpcCompanion
   abstract class FullRPCInfo[T] extends BaseFullRPCInfo[T]
 
   type RawValue = ByteString
@@ -26,6 +28,23 @@ object AkkaRPCFramework extends GetterRPCFramework with ProcedureRPCFramework wi
   type ParamTypeMetadata[T] = DummyImplicit
   type ResultTypeMetadata[T] = DummyImplicit
 
+  case class RPCMetadata[T](
+    @reifyName name: String,
+    @reifyAnnot @multi annotations: List[MetadataAnnotation],
+    @multi @verbatim procedureSignatures: Map[String, ProcedureSignature],
+    @multi functionSignatures: Map[String, FunctionSignature[_]],
+    @multi observeSignatures: Map[String, ObserveSignature[_]],
+    @multi getterSignatures: Map[String, GetterSignature[_]]
+  )
+  object RPCMetadata extends RpcMetadataCompanion[RPCMetadata]
+
+  case class ObserveSignature[T](
+    name: String,
+    paramMetadata: List[ParamMetadata[_]],
+    annotations: List[MetadataAnnotation],
+    @infer resultTypeMetadata: ResultTypeMetadata[T]
+  ) extends Signature with TypedMetadata[Observable[T]]
+
   def read[T: Reader](raw: RawValue): T =
     GenCodec.read[T](new StreamInput(new DataInputStream(raw.iterator.asInputStream)))
   def write[T: Writer](value: T): RawValue = {

commons-akka/src/main/scala/com/avsystem/commons/rpc/akka/MonixRPCFramework.scala

@@ -1,19 +1,18 @@
 package com.avsystem.commons
 package rpc.akka
 
-import com.avsystem.commons.rpc.RPCFramework
+import com.avsystem.commons.rpc._
 import monix.reactive.Observable
 
 trait MonixRPCFramework extends RPCFramework {
   override type RawRPC <: MonixRawRPC
 
-  trait MonixRawRPC {this: RawRPC =>
-    def observe(rpcName: String, argLists: List[List[RawValue]]): Observable[RawValue]
+  trait MonixRawRPC { this: RawRPC =>
+    @multi def observe(rpcName: String)(@multi args: List[RawValue]): Observable[RawValue]
   }
 
-  implicit def ObservableRealHandler[A: Writer]: RealInvocationHandler[Observable[A], Observable[RawValue]] =
-    RealInvocationHandler[Observable[A], Observable[RawValue]](_.map(write[A]))
-
-  implicit def ObservableRawHandler[A: Reader]: RawInvocationHandler[Observable[A]] =
-    RawInvocationHandler[Observable[A]]((rawRpc, rpcName, argLists) => rawRpc.observe(rpcName, argLists).map(read[A]))
+  implicit def readerBasedObservableAsReal[T: Reader]: AsReal[Observable[RawValue], Observable[T]] =
+    AsReal.create(_.map(read[T]))
+  implicit def writerBasedObservableAsRaw[T: Writer]: AsRaw[Observable[RawValue], Observable[T]] =
+    AsRaw.create(_.map(write[T]))
 }

commons-akka/src/main/scala/com/avsystem/commons/rpc/akka/RemoteMessage.scala

@@ -28,16 +28,16 @@ private object RemoteMessage {
   implicit val heatBeatCodec: GenCodec[MonixProtocol.Heartbeat.type] = GenCodec.materialize[MonixProtocol.Heartbeat.type]
 }
 
-private final case class RawInvocation(rpcName: String, argLists: List[List[RawValue]]) extends RemoteMessage
+private final case class RawInvocation(rpcName: String, args: List[RawValue]) extends RemoteMessage
 
 private sealed trait InvocationMessage extends RemoteMessage {
   def getterChain: Seq[RawInvocation]
   def name: String
-  def argLists: List[List[RawValue]]
+  def args: List[RawValue]
 }
-private final case class ProcedureInvocationMessage(name: String, argLists: List[List[RawValue]], getterChain: Seq[RawInvocation]) extends InvocationMessage
-private final case class FunctionInvocationMessage(name: String, argLists: List[List[RawValue]], getterChain: Seq[RawInvocation]) extends InvocationMessage
-private final case class ObservableInvocationMessage(name: String, argLists: List[List[RawValue]], getterChain: Seq[RawInvocation]) extends InvocationMessage
+private final case class ProcedureInvocationMessage(name: String, args: List[RawValue], getterChain: Seq[RawInvocation]) extends InvocationMessage
+private final case class FunctionInvocationMessage(name: String, args: List[RawValue], getterChain: Seq[RawInvocation]) extends InvocationMessage
+private final case class ObservableInvocationMessage(name: String, args: List[RawValue], getterChain: Seq[RawInvocation]) extends InvocationMessage
 
 private sealed trait InvocationResult extends RemoteMessage
 private final case class InvocationSuccess(value: RawValue) extends InvocationResult

commons-akka/src/main/scala/com/avsystem/commons/rpc/akka/client/ClientRawRPC.scala

@@ -4,7 +4,7 @@ package rpc.akka.client
 import akka.actor.ActorSystem
 import akka.pattern.ask
 import akka.util.Timeout
-import com.avsystem.commons.rpc.akka.AkkaRPCFramework.RawRPC
+import com.avsystem.commons.rpc.akka.AkkaRPCFramework.{RawRPC, RawValue}
 import com.avsystem.commons.rpc.akka._
 import monix.execution.Cancelable
 import monix.reactive.{Observable, OverflowStrategy}
@@ -14,12 +14,12 @@ import monix.reactive.{Observable, OverflowStrategy}
   */
 private[akka] final class ClientRawRPC(config: AkkaRPCClientConfig, getterChain: Seq[RawInvocation] = Nil)(implicit system: ActorSystem) extends AkkaRPCFramework.RawRPC {
 
-  override def fire(rpcName: String, argLists: List[List[AkkaRPCFramework.RawValue]]): Unit = {
-    system.actorSelection(config.serverPath) ! ProcedureInvocationMessage(rpcName, argLists, getterChain)
+  override def fire(rpcName: String)(args: List[RawValue]): Unit = {
+    system.actorSelection(config.serverPath) ! ProcedureInvocationMessage(rpcName, args, getterChain)
   }
-  override def call(rpcName: String, argLists: List[List[AkkaRPCFramework.RawValue]]): Future[AkkaRPCFramework.RawValue] = {
+  override def call(rpcName: String)(args: List[RawValue]): Future[RawValue] = {
     implicit val timeout: Timeout = Timeout(config.functionCallTimeout)
-    val future = system.actorSelection(config.serverPath) ? FunctionInvocationMessage(rpcName, argLists, getterChain)
+    val future = system.actorSelection(config.serverPath) ? FunctionInvocationMessage(rpcName, args, getterChain)
 
     import com.avsystem.commons.concurrent.RunNowEC.Implicits.executionContext
 
@@ -29,13 +29,13 @@ private[akka] final class ClientRawRPC(config: AkkaRPCClientConfig, getterChain:
       case value => Future.failed(new IllegalStateException(s"Illegal message type. Should be InvocationResult, but received value was: $value"))
     }
   }
-  override def get(rpcName: String, argLists: List[List[AkkaRPCFramework.RawValue]]): RawRPC =
-    new ClientRawRPC(config, getterChain :+ RawInvocation(rpcName, argLists))
+  override def get(rpcName: String)(args: List[RawValue]): RawRPC =
+    new ClientRawRPC(config, getterChain :+ RawInvocation(rpcName, args))
 
-  override def observe(rpcName: String, argLists: List[List[AkkaRPCFramework.RawValue]]): Observable[AkkaRPCFramework.RawValue] = {
+  override def observe(rpcName: String)(args: List[RawValue]): Observable[RawValue] = {
     Observable.create(OverflowStrategy.Unbounded) { s =>
       val actor = system.actorOf(MonixClientActor.props(s, config))
-      actor ! ObservableInvocationMessage(rpcName, argLists, getterChain)
+      actor ! ObservableInvocationMessage(rpcName, args, getterChain)
       Cancelable.empty // TODO implement proper canceling
     }
 

commons-akka/src/main/scala/com/avsystem/commons/rpc/akka/server/ServerActor.scala

@@ -16,11 +16,11 @@ private final class ServerActor(rawRPC: AkkaRPCFramework.RawRPC, config: AkkaRPC
 
   override def receive: Receive = {
     case msg@ProcedureInvocationMessage(name, argLists, getterChain) =>
-      resolveRpc(msg).fire(name, argLists)
+      resolveRpc(msg).fire(name)(argLists)
     case msg@FunctionInvocationMessage(name, argLists, getterChain) =>
       import com.avsystem.commons.concurrent.RunNowEC.Implicits.executionContext
       val s = sender()
-      resolveRpc(msg).call(name, argLists).onComplete {
+      resolveRpc(msg).call(name)(argLists).onComplete {
         case Success(value) => s ! InvocationSuccess(value)
         case Failure(e) =>
           logError(e, name)
@@ -37,7 +37,7 @@ private final class ServerActor(rawRPC: AkkaRPCFramework.RawRPC, config: AkkaRPC
           Ack.Continue
         }
 
-      resolveRpc(msg).observe(name, argLists).subscribe(
+      resolveRpc(msg).observe(name)(argLists).subscribe(
         value => {
           val result = s ? InvocationSuccess(value)
           result.mapTo[MonixProtocol.RemoteAck].map {
@@ -64,7 +64,8 @@ private final class ServerActor(rawRPC: AkkaRPCFramework.RawRPC, config: AkkaRPC
       )
   }
 
-  private def resolveRpc(msg: InvocationMessage) = rawRPC.resolveGetterChain(msg.getterChain.map(r => AkkaRPCFramework.RawInvocation(r.rpcName, r.argLists)).toList)
+  private def resolveRpc(msg: InvocationMessage) =
+    rawRPC.resolveGetterChain(msg.getterChain.map(r => AkkaRPCFramework.RawInvocation(r.rpcName, r.args)).toList)
 
   private def logError(e: Throwable, methodName: String): Unit = {
     log.error(e,
@@ -77,4 +78,4 @@ private final class ServerActor(rawRPC: AkkaRPCFramework.RawRPC, config: AkkaRPC
 
 private[akka] object ServerActor {
   def props(rawRPC: AkkaRPCFramework.RawRPC, config: AkkaRPCServerConfig): Props = Props(new ServerActor(rawRPC, config))
-}
+}

commons-akka/src/test/scala/com/avsystem/commons/rpc/akka/TestRPC.scala

@@ -1,29 +1,22 @@
 package com.avsystem.commons
 package rpc.akka
 
-import com.avsystem.commons.rpc.RPC
 import com.avsystem.commons.rpc.akka.AkkaRPCFramework._
 import monix.reactive.Observable
 
 
 /**
   * @author Wojciech Milewski
   */
-@RPC
 trait TestRPC {
   def fireAndForget(): Unit
   def echoAsString(int: Int): Future[String]
   def stream: Observable[Int]
   def inner: InnerRPC
 }
-object TestRPC {
-  implicit val fullRPCInfo: BaseFullRPCInfo[TestRPC] = materializeFullInfo
-}
+object TestRPC extends RPCCompanion[TestRPC]
 
-@RPC
 trait InnerRPC {
   def innerFire(): Unit
 }
-object InnerRPC {
-  implicit val fullRPCInfo: BaseFullRPCInfo[InnerRPC] = materializeFullInfo
-}
+object InnerRPC extends RPCCompanion[InnerRPC]

commons-annotations/src/main/scala/com/avsystem/commons/annotation/AnnotationAggregate.scala

@@ -4,20 +4,47 @@ package annotation
 import scala.annotation.StaticAnnotation
 
 /**
-  * When an annotation class extends this trait, annotation processing macros (e.g. for `GenCodec` materialization)
-  * will look into annotations of the aggregating annotation itself and apply these annotations as if they were
-  * applied directly on the same target as the aggregating annotation. Example:
+  * Base trait for annotations which aggregate multiple other annotations. This way annotation aggregates
+  * work like "annotation functions" - they are annotations that yield more annotations.
+  *
+  * In order to specify aggregated annotations, the class that extends `AnnotationAggregate` must
+  * redefine the `Implied` dummy type member and apply the aggregated annotations on it. Macro engines
+  * used in `GenCodec` materialization and RPC framework will automatically pick up these annotations.
   *
   * {{{
   *   import com.avsystem.commons.serialization._
   *
-  *   @name("_id") @outOfOrder
-  *   class mongoId extends AnnotationAggregate
+  *   class mongoId extends AnnotationAggregate {
+  *     @name("_id") @outOfOrder
+  *     type Implied
+  *   }
   *
   *   case class SomeMongoEntity(@mongoId id: String, data: String)
   * }}}
   *
   * In the above example, applying `@mongoId` annotation on the `id` field has the same effect as if
   * annotations `@name("_id") @outOfOrder` were applied directly on that field.
+  *
+  * NOTE: thanks to the fact that aggregated annotations are applied on a type member
+  * you can pass the arguments of original annotation to aggregated annotations, e.g.
+  *
+  * {{{
+  *   class rpcNameAndDescription(name: String, description: String) extends AnnotationAggregate {
+  *     @rpcName(name) // passing `name` to aggregated annotation
+  *     type Implied
+  *   }
+  * }}}
   */
-trait AnnotationAggregate extends StaticAnnotation
+trait AnnotationAggregate extends StaticAnnotation {
+  /**
+    * Dummy type member meant to be redefined in order to have aggregated annotations applied on it.
+    * These annotations will be automatically picked up by macro engines each time they encounter
+    * the aggregating annotation itself.
+    * Other than being an "anchor" for annotations, this type member has no actual meaning and there is no
+    * reason to ever actually use it.
+    * NOTE: a less weird solution would be to put aggregated annotations on the
+    * aggregating annotation class itself, but this would make it impossible to access the arguments
+    * of aggregating annotation in aggregated annotations.
+    */
+  type Implied
+}