Skip to content

Instantly share code, notes, and snippets.

/post.scala

Created May 31, 2017 19:59
Show Gist options
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save anonymous/cf1e5cdb071162b6bdcdc034e1510aaf to your computer and use it in GitHub Desktop.
Save anonymous/cf1e5cdb071162b6bdcdc034e1510aaf to your computer and use it in GitHub Desktop.
Download ZIP
the description for this gist
Raw
post.scala
monixObservable
.mapAsync(1) {
case CommitableMsg(MsgA(a), offset) =>
someBusinessTask(a).map(_ => offset)
case CommitableMsg(_, offset) =>
Task.now(offset)
}
.bufferTimedAndCounted(1.second, 100)
.filter(_.nonEmpty)
.mapAsync(1)(offsets => Task.fromFuture(offsets.last.commitOffset))
.takeWhileNotCanceled(injectedCancellable)
.consumeWith(Consumer.complete)
.runAsync(monix.execution.Scheduler.computation())
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment