dsebban

// val skunkCore              = "0.3.2"
// val zioCats                = "3.3.0"
// val zioStreams             = "2.0.0"
// val skunkCore               = "org.tpolecat"                  %% "skunk-core"                    % V.skunkCore
// val zioCats                 = "dev.zio"                       %% "zio-interop-cats"              % V.zioCats
// val zioStreams              = "dev.zio"                       %% "zio-streams"                   % V.zioStreams
object Cats {
  import skunk._
  import skunk.implicits._
  import skunk.codec.all._

dsebban

def task(id: Int): Runnable = () => 
{
  println(s"${Thread.currentThread().getName()} start-$id")
  Thread.sleep(10000)
  println(s"${Thread.currentThread().getName()} end-$id")
}

dsebban

A Streaming library with a superpower: Functional Programming

Scala has a very unique streaming library called FS2 (Functional Streams for Scala). This library embodies all the advantages of functional programming(FP). By understanding its design goals you will get exposure to the core ideas that makes FP so appealing.

FS2 has one central type:

Stream[Effect,Output]

dsebban

import $ivy.`org.typelevel::cats-core:1.6.0`, cats.implicits._

def rule[A](n: Int, default: A): Int => Option[A] = x => if ( x % n == 0) default.some else None
val fizz = rule(3,"Fizz")
val buzz = rule(5,"Buzz")
val boom = rule(7,"Boom")

def fizzBuzz(rules: List[Int => Option[String]]): List[Int] => List[String] = xs => {
   val rule: Int => Option[String] = rules.foldMap(identity)
   xs.map(x => rule(x).getOrElse(x.show))

dsebban

def flip[A,B,C]: (A => B => C) => (B => A => C) = f => x => y => f(y)(x)
def f: Int => String => Int=  a => b => a + b.toInt
flip(f)("1")(2)
def ++ : String => String => String = a => b => a + b
flip(++)("Hello")("World")

dsebban

def notSafeFunc: String  => Int = _.toInt
notSafeFunc("1") // 1
notSafeFunc("a") // java.lang.NumberFormatException: For input string: "a"
import $ivy.`org.typelevel::cats-core:1.5.0`
import cats.implicits._
import cats.Applicative

def makeSafe[A,B,F[_]: Applicative](f: A => B)(implicit F: Applicative[F]): A => F[B] = a => F.lift(f)((F.pure(a)))

dsebban

 import $ivy.`org.typelevel::cats-core:1.5.0`
 import cats.Contravariant
 import cats.implicits._

{
  trait Predicate[A] {
    def getPredicate(a:A): Boolean
  }
  object Predicate {
    def apply[A](f: A => Boolean): Predicate[A] = new Predicate[A]{ def getPredicate(a: A) = f(a) }

dsebban

import $ivy.`com.lightbend.akka::akka-stream-alpakka-dynamodb:0.20`
import akka.actor.ActorSystem
import com.typesafe.config.Config
import com.typesafe.config.ConfigFactory
import akka.stream.ActorMaterializer
import akka.stream.alpakka.dynamodb.scaladsl._
import akka.stream.alpakka.dynamodb.impl.DynamoSettings
import akka.stream.scaladsl._
import scala.concurrent.{ Await , Future }
import scala.concurrent.duration._

dsebban

import $ivy.`co.fs2:fs2-core_2.12:0.10.5` , fs2._

def dedupeConsecutiveWhen[F[_], I](f: (I, I) => Boolean): Pipe[F, I, I] = in =>
    in.zipWithNext.map {
       case (curr, Some(next)) if f(curr, next) => (None: Option[I])
       case (curr, Some(next))                  => Some(curr)
       case (curr, None)                        => Some(curr)
        }.unNone

Stream(1,1,2,2,3,4,5,2).through(dedupeConsecutiveWhen(_ == _ )).toList

dsebban

Undertsanding the Pull type from fs2

From the scaladocs

class Stream[+F[_], +O] extends AnyVal

• A stream producing output of type O
• May evaluate F effects.