SystemFw

/*
 Here's a derivation for a tail recursive fibonacci function.
 Let's start from the Maths definition:
   fib(0) = 0
   fib(1) = 1
   fib(n) = fib(n - 1)  + fib(n - 2)
*/


// Let's translate the structure, leaving the recursive case unspecified:

SystemFw

-- The assumption here is that stream based I/O is a better fit for 
-- distributed systems than monadic I/O, because there is a greater
-- separation between logic, which can be idempotent and distributed
-- via something like differential dataflow, and actions, which can 
-- be represented like ports and actuators

-- We already know that stream based I/O is a good fit for reactive programs
-- but what about more "imperative" ones that fit monads well? I took an imperative
-- example and tried to express it via stream I/O

SystemFw

object Ex {
  import Lib._

  case class Person(name: String, age: Int)
  case class PersonOpt(name: Option[String], age: Option[Int])

  def mergePerson(p: Person, po: PersonOpt) =
    Person(po.name.getOrElse(p.name), po.age.getOrElse(p.age))

  case class Contact(person: Person, phone: String)

SystemFw

// stemming from
// https://gitter.im/typelevel/general?at=5f987f8161007f7d1b9b4b92
// The idea is
// read the JSON files in a directory
// try to decode each of them as Event which could be either One or Two, we don't know this at compile time
// serialize the values we got, again could be either One or Two and compare it with the content of the JSON file
// verify that there was at least a JSON file for One and Two (exhaustivity check)

import shapeless._

SystemFw

// This is the traditional encoding, using abstract type members
// to encode existentials
object AbstractExistentials {
  trait E {
    type X
    val x: X
    def f(p: X): String
  }
   

SystemFw

import shapeless._

@annotation.implicitNotFound("Make sure all cases of ${C} have an instance of ${TC}")
trait CopTCs[C <: Coproduct, TC[_]] {
  type Out <: HList
  def result: Out
}
object CopTCs {
  type Aux[C <: Coproduct, TC[_], O <: HList] = CopTCs[C, TC] { type Out = O }
  def apply[C <: Coproduct, TC[_]](implicit ev: CopTCs[C, TC]) = ev

SystemFw

 // Grows with the number of distinct `K`
  def partitions[F[_], A, K](selector: A => F[K])(implicit F: Effect[F], ec: ExecutionContext)  : Pipe[F, A, (K, Stream[F, A])] = in => 
  Stream.eval(async.refOf[F, Map[K, Queue[F, Option[A]]]](Map.empty)).flatMap { st =>
    val cleanup = {
      import alleycats.std.all._
      st.get.flatMap(_.traverse_(_.enqueue1(None)))
    }

    (in ++ Stream.eval_(cleanup)).evalMap { el =>
      (selector(el), st.get).mapN { (key, queues) =>

SystemFw

cats-effect

The cats-effect project defines a purely functional effect type (IO[A]), and associated typeclasses defining its behaviour. The ones we care about for this example are:

trait Sync[F[_]] extends MonadError[F, Throwable] {
   def delay[A](a: => A): F[A]
   ...
}

SystemFw

object Test {
  import inc._

  case class Foo(i: Int, s: String)
  case class Bar(f: Int, a: Foo, n: Int)

  val res = Bar(0, Foo(1, "hello"), 2).incBy(2)
  //res0: Test.Bar = Bar(2,Foo(3,hello),4)
}

SystemFw

object Test {
  import cats.implicits._
  import cats.kernel.CommutativeMonoid
  import cats.effect._
  import fs2._
  import scala.concurrent.ExecutionContext

  // same as parallelFoldMonoid, but with some logging to show the parallelism
  def loggedParallelFoldMonoid[F[_], A: CommutativeMonoid](n: Int)(
      input: Stream[F, A])(implicit F: Effect[F], ec: ExecutionContext) =