Creazione di un chunking basato sul tempo Enumerato

Question

Desidero creare un enumerato di riproduzione 2 che acquisisca valori e li emetta, suddivisi in blocchi, ogni x secondi/millisecondi. In questo modo, in un ambiente Websocket multiutente con un sacco di input dell'utente, si potrebbe limitare il numero di frame ricevuti al secondo.

So che è possibile raggruppare un certo numero di elementi insieme in questo modo:

val chunker = Enumeratee.grouped(
    Traversable.take[Array[Double]](5000) &>> Iteratee.consume() 
) 

Esiste un modo integrato per effettuare questa operazione in base al tempo piuttosto che in base al numero di articoli?

Stavo pensando di farlo in qualche modo con un lavoro Akka programmato, ma a prima vista questo sembra inefficiente, e non sono sicuro che si verifichino problemi di concurenza.

Answer 1

Che ne dici di questo? Spero che questo sia utile per te.

package controllers 

import play.api._ 
import play.api.Play.current 
import play.api.mvc._ 
import play.api.libs.iteratee._ 
import play.api.libs.concurrent.Akka 
import play.api.libs.concurrent.Promise 

object Application extends Controller { 

    def index = Action { 
    val queue = new scala.collection.mutable.Queue[String] 
    Akka.future { 
     while(true){ 
     Logger.info("hogehogehoge") 
     queue += System.currentTimeMillis.toString 
     Thread.sleep(100) 
     } 
    } 
    val timeStream = Enumerator.fromCallback {() => 
     Promise.timeout(Some(queue), 200) 
    } 
    Ok.stream(timeStream.through(Enumeratee.map[scala.collection.mutable.Queue[String]]({ queue => 
     var str = "" 
     while(queue.nonEmpty){ 
     str += queue.dequeue + ", " 
     } 
     str 
    }))) 
    } 

} 

E questo documento è anche utile per voi. http://www.playframework.com/documentation/2.0/Enumerators

UPDATE Questo è per la versione play2.1.

package controllers 

import play.api._ 
import play.api.Play.current 
import play.api.mvc._ 
import play.api.libs.iteratee._ 
import play.api.libs.concurrent.Akka 
import play.api.libs.concurrent.Promise 
import scala.concurrent._ 
import ExecutionContext.Implicits.global 

object Application extends Controller { 

    def index = Action { 
    val queue = new scala.collection.mutable.Queue[String] 
    Akka.future { 
     while(true){ 
     Logger.info("hogehogehoge") 
     queue += System.currentTimeMillis.toString 
     Thread.sleep(100) 
     } 
    } 
    val timeStream = Enumerator.repeatM{ 
     Promise.timeout(queue, 200) 
    } 
    Ok.stream(timeStream.through(Enumeratee.map[scala.collection.mutable.Queue[String]]({ queue => 
     var str = "" 
     while(queue.nonEmpty){ 
     str += queue.dequeue + ", " 
     } 
     str 
    }))) 
    } 

} 

Answer 2

Qui ho rapidamente definito un iteratee che avrà valori da un ingresso per una lunghezza t tempo fisso misurata in millisecondi e un enumeratee che permetterà di gruppo e ulteriore processo un flusso di input diviso in segmenti costruiti entro tale lunghezza t. Si affida a JodaTime per tenere traccia di quanto tempo è trascorso da quando è iniziata l'iteratee.

def throttledTakeIteratee[E](timeInMillis: Long): Iteratee[E, List[E]] = { 
    var startTime = new Instant() 

    def step(state: List[E])(input: Input[E]): Iteratee[E, List[E]] = { 
    val timePassed = new Interval(startTime, new Instant()).toDurationMillis 

    input match { 
     case Input.EOF => { startTime = new Instant; Done(state, Input.EOF) } 
     case Input.Empty => Cont[E, List[E]](i => step(state)(i)) 
     case Input.El(e) => 
     if (timePassed >= timeInMillis) { startTime = new Instant; Done(e::state, Input.Empty) } 
     else Cont[E, List[E]](i => step(e::state)(i)) 
    } 
    } 

    Cont(step(List[E]())) 
} 

def throttledTake[T](timeInMillis: Long) = Enumeratee.grouped(throttledTakeIteratee[T](timeInMillis))