我正在Scala中构建一个基于actor的服务，消费者可以在其中查询客户端是否已获得授权并且还可以授权客户端 .
如果消费者查询客户端的授权状态并且该客户端尚未被授权，则该actor应该在指定的超时内等待传入的消息，然后发送回复 . IsAuthorized 应该能够在消费者代码中同步执行，以便阻止并等待回复 . 就像是
service !? IsAuthorized(client) => {
case IsAuthorizedResponse(_, authorized) => // do something
}
但是我的演员中的 receiveWithin() 永远不会收到消息并且总是会遇到超时 .
这是我的代码
case object WaitingForAuthorization
case class WaitingForAuthorizationResponse(clients: immutable.Set[Client])
case class IsAuthorized(client: Client)
case class IsAuthorizedResponse(client: Client, authorized: Boolean)
case class Authorize(client: Client)
class ClientAuthorizationService {
private val authorized: mutable.Set[Client] = new mutable.HashSet[Client] with mutable.SynchronizedSet[Client]
private val waiting: mutable.Set[Client] = new mutable.HashSet[Client] with mutable.SynchronizedSet[Client]
def actor = Actor.actor {
loop {
react {
case IsAuthorized(client: Client) => reply {
if (authorized contains client) {
IsAuthorizedResponse(client, true)
} else {
waiting += client
var matched = false;
val end = Instant.now.plus(ClientAuthorizationService.AUTH_TIMEOUT)
while (!matched && Instant.now.isBefore(end)) {
// ERROR HERE: Never receives Authorize messages
receiveWithin(ClientAuthorizationService.AUTH_TIMEOUT) {
case Authorize(authorizedClient: Client) => {
authorizeClient(authorizedClient)
if (authorizedClient == client) matched = true
}
case TIMEOUT => // do nothing since we handle the timeout in the while loop
}
}
IsAuthorizedResponse(client, matched)
}
}
case Authorize(client: Client) => authorizeClient(client)
case WaitingForAuthorization => reply {
WaitingForAuthorizationResponse(immutable.Set() ++ waiting)
}
}
}
}
private def authorizeClient(client: Client) = synchronized {
authorized += client
waiting -= client
}
}
object ClientAuthorizationService {
val AUTH_TIMEOUT: Long = 60 * 1000;
}
当我在播放器中发送 Authorize 消息时，它在receiveWithin块中，消息被下面的第二个case语句捕获，实际上只有当没有人在那个时候等待响应时才捕获这些消息 .
我的代码出了什么问题？
Update:
这是相关代码的缩短版本，它实际上代表了一个更简单和不同的逻辑，但可能更好地澄清了问题：
loop {
react {
case IsAuthorized(client: Client) => reply {
var matched = false
// In the "real" logic we would actually loop here until either the
// authorized client matches the requested client or the timeout is hit.
// For the sake of the demo we only take the first Authorize message.
receiveWithin(60*1000) {
// Although Authorize is send to actor it's never caught here
case Authorize(authorizedClient: Client) => matched = authorizedClient == client
case TIMEOUT =>
}
IsAuthorizedResponse(client, matched)
}
case Authorize(client: Client) => // this case is hit
}
}
Update 2:
我终于解决了这个问题 . 我认为问题是当试图在对前面的 IsAuthorized 消息的回复中收到 Authorize 消息时，该演员正在阻止 .
我重写了代码，以便在我们等待 Authorized 时启动匿名Actor . 这是感兴趣的人的代码 . waiting 是 Map[Client, Actor] .
loop {
react {
case IsAuthorized(client: Client) =>
if (authorized contains client) {
sender ! IsAuthorizedResponse(client, true)
} else {
val receipient = sender
// Start an anonymous actor that waits for an Authorize message
// within a given timeout and sends a reply to the consumer.
// The actor will be notified by the parent actor below.
waiting += client -> Actor.actor {
val cleanup = () => {
waiting -= client
exit()
}
receiveWithin(ClientAuthorizationService.AUTH_TIMEOUT) {
case Authorize(c) =>
receipient ! IsAuthorizedResponse(client, true)
cleanup()
case TIMEOUT =>
receipient ! IsAuthorizedResponse(client, false)
cleanup()
}
}
}
case Authorize(client: Client) =>
authorized += client
waiting.get(client) match {
case Some(actor) => actor ! Authorize(client)
case None =>
}
case WaitingForAuthorization => sender ! WaitingForAuthorizationResponse(immutable.Set() ++ waiting.keySet)
}
}
如果有更好的方法来解决这个问题，请告诉我！
2 回答
是不是回复了问题？在
所有代码都在回复块的参数中，因此在实际发送回复之前执行它（包括receiveWithing） . 这意味着当您的客户处理您的回复时，您将不再等待它 .
在您的原始代码中，它应该是类似的东西
我终于解决了这个问题 . 我认为问题是当试图在对前面的
IsAuthorized消息的回复中收到Authorize消息时，该演员正在阻止 .我重写了代码，以便在我们等待
Authorized时启动匿名Actor . 这是感兴趣的人的代码 .waiting是Map[Client, Actor].如果有更好的方法来解决这个问题，请告诉我！