Feature/streaming take2

benchmark/client/src/main/scala/Main.scala

@@ -159,6 +159,11 @@ object BenchmarkClientMain extends TestData with transformations {
       new Test(results, server.BenchmarkClient.identityBig(toBigW(bigIn)).runWithoutContext(endpoint))
 //      new Test(results, BenchmarkClient.identityMedium(toMediumW(medIn)).runWithoutContext(endpoint))
 //      new Test(results, BenchmarkClient.identityLarge(toLargeW(largeIn)).runWithoutContext(endpoint))
+      // We are currently measuring how fast we can send receive streams of a single element.
+      // It might be more useful to measure how fast we can stream elements of a single stream
+//      new Test(results, server.BenchmarkClient.streamBig(localStreamToRemote(bigStream.map(toBigW))).run(endpoint))
+//      new Test(results, server.BenchmarkClient.streamMedium(medStream.map(toMediumW)).runWithoutContext(endpoint))
+//      new Test(results, server.BenchmarkClient.streamLarge(largeStream.map(toLargeW)).runWithoutContext(endpoint))
     }
     val threads = testers.map(new Thread(_))
 

benchmark/protocol/src/main/scala/Protocol.scala

@@ -59,7 +59,10 @@ object protocol {
       .codec[MediumW]
       .codec[SmallW]
       .codec[BigW]
-      .specify1("identityLarge", Field.strict[LargeW]("in"), Type[LargeW])
-      .specify1[MediumW, MediumW]("identityMedium", Field.strict[MediumW]("in"), Type[MediumW])
-      .specify1[BigW, BigW]("identityBig", Field.strict[BigW]("in"), Type[BigW])
+      .specify1("identityLarge", Field.strict[LargeW]("in"), Type.strict[LargeW])
+      .specify1("identityMedium", Field.strict[MediumW]("in"), Type.strict[MediumW])
+      .specify1("identityBig", Field.strict[BigW]("in"), Type.strict[BigW])
+      .specify1("streamLarge", Field.stream[LargeW]("in"), Type.stream[LargeW])
+      .specify1("streamMedium", Field.stream[MediumW]("in"), Type.stream[MediumW])
+      .specify1("streamBig", Field.stream[BigW]("in"), Type.stream[BigW])
 }

benchmark/protocol/src/main/scala/TestData.scala

@@ -18,6 +18,9 @@
 package remotely
 package example.benchmark
 
