added possibility to register a remote actor by explicit handle id
This commit is contained in:
parent
963d76beae
commit
d76d69f620
3 changed files with 18 additions and 57 deletions
|
|
@ -26,55 +26,6 @@ import java.util.concurrent.atomic.AtomicLong
|
||||||
|
|
||||||
import scala.collection.mutable.{HashSet, HashMap}
|
import scala.collection.mutable.{HashSet, HashMap}
|
||||||
|
|
||||||
/*
|
|
||||||
class RemoteActorHandle(id: String, className: String, timeout: Long, hostname: String, port: Int) extends Actor {
|
|
||||||
start
|
|
||||||
val remoteClient = RemoteClient.clientFor(hostname, port)
|
|
||||||
|
|
||||||
override def postMessageToMailbox(message: Any, sender: Option[Actor]): Unit = {
|
|
||||||
val requestBuilder = RemoteRequest.newBuilder
|
|
||||||
.setId(RemoteRequestIdFactory.nextId)
|
|
||||||
.setTarget(className)
|
|
||||||
.setTimeout(timeout)
|
|
||||||
.setUuid(id)
|
|
||||||
.setIsActor(true)
|
|
||||||
.setIsOneWay(true)
|
|
||||||
.setIsEscaped(false)
|
|
||||||
if (sender.isDefined) {
|
|
||||||
val s = sender.get
|
|
||||||
requestBuilder.setSourceTarget(s.getClass.getName)
|
|
||||||
requestBuilder.setSourceUuid(s.uuid)
|
|
||||||
val (host, port) = s._replyToAddress.map(a => (a.getHostName, a.getPort)).getOrElse((Actor.HOSTNAME, Actor.PORT))
|
|
||||||
requestBuilder.setSourceHostname(host)
|
|
||||||
requestBuilder.setSourcePort(port)
|
|
||||||
}
|
|
||||||
RemoteProtocolBuilder.setMessage(message, requestBuilder)
|
|
||||||
remoteClient.send(requestBuilder.build, None)
|
|
||||||
}
|
|
||||||
|
|
||||||
override def postMessageToMailboxAndCreateFutureResultWithTimeout(
|
|
||||||
message: Any,
|
|
||||||
timeout: Long,
|
|
||||||
senderFuture: Option[CompletableFutureResult]): CompletableFutureResult = {
|
|
||||||
val requestBuilder = RemoteRequest.newBuilder
|
|
||||||
.setId(RemoteRequestIdFactory.nextId)
|
|
||||||
.setTarget(className)
|
|
||||||
.setTimeout(timeout)
|
|
||||||
.setUuid(id)
|
|
||||||
.setIsActor(true)
|
|
||||||
.setIsOneWay(false)
|
|
||||||
.setIsEscaped(false)
|
|
||||||
RemoteProtocolBuilder.setMessage(message, requestBuilder)
|
|
||||||
val future = remoteClient.send(requestBuilder.build, senderFuture)
|
|
||||||
if (future.isDefined) future.get
|
|
||||||
else throw new IllegalStateException("Expected a future from remote call to actor " + toString)
|
|
||||||
}
|
|
||||||
|
|
||||||
def receive = { case _ => {} }
|
|
||||||
}
|
|
||||||
|
|
||||||
*/
|
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* @author <a href="http://jonasboner.com">Jonas Bonér</a>
|
* @author <a href="http://jonasboner.com">Jonas Bonér</a>
|
||||||
*/
|
*/
|
||||||
|
|
@ -96,13 +47,13 @@ object RemoteClient extends Logging {
|
||||||
private val remoteActors = new HashMap[RemoteServer.Address, HashSet[String]]
|
private val remoteActors = new HashMap[RemoteServer.Address, HashSet[String]]
|
||||||
|
|
||||||
// FIXME: simplify overloaded methods when we have Scala 2.8
|
// FIXME: simplify overloaded methods when we have Scala 2.8
|
||||||
/*
|
|
||||||
def actorFor(className: String, hostname: String, port: Int): Actor =
|
def actorFor(className: String, hostname: String, port: Int): Actor =
|
||||||
actorFor(className, className, 5000, hostname, port)
|
actorFor(className, className, 5000L, hostname, port)
|
||||||
|
|
||||||
def actorFor(actorId: String, className: String, hostname: String, port: Int): Actor =
|
def actorFor(actorId: String, className: String, hostname: String, port: Int): Actor =
|
||||||
actorFor(actorId, className, 5000, hostname, port)
|
actorFor(actorId, className, 5000L, hostname, port)
|
||||||
*/
|
|
||||||
def actorFor(className: String, timeout: Long, hostname: String, port: Int): Actor =
|
def actorFor(className: String, timeout: Long, hostname: String, port: Int): Actor =
|
||||||
actorFor(className, className, timeout, hostname, port)
|
actorFor(className, className, timeout, hostname, port)
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -191,10 +191,22 @@ class RemoteServer extends Logging {
|
||||||
}
|
}
|
||||||
|
|
||||||
// TODO: register active object in RemoteServer as well
|
// TODO: register active object in RemoteServer as well
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Register Remote Actor by the Actor's 'id' field.
|
||||||
|
*/
|
||||||
def register(actor: Actor) = if (isRunning) {
|
def register(actor: Actor) = if (isRunning) {
|
||||||
log.info("Registering server side remote actor [%s] with id [%s]", actor.getClass.getName, actor.id)
|
log.info("Registering server side remote actor [%s] with id [%s]", actor.getClass.getName, actor.id)
|
||||||
RemoteServer.actorsFor(RemoteServer.Address(hostname, port)).actors.put(actor.id, actor)
|
RemoteServer.actorsFor(RemoteServer.Address(hostname, port)).actors.put(actor.id, actor)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Register Remote Actor by a specific 'id' passed as argument.
|
||||||
|
*/
|
||||||
|
def register(id: String, actor: Actor) = if (isRunning) {
|
||||||
|
log.info("Registering server side remote actor [%s] with id [%s]", actor.getClass.getName, id)
|
||||||
|
RemoteServer.actorsFor(RemoteServer.Address(hostname, port)).actors.put(id, actor)
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
case class Codec(encoder : ChannelHandler, decoder : ChannelHandler)
|
case class Codec(encoder : ChannelHandler, decoder : ChannelHandler)
|
||||||
|
|
|
||||||
|
|
@ -16,7 +16,7 @@ object ServerInitiatedRemoteActorServer {
|
||||||
def run = {
|
def run = {
|
||||||
val server = new RemoteServer()
|
val server = new RemoteServer()
|
||||||
server.start("localhost", 9999)
|
server.start("localhost", 9999)
|
||||||
server.register(new HelloWorldActor)
|
server.register("hello-service", new HelloWorldActor)
|
||||||
}
|
}
|
||||||
|
|
||||||
def main(args: Array[String]) = run
|
def main(args: Array[String]) = run
|
||||||
|
|
@ -25,9 +25,7 @@ object ServerInitiatedRemoteActorServer {
|
||||||
object ServerInitiatedRemoteActorClient extends Logging {
|
object ServerInitiatedRemoteActorClient extends Logging {
|
||||||
|
|
||||||
def run = {
|
def run = {
|
||||||
val actor = RemoteClient.actorFor(
|
val actor = RemoteClient.actorFor("hello-service", "localhost", 9999)
|
||||||
"sample.HelloWorldActor",
|
|
||||||
5000L, "localhost", 9999)
|
|
||||||
val result = actor !! "Hello"
|
val result = actor !! "Hello"
|
||||||
log.info("Result from Remote Actor: %s", result)
|
log.info("Result from Remote Actor: %s", result)
|
||||||
}
|
}
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue