atamborrino

/// Working version
val fixedThreadPool = ExecutionContext.fromExecutorService(
  Executors.newFixedThreadPool(10)
)

case class KafkaRecord(partition: Int, value: Int)

val records = (1 until 1000).map(i => KafkaRecord(i % 10, i))

val flow = Flow[KafkaRecord].mapAsync(1) { i =>

atamborrino

def sequentialTraverse[A, B](as: Seq[A])(f: A => Future[B]): Future[Seq[B]] = {
  as.foldLeft(Future.sucessful(Vector.empty[B])) { (futAcc, a) =>
    for {
      acc <- futAcc
      b <- f(a)
    } yield acc :+ b
  }
}

atamborrino

scala> sealed trait Test
defined trait Test

scala> case class Test1(a: Boolean) extends Test
defined class Test1

scala> case class Test2(a: Boolean) extends Test
defined class Test2

scala> :paste

atamborrino

case class Data(s: String, i: Int)
object Circe {
  import io.circe._
  import cats.implicits._
  
  implicit val decoder: AccumulatingDecoder[Data] = (
    Decoder[String].prepare(_.downField("s")) |@|
    Decoder[Int].prepare(_.downField("i"))
  ).map(Data.apply)
   .accumulating

atamborrino

def mapPartition[A, B, C](as: Seq[A])(f: A => Either[B, C]): (Seq[B], Seq[C]) = {
  as.foldLeft((Vector.empty[B], Vector.empty[C])) { (acc, a) =>
    f(a) match {
      case Left(b) => (acc._1 :+ b, acc._2)
      case Right(c) => (acc._1, acc._2 :+ c)
    }
  }
}

val seq: Seq[Either[String, Int]] = Seq(Left(1), Right("a"))

atamborrino

import akka.stream.Materializer
import akka.stream.scaladsl._
import scala.concurrent.{Future, ExecutionContext}

// This will execute futures in parallel with a max parallelism
def traverse[A, B](as: Seq[A], parallelism: Int)(f: A => Future[B])
                  (implicit ec: ExecutionContext, mat: Materializer): Future[Seq[B]] = {
  Source(as.toList)
    .mapAsync(parallelism)(f)
    .runWith(Sink.seq)

atamborrino

import scala.concurrent._
import scala.concurrent.duration._
import scala.util.{Failure, Success}
import scala.util.control.NonFatal
import akka.actor.Scheduler
import java.util.concurrent.atomic.AtomicInteger
import akka.stream.ActorMaterializer
import akka.stream.scaladsl._

object FutureHelper {

atamborrino

import scala.concurrent.{ExecutionContext, Future}
import scala.util.control.NonFatal

// private constructor guarantees that we can not build a semantically incorrect UnfailableFuture
class UnfailableFuture[A] private (val value: Future[A]) extends AnyVal

object UnfailableFuture {
  def apply[A](fut: Future[A], fallback: Throwable => A)(implicit ec: ExecutionContext): UnfailableFuture[A] = {
    val futWithFallback = fut.recover {
      case NonFatal(failure) => fallback(failure)

atamborrino

object TestCats {

  import cats._
  import cats.data._
  import cats.instances.future._
  import cats.syntax.either._
  import cats.syntax.cartesian._
  import cats.instances.list._
  import cats.syntax.traverse._

atamborrino

import org.scalactic._
import org.scalactic.Accumulation._

trait Error
type Errors = Every[Error]
case object AnError extends Error

object ValidationImplicits {

  implicit class ValidationOps[E](val validLeft: Validation[Every[E]]) extends AnyVal {