atamborrino

import akka.actor.Scheduler
import scala.concurrent.{ExecutionContext, Future, Promise}
import scala.concurrent.duration.FiniteDuration

object FutureHelper {
  private val log = ScalaLogger.get(this.getClass)

  /**
    * Retry concurrently the async execution ''f'' every ''delay'' while no execution completes. 
    * Number of retries is bounded by ''maxSpeculativeExecutions''. 

atamborrino

import akka.NotUsed
import akka.stream.scaladsl.Source
import redis.clients.jedis.{Jedis, JedisCluster, ScanParams}

import scala.collection.JavaConverters._
import scala.concurrent.Future

class RedisStream(maxNodeParallelism: Int)
                 (implicit jedisCluster: JedisCluster,
                  blockingEC: RedisBlockingEC) {

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 {

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 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

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 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

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

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

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