Akka-Stream-Acknowledgements

ack-via-helper.scala

  val source = Source(RabbitSource(
    "Worker",
    rabbitMq,
    channel(qos = 100),
    consume(topic("app.domain.class", topics = List(Pg.Table("MainTable").all, Pg.Table("RelatedTable").all))),
    body(as[PgChange]).map(getPrimaryId)))

  source.
    map { case (p, primaryId) => Delivery(p, primaryId) }.
    mapConcat { delivery =>
      Delivery.collect(delivery :: Nil) { case Some(i) => i }
    }.
    groupedWithin(100, 60 seconds).
    mapConcat {
      _.groupBy(_.data).toList.map { case (primaryId, deliveries) =>
        Delivery.reduce(deliveries)(Keep.left)
      }
    }.
    mapAsync(4) { case d @ Delivery(_, primaryId) =>
      d.ack(api.pushChangesFor(primaryId))
    }