El mejor método para echar un vistazo al buzón de Scala Actor

Question

Utilizando Scala 2.8 RC1 o posterior, ¿cuál es el mejor método (más fácil y directo) para "mirar" los mensajes que esperan en el buzón de un actor (desde el mismo actor método act()) para examinar lo que está en la cola, sin tener que reaccionar/recibir los mensajes y/o alterar el contenido actual del buzón de ninguna manera.

El propósito de esto es que un actor pueda determinar si es seguro procesar una solicitud de salida determinando primero si alguno de los mensajes de buzón restantes son los que deben procesarse, en lugar de simplemente dejar de actuar al actor inmediatamente.

Answer 1

No necesita mirar hacia adelante. Simplemente haga un seguimiento del hecho de que se ha solicitado una salida y use un reactionWithin (0) para determinar cuándo la cola está vacía después de haber solicitado una salida.

import scala.actors._ 

sealed case class Message 
case object Exit extends Message 
case class Unimportant(n:Int) extends Message 
case class Important(n:Int) extends Message 

class SafeExitingActor extends Actor { 
    def act : Nothing = react { 
     case Exit => { 
      println("exit requested, clearing the queue") 
      exitRequested 
     } 
     case message => { 
      processMessage(message, false) 
      act 
     } 
    } 

    // reactWithin(0) gives a TIMEOUT as soon as the mailbox is empty 
    def exitRequested : Nothing = reactWithin(0) { 
    case Exit => { 
     println("extra exit requested, ignoring") 
     exitRequested // already know about the exit, keep processing 
    } 
    case TIMEOUT => { 
     println("timeout, queue is empty, shutting down") 
     exit // TIMEOUT so nothing more to process, we can shut down 
    } 
    case message => { 
     processMessage(message, true) 
     exitRequested 
    } 
    } 

    // process is a separate method to avoid duplicating in act and exitRequested 
    def processMessage(message : Any, importantOnly : Boolean) = { 
    message match { 
     case Unimportant(n) if !importantOnly => println("Unimportant " + n) 
     case Unimportant(n) =>() // do nothing 
     case Important(n) => println("Important! " + n) 
    } 
    Thread sleep 100 // sleep a little to ensure mailbox backlog 
    } 
} 

object TestProcessing { 
    def main(args : Array[String]) { 
    val actor = new SafeExitingActor() 
    actor.start() 
    for (i <- 1 to 10) { 
     actor ! Unimportant(i) 
     actor ! Important(i) 
    } 
    actor ! Exit 
    for (i <- 11 to 20) { 
     actor ! Unimportant(i) 
     actor ! Important(i) 
    } 
    actor ! Exit 
    actor ! Important(100) 
    } 
} 

Eso debería de salida

Unimportant 1 
Important! 1 
Unimportant 2 
Important! 2 
Unimportant 3 
Important! 3 
Unimportant 4 
Important! 4 
Unimportant 5 
Important! 5 
Unimportant 6 
Important! 6 
Unimportant 7 
Important! 7 
Unimportant 8 
Important! 8 
Unimportant 9 
Important! 9 
Unimportant 10 
Important! 10 
exit requested, clearing the queue 
Important! 11 
Important! 12 
Important! 13 
Important! 14 
Important! 15 
Important! 16 
Important! 17 
Important! 18 
Important! 19 
Important! 20 
extra exit requested, ignoring 
Important! 100 
timeout, queue is empty, shutting down 

Answer 2

Esto suena como una operación peligrosa en general, ya que si hay mensajes críticos, el actor que los procesa podría verificar y no encontrar ninguno, pero antes de salir podría obtener otro de otro hilo.

Si sabe con certeza que esto no puede suceder, y no necesita muchos interruptores de mensajes increíblemente rápidos, probablemente escribiría un guardia-actor que cuenta y realiza un seguimiento de los mensajes críticos, pero de lo contrario simplemente pasa ellos a otro actor para el manejo.

Si no desea hacer eso, tenga en cuenta que los detalles de las partes internas deberían cambiar y es posible que tenga que pasar por el código fuente de Actor, Reactor, MessageQueue, etc. para obtener usted quiere. Por ahora, algo como esto debería funcionar (advertencia, no probado):

package scala.actors 
package myveryownstuff 
trait CriticalActor extends Actor { 
    def criticalAwaits(p: Any => Boolean) = { 
    synchronized { 
     drainSendBuffer(mailbox) 
     mailbox.get(0)(p).isDefined 
    } 
    } 
} 

Tenga en cuenta que tenemos que colocar el rasgo extendido en el paquete scala.actors, porque todas las partes internas de buzón se declaran privado a los scala.actors paquete. (Esta es una buena advertencia de que debe tener cuidado antes de meterse con las partes internas.) Luego agregamos un nuevo método que toma una función que puede probar un mensaje crítico y lo busca utilizando el método integrado mailbox.get(n) que devuelve el n th mensaje cumpliendo algún predicado.