1 - triagedhelp wantedt:stream
Repository metrics
- Stars
- (13,277 stars)
- PR merge metrics
- (Avg merge 8d 19h) (10 merged PRs in 30d)
Description
Recently someone asked me for a way to "not send the same thing 100 times to an actor", this is basically the debounce operation.
It can be implemented such:
import scala.concurrent.duration._
/**
* Groups items within a given time interval (unless the max size is reached, then earlier),
* and picks the single element to signal downstream using the provided `pick` function.
*/
def debounceSelect[A](interval: FiniteDuration, pick: immutable.Seq[A] => A, max: Int = 100) =
Flow[A].groupedWithin(max, interval).map { group => pick(group) }
case class RefreshSignal(msg: Any)
val queue: SourceQueueWithComplete[RefreshSignal] =
Source.queue[RefreshSignal](4, OverflowStrategy.dropHead)
.via(debounceSelect(1.second, _.head)) // picking any of the refresh signals
.to(Sink.actorRef(target, onCompleteMessage = RefreshSignal("done")))
.run()
val offer1: Future[QueueOfferResult] = queue.offer(RefreshSignal("refresh"))
val offer2: Future[QueueOfferResult] = queue.offer(RefreshSignal("refresh"))
val offer3: Future[QueueOfferResult] = queue.offer(RefreshSignal("refresh"))
val offer4: Future[QueueOfferResult] = queue.offer(RefreshSignal("refresh"))
Shall we either make it a recipe or even add as an API? I think i've bumped into this a few times already