Scala Actor : receiveWithin() doesn't receive messages

Question

我正在 Scala 中构建基于参与者的服务，消费者可以在其中查询客户端是否已获得授权，也可以向客户端授权。

如果消费者查询客户端的授权状态并且该客户端尚未获得授权，则参与者应在指定的超时时间内等待传入的 Authorize 消息，然后发送回复。 IsAuthorized 应该能够在消费者代码中同步执行，以便它阻塞并等待回复。有点像

service !? IsAuthorized(client) => {
  case IsAuthorizedResponse(_, authorized) => // do something
}

但是，我的 actor 中的 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;
}

当我在 receiveWithin block 中向 actor 发送 Authorize 消息时，消息被下面的第二个 case 语句捕获，实际上应该只在没有人等待时捕获这些消息当时的回应。

我的代码有什么问题？

更新:

这是相关代码的简化版本，它实际上代表了一个更简单和不同的逻辑，但可能更好地阐明了问题:

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
  }
}

更新 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)
  }
}

如果有更好的方法解决这个问题请告诉我!

Answer 1

不是回复问题吗？在

case IsAuthorized(client: Client) => reply { ... }

所有代码都在回复 block 的参数中，因此在实际发送回复之前执行(包括 receiveWithing)。这意味着当您的客户处理您的回复时，您将不再等待它。

在你的原始代码中，它可能应该是这样的

case IsAuthorized(client: Client) =>
  if(ok) reply(AuthorizedReply(client, true))
  else {
     reply(AuthorizedReply(client, false))
     receiveWithin(...)
  }