Merge pull request #78 from AVSystem/rpc-rest

RPC based REST

Author: ghik

.travis.yml

@@ -1,7 +1,7 @@
 language: scala
 
 scala:
-  - 2.11.11
+  - 2.11.12
   - 2.12.6
 
 jdk:

README.md

@@ -14,6 +14,7 @@
     derivation for case classes and sealed hierarchies
   * [RPC framework](docs/RPCFramework.md): **typesafe** RPC/proxy framework used in particular by
     [Udash Framework](http://guide.udash.io/#/rpc) for client-server communication
+  * [REST framework](docs/REST.md) based on RPC framework
   * Better enumeration support for Scala -
     [`ValueEnum`](http://avsystem.github.io/scala-commons/api/com/avsystem/commons/misc/ValueEnum.html),
     [`SealedEnumCompanion`](http://avsystem.github.io/scala-commons/api/com/avsystem/commons/misc/SealedEnumCompanion.html),

build.sbt

@@ -9,7 +9,7 @@ cancelable in Global := true
 val forIdeaImport = System.getProperty("idea.managed", "false").toBoolean && System.getProperty("idea.runid") == null
 
 // for binary compatibility checking
-val previousVersion = "1.27.3"
+val previousCompatibleVersions = Set.empty[String]
 
 val silencerVersion = "1.1"
 val guavaVersion = "23.0"
@@ -100,8 +100,8 @@ val jvmCommonSettings = commonSettings ++ Seq(
   libraryDependencies ++= Seq(
     "org.apache.commons" % "commons-io" % commonsIoVersion % Test,
   ),
-  mimaPreviousArtifacts := {
-    Set(organization.value % s"${name.value}_${scalaBinaryVersion.value}" % previousVersion)
+  mimaPreviousArtifacts := previousCompatibleVersions.map { previousVersion =>
+    organization.value % s"${name.value}_${scalaBinaryVersion.value}" % previousVersion
   },
 )
 
@@ -192,7 +192,7 @@ lazy val `commons-macros` = project.settings(
 )
 
 lazy val `commons-core` = project
-  .dependsOn(`commons-macros`, `commons-annotations`)
+  .dependsOn(`commons-macros`, `commons-annotations` % CompileAndTest)
   .settings(
     jvmCommonSettings,
     sourceDirsSettings(_ / "jvm"),
@@ -206,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`, `commons-annotations-js`)
+  .dependsOn(`commons-macros`, `commons-annotations-js` % CompileAndTest)
   .settings(
     jsCommonSettings,
     name := (name in `commons-core`).value,
@@ -229,13 +229,16 @@ lazy val `commons-analyzer` = project
   )
 
 lazy val `commons-jetty` = project
-  .dependsOn(`commons-core`)
+  .dependsOn(`commons-core` % CompileAndTest)
   .settings(
     jvmCommonSettings,
     libraryDependencies ++= Seq(
       "org.eclipse.jetty" % "jetty-client" % jettyVersion,
       "org.eclipse.jetty" % "jetty-server" % jettyVersion,
       "com.typesafe.scala-logging" %% "scala-logging" % scalaLoggingVersion,
+
+      "org.eclipse.jetty" % "jetty-servlet" % jettyVersion % Test,
+      "org.slf4j" % "slf4j-simple" % "1.7.25" % Test,
     ),
   )
 
@@ -283,7 +286,7 @@ lazy val `commons-benchmark-js` = project.in(`commons-benchmark`.base / "js")
   )
 
 lazy val `commons-mongo` = project
-  .dependsOn(`commons-core`)
+  .dependsOn(`commons-core` % CompileAndTest)
   .settings(
     jvmCommonSettings,
     libraryDependencies ++= Seq(
@@ -297,7 +300,7 @@ lazy val `commons-mongo` = project
   )
 
 lazy val `commons-kafka` = project
-  .dependsOn(`commons-core`)
+  .dependsOn(`commons-core` % CompileAndTest)
   .settings(
     jvmCommonSettings,
     libraryDependencies ++= Seq(
@@ -306,7 +309,7 @@ lazy val `commons-kafka` = project
   )
 
 lazy val `commons-redis` = project
-  .dependsOn(`commons-core`)
+  .dependsOn(`commons-core` % CompileAndTest)
   .settings(
     jvmCommonSettings,
     libraryDependencies ++= Seq(
@@ -318,7 +321,7 @@ lazy val `commons-redis` = project
   )
 
 lazy val `commons-spring` = project
-  .dependsOn(`commons-core`)
+  .dependsOn(`commons-core` % CompileAndTest)
   .settings(
     jvmCommonSettings,
     libraryDependencies ++= Seq(
@@ -328,7 +331,7 @@ lazy val `commons-spring` = project
   )
 
 lazy val `commons-akka` = project
-  .dependsOn(`commons-core`)
+  .dependsOn(`commons-core` % CompileAndTest)
   .settings(
     jvmCommonSettings,
     libraryDependencies ++= Seq(

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

@@ -8,7 +8,7 @@ trait MonixRPCFramework extends RPCFramework {
   override type RawRPC <: MonixRawRPC
 
   trait MonixRawRPC { this: RawRPC =>
-    @multi def observe(rpcName: String)(@multi args: List[RawValue]): Observable[RawValue]
+    @multi def observe(@composite invocation: RawInvocation): Observable[RawValue]
   }
 
   implicit def readerBasedObservableAsReal[T: Reader]: AsReal[Observable[RawValue], Observable[T]] =

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

@@ -2,7 +2,7 @@ package com.avsystem.commons
 package rpc.akka
 
 import akka.util.ByteString
-import com.avsystem.commons.rpc.akka.AkkaRPCFramework.RawValue
+import com.avsystem.commons.rpc.akka.AkkaRPCFramework._
 import com.avsystem.commons.serialization.GenCodec
 
 /**
@@ -11,8 +11,8 @@ import com.avsystem.commons.serialization.GenCodec
 private sealed trait RemoteMessage extends Serializable
 
 private object RemoteMessage {
-  implicit val byteStringCodec = GenCodec.create[ByteString](input => ByteString(input.readBinary()), (output, byteString) => output.writeBinary(byteString.toArray))
-  implicit val rawInvocationCodec = GenCodec.materialize[RawInvocation]
+  implicit val byteStringCodec: GenCodec[ByteString] =
+    GenCodec.create[ByteString](input => ByteString(input.readBinary()), (output, byteString) => output.writeBinary(byteString.toArray))
 
   implicit val procedureInvocationMessageCodec: GenCodec[ProcedureInvocationMessage] = GenCodec.materialize[ProcedureInvocationMessage]
   implicit val functionInvocationMessageCodec: GenCodec[FunctionInvocationMessage] = GenCodec.materialize[FunctionInvocationMessage]
@@ -28,16 +28,13 @@ private object RemoteMessage {
   implicit val heatBeatCodec: GenCodec[MonixProtocol.Heartbeat.type] = GenCodec.materialize[MonixProtocol.Heartbeat.type]
 }
 
-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 args: List[RawValue]
+  def invocation: RawInvocation
 }
-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 final case class ProcedureInvocationMessage(invocation: RawInvocation, getterChain: Seq[RawInvocation]) extends InvocationMessage
+private final case class FunctionInvocationMessage(invocation: RawInvocation, getterChain: Seq[RawInvocation]) extends InvocationMessage
+private final case class ObservableInvocationMessage(invocation: RawInvocation, getterChain: Seq[RawInvocation]) extends InvocationMessage
 
 private sealed trait InvocationResult extends RemoteMessage
 private final case class InvocationSuccess(value: RawValue) extends InvocationResult
@@ -52,4 +49,4 @@ private object MonixProtocol {
   case object StreamCompleted extends RemoteMessage
 
   case object Heartbeat extends RemoteMessage
-}
+}

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, RawValue}
+import com.avsystem.commons.rpc.akka.AkkaRPCFramework._
 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)(args: List[RawValue]): Unit = {
-    system.actorSelection(config.serverPath) ! ProcedureInvocationMessage(rpcName, args, getterChain)
+  override def fire(invocation: RawInvocation): Unit = {
+    system.actorSelection(config.serverPath) ! ProcedureInvocationMessage(invocation, getterChain)
   }
-  override def call(rpcName: String)(args: List[RawValue]): Future[RawValue] = {
+  override def call(invocation: RawInvocation): Future[RawValue] = {
     implicit val timeout: Timeout = Timeout(config.functionCallTimeout)
-    val future = system.actorSelection(config.serverPath) ? FunctionInvocationMessage(rpcName, args, getterChain)
+    val future = system.actorSelection(config.serverPath) ? FunctionInvocationMessage(invocation, 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)(args: List[RawValue]): RawRPC =
-    new ClientRawRPC(config, getterChain :+ RawInvocation(rpcName, args))
+  override def get(invocation: RawInvocation): RawRPC =
+    new ClientRawRPC(config, getterChain :+ invocation)
 
-  override def observe(rpcName: String)(args: List[RawValue]): Observable[RawValue] = {
+  override def observe(invocation: RawInvocation): Observable[RawValue] = {
     Observable.create(OverflowStrategy.Unbounded) { s =>
       val actor = system.actorOf(MonixClientActor.props(s, config))
-      actor ! ObservableInvocationMessage(rpcName, args, getterChain)
+      actor ! ObservableInvocationMessage(invocation, getterChain)
       Cancelable.empty // TODO implement proper canceling
     }
 

commons-akka/src/main/scala/com/avsystem/commons/rpc/akka/serialization/RemoteMessageSerializer.scala

@@ -26,7 +26,6 @@ final class RemoteMessageSerializer extends Serializer {
       case c if c == classOf[ProcedureInvocationMessage] => GenCodec.read[ProcedureInvocationMessage](input)
       case c if c == classOf[FunctionInvocationMessage] => GenCodec.read[FunctionInvocationMessage](input)
       case c if c == classOf[ObservableInvocationMessage] => GenCodec.read[ObservableInvocationMessage](input)
-      case c if c == classOf[RawInvocation] => GenCodec.read[RawInvocation](input)
       case c if c == classOf[InvocationSuccess] => GenCodec.read[InvocationSuccess](input)
       case c if c == classOf[InvocationFailure] => GenCodec.read[InvocationFailure](input)
       case c if c == MonixProtocol.Continue.getClass => GenCodec.read[MonixProtocol.Continue.type](input)
@@ -46,7 +45,6 @@ final class RemoteMessageSerializer extends Serializer {
       case m: ProcedureInvocationMessage => GenCodec.write(output, m)
       case m: FunctionInvocationMessage => GenCodec.write(output, m)
       case m: ObservableInvocationMessage => GenCodec.write(output, m)
-      case m: RawInvocation => GenCodec.write(output, m)
       case m: InvocationSuccess => GenCodec.write(output, m)
       case m: InvocationFailure => GenCodec.write(output, m)
       case MonixProtocol.Continue => GenCodec.write(output, MonixProtocol.Continue)

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

@@ -5,27 +5,28 @@ import akka.actor.{Actor, ActorLogging, Props}
 import akka.pattern.{AskTimeoutException, ask}
 import akka.util.Timeout
 import com.avsystem.commons.concurrent.RunNowEC
+import com.avsystem.commons.rpc.akka.AkkaRPCFramework._
 import com.avsystem.commons.rpc.akka._
 import monix.execution.{Ack, Scheduler}
 import monix.reactive.Observable
 
 /**
   * @author Wojciech Milewski
   */
-private final class ServerActor(rawRPC: AkkaRPCFramework.RawRPC, config: AkkaRPCServerConfig) extends Actor with ActorLogging {
+private final class ServerActor(rawRPC: RawRPC, config: AkkaRPCServerConfig) extends Actor with ActorLogging {
 
   override def receive: Receive = {
-    case msg@ProcedureInvocationMessage(name, argLists, getterChain) =>
-      resolveRpc(msg).fire(name)(argLists)
-    case msg@FunctionInvocationMessage(name, argLists, getterChain) =>
+    case ProcedureInvocationMessage(invocation, getterChain) =>
+      resolveRpc(getterChain).fire(invocation)
+    case FunctionInvocationMessage(invocation, getterChain) =>
       val s = sender()
-      resolveRpc(msg).call(name)(argLists).onCompleteNow {
+      resolveRpc(getterChain).call(invocation).onCompleteNow {
         case Success(value) => s ! InvocationSuccess(value)
         case Failure(e) =>
-          logError(e, name)
+          logError(e, invocation.rpcName)
           s ! InvocationFailure(e.getClass.getCanonicalName, e.getMessage)
       }
-    case msg@ObservableInvocationMessage(name, argLists, getterChain) =>
+    case ObservableInvocationMessage(invocation, getterChain) =>
       implicit val scheduler: Scheduler = Scheduler(RunNowEC)
       implicit val timeout: Timeout = Timeout(config.observableAckTimeout)
       val s = sender()
@@ -36,7 +37,7 @@ private final class ServerActor(rawRPC: AkkaRPCFramework.RawRPC, config: AkkaRPC
           Ack.Continue
         }
 
-      resolveRpc(msg).observe(name)(argLists).subscribe(
+      resolveRpc(getterChain).observe(invocation).subscribe(
         value => {
           val result = s ? InvocationSuccess(value)
           result.mapTo[MonixProtocol.RemoteAck].map {
@@ -53,7 +54,7 @@ private final class ServerActor(rawRPC: AkkaRPCFramework.RawRPC, config: AkkaRPC
         },
         e => {
           heartbeat.cancel()
-          logError(e, name)
+          logError(e, invocation.rpcName)
           s ! InvocationFailure(e.getClass.getCanonicalName, e.getMessage)
         },
         () => {
@@ -63,8 +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.args)).toList)
+  private def resolveRpc(getterChain: Seq[RawInvocation]): RawRPC =
+    rawRPC.resolveGetterChain(getterChain)
 
   private def logError(e: Throwable, methodName: String): Unit = {
     log.error(e,
@@ -76,5 +77,5 @@ 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))
+  def props(rawRPC: RawRPC, config: AkkaRPCServerConfig): Props = Props(new ServerActor(rawRPC, config))
 }

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

@@ -14,10 +14,9 @@ import scala.concurrent.duration._
   * @author Wojciech Milewski
   */
 abstract class AkkaRPCFrameworkTest(
-  serverSystem: ActorSystem,
-  clientSystem: ActorSystem,
-  serverSystemPath: Option[String] = None)
-  extends FlatSpec with RPCFrameworkTest with ProcedureRPCTest with FunctionRPCTest with GetterRPCTest with ObservableRPCTest with BeforeAndAfterAll {
+  serverSystem: ActorSystem, clientSystem: ActorSystem, serverSystemPath: Option[String] = None)
+  extends FlatSpec with RPCFrameworkTest with ProcedureRPCTest with FunctionRPCTest with GetterRPCTest with ObservableRPCTest
+    with BeforeAndAfterAll {
 
   /**
     * Servers as identifier supplier for each test case to allow tests parallelization.

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

@@ -21,9 +21,10 @@ trait RPCFrameworkTest extends FlatSpecLike with Matchers with MockitoSugar with
 
   import RPCFrameworkTest._
 
-  val callTimeout = 200.millis
+  val callTimeout: FiniteDuration = 200.millis
 
-  override implicit val patienceConfig: PatienceConfig = PatienceConfig(timeout = Span.convertDurationToSpan(3.seconds))
+  override implicit val patienceConfig: PatienceConfig =
+    PatienceConfig(timeout = Span.convertDurationToSpan(10.seconds))
 
   /**
     * Run tests with connection between client and server.