raboof/pekko
1
0
Fork 0
Code Issues Pull requests Projects Releases 6 Packages Wiki Activity Actions 10

Fixing #437 by adding "Remote" Future

Browse source
This commit is contained in:
Viktor Klang 2010-09-16 15:58:46 +02:00
parent 1960241b87
commit 971ebf4501
2 changed files with 31 additions and 17 deletions
Download patch file Download diff file Expand all files Collapse all files

4
akka-actor/src/main/scala/dispatch/Future.scala
View file

@ -160,6 +160,7 @@ class DefaultCompletableFuture[T](timeout: Long) extends CompletableFuture[T] {
if (!_completed) { if (!_completed) {
_completed = true _completed = true
_result = Some(result) _result = Some(result)
onComplete(result)
} }
} finally { } finally {
_signal.signalAll _signal.signalAll
@ -171,6 +172,7 @@ class DefaultCompletableFuture[T](timeout: Long) extends CompletableFuture[T] {
if (!_completed) { if (!_completed) {
_completed = true _completed = true
_exception = Some(exception) _exception = Some(exception)
onCompleteException(exception)
} }
} finally { } finally {
_signal.signalAll _signal.signalAll
@ -178,4 +180,6 @@ class DefaultCompletableFuture[T](timeout: Long) extends CompletableFuture[T] {
} }
private def currentTimeInNanos: Long = TIME_UNIT.toNanos(System.currentTimeMillis) private def currentTimeInNanos: Long = TIME_UNIT.toNanos(System.currentTimeMillis)
protected def onComplete(result: T) {}
protected def onCompleteException(exception: Throwable) {}
} }

44
akka-remote/src/main/scala/remote/RemoteServer.scala
View file

@ -30,6 +30,7 @@ import org.jboss.netty.handler.ssl.SslHandler
import scala.collection.mutable.Map import scala.collection.mutable.Map
import scala.reflect.BeanProperty import scala.reflect.BeanProperty
import se.scalablesolutions.akka.dispatch.{DefaultCompletableFuture, CompletableFuture}
/** /**
* Use this object if you need a single remote server on a specific node. * Use this object if you need a single remote server on a specific node.
@ -498,27 +499,36 @@ class RemoteServerHandler(
case RemoteActorSystemMessage.Stop => actorRef.stop case RemoteActorSystemMessage.Stop => actorRef.stop
case _ => // then match on user defined messages case _ => // then match on user defined messages
if (request.getIsOneWay) actorRef.!(message)(sender) if (request.getIsOneWay) actorRef.!(message)(sender)
else { else actorRef.postMessageToMailboxAndCreateFutureResultWithTimeout(message,request.getActorInfo.getTimeout,None,Some(
try { new DefaultCompletableFuture[AnyRef](request.getActorInfo.getTimeout){
val resultOrNone = (actorRef.!!(message)(sender)).as[AnyRef] override def onComplete(result: AnyRef) {
val result = if (resultOrNone.isDefined) resultOrNone.get else null log.debug("Returning result from actor invocation [%s]", result)
val replyBuilder = RemoteReplyProtocol.newBuilder
.setId(request.getId)
.setMessage(MessageSerializer.serialize(result))
.setIsSuccessful(true)
.setIsActor(true)
log.debug("Returning result from actor invocation [%s]", result) if (request.hasSupervisorUuid) replyBuilder.setSupervisorUuid(request.getSupervisorUuid)
val replyBuilder = RemoteReplyProtocol.newBuilder
.setId(request.getId)
.setMessage(MessageSerializer.serialize(result))
.setIsSuccessful(true)
.setIsActor(true)
if (request.hasSupervisorUuid) replyBuilder.setSupervisorUuid(request.getSupervisorUuid) try {
channel.write(replyBuilder.build) channel.write(replyBuilder.build)
} catch {
case e: Throwable =>
server.notifyListeners(RemoteServerError(e, server))
}
}
} catch { override def onCompleteException(exception: Throwable) {
case e: Throwable => try {
channel.write(createErrorReplyMessage(e, request, true)) channel.write(createErrorReplyMessage(exception, request, true))
server.notifyListeners(RemoteServerError(e, server)) } catch {
} case e: Throwable =>
server.notifyListeners(RemoteServerError(e, server))
}
}
} }
))
} }
} }