avoid wrapping messages when ask returns single type
This commit is contained in:
Roland
parent
859589b9d4
commit
9b0c3a486f
5 changed files with 106 additions and 44 deletions
|
|
@ -32,7 +32,7 @@ class ChannelRef[+T <: ChannelList](val actorRef: ActorRef) extends AnyVal {
|
|||
* which will be completed with the reply message or a TimeoutException.
|
||||
* If the message is a Future itself, eventually send the Future’s value.
|
||||
*/
|
||||
def <-?-[M](msg: M): Future[_] = macro macros.Ask.impl[ChannelList, Any, T, M]
|
||||
def <-?-[M](msg: M): Future[_] = macro macros.Ask.impl[ChannelList, Any, Any, T, M]
|
||||
|
||||
/**
|
||||
* Narrow this ChannelRef by removing channels or narrowing input types or
|
||||
|
|
|
|||
|
|
@ -21,11 +21,12 @@ object Ask {
|
|||
def impl[ //
|
||||
ReturnChannels <: ChannelList, // the precise type union describing the reply
|
||||
ReturnLUB, // the least-upper bound for the reply types
|
||||
ReturnT, // the return type if it is just a single type
|
||||
Channel <: ChannelList: c.WeakTypeTag, // the channel being asked
|
||||
Msg: c.WeakTypeTag // the message being sent down the channel
|
||||
](c: Context {
|
||||
type PrefixType = ChannelRef[Channel]
|
||||
})(msg: c.Expr[Msg]): c.Expr[Future[WrappedMessage[ReturnChannels, ReturnLUB]]] = {
|
||||
})(msg: c.Expr[Msg]): c.Expr[Future[_]] = {
|
||||
import c.universe._
|
||||
|
||||
val tpeChannel = weakTypeOf[Channel]
|
||||
|
|
@ -41,18 +42,32 @@ object Ask {
|
|||
|
||||
Tell.verify(c)(null, unwrapped, typeOf[(Any, Nothing) :+: TNil], tpeChannel)
|
||||
|
||||
implicit val ttReturnChannels = c.TypeTag[ReturnChannels](toChannels(c.universe)(out, weakTypeOf[Nothing]))
|
||||
implicit val ttReturnLUB = c.TypeTag[ReturnLUB](c.universe.lub(out))
|
||||
if (isFuture)
|
||||
if (unwrapped <:< typeOf[ChannelList])
|
||||
reify(askFutureWrapped[WrappedMessage[ReturnChannels, ReturnLUB]](
|
||||
c.prefix.splice.actorRef, msg.splice.asInstanceOf[Future[WrappedMessage[TNil, Any]]])(imp[Timeout](c).splice))
|
||||
else
|
||||
reify(askFuture[WrappedMessage[ReturnChannels, ReturnLUB]](
|
||||
c.prefix.splice.actorRef, msg.splice.asInstanceOf[Future[Any]])(imp[Timeout](c).splice))
|
||||
else
|
||||
reify(askOps[WrappedMessage[ReturnChannels, ReturnLUB]](
|
||||
c.prefix.splice.actorRef, toMsg(c)(msg, tpeMsg).splice)(imp[Timeout](c).splice))
|
||||
implicit lazy val ttReturn = c.TypeTag[ReturnT](out.head)
|
||||
implicit lazy val ttReturnChannels = c.TypeTag[ReturnChannels](toChannels(c.universe)(out, weakTypeOf[Nothing]))
|
||||
implicit lazy val ttReturnLUB = c.TypeTag[ReturnLUB](c.universe.lub(out))
|
||||
|
||||
out match {
|
||||
case x :: Nil if isFuture ⇒
|
||||
if (unwrapped <:< typeOf[ChannelList])
|
||||
reify(askFutureWrappedNoWrap[ReturnT](
|
||||
c.prefix.splice.actorRef, msg.splice.asInstanceOf[Future[WrappedMessage[TNil, Any]]])(imp[Timeout](c).splice))
|
||||
else
|
||||
reify(askFutureNoWrap[ReturnT](
|
||||
c.prefix.splice.actorRef, msg.splice.asInstanceOf[Future[Any]])(imp[Timeout](c).splice))
|
||||
case x :: Nil ⇒
|
||||
reify(askOpsNoWrap[ReturnT](
|
||||
c.prefix.splice.actorRef, toMsg(c)(msg, tpeMsg).splice)(imp[Timeout](c).splice))
|
||||
case _ if isFuture ⇒
|
||||
if (unwrapped <:< typeOf[ChannelList])
|
||||
reify(askFutureWrapped[WrappedMessage[ReturnChannels, ReturnLUB]](
|
||||
c.prefix.splice.actorRef, msg.splice.asInstanceOf[Future[WrappedMessage[TNil, Any]]])(imp[Timeout](c).splice))
|
||||
else
|
||||
reify(askFuture[WrappedMessage[ReturnChannels, ReturnLUB]](
|
||||
c.prefix.splice.actorRef, msg.splice.asInstanceOf[Future[Any]])(imp[Timeout](c).splice))
|
||||
case _ ⇒
|
||||
reify(askOps[WrappedMessage[ReturnChannels, ReturnLUB]](
|
||||
c.prefix.splice.actorRef, toMsg(c)(msg, tpeMsg).splice)(imp[Timeout](c).splice))
|
||||
}
|
||||
}
|
||||
|
||||
def opsImpl[ //
|
||||
|
|
@ -62,7 +77,7 @@ object Ask {
|
|||
Msg: c.WeakTypeTag // the message being sent down the channel
|
||||
](c: Context {
|
||||
type PrefixType = AnyOps[Msg]
|
||||
})(channel: c.Expr[ChannelRef[Channel]]): c.Expr[Future[WrappedMessage[ReturnChannels, ReturnLUB]]] = {
|
||||
})(channel: c.Expr[ChannelRef[Channel]]): c.Expr[Future[_]] = {
|
||||
import c.universe._
|
||||
|
||||
val tpeChannel = weakTypeOf[Channel]
|
||||
|
|
@ -72,11 +87,18 @@ object Ask {
|
|||
|
||||
Tell.verify(c)(null, unwrapped, typeOf[(Any, Nothing) :+: TNil], tpeChannel)
|
||||
|
||||
implicit val ttReturnChannels = c.TypeTag[ReturnChannels](toChannels(c.universe)(out, weakTypeOf[Nothing]))
|
||||
implicit val ttReturnLUB = c.TypeTag[ReturnLUB](c.universe.lub(out))
|
||||
val msg = reify(c.prefix.splice.value)
|
||||
reify(askOps[WrappedMessage[ReturnChannels, ReturnLUB]](
|
||||
channel.splice.actorRef, toMsg(c)(msg, tpeMsg).splice)(imp[Timeout](c).splice))
|
||||
out match {
|
||||
case x :: Nil ⇒
|
||||
implicit val ttReturnLUB = c.TypeTag[ReturnLUB](x)
|
||||
reify(askOpsNoWrap[ReturnLUB](
|
||||
channel.splice.actorRef, toMsg(c)(msg, tpeMsg).splice)(imp[Timeout](c).splice))
|
||||
case _ ⇒
|
||||
implicit val ttReturnChannels = c.TypeTag[ReturnChannels](toChannels(c.universe)(out, weakTypeOf[Nothing]))
|
||||
implicit val ttReturnLUB = c.TypeTag[ReturnLUB](c.universe.lub(out))
|
||||
reify(askOps[WrappedMessage[ReturnChannels, ReturnLUB]](
|
||||
channel.splice.actorRef, toMsg(c)(msg, tpeMsg).splice)(imp[Timeout](c).splice))
|
||||
}
|
||||
}
|
||||
|
||||
// this is the implementation for Future[_] -?-> ChannelRef[_]
|
||||
|
|
@ -87,7 +109,7 @@ object Ask {
|
|||
Msg: c.WeakTypeTag // the message being sent down the channel
|
||||
](c: Context {
|
||||
type PrefixType = FutureOps[Msg]
|
||||
})(channel: c.Expr[ChannelRef[Channel]]): c.Expr[Future[WrappedMessage[ReturnChannels, ReturnLUB]]] = {
|
||||
})(channel: c.Expr[ChannelRef[Channel]]): c.Expr[Future[_]] = {
|
||||
import c.universe._
|
||||
|
||||
val tpeChannel = weakTypeOf[Channel]
|
||||
|
|
@ -97,19 +119,30 @@ object Ask {
|
|||
|
||||
Tell.verify(c)(null, unwrapped, typeOf[(Any, Nothing) :+: TNil], tpeChannel)
|
||||
|
||||
implicit val ttReturnChannels = c.TypeTag[ReturnChannels](toChannels(c.universe)(out, weakTypeOf[Nothing]))
|
||||
implicit val ttReturnLUB = c.TypeTag[ReturnLUB](c.universe.lub(out))
|
||||
if (tpeMsg <:< typeOf[WrappedMessage[_, _]])
|
||||
reify(askFutureWrapped[WrappedMessage[ReturnChannels, ReturnLUB]](
|
||||
channel.splice.actorRef, c.prefix.splice.future.asInstanceOf[Future[WrappedMessage[TNil, Any]]])(imp[Timeout](c).splice))
|
||||
else
|
||||
reify(askFuture[WrappedMessage[ReturnChannels, ReturnLUB]](
|
||||
channel.splice.actorRef, c.prefix.splice.future)(imp[Timeout](c).splice))
|
||||
out match {
|
||||
case x :: Nil ⇒
|
||||
implicit val ttReturnLUB = c.TypeTag[ReturnLUB](x)
|
||||
if (tpeMsg <:< typeOf[WrappedMessage[_, _]])
|
||||
reify(askFutureWrappedNoWrap[ReturnLUB](
|
||||
channel.splice.actorRef, c.prefix.splice.future.asInstanceOf[Future[WrappedMessage[TNil, Any]]])(imp[Timeout](c).splice))
|
||||
else
|
||||
reify(askFutureNoWrap[ReturnLUB](
|
||||
channel.splice.actorRef, c.prefix.splice.future)(imp[Timeout](c).splice))
|
||||
case _ ⇒
|
||||
implicit val ttReturnChannels = c.TypeTag[ReturnChannels](toChannels(c.universe)(out, weakTypeOf[Nothing]))
|
||||
implicit val ttReturnLUB = c.TypeTag[ReturnLUB](c.universe.lub(out))
|
||||
if (tpeMsg <:< typeOf[WrappedMessage[_, _]])
|
||||
reify(askFutureWrapped[WrappedMessage[ReturnChannels, ReturnLUB]](
|
||||
channel.splice.actorRef, c.prefix.splice.future.asInstanceOf[Future[WrappedMessage[TNil, Any]]])(imp[Timeout](c).splice))
|
||||
else
|
||||
reify(askFuture[WrappedMessage[ReturnChannels, ReturnLUB]](
|
||||
channel.splice.actorRef, c.prefix.splice.future)(imp[Timeout](c).splice))
|
||||
}
|
||||
}
|
||||
|
||||
val wrapMessage = (m: Any) ⇒ (new WrappedMessage[TNil, Any](m): Any)
|
||||
|
||||
@inline def askOps[T <: WrappedMessage[_, _]](target: ActorRef, msg: Any)(implicit t: Timeout): Future[T] = {
|
||||
def askOps[T <: WrappedMessage[_, _]](target: ActorRef, msg: Any)(implicit t: Timeout): Future[T] = {
|
||||
implicit val ec = ExecutionContexts.sameThreadExecutionContext
|
||||
akka.pattern.ask(target, msg).map(wrapMessage).asInstanceOf[Future[T]]
|
||||
}
|
||||
|
|
@ -124,4 +157,17 @@ object Ask {
|
|||
future flatMap (w ⇒ akka.pattern.ask(target, w.value).map(wrapMessage).asInstanceOf[Future[T]])
|
||||
}
|
||||
|
||||
def askOpsNoWrap[T](target: ActorRef, msg: Any)(implicit t: Timeout): Future[T] =
|
||||
akka.pattern.ask(target, msg).asInstanceOf[Future[T]]
|
||||
|
||||
def askFutureNoWrap[T](target: ActorRef, future: Future[_])(implicit t: Timeout): Future[T] = {
|
||||
implicit val ec = ExecutionContexts.sameThreadExecutionContext
|
||||
future flatMap (m ⇒ akka.pattern.ask(target, m).asInstanceOf[Future[T]])
|
||||
}
|
||||
|
||||
def askFutureWrappedNoWrap[T](target: ActorRef, future: Future[WrappedMessage[_, _]])(implicit t: Timeout): Future[T] = {
|
||||
implicit val ec = ExecutionContexts.sameThreadExecutionContext
|
||||
future flatMap (w ⇒ akka.pattern.ask(target, w.value).asInstanceOf[Future[T]])
|
||||
}
|
||||
|
||||
}
|
||||
Loading…
Add table
Add a link
Reference in a new issue