+import scalaz.concurrent.Task
+import scalaz.stream.Process
+
 trait TestData {
   val sm: Small = Small((for(i <- 1 to 10) yield i.toString -> i.toString).toMap, (for(i <- 1 to 10) yield i.toString).toList)
 
@@ -27,4 +30,10 @@ trait TestData {
 
   val bigIn: Big = Big(1)
 
+  val bigStream: Process[Task, Big] = Process(bigIn)
+
+  val medStream: Process[Task, Medium] = Process(medIn)
+
+  val largeStream: Process[Task, Large] = Process(largeIn)
+
 }

benchmark/server/src/main/scala/ServerImpl.scala

@@ -20,14 +20,16 @@ package example.benchmark
 package server
 
 import scalaz.concurrent._
-import java.util.concurrent._
+import scalaz.stream.Process
 import java.util.concurrent.atomic.AtomicInteger
 
 class BenchmarkServerImpl extends BenchmarkServer with transformations {
-  override def identityLarge = (large: LargeW) => Response[LargeW]((c: Response.Context) => Task.now{toLargeW(fromLargeW(large))})
-  override def identityMedium = (med: MediumW) => Response[MediumW]((c: Response.Context) => Task.now{toMediumW(fromMediumW(med))})
-  override def identityBig = (big: BigW) => Response[BigW]((c: Response.Context) => Task.now{toBigW(fromBigW(big))})
-
+  override def identityLarge = (large: LargeW) => Response.now(toLargeW(fromLargeW(large)))
+  override def identityMedium = (med: MediumW) => Response.now(toMediumW(fromMediumW(med)))
+  override def identityBig = (big: BigW) => Response.now(toBigW(fromBigW(big)))
+  override def streamLarge = (largeStream: Process[Task, LargeW]) => Response.now(largeStream.map(large => toLargeW(fromLargeW(large))))
+  override def streamMedium = (mediumStream: Process[Task, MediumW]) => Response.now(mediumStream.map(medium => toMediumW(fromMediumW(medium))))
+  override def streamBig = (bigStream: Process[Task, BigW]) => Response.now(bigStream.map(big => toBigW(fromBigW(big))))
 }
 
 object Main {

core/build.sbt

@@ -21,7 +21,8 @@ scalacOptions ++= Seq(
 libraryDependencies ++= Seq(
   "org.scodec"         %% "scodec-core"   % "1.7.0",
   "org.scodec"         %% "scodec-scalaz" % "1.0.0",
-  "org.scalaz"         %% "scalaz-core"   % "7.1.0",
+  "org.scalaz"         %% "scalaz-core"   % "7.1.2",
+  "org.scalaz"         %% "scalaz-concurrent" % "7.1.2",
   "org.scalaz.stream"  %% "scalaz-stream" % "0.7a",
   "org.apache.commons" % "commons-pool2"  % "2.2",
   "io.netty"           % "netty-handler"  % "4.0.25.Final",

core/src/main/scala/Environment.scala

@@ -62,12 +62,8 @@ case class Environment[H <: HList](codecs: Codecs[H], values: Values) {
     this.populate(_ => v)
 
   private def serverHandler(monitoring: Monitoring): Handler = { bytes =>
-      // we assume the input is a framed stream, and encode the response(s)
-      // as a framed stream as well
-      bytes pipe Process.await1[BitVector] /*server.Handler.deframe*/ evalMap { bs =>
-        Server.handle(this)(bs)(monitoring)
-      }
-    }
+    Server.handle(this)(bytes)(monitoring)
+  }
 
   /**
     * start a netty server listening to the given address

core/src/main/scala/Gen.scala

@@ -26,12 +26,17 @@ private[remotely] object Gen extends MacrosCompatibility {
   def liftSignature(c: Context)(signature: Signature): c.universe.Tree = {
     import c.universe._
     val s = signature
-    val t: Tree = q"_root_.remotely.Signature(${s.name}, List(..${s.params.map(liftField(c)(_))}), ${s.outType})"
+    val t: Tree = q"_root_.remotely.Signature(${s.name}, List(..${s.params.map(liftField(c)(_))}), ${liftType(c)(s.out)})"
     t
   }
 
   def liftField(c: Context)(field: Field[Any]): c.universe.Tree = {
     import c.universe._
-    q"_root_.remotely.Field(${field.name}, ${field.typeString})"
+    q"_root_.remotely.Field(${field.name}, ${liftType(c)(field.type_)})"
+  }
+
+  def liftType(c: Context)(type_ : Type[Any]): c.universe.Tree = {
+    import c.universe._
+    q"_root_.remotely.Type(${type_.name}, ${type_.isStream})"
   }
 }

core/src/main/scala/GenServer.scala

@@ -56,12 +56,12 @@ object GenServer extends MacrosCompatibility {
 
     // Creates name/type pairs from the signatures in the protocol.
     val signatures = p.signatures.signatures.map { s =>
-      val typ = parseType(c)(Signatures.wrapResponse(s.typeString))
-      (s.name, typ)
+      val typ = parseType(c)(s.wrapResponse)
+      (s.name, typ, s.out.isStream)
     }
 
     // Generates the method defs for the generated class.
-    val sigDefs = signatures.collect { case (n,t) if n != "describe" =>
+    val sigDefs = signatures.collect { case (n,t, isStream) if n != "describe" =>
       genSig(n, t)
     }
 
@@ -93,7 +93,8 @@ object GenServer extends MacrosCompatibility {
 
              private def populateDeclarations(env: Values): Values =
                 ${ signatures.foldLeft(q"env":c.Tree)((e,s) =>
-                    q"$e.declare(${Literal(Constant(s._1))},${Ident(createTermName(c)(s._1))})"
+                    if (s._3)  q"$e.declareStream(${Literal(Constant(s._1))},${Ident(createTermName(c)(s._1))})"
+                    else q"$e.declare(${Literal(Constant(s._1))},${Ident(createTermName(c)(s._1))})"
                   )}
 
               ..$body

core/src/main/scala/Monitoring.scala

@@ -20,7 +20,9 @@ package remotely
 import java.net.InetSocketAddress
 import java.net.SocketAddress
 import scala.concurrent.duration._
-import scalaz.\/
+import scalaz.{\/-, \/}
+import scalaz.concurrent.Task
+import scalaz.stream.Process
 
 /**
  * A collection of callbacks that can be passed to `[[remotely.Environment#serve]]`
@@ -39,6 +41,11 @@ trait Monitoring { self =>
                  result: Throwable \/ A,
                  took: Duration): Unit
 
+  def handle[A](ctx: Response.Context, req: Remote[A], references: Iterable[String], result: Throwable \/ A, startTime: Long) = Task.delay {
+    val duration = Duration.fromNanos(System.nanoTime() - startTime)
+    handled(ctx, req, references, result, duration)
+  }
+
   def negotiating(addr: Option[SocketAddress],
                   what: String,
                   error: Option[Throwable]): Unit
@@ -48,6 +55,10 @@ trait Monitoring { self =>
     case _ => "<unknown addr>"
   }
 
+  def sink[A](ctx: Response.Context, req: Remote[A], references: Iterable[String], startTime: Long) = Process.constant { a: A =>
+    handle(ctx, req, references, \/-(a), startTime)
+  }
+
   /**
    * Return a new `Monitoring` instance that send statistics
    * to both `this` and `other`.

core/src/main/scala/Protocol.scala

@@ -29,19 +29,19 @@ case class Protocol[H <: HList](codecs: Codecs[H], signatures: Signatures) {
     this.copy(codecs = codecs.codec[A])
 
   def specify0[O](name: String, out: Type[O])(implicit evidenceA : Selector[H, O]): Protocol[H] =
-    this.copy(signatures = signatures.specify0(name, out.name))
+    this.copy(signatures = signatures.specify0(name, out))
 
   def specify1[A,O](name: String, in: Field[A], out: Type[O])(implicit evidenceA : Selector[H, A], evidenceB : Selector[H, O]): Protocol[H] =
-    this.copy(signatures = signatures.specify1(name, in, out.name))
+    this.copy(signatures = signatures.specify1(name, in, out))
 
   def specify2[A,B,O](name: String, in1: Field[A], in2: Field[B], out: Type[O])(implicit evidenceA : Selector[H, A], evidenceB : Selector[H, B], evidenceC : Selector[H, O]): Protocol[H] =
-    this.copy(signatures = signatures.specify2(name, in1, in2, out.name))
+    this.copy(signatures = signatures.specify2(name, in1, in2, out))
 
   def specify3[A,B,C,O](name: String, in1: Field[A], in2: Field[B], in3: Field[C], out: Type[O])(implicit evidenceA : Selector[H, A], evidenceB : Selector[H, B], evidenceC : Selector[H, C], evidenceD: Selector[H, O]): Protocol[H] =
-    this.copy(signatures = signatures.specify3(name, in1, in2, in3, out.name))
+    this.copy(signatures = signatures.specify3(name, in1, in2, in3, out))
 
   def specify4[A,B,C,D,O](name: String, in1: Field[A], in2: Field[B], in3: Field[C], in4: Field[D], out: Type[O])(implicit evidenceA : Selector[H, A], evidenceB : Selector[H, B], evidenceC : Selector[H, C], evidenceD: Selector[H, D], evidenceE: Selector[H, O]): Protocol[H] =
-    this.copy(signatures = signatures.specify4(name, in1, in2, in3, in4, out.name))
+    this.copy(signatures = signatures.specify4(name, in1, in2, in3, in4, out))
 
   def pretty: String =
     "Protocol(\n" +

core/src/main/scala/Remote.scala

@@ -25,6 +25,7 @@ import scala.reflect.runtime.universe.TypeTag
 import scodec.{Codec,Decoder,Encoder}
 import scodec.bits.BitVector
 import shapeless._
+import scalaz.stream.Process
 
 /**
  * Represents a remote computation which yields a
@@ -38,6 +39,18 @@ sealed trait Remote[+A] {
   def pretty: String = "Remote {\n  " +
     Remote.refs(this).mkString("\n  ") + "\n  " +
     toString + "\n}"
+
+  /**
+   * Builds a List by applying a partial function to all children of this Remote on which the function is defined.
+   */
+  def collect[B](partial: PartialFunction[Remote[Any],B]): List[B] = {
+    val me = partial.lift(this).toList
+    val rest = this match {
+      case app: Remote.Ap[_] => (app.f :: app.args).map(_.collect(partial)).flatten
+      case _ => Nil
+    }
+    me ++ rest
+  }
 }
 
 object Remote {
@@ -52,6 +65,15 @@ object Remote {
   def local[A:Encoder:TypeTag](a: A): Remote[A] =
     Remote.Local(a, Some(Encoder[A]), Remote.toTag[A])
 
+  def localStream[A: Encoder:TypeTag](stream: Process[Task, A]): Remote[Process[Task,A]] =
+    Remote.LocalStream(stream, Some(Encoder[A]), Remote.toTag[A])
+
+  implicit class RunSyntaxForStreaming[A](self: Remote[Process[Task,A]]) {
+    /** Call `self.run(at, M).apply(ctx)` to get back a `Task[A]`. */
+    def run(at: Endpoint, ctx: Response.Context = Response.Context.empty, M: Monitoring = Monitoring.empty)(implicit A: TypeTag[A], C: Codec[A]): Task[Process[Task,A]] =
+      evaluateStream(at,M)(self).apply(ctx)
+  }
+
   /** Provides the syntax `expr.run(endpoint)`, where `endpoint: Endpoint`. */
   implicit class RunSyntax[A](self: Remote[A]) {
     /**
@@ -66,8 +88,8 @@ object Remote {
       run(at, M).apply(ctx)
 
     /** Run this with an empty context */
-    def runWithoutContext(at: Endpoint)(implicit A: TypeTag[A], C: Codec[A]): Task[A] =
-      runWithContext(at, Response.Context.empty)
+    def runWithoutContext(at: Endpoint, M: Monitoring = Monitoring.empty)(implicit A: TypeTag[A], C: Codec[A]): Task[A] =
+      runWithContext(at, Response.Context.empty, M)
   }
   implicit class Ap1Syntax[A,B](self: Remote[A => B]) {
     def apply(a: Remote[A]): Remote[B] =
@@ -95,64 +117,58 @@ object Remote {
     override def toString = a.toString
   }
 
+  private[remotely] case class LocalStream[A](
+    stream: Process[Task,A],
+    format: Option[Encoder[A]],
+    tag: String
+  ) extends Remote[Process[Task,A]] {
+    override def toString = stream.toString
+  }
+
   /**
    * Reference to a remote value on the server.
    */
   private[remotely] case class Ref[A](name: String) extends Remote[A] {
     override def toString = name.takeWhile(_ != ':')
   }
 
+  private[remotely] sealed abstract class Ap[A](val f: Remote[Any],val args: List[Remote[Any]]) extends Remote[A] {
+    override def toString = s"""$f({$args.mkString(","})"""
+  }
+
   // we require a separate constructor for each function
   // arity, since remote invocations must be fully saturated
   private[remotely] case class Ap1[A,B](
-    f: Remote[A => B],
-    a: Remote[A]) extends Remote[B] {
-    override def toString = s"$f($a)"
-  }
+    override val f: Remote[A => B],
+    a: Remote[A]) extends Ap[B](f,List(a))
 
   private[remotely] case class Ap2[A,B,C](
-    f: Remote[(A,B) => C],
+    override val f: Remote[(A,B) => C],
     a: Remote[A],
-    b: Remote[B]) extends Remote[C] {
-    override def toString = s"$f($a, $b)"
-  }
+    b: Remote[B]) extends Ap[C](f,List(a,b))
 
   private[remotely] case class Ap3[A,B,C,D](
-    f: Remote[(A,B,C) => D],
+    override val f: Remote[(A,B,C) => D],
     a: Remote[A],
     b: Remote[B],
-    c: Remote[C]) extends Remote[D] {
-    override def toString = s"$f($a, $b, $c)"
-  }
+    c: Remote[C]) extends Ap[D](f,List(a,b,c))
 
   private[remotely] case class Ap4[A,B,C,D,E](
-    f: Remote[(A,B,C,D) => E],
+    override val f: Remote[(A,B,C,D) => E],
     a: Remote[A],
     b: Remote[B],
     c: Remote[C],
-    d: Remote[D]) extends Remote[E] {
-    override def toString = s"$f($a, $b, $c, $d)"
-  }
+    d: Remote[D]) extends Ap[E](f,List(a,b,c,d))
 
   /** Collect up all the `Ref` names referenced by `r`. */
-  def refs[A](r: Remote[A]): SortedSet[String] = r match {
-    case Local(a,e,t) => SortedSet.empty
+  def refs[A](r: Remote[A]): SortedSet[String] = (r collect {
     case Ref(t) => SortedSet(t)
-    case Ap1(f,a) => refs(f).union(refs(a))
-    case Ap2(f,a,b) => refs(f).union(refs(b)).union(refs(b))
-    case Ap3(f,a,b,c) => refs(f).union(refs(b)).union(refs(b)).union(refs(c))
-    case Ap4(f,a,b,c,d) => refs(f).union(refs(b)).union(refs(b)).union(refs(c)).union(refs(d))
-  }
+  }).fold(SortedSet.empty[String])(_.union(_))
 
   /** Collect up all the formats referenced by `r`. */
-  def formats[A](r: Remote[A]): SortedSet[String] = r match {
+  def formats[A](r: Remote[A]): SortedSet[String] = r.collect {
     case Local(a,e,t) => SortedSet(t)
-    case Ref(t) => SortedSet.empty
-    case Ap1(f,a) => formats(f).union(formats(a))
-    case Ap2(f,a,b) => formats(f).union(formats(b)).union(formats(b))
-    case Ap3(f,a,b,c) => formats(f).union(formats(b)).union(formats(b)).union(formats(c))
-    case Ap4(f,a,b,c,d) => formats(f).union(formats(b)).union(formats(b)).union(formats(c)).union(formats(d))
-  }
+  }.fold(SortedSet.empty[String])(_.union(_))
 
   def toTag[A:TypeTag]: String = {
     val tt = typeTag[A]
@@ -175,6 +191,8 @@ object Remote {
 
     /** Implicitly promote a local value to a `Remote[A]`. */
     implicit def localToRemote[A:Encoder:TypeTag](a: A): Remote[A] = local(a)
+
+    implicit def localStreamToRemote[A: Encoder:TypeTag](stream: Process[Task, A]): Remote[Process[Task,A]] = localStream(stream)
   }
 }