Split Commands and Replies

gideondk · gideondk · commit 99ec0a9e7256 · 2013-10-20T15:33:56.000+02:00
diff --git a/src/main/scala/nl/gideondk/sentinel/Antenna.scala b/src/main/scala/nl/gideondk/sentinel/Antenna.scala
@@ -58,16 +58,15 @@ class Antenna[Cmd, Evt](init: Init[WithinActorContext, Cmd, Evt], Resolver: Sent
         consumer ! x.registration
         tcpHandler ! init.Command(x.payload)
 
-      case x: Command.Reply[Cmd] ⇒
+      case x: Reply.Response[Cmd] ⇒
         tcpHandler ! init.Command(x.payload)
 
-      case x: Command.StreamReply[Cmd] ⇒
+      case x: Reply.StreamResponseChunk[Cmd] ⇒
         tcpHandler ! init.Command(x.payload)
 
       case init.Event(data) ⇒ {
         Resolver.process(data) match {
           case x: ResponderAction.Reaction[Evt, Cmd] ⇒ responder ! x
-
           case ConsumerAction.Consume                ⇒ consumer ! init.Event(data) // Pass through
           case ConsumerAction.Ignore                 ⇒ ()
         }
diff --git a/src/main/scala/nl/gideondk/sentinel/Messages.scala b/src/main/scala/nl/gideondk/sentinel/Messages.scala
@@ -13,6 +13,7 @@ object Registration {
 }
 
 trait Command[Cmd]
+trait Reply[Cmd]
 
 object Command {
   import Registration._
@@ -23,9 +24,11 @@ object Command {
   case class SendStream[Cmd, Evt](command: Cmd, stream: Process[Future, Cmd], registration: ReplyRegistration[Evt]) extends Command[Cmd]
 
   case class Conversate[Cmd, Evt](command: Cmd, stream: Process[Future, Cmd], registration: StreamReplyRegistration[Evt]) extends Command[Cmd]
+}
 
-  case class Reply[Cmd](payload: Cmd) extends Command[Cmd]
-  case class StreamReply[Cmd](payload: Cmd) extends Command[Cmd]
+object Reply {
+  case class Response[Cmd](payload: Cmd) extends Reply[Cmd]
+  case class StreamResponseChunk[Cmd](payload: Cmd) extends Reply[Cmd]
 }
 
 object Management {
diff --git a/src/main/scala/nl/gideondk/sentinel/Resolver.scala b/src/main/scala/nl/gideondk/sentinel/Resolver.scala
@@ -21,6 +21,8 @@ trait ResponseResolver[Evt, Cmd] extends Resolver[Evt, Cmd] {
   import ResponderAction._
   def answer(f: ⇒ Future[Cmd]): Answer[Evt, Cmd] = Answer(f)
 
+  def handle(f: ⇒ Unit): Handle[Evt, Cmd] = Handle()
+
   def produce(p: ⇒ Future[Process[Future, Cmd]]): ProduceStream[Evt, Cmd] = ProduceStream(p)
 
   def react(p: ⇒ Future[Process[Future, Cmd]]): ReactToStream[Evt, Cmd] = ReactToStream(p)
diff --git a/src/main/scala/nl/gideondk/sentinel/Responder.scala b/src/main/scala/nl/gideondk/sentinel/Responder.scala
@@ -66,6 +66,7 @@ class Responder[Cmd, Evt](init: Init[WithinActorContext, Cmd, Evt], streamChunkT
           log.error(e, e.getMessage)
           context.stop(self)
       }
+
     case x: ProduceStream[Evt, Cmd] ⇒
       val me = self
       val promise = Promise[HandleResult]()
@@ -108,7 +109,7 @@ class Responder[Cmd, Evt](init: Init[WithinActorContext, Cmd, Evt], streamChunkT
   }
 
   def handleRequestAndResponse: Receive = handleRequest orElse handleDequeue orElse {
-    case x: HandleAsyncResult[Cmd] ⇒ context.parent ! Command.Reply(x.response)
+    case x: HandleAsyncResult[Cmd] ⇒ context.parent ! Reply.Response(x.response)
     case x: HandleStreamResult[Cmd] ⇒
       val worker = self
       implicit val timeout = streamChunkTimeout
@@ -124,7 +125,7 @@ class Responder[Cmd, Evt](init: Init[WithinActorContext, Cmd, Evt], streamChunkT
       sender ! ReadyForStream
     case StreamProducerChunk(c) ⇒
       sender ! StreamProducerChunkReceived
-      context.parent ! Command.StreamReply(c)
+      context.parent ! Reply.StreamResponseChunk(c)
     case StreamProducerEnded ⇒
       sender ! StreamProducerChunkReceived
       context.become(handleRequestAndResponse)
diff --git a/src/main/scala/nl/gideondk/sentinel/client/Core.scala b/src/main/scala/nl/gideondk/sentinel/client/Core.scala
@@ -37,11 +37,17 @@ trait Client[Cmd, Evt] {
     promise.future
   }
 
-  // def sendStream(command: Cmd, source: Process[Future, Evt]): Task[Evt] = Task {
-  //   val promise = Promise[Evt]()
-  //   actor ! Command.AskStream(command, StreamReplyRegistration(terminator, includeTerminator, promise))
-  //   promise.future
-  // }
+  def sendStream(command: Cmd, source: Process[Future, Evt]): Task[Evt] = Task {
+    val promise = Promise[Evt]()
+    actor ! Command.SendStream(command, source, ReplyRegistration(promise))
+    promise.future
+  }
+
+  def conversate(command: Cmd, source: Process[Future, Evt], terminator: Evt ⇒ Boolean, includeTerminator: Boolean): Task[Process[Future, Evt]] = Task {
+    val promise = Promise[Process[Future, Evt]]()
+    actor ! Command.Conversate(command, source, StreamReplyRegistration(terminator, includeTerminator, promise))
+    promise.future
+  }
 }
 
 object Client {
