akka/akka-core

Consider recipe or API for debounce

開放

#22,220 建立於 2017年1月25日

 (4 則留言) (0 個反應) (0 位負責人)Scala (3,547 個分叉)batch import
1 - triagedhelp wantedt:stream

倉庫指標

星標
 (13,277 顆星)
PR 合併指標
 (PR 指標待抓取)

描述

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

貢獻者指南