Scala-Stream lesen, bis die Kriterien erfüllt sind

Question

Ich lese einen Stream, in dem ich in einem bestimmten Zeitintervall nach Json-Werten suche. Alles gelesen, vor und während der Werte im Zeitintervall möchte ich cachen. Sobald der Stream alle Werte im Zeitintervall gelesen hat (chronologische Reihenfolge vorausgesetzt), sollte ich aufhören zu lesen.

Dieser Code speichert alle Werte in einer veränderbaren Liste und filtert dann nur nach json im Datumsbereich. Aber es hört nicht auf zu lesen, sobald der Filter abgeschlossen ist, es speichert alles im Stream zwischen.

val cache: MutableList[JsValue] = MutableList.empty 

pg.pins.flatten 
    .map(record => {cache += record; record}) 
    .filter(js => (js \ "created_at").as[Date].after(start) && (js \ "created_at").as[Date].before(end)) 

Jede Hilfe wird geschätzt.

Answer 1

Sie brauchen keine veränderbare Liste dafür (vermeiden Sie die Verwendung von veränderbaren Strukturen, es sei denn, es gibt einen wirklichen Grund, warum Sie müssen). Unter der Annahme der pg.pins ist ein Stream und dass die Daten nach dem Datum sortiert wird (das Ihre Frage impliziert zu sein scheint), so etwas wie dies tun sollten, was Sie wollen:

val (after, before) = pg.pins.flatten 
    .takeWhile { js => (js \ "createdAt").as[Date].before(end) } 
    .partition { js => (js \ "createdAt").as[Date].after(start) }