akka/akka-core

Add an optional buffer/parallelism parameter for actorRefWithAck

Aberta

#26.129 aberto em 17 de dez. de 2018

 (6 comentários) (0 reação) (0 responsável)Scala (3.547 forks)batch import
1 - triagedhelp wantedt:stream

Métricas do repositório

Stars
 (13.277 estrelas)
Métricas de merge de PR
 (Mesclagem média 8d 19h) (10 fundiu PRs em 30d)

Description

Flow.ask currently takes in a parallelism factor as an optional parameter. This allows messages to buffer in the destination ActorRef's mailbox. E.g.,

def myFlow(ref: ActorRef) = {
  Source(1 to 100)
    .ask[Response](4)(ref) //allows 4 messages to send before waiting for a Response
    .to(Sink.ignore)
}

Sink.actorRefWithAck does not have any such functionality. Messages do not buffer.

def myAckFlow(ref: ActorRef) = {
  Source(1 to 100)
    //waits for a Response before sending the next element
    .runWith(Sink.actorRefWithAck(ref,Init,Response,Completed, ex => Error(ex))
}

Johan and I discussed this briefly, and he felt like it would be a reasonable thing to add a parallelism (or buffer) parameter to actorRefWithAck.

Guia do colaborador