Loïc's Blog

A Scala REST API in 2026, and beyond

2026-03-11T00:00:00+00:00

Direct-Style REST in Scala 3: Virtual Threads, sttp, and the Road to Caprese

The conventional wisdom in Scala has long been: if you do I/O, you need Future (or an effect type like IO). Your return types get wrapped, your signatures are impacted, and every layer of the stack has to “speak async”. Project Loom changes that equation. This post walks through a REST service that drops Future from the domain layer entirely — and closes with a look at where Caprese / saferExceptions takes things next.

Disclaimer: IO monads have other benefits such as referential transparency that are not discussed in this post.

The Stack

Layer	Technology
HTTP server	Apache Pekko HTTP 1.3.0
HTTP client	sttp client4 `DefaultSyncBackend`
Error handling	`Either[AppError, A]`
DI	MacWire (compile-time)
Runtime	JDK 21 virtual threads

Why Virtual Threads Change the Signature

With Future, the effect type is viral: every layer that calls async code must return Future[_], carry an implicit ExecutionContext, and use map/flatMap for composition.

// The "classic" async style — effect type spreads everywhere
trait UserApiClient:
  def fetchUser(id: Int)(using ExecutionContext): Future[Either[AppError, User]]

trait UserService:
  def getUserWithPosts(id: Int)(using ExecutionContext): Future[Either[AppError, UserWithPosts]]

Virtual threads (JEP 444, JDK 21) break the assumption that drove this pattern. When a virtual thread blocks on I/O, the JVM simply parks it and reassigns the underlying OS thread to another task. Blocking is free. A synchronous call that waits for a network response no longer monopolises an OS thread.

sttp’s DefaultSyncBackend is built on java.net.HttpClient and blocks the calling thread. On a virtual thread, that cost disappears:

// Direct-style: no Future, no ExecutionContext
trait UserApiClient:
  def fetchUser(id: Int): Either[AppError, User]

trait UserService:
  def getUserWithPosts(id: Int): Either[AppError, UserWithPosts]

The I/O still happens; it just doesn’t need to be announced in the type signature.

Architecture

HTTP Request
    │
    ▼
Pekko HTTP (NIO event-loop)         ← OS threads, non-blocking I/O
    │
    │   Future { ... }              ← hand off to loom Execution Context
    ▼
Virtual Thread (Loom)               ← one per request, ~1 KB heap
    │
    ├─► UserService.getUserWithPosts  ← plain synchronous call
    │       │
    │       ├─► sttp DefaultSyncBackend.send(request)
    │       │       VT parked during network wait ✓
    │       │       OS thread freed ✓
    │       │
    │       └─► Either[AppError, UserWithPosts]
    │
    ▼
Pekko HTTP completes the Future and sends the HTTP response

Pekko HTTP is a reactive framework built on Pekko Streams — it cannot natively dispatch handlers on virtual threads. The solution is a thin boundary in the route layer:

// UserRoutes.scala — the ONLY place Future appears
complete:
  Future:                                   // dispatched on loom Execution Context
    userService.getUserWithPosts(userId) match
      case Right(result) => OK  -> result.asJson.noSpaces
      case Left(e: AppError.NotFound) => NotFound -> e.asJson.noSpaces
      case Left(e) => InternalServerError -> e.asJson.noSpaces

Everything inside the Future { } block is synchronous, direct-style code.

Error Handling with `Either`

The service layer uses a sealed AppError ADT and Either for typed, composable error propagation. The for-comprehension on Either short-circuits on the first Left:

def getUserWithPosts(id: Int): Either[AppError, UserWithPosts] =
  for
    user  <- getUser(id)                        // Left here → stops immediately
    posts <- userApiClient.fetchPostsForUser(id) // only runs on Right
  yield UserWithPosts(user, posts.map(postToDomain(user.id)))

Compared to Future[Either[_, _]], this reads like ordinary sequential code. The compiler still enforces exhaustive error handling at every call site.

MacWire: Zero-Reflection DI

MacWire resolves the dependency graph at compile time via macros:

trait AppModule:
  lazy val userApiClient: UserApiClient = wire[UserApiClientImpl]
  lazy val userService: UserService     = wire[UserServiceImpl]
  lazy val userRoutes: UserRoutes       = wire[UserRoutes]

A missing dependency or type mismatch is a compile error, not a runtime crash. There is no reflection, no classpath scanning, no startup overhead.

Testing

Because the service layer is purely synchronous, tests need no async machinery at all:

// No ScalaFutures, no PatienceConfig, no .futureValue
"UserService" should "short-circuit Left when user not found" in:
  val svc = makeService(userStub = _ => Left(AppError.NotFound("not found")))
  svc.getUserWithPosts(99) shouldBe Left(AppError.NotFound("not found"))

The sttp stub also runs synchronously:

val stub = SyncBackendStub
  .whenRequestMatches(_.uri.toString.contains("/users/1"))
  .thenRespondAdjust(sampleUser.asJson.noSpaces)

val client  = UserApiClientImpl("https://api.example.com", stub)
val service = UserServiceImpl(client)
service.getUser(1) shouldBe Right(User(UserId("1"), sampleUser.name, sampleUser.email))

Identity is the effect type for synchronous backends — no wrapper, no monad, just a plain value.

The Road Ahead: Caprese and `saferExceptions`

Caprese is Martin Odersky’s research program on capabilities and effects for Scala. Its first concrete output in the compiler is language.experimental.saferExceptions (available since Scala 3.1, stabilising toward 3.9 LTS), along with language.experimental.captureChecking (Scala 3.4+, improved in 3.8).

The core idea: instead of wrapping errors in a container type (Either, Option, Try), you declare the capability to throw as a compile-time-only implicit parameter. The CanThrow[E] capability is erased before code generation — zero runtime cost.

What the current service looks like with `Either`

// Current approach — Either monad
def fetchUser(id: Int): Either[AppError, User] =
  try
    val response = request.send(backend)
    response.body match
      case Right(user) => Right(toDomain(user))
      case Left(_)     => Left(AppError.NotFound(s"User $id not found"))
  catch
    case ex: Exception => Left(AppError.Unexpected(ex.getMessage, ex))

Error types live in the return value. The caller must unwrap Either explicitly.

The same code with `saferExceptions` (Scala 3.x experimental)

import scala.language.experimental.saferExceptions

// Typed exception classes — must extend Exception, not RuntimeException
class NotFoundException(msg: String)         extends Exception(msg)
class ExternalServiceException(msg: String)  extends Exception(msg)

// The `throws` clause is syntactic sugar for (using CanThrow[E])
def fetchUser(id: Int): User throws NotFoundException | ExternalServiceException =
  val response = request.send(backend)
  response.body match
    case Right(user) => toDomain(user)
    case Left(_) if response.code.code == 404 =>
      throw NotFoundException(s"User $id not found")
    case Left(err) =>
      throw ExternalServiceException(err.getMessage)

And the service layer composes naturally:

def getUserWithPosts(id: Int): UserWithPosts
    throws NotFoundException | ExternalServiceException =
  val user  = fetchUser(id)        // throws propagate automatically
  val posts = fetchPostsForUser(id)
  UserWithPosts(user, posts.map(postToDomain(user.id)))

The throws clause desugars to (using CanThrow[NotFoundException | ExternalServiceException]). This capability is erased at runtime — it exists only to make the compiler verify that every call site either handles the exception or declares it in its own throws clause.

The route layer becomes the explicit error boundary:

complete:
  Future:
    try
      val result = userService.getUserWithPosts(userId)
      OK -> result.asJson.noSpaces
    catch
      case _: NotFoundException        => NotFound      -> ...
      case _: ExternalServiceException => ServiceUnavailable -> ...

`Either` vs `saferExceptions` — a practical comparison

	`Either[AppError, A]`	`saferExceptions` (`throws`)
Error visibility	In return type	In `throws` clause
Composition	`for`-comprehension	Sequential (direct-style)
Runtime cost	Allocation of `Right`/`Left`	Zero (erased)
Exhaustiveness	Pattern match on `Either`	Compiler checks `CanThrow` at call sites
Maturity	Stable, production-ready	Experimental

Readers familiar with Java’s checked exceptions will notice the surface similarity: both systems force the compiler to track which errors a method can produce. Java checked exceptions issues that Caprese aims to avoid are discussed here.

The bigger picture: `captureChecking`

saferExceptions is just one application of Caprese’s broader capability model — and a limited one: it cannot track exceptions thrown inside lambdas passed to higher-order functions. captureChecking lifts this restriction and generalises beyond exceptions: with language.experimental.captureChecking, the compiler can verify that any capability — as an example, a database connection — does not escape the scope where it was introduced. See this page for more details.

Summary

Concern	Solution	Key benefit
Non-blocking I/O	JDK 21 virtual threads	No async wrapper needed
HTTP client	`sttp DefaultSyncBackend`	Plain synchronous calls
Error handling	`Either[AppError, A]`	Typed, composable, stable
DI	MacWire	Compile-time safety
Future direction	Caprese `saferExceptions`	Zero-cost, direct-style typed errors

The architecture makes a deliberate bet: Loom removes the need for async types in the domain layer. The Future boundary is pushed to the very edge — where Pekko HTTP hands off to the application — and the rest of the code is plain, readable, synchronous Scala. As Caprese matures, Either can gradually give way to throws, further simplifying return types while keeping the same compile-time safety guarantees.

Some Scala & ZIO tips

2024-01-19T00:00:00+00:00

To start this new year, let’s see some Scala and ZIO tips snippets!

How to cache queries with ZIO and ZQuery

Let’s say we have some employees and departments data.
For each employee I want departement info, but I don’t want to query multiple time for the same departement.

How to write a generic micro service with Scala 3

Here, I want to write a service that would be able to read, transform and write any type of data, as soon as the data types and the transformers are defined.

To learn more about the transformer pattern, you can read this post about type classes.

Happy new year everyone!

How to define a flexible String decoder with ZIO JSON

2023-01-20T00:00:00+00:00

I am using ZIO JSON to decode responses from a Web service and I’ve faced an issue : some objects have fields that can sometimes be strings and sometime be integers.

This can be solved with decoders orElse method like this :

case class SoftwareVersion(
                            name: String,
                            // ...
                          )

case class SoftwareVersionWithIntName(
                            name: Int,
                            // ...
                          )

object Decoders{
  implicit val softwareVersionDecoder = DeriveJsonDecoder.gen[SoftwareVersion].orElse(
    DeriveJsonDecoder.gen[SoftwareVersionWithIntName].map(decoded =>
      SoftwareVersion(decoded.name.toString)
    )
  )
}

It is working but if there is a lot of changing fields, it may lead to a lot of cases to manage.

To make it work at any level with the minimum amount of code, you can hack the String decoder itself :

implicit val flexStringDecoder = JsonDecoder[String].orElse(
  JsonDecoder[Int].map(decoded => decoded.toString)
)

Use it with caution!

The same trick should also work with Circe.

All We Leave Behind (my doom metal band)

2022-11-08T00:00:00+00:00

For once I’m not going to talk to you about code but about music. I play guitar in a doom metal band called All We Leave Behind.

Here is a small video extract of our last gig :

You can also follow us on Instagram at this address to be informed about new track releases or concert dates : All We Leave Behind Official

Type classes and extension methods in Scala 3

2022-09-29T00:00:00+00:00

Type classes in Scala

A type class is a type system construct that supports “ad hoc polymorphism”. It defines features than can be added to any type A if an instance of the type class for the A type is provided.

Type class definition

Let’s define a type class with an “add” method :

trait Addable[A]:
  extension (a1: A) def add(a2: A): A

The extension keyword means that we will augment the capabilties of a type A with an add method, that takes another A to produce a sum (also of type A).

Instances

We also need instances to use this type class with concrete types. We will start to define instances to add integers and strings :

given Addable[Int] with
  extension (i1: Int) def add(i2: Int) = i1 + i2

given Addable[String] with
  extension (s1: String) def add(s2: String) = s"$s1 . $s2"

Then we can use the extension method, directly on any integer or string :

println(
  "Hello".add("How are you?")
)

// This prints: Hello . How are you?

We can add this capability to any data type, without changing the type directly. If you know Domain Driven Design (DDD) you know it is important to be able to keep the domain pure. Type classes are very useful for this in a lot of cases, like JSON serialization for example (Scala JSON librairies defined typeclass that offer toJson methods, which can be added to any types).

It is also possible to add given instances to data type you can’t change (from libs/external modules for example).

For example, if we have a Color type, defined like that :

  enum Color :
    case Red 
    case Green
    case Blue
    case Purple
    case Yellow
    case Unknown

Without changing the Color type, we can add new capabilities to it :

given Addable[Color] with
  extension (c1: Color) def add(c2: Color) = 

   // dumb color addition example
   (c1, c2) match {
     case (Color.Red, Color.Green) => Color.Yellow
     case (Color.Red, Color.Blue) => Color.Purple
     //... etc.
     case _=> Color.Unknown
   }

println(
  Color.Red.add(Color.Blue) 
)

// This prints: Purple

More powerful functions

Finally, we can write methods that require a type with some given instances to work, so we can make operations using the augmented capabilities of this type.

For example, we can write a addAll method, that will take any list of A, if there a given instance of Addable[A], and make the sum of all the A.

def addAll[A: Addable](l: List[A]) = l.reduceLeft(((total, current) => total.add(current)))

// or simply :
def addAll[A: Addable](l: List[A]) = l.reduceLeft(_.add(_))

println(addAll(List("Coucou", "comment", "ça", "va")))
   
// This prints : Coucou . comment . ça . va

A: Addable means that an instance of Addable[A] must be defined.

Note : this is exactly how the sum method of the Scala List type works.

How to traverse optional values in Scala, Kotlin and Java

2022-09-09T00:00:00+00:00

Let’s study a real life, quite simple problem :

We have a list of optional values, and we want to produce an optional single output, made from present values (if there is any). We will also join values using a separator character.

Exemple 1 : Using a List(Some("A"), None, Some("B")) should produce Some("A,B").

Exemple 2 : Using List(None) should produce None.

Scala version

We can use the flip method from ZIO-prelude (or the sequence method from Cats) to turn a List[Option[String]] into a Option[List[String]].

val list = List(Some("A"), None, Some("B"))
val flippedList = list.filter(_.isDefined).flip // <- Option(List("A","B"))
flippedList.map(_.mkString(",")) // <- Some("A,B")

Note that if we just wanted to produce a String (that may be empty), we could have done this to remove the Option wrapper and undefined values :

val flattenList = List(Some("A"), None, Some("B")).flatten // <- List("A","B")
flattenList.mkString(",")

Kotlin version

In Kotlin, we usually just use nullable types to replace Option.

val list = listOf("a", null, "b")
val result = list.filterNotNull().joinToString(separator=",").ifBlank { null }

Java version

In Java we will use the stream API, which is a bit more verbose but still working fine. Stream::flatMap will remove all the undefined elements and remove the Optional wrapper (like Scala’s flatten).

List<Optional<String>> list = List.of(Optional.of("A"), Optional.empty(), Optional.of("B"));
     
var result = Optional.of(
         list.stream()
        .flatMap(Optional::stream)
        .collect(Collectors.joining( "," ))
).filter(Predicate.not(String::isEmpty));

Memoization with ZIO

2021-04-21T00:00:00+00:00

ZIO tip : How to memoize the result of an effectful computation :

ZIO values are lazy and referentially transparent. So a IO[E,A] will be computed each time it needs to be evaluated.
It is a very intersting property in functional programming, but sometimes for long and cachable computations, it can be useful to memoize a result. It can be done using ZIO.memoize.

Here is an example :

val compute: Int => UIO[Int] = (x: Int) => ???
for {
      m <- ZIO.memoize(compute)
      a <- m(1)
      b <- m(1)
      c <- m(1)
      d <- m(2)
} yield(a, b, c, d)

With this code, m(1) result will only be computed 1 time.

ZIO HTTP basic app example

2021-03-17T00:00:00+00:00

ZIO-HTTP is a new ZIO lib for HTTP applications creation.
It integrates very naturally with ZIO.

Here is a very simple example using ZIO effects and dependency injection with ZLayer :

Data Validation in Scala with ZIO Prelude

2020-12-20T00:00:00+00:00

ZIO Validation is a new library made to help functional programming with algebraic constructions in Scala.
It aims to be simpler and to fit better to the Scala language than Cats and Scalaz.

Here is a simple data validation example (using Scala 3 indentation-based syntax) :

As you can see it is really easy to mix monadic error handling (stop at first error) with Either (or Option) and validation of multiple errors (parallel validation instead of stpping at first error) with the ZIO Prelude Validation data type.

P.S. : I’d like to thank the Univalence team for their youtube channel and their unboxing video of ZIO Prelude.

Resources to run Spark and HDFS in Kubernetes

2020-12-17T00:00:00+00:00

Apache Spark and the Hadoop ecosystem are not really easy to install and manage manually, but using a few resources it is quite convienent to deploy an HDFS + Spark cluster on Kubernetes and submit applications on it.

HDFS

You can use the Gradiant Helm charts provided here to install HDFS in your cluster : https://github.com/Gradiant/charts. It is possible to override the number of datanodes and other settings using a values.yaml file, see all the defaults here.

Spark

Gradiant is also providing Spark images : https://hub.docker.com/r/gradiant/spark.

Submit apps

You can use a local Spark installation to run spark-submit, or use it directly from the Docker image as suggested in the Gradiant image README file.

Example :

spark-submit \
    --master k8s://https://127.0.0.1:6443 \
    --deploy-mode cluster \
    --name hello-spark \
    --class Hello \
    --conf spark.executor.instances=2 \
    --conf spark.kubernetes.container.image.pullPolicy=IfNotPresent \
    --conf spark.kubernetes.authenticate.driver.serviceAccountName=spark \
    --conf spark.kubernetes.container.image=gradiant/spark:2.4.4 hdfs://hdfs-namenode/user/loic/jars/helloSpark.jar

As you can see, once the HDFS service are deployed in Kubernetes, you can use it in the cluster using the ‘hdfs-namenode’ service. (Use kubectl get services to have the list of the deployed services).
HDFS can be reached from your Spark applications in the same way. Please note that you will need to create a Kubernetes service account with permissions to create pods and services. See the image README file for more details.

To put your application jar file in HDFS, you will be able to use the httpfs service included in the HDFS Helm chart. The WebHDFS REST API documentation can be found here : https://hadoop.apache.org/docs/r1.0.4/webhdfs.html.

Finally, it is possible to add custom labels to driver and executor pods, set cpu resources limits etc., using this configuration keys : https://spark.apache.org/docs/latest/running-on-kubernetes.html#configuration.

Enjoy !

A ZIO Akka HTTP giter8 template

2020-05-19T00:00:00+00:00

I have just published a small Scala project : a ZIO Akka HTTP giter8 template, using Zlayer for dependency injection, Circe for JSON and Doobie for DB access.

You can find it on this Github repository.

To generate a new Akka HTTP ZIO app, run the following command : sbt new KaizenLab-ScalaFP/zioAkkaHttpTemplate.g8.

How to add elements dynamically to a ZIO stream using a queue

2019-12-03T00:00:00+00:00

Here is a small example of parallel pull/push into a ZIO queue. The data is consumed via a stream created from the queue.

Scala snippets (ZIO, Cats, Slinky, Akka Streams)

2019-11-07T00:00:00+00:00

I write this snippets as memory aid, but maybe it can be useful for someone else :)

Change delimiter in a ZIO Stream

Example, group/split by lines :

Accumulate errors with Either and Cats

Based on Cats documentation but with Error ADT and NonEmptyChain

Slinky React Hooks example

Akka Streams : Consume a source with two sinks and get the result of one sink

Akka Streams and Play Framework : add elements to a source dynamically, using a queue

HTTP streaming in Scala with STTP and ZIO

2019-05-08T00:00:00+00:00

It is now possible to stream HTTP responses with Scala STTP client using ZIO and ZIO streams.

I’ve also updated my STTP streaming examples with a ZIO use case.

How to manage UI state with immutable data structures in Scala

2019-03-21T00:00:00+00:00

A friend asked me how he could create a chess game in Scala, without relying on mutable data structures. I made a small user interface example, that displays a state for each user input, depending on the previous state, without any mutation.

Here is the sample code :

Playing with ReasonML and ReasonReact

2019-02-24T00:00:00+00:00

This days I’ve decided to take a look at the Reason language. This is a very nice functional programming language developped by Facebook and based on OCaml.

A good way to play with this language is to try ReasonReact, the Reason version of React. You can find a small example I did here.

As you can see, there is a lot of nice things for a Scala developer, like pattern matching or the Option type.

This example can be compared with the similar example I’ve done using Binding.scala or with React.js.

Introduction to functional programming and error management in Kotlin and TypeScript (french)

2019-02-24T00:00:00+00:00

After an introduction to functional programming and error management in Scala, I’ve written a post on my company’s blog about Arrow (a Kotlin FP library) and fp-ts(a TypeScript FP library).
This posts are both written in french language. I hope it can help to let people start using FP in mainstream languages :)

Lightbend VS Typelevel at ScalaIO - code samples and slides

2018-11-04T00:00:00+00:00

Last week, I’ve given a talk at the awesome ScalaIO conference, called “Lightbend vs Typelevel”.
I’ve published the streaming code examples on github.
You can also find the slides (in french) here.

Smart function application in Scala using currying and pipe operator

2018-07-29T00:00:00+00:00

First, let’s define a pipe operator to chain functions :

def f(l: List[Int]) = l.filter(x => x >1)
def g(l: List[Int]) = l.filter(x => x <3)

// apply f then g
List(1,2,3) |> f |> g 
//result : List(2)

Now, let’s define a filter function, taking predicates for a type T in a first parameter block and a list of T in a second parameter block :

def filterByField[T](selectField: Item => T, predicates: Set[String])(items: List[Item]) = {
  items.filter { item =>
    predicates.exists(included => selectField(item).toString.matches(included))
  }
}

With the pipe operator, it’s now really easy to chain such functions in a very readable way :

val items: List[Item] = ???

val filteredItems = items |>
  filterByField(_.name, nameFilters) |>
  filterByField(_.value, valueFilters) // chain as many filters as you want...

Here is an implementation for |> (found on Reddit) :

implicit class AnyEx[T](val v: T) extends AnyVal {
  def |>[U](f: T ⇒ U): U = f(v)
}

How to build a purely functional webapp in Scala

2017-12-31T00:00:00+00:00

I’ve just published minimal demo application to show how to build a purely functional web application in Scala.
To learn more about the benefits of this approach, you can read the Readme file. I’ve used Typelevel libraries to build this application : http4s, Doobie, CatsEffect and Circe.

You can find the project here : https://github.com/loicdescotte/pureWebappSample

Small update about Scala Hamsters

2017-12-31T00:00:00+00:00

The end of 2017 has been quite busy for the Hamsters team!

In November, I went to the awesome ScalaIO conference and had the chance to do a talk about Hamsters with Damien Gouyette (also a member of the Hamsters team). You can see the video here (in french) : https://www.youtube.com/watch?v=_SaD28-1ds4.

The last day of the conference, several persons have made great contributions to the library during an OSS workshop organized by the Scala center, and we’ve been able to release the 2.0 version just after that.

A week ago, 2.4.0 has been released with a new feature, mapN.

Scala Hamsters 1.4.0 is out with Scala.js support

2017-06-16T00:00:00+00:00

To use Hamsters in your Scala.js project, you just have to add this dependency in your build.sbt :

libraryDependencies ++= Seq(
  "io.github.scala-hamsters" %%% "hamstersjs" % "1.4.0"
)

See the Readme file for more information at https://github.com/scala-hamsters/hamsters.

Scala Hamsters 1.3.1 is out

2017-05-15T00:00:00+00:00

This new version includes improvements in the Validation and HList APIs.

See the Readme file for more information at https://github.com/scala-hamsters/hamsters.

Creating user interfaces with Scala.js and Binding.scala

2017-04-12T00:00:00+00:00

Binding.scala is a UI library for Scala.js, that is quite similar to React.js.

Let’s see how to use this library to build a simple list of elements, with a search box and add/remove features.

Binding.scala allows to mix HTML with Scala code to dynamically generate content on the browser side. Let’s create a first version of an item list component :

@dom def itemList(items: Vars[String]) = {
  <section>
    <ol>{
      for (item <- items) yield
        <li>
          { item }         
        </li>
    }</ol>
  </section>
}

Vars and Var types can encapsulate data and states. When a change is done in this structures, the UI is updated.

Now let’s add a delete button to remove items :

@dom def itemList(items: Vars[String]) = {
  <section>
    <ol>{
      for (item <- items) yield
        <li>
          { item }         
          <button onclick={ event: Event => items.get -= item }>x</button>
        </li>
    }</ol>
  </section>
}

You can notice that Vars defines some operators (like -=) to update the state.

It’s also quite easy to add a search box :

@dom def itemList(items: Vars[String]) = {

  val search: Var[String] = Var("")
  val searchInput: Input = {
    <input type="text" oninput={event: Event => search := dom.currentTarget[Input].value}/>
  }

  <section>
    <div> Search: { searchInput } </div>
    <ol>{
      //bind search value to item list
      for (item <- items if (item.startsWith(search.bind))) yield
        <li>
          { item }
          <button onclick={ event: Event => items.get -= item }>x</button>
        </li>
    }</ol>
    {add(items).bind}
  </section>
}

Using a simple if in items rendering, we can bind a filter to the search box value.

Now we would like to add new items in our list. When the UI becomes more complex, it’s better to split it into several sub-components, so we will create a distinct add component :

@dom def add(items: Vars[String]) = {

 val addInput: Input = <input type="text"/>

 val addHandler = { event: Event =>
    if (addInput.value != "" && !items.get.contains(addInput.value)) {
      items.get += addInput.value
      addInput.value = ""
    }
  }

  <div>{ addInput } <button onclick={ addHandler }>Add</button></div>
}

To ensure communication between our two components, we use the items Vars as a parameter. Now we can include the add component into our itemList component :

@dom def itemList(items: Vars[String]) = {

  val search: Var[String] = Var("")
  val searchInput: Input = {
    <input type="text" oninput={event: Event => search := dom.currentTarget[Input].value}/>
  }

  <section>
    <div> Search: { searchInput } </div>
    <ol>{
      for (item <- items if (item.startsWith(search.bind))) yield
        <li>
          { item }
          <button onclick={ event: Event => items.get -= item }>x</button>
        </li>
    }</ol>
    {add(items).bind}
  </section>
}

Finally we can initialize our components and build our application in a simple HTML page :

@dom def render() = {
  val items = Vars("bob", "joe", "john")
  <div>
    <h3>List of elements</h3>
    { itemList(items).bind }
  </div>
}

dom.render(document.body, render)

In conclusion, I think it’s pretty easy to build and assemble reactive UI components with Binding.scala. It’s also nice (and more powerful) to be able to use Scala to iterate over data, format and filter data, rather than HTML tags (à la Angular).

You can run the full example here.

You can also compare this example with this similar application written in JavaScript with React.js. The main difference is that React is using a virtual dom while Binding.scala is not. The more direct consequence is the use of Var and Vars types to handle states and data lifecycle in Binding.scala, while React only allows to update the state of a component to update the (virtual) dom.

JavaScript for Scala Developers

2017-01-28T00:00:00+00:00

Disclaimer : this post is not about Scala.js, but about the JavaScript language (+ Typescript).

As you may know, the last published JavaScript standard is ECMAScript 2016 (or ECMAScript 7, or ES7). Since ES6 (the previous standard, published in 2015), JS has a lot new syntactic possibilities. Many of this features will make Scala developers feel at home, at least a lot more than previous JS versions. Important note : it is possible to target browsers that don’t fully support ES6 or ES7 using the Babel compiler.

Basic examples

Define a variable

Scala :

var x = 1

JS :

let x = 1

Note : let is aimed at replacing JS var, as it defines block scoped variables.

Define a constant (a value)

Scala :

val x = 1

JS :

const x = 1

Both of this values are not reassignable. In JS you will have an “invalid assignment” error if you try to redefine a const.

Note : like let, const is block scoped.

Functions

// old school JS functions
const f = function(x) {
  return x +1
};

// Or arrow functions
const f = (x => x +1);

The second syntax is closer to Scala, and is nice when you need to pass a function argument to another function.

Classes

Since ES6, it is possible to define classes in JavaScript.

class Point {
       constructor(x, y) {
           this.x = x;
           this.y = y;
       }
       toString() {
           return '(' + this.x + ', ' + this.y + ')';
       }
   }

const p = new Point(1,2);

Default parameters

function multiply(a, b = 1) {
  return a * b;
}

multiply(5, 2); // 10
multiply(5, 1); // 5
multiply(5);    // 5

Generators

Generators are functions that can produce a sequence of values :

function* idMaker(index = 0, max=10){
    yield index;
    if(index < max)
      yield* idMaker(index+1);
}
let gen = idMaker();

console.log(gen.next().value); // 0
console.log(gen.next().value); // 1
console.log(gen.next().value); // 2

Generator functions are prefixed with a *. yield may remind you of Scala for comprehension. This keyword is used to produce an element in the generator. yield* is used to append elements from another generator.

Destructuring

//define x and y coordinates from a point :
let { x, y } = new Point(1,2);

Note : unlike Scala, JS asks you to use the same parameter names in the left expression as in your object.

Spread operator

This operator can be used to structure and destructure arrays.

let [...values] = idMaker();
console.log(values); //0 to 10

let [head, ...tail] = values;
console.log(head); //0
console.log(tail); //1 to 10

Rest operator

let f = function(a, b, ...rest) {
  // ...
}

This operator works like Scala varargs. parameters is retrieved as an Array object.

Usage :

f("a","b", 1 ,2 ,3) //1, 2, 3 will be in the rest array

Modules and imports

In lib.js :

export function f(x) {
    return x+1;
};

// other functions ...

In your JS file :

import { f, g } from 'lib';

const a = f(1);

Promises

JS Promises are close to Scala Futures :

let result = Promise.resolve("Success")
.then((value) =>"result :" + value);
//Promise { <state>: "fulfilled", <value>: "result :Success" }

//combine 2 Promises :
let result2 = Promise.resolve("Success")
 .then((value) => Promise.resolve("result " + value + " Success 2"));
//Promise { <state>: "fulfilled", <value>: "result Success Success 2" }

//print result :
Promise.resolve("Success")
.then((value) => Promise.resolve("result " + value + " Success 3"))
.then((value) => console.log(value));

Where are my types ?

One of the biggest difference with Scala is the absence of types. TypeScript is a typed superset of JavaScript, that can be compiled and transformed (transpiled) to JS. It is the language that has been used to write the Angular framework (since Angular 2), and it is the recommended language to develop Angular applications as well.

Let’s see how it looks :

let result: Promise<string> = Promise.resolve("Success")
.then((value) =>"result :" + value);

let result2: Promise<string>  = Promise.resolve("Success")
.then((value) => Promise.resolve("result " + value + " Success 2"));

You can see that I’ve added some type definitions to my result variable. The type notation is similar to Scala type notation. let result: Promise<string> tells the compiler that this variable must be a Promise of string. If it’s not the case, it will produce a compilation error.

Note: You can notice that there is an automatic flatten of JS Promises, as my result2 is a Promise<string> and not a Promise<Promise<string>>. This is not specific to the TypeScript language (pure JS promises are also flatten).

Type aliases with TypeScript

TypeScript type aliases look like Scala aliases :

type Name = string;
type NameResolver = () => string;

Immutable data

We have const to define values, but it is possible to get further with Immutable.js collections and records.

React, Redux and Angular are frameworks that rely a lot on immutability.

ES proposal : rest/spread properties

This proposal would be very convenient to copy objects.

Object copy in Scala (using case classes) :

case class Person(name: String, age: Int)

val bob = Person("bob", 30)
val john = bob.copy(name="john")

JS :

const bob = {name: "bob", age: 30};
const obj1 = {...bob, name: "john", address: "25 5th street NYC"}; // copy bob with updated name and address added

And it can already be used with Babel!

More advanced examples

Extensions methods

With JS you can add methods on any class via its prototype (the concept behind JS object oriented programming):

let add = (x, y) => x.sum(y);

//extension
Number.prototype.sum = function (x) { return this + x };

console.log(add(1, 2)); //3

With TypeScript you can declare that a function needs an object with a specific method (or several methods), using an interface :

interface CanSum {
    sum(x: CanSum): Number;
}

let add = (x: CanSum, y: CanSum) => x.sum(y);

//extension
Number.prototype.sum = function (x) { return this + x };

console.log(add(1, 2)); //3

Note : interfaces are implicitly implemented if an object defines all methods of the interface (it is called structural subtyping).

With extension methods and structural subtyping, you can emulate Scala implicit classes and be close to type classes, but without the recursive power of implicit resolution of type class instances. For example, if I need a class P to be serializable to a format X, I can add a serializeToX method on it instead of defining an implicit XSerializer[P], and P could be seen as implementing a CanBeSerialized interface. But, if P contains other types that need to be serialized, serializers won’t be discovered implicitly and recursively. You will have to define methods on each type.

Other functional goodness

If you miss functional abstractions like Option, Either and even Free monad, take a look at Monet.js, an amazing functional JS library. As types are really important for this kind of library (Maybe/Option monad for instance is not very useful without types), you can use Monet with its TypeScript bindings.

Spark Datasets and type-safety

2017-01-22T00:00:00+00:00

Spark 2.0 has introduced the Datasets API (in a stable version). Datasets promise is to add type-safety to dataframes, that are a more SQL oriented API. I used to rely on the lower level RDD API (distributed Spark collections) on some parts of my code when I wanted more type-safety but it lacks some of the dataframe optimizations (for example on groupBy and aggregations operations). The recommended way is now to use datasets everywhere (except if you’re doing something very specific and if you need to use the low level RDD funcstions). Let’s see how it looks.

This is the classical word count using RDD :

val textFile = sc.textFile("hdfs://...")
val counts = textFile.flatMap(line => line.split(" "))
             .map(word => (word.toLowerCase, 1)) // put 1 with each word instance
             .reduceByKey((accumulator, current) => accumulator+current) // add all words, grouped by value (by key)

And the word count with dataframes :

val df = sparkSession.read.text("hdfs://...")
val wordsDF = df.select(split(df("value")," ").alias("words"))
val wordDF = wordsDF.select(explode(wordsDF("words")).alias("word"))
val wordCount = wordDF.groupBy(lower($"word")).count

The drawback is that you loose the type information and the field names, so you need to use columns names as strings, which can be error-prone. Also, my personal opinion for this kind of example is that map/flatMap operations are easier to read.

With datasets, you have a mix of RDD and dataframes, with an high level API while retaining some type information :

sparkSession.read.text("hdfs://...").as[String]
  .flatMap(_.split(" "))
  .groupByKey(_.toLowerCase)
  .count()

This version is the easiest to read and to understand. So, it seems perfect! But… if you try to do some more complex operations on rows, you will quick fallback on the non type-safe API.

For example, if you want to order the words by count :

sparkSession.read.text("hdfs://...").as[String]
  .flatMap(_.split(" "))
  .groupByKey(_.toLowerCase)
  .count()
  .orderBy($"count(1)".desc) // WTF

Note : The count column name is pretty weird, but it gets a little better if you use groupBy (which is not type-safe) instead of groupByKey :

sparkSession.read.text("hdfs://...").as[String]
  .flatMap(_.split(" "))
  .map(_.toLowerCase)
  .groupBy($"value") // value is the default column name
  .count()
  .orderBy($"count".desc)

You will also have this kind of problem if you want to join 2 datasets :

//dataset creation from a Seq
val departments = Seq(
  Department(1, "rh"),
  Department(2, "it"),
  Department(3, "marketing")
).toDS

val people = Seq(
  Person("jane", 28, "female", 2000, 2),
  Person("bob", 31, "male", 2000, 1),
  //...
).toDS

people.joinWith(departments, people("deptId") === departments("id"))

Here again you have to pass the column names as strings. To be honest, it’s still easier than the RDD equivalent, where you would have to create (key, value) pair RDDs to be able to join data :

val departmentsById = departments.rdd.map{ department =>
  (department.id, department)
}

val peopleByDepartmentId = people.rdd.map{ person =>
  (person.deptId, person)
}

peopleByDepartmentId.join(departmentsById)

Let’s go back to the main subject of this post : datasets. With more complex operations, type-safety is still far from perfect :

people.filter(_.age > 30)
    .join(departments, people("deptId") === departments("id"))
    .group(departments("name"), $"gender")
    .agg(avg(people("salary")), max(people("age"))) // average salary and max age

Yes, that can be disappointing… but, there is a library named Frameless, based on the awesome Shapeless, that can add more type-safety to datasets!

import frameless.TypedDataset

val departments = TypedDataset.create(Seq(
  Department(1, "rh"),
  Department(2, "it"),
  Department(3, "marketing")
))

val people = TypedDataset.create(Seq(
  Person("jane", 28, "female", 2000, 2),
  Person("bob", 31, "male", 2000, 1),
  //...
))

val joined = people.join(departments, people('deptId), departments('id))

// val joined = people.join(departments, people('detppid), departments('id)) <-- Won't compile as 'detppid symbol is wrong

As you can see on the last line, if you provide a bad column name, you will get a compilation error. Pretty great isn’t it?

Another problem with regular datasets is that they can produce null values, for example using left outer joins :

departments.joinWith(people, departments("id") === people("deptId"), "left_outer").show

The output is :

//    +-------------+--------------------+
//    |           _1|                  _2|
//    +-------------+--------------------+
//    |       [1,rh]|[linda,37,female,...|
//    |       [1,rh]|[john,45,male,300...|
//    |       [1,rh]|[bob,35,male,2200,1]|
//    |       [1,rh]|[bob,31,male,2000,1]|
//    |       [2,it]|[joe,40,male,3000,2]|
//    |       [2,it]|[jane,28,female,2...|
//    |[3,marketing]|                null| <--- WARNING : null
//      +-------------+--------------------+
//

With Frameless, you won’t have this problem as it gives you Options :

val leftJoined: TypedDataset[(Department, Option[Person])] = departments.joinLeft(people, departments('id), people('deptId))

Frameless is still described as a “proof of concept” and does not cover all dataset operations, but I think it’s an interesting library to follow!

A small update on Scala Hamsters

2016-09-21T00:00:00+00:00

I’ve just published Hamsters 1.0.7, with a few new features.

Now you can automatically catch exceptions into a KO object :

import io.github.hamsters.Validation._

def compute(x: Int):Int = 2/x

fromCatchable(compute(1)) //OK(2)
fromCatchable(compute(0)) //KO("/ by zero")

fromCatchable(compute(0), (t: Throwable) => t.getClass.getSimpleName) //KO("ArithmeticException")

To see how to use the OK/KO monad and Validation helpers, see the Readme file.

Another new feature allows to ask for a specifc type and retrieve an option on an Union type :

import io.github.hamsters.{Union3, Union3Type}

//json element can contain a String, a Int or a Double
val jsonUnion = new Union3Type[String, Int, Double]
import jsonUnion._

def jsonElement(x: Int): Union3[String, Int, Double] = {
  if(x == 0) "0"
  else if (x % 2 == 0) 1
  else 2.0
}

jsonElement(0).get[String] // Some("0")
jsonElement(1).getOrElse("not found") // get String value or "not found" if get[String] is undefined

How to compose Future and Nullable in Kotlin

2016-05-17T00:00:00+00:00

In a previous post, we’ve seen how to compose Future and Option in Scala. Let’s try to do the same with Kotlin using nullables types and Java 8 completable futures.

This is how you can compose nullable functions :

  fun giveInt(x: Int):Int? = x+1

  fun giveInt2(x: Int):Int? = x+2

  fun combine(x: Int): Int? = giveInt(x)?.let(::giveInt2)

  combine(1) //4

The result of giveInt is passed to giveInt2. But let only applies if the first part is not null. Either a nullable is returned.

This is how you can compose futures (using Java 8 CompletableFuture API):

  fun giveInt(x: Int): CompletableFuture<Int> = CompletableFuture.supplyAsync({ x + 1 })

  fun giveInt2(x: Int): CompletableFuture<Int> = CompletableFuture.supplyAsync({ x + 2 })

  fun combine(x: Int): CompletableFuture<Int> =  giveInt(x).thenCompose(::giveInt2)

  combine(1).get() //4

This is how you can compose futures of nullable types :

  fun giveInt(x: Int): CompletableFuture<Int?> = CompletableFuture.supplyAsync({ x + 1 })

  fun giveInt2(x: Int): CompletableFuture<Int?> = CompletableFuture.supplyAsync({ x + 2 })

  fun combine(x: Int): CompletableFuture<Int?> = giveInt(x).thenCompose({ it?.let(::giveInt2) })

  combine(1).get() //4

And this how we would have done it with Scala and with the Hamsters library :

def giveInt(x: Int): Future[Option[Int]] = Future.successful(Some(x+1))

def giveInt2(x: Int): Future[Option[Int]] = Future.successful(Some(x+2))

def combine(x: Int): Future[Option[Int]] = for {
  y <- FutureOption(giveInt(x))
  z <- FutureOption(giveInt2(y))
} yield z

Await.result(combine(1), 1 second) //Some(4)

Finally even if it’s not possible to use monad transformers as we would do in Scala, if think Kotlin is giving out of the box a very simple and readable solution.

And this may be even simpler in the future using Kotlin coroutines.

Scala Hamsters 1.0

2016-05-08T00:00:00+00:00

I’ve just published Scala hamsters first stable version. It’s a simple Scala utility library featuring Lists, Union types, Monad transformers and Validation.
Hamsters is fully compatible with functional programming beginners!

It may help you understand how to use - and how to implement (looking at the code) - some popular functional programming features. Il also provides some useful utilities.

You can find it here : https://github.com/loicdescotte/hamsters

Don’t hesitate to contribute ;)

Akka Streams, Play Framework and queues

2016-04-10T00:00:00+00:00

Update (2019/11/06) : Source.preMaterialize can replace peekMatValue.
See the scaladoc for this method :

Materializes this Source, immediately returning (1) its materialized value, and (2) a new Source that can be used to consume elements from the newly materialized Source.

Usage example : https://gist.github.com/loicdescotte/5f3ed7a56b8d3fa8eb2982e9e97dcb36

In the previous post, we’ve seen how to stream, transform and combine data using Akka Streams on Play Framework. Now let’s see how to use a queue in which we can post new data dynamically. This can be very useful if you want to plug your application on a third party API that is sending events periodically and that is not using Akka Streams or Reactive Streams to emit data.

This is the common way to use an Akka Streams SourceQueue :

val bufferSize = 10

val queue = Source.queue[Int](bufferSize, OverflowStrategy.fail)
  .to(Sink.foreach(println))
  .run()

for (i ← 1 to 3) {
  queue.offer(i)
}

But our case is a little more complex. Play controller needs a Source to be able to stream a response using the chunked method. As Play uses its own Akka Stream Sink (i.e. a reactive data consumer) under the hood, we can’t materialize the source queue ourselves using a Sink because the source would be consumed before it’s used by the chunked method.

The solution is to use mapMaterializedValue on the source returned by Source.queue to get a future of its queue materialization.

Here is an helper method to do it (thanks to the Akka team who gave me this hint on github) :

 //T is the source type, here String
 //M is the materialization type, here a SourceQueue[String]
 def peekMatValue[T, M](src: Source[T, M]): (Source[T, M], Future[M]) = {
   val p = Promise[M]
   val s = src.mapMaterializedValue { m =>
     p.trySuccess(m)
     m
   }
   (s, p.future)
 }

We will use an Akka scheduler to simulate an API that would produce data periodically. To begin, let’s declare the actor that will post in our queue :

  val Tick = "tick"

  class TickActor(queue: SourceQueue[String]) extends Actor {
    def receive = {
      case Tick => queue.offer("tack")
    }
  }

Note that this method can materialize any kind of source, not only a queue source.

Now we can use this helper with a queue source and start a scheduler to post data into the queue using the TickActor :

def queueAction = Action {

  val (queueSource, futureQueue) = peekMatValue(Source.queue[String](10, OverflowStrategy.fail))

  futureQueue.map { queue =>

    val tickActor = actorSystem.actorOf(Props(new TickActor(queue)))
    val tickSchedule =
      actorSystem.scheduler.schedule(0 milliseconds,
        1 second,
        tickActor,
        Tick)

      queue.watchCompletion().map{ done =>
        Logger.debug("Client disconnected")
        tickSchedule.cancel
        Logger.debug("Scheduler canceled")
      }
  }

  Ok.chunked(
    queueSource.map{e => 
      Logger.debug("queue source element : " + e)
      e
    } 
    via EventSource.flow
  )

}

If you close your browser tab, queueSource will be canceled automatically. As we catch the watchCompletion event ont the underlying queue, we can also stop the Akka scheduler. Then you will see the following logs in your console :

[debug] application - queue source : tack
[debug] application - queue source : tack
[debug] application - queue source : tack
[debug] application - Client disconnected
[debug] application - Scheduler canceled

You can see the full code of this example here.

Akka Streams integration in Play Framework 2.5

2016-02-05T00:00:00+00:00

Update 1 (February 10, 2016) : Using latest Play 2.5 SNAPSHOT to benefit from EventSource helper
Update 2 (February 18, 2016) : Upgrade to Play 2.5 RC1 and refactor code (thank you Ahmed Mushtaq for the pull request in the sample app!)
Update 3 (April 10, 2016) : Adding part 2 : Akka Streams, Play Framework and queues

From version 2.5, Play’s default stream processing library is Akka Streams. Akka Streams is an implementation of the Reactive Streams standard. With Play 2.4 it was possible to use Akka Streams via bindings to the Play Iteratee lib (see this post). Now we don’t need this Iteratee bindings anymore, we can use Akka Streams natively in Play controllers. Play 2.5 also comes with Akka Streams 2, which is easier to use and in many cases faster than the first version.

A fake tweet API

In this example, we will create a fake Twitter service with Play. This service will stream messages using an Akka Streams Source :

  def timeline(keyword: String) = Action {
    val source = Source.tick(initialDelay = 0 second, interval = 1 second, tick = "tick")
    Ok.chunked(source.map { tick =>
      val (prefix, author) = prefixAndAuthor
      Json.obj("message" -> s"$prefix $keyword", "author" -> author).toString + "\n"
    }.limit(100))
  }

  private def prefixAndAuthor = {
      // random message and author
  }

The tick method creates a simple source with 1 second of delay between 2 message emissions. We can transform this source to get a message per second, related to a keyword using map. Finally we are limiting the feed to 100 elements.

A tweet consumer

Now we need another Web application that will be plugged on our fake Twitter service to make several queries in parallel and mix the results in a single source. This source will be streamed using Server Sent Events, which is simpler than WebSockets and a good fit to push data from server to browser.

case class TweetInfo(searchQuery: String, message: String, author: String)

object TweetInfo {
  implicit val tweetInfoFormat = Json.format[TweetInfo]
}

def mixedStream(queryString: String) = Action {
  val keywordSources = Source(queryString.split(",").toList)
  val responses = keywordSources.flatMapMerge(10, queryToSource)
  Ok.chunked(responses via EventSource.flow)
}

The first step is to call the fake Tweet API for each user query. Then we need to transform the WebService responses into Akka Streams sources. The list of sources we get in result can be merged in a single stream thanks to the flatMapMerge method.

Let’s see how to build this sources in detail :

private def queryToSource(keyword: String) = {
  val request = wSClient
    .url(s"http://localhost:9000/timeline")
    .withQueryString("keyword" -> keyword)

  streamResponse(request)
    .via(Framing.delimiter(ByteString("\n"), maximumFrameLength = 100, allowTruncation = true))
    .map { byteString =>
      val json = Json.parse(byteString.utf8String)
      val tweetInfo = TweetInfo(keyword, (json \ "message").as[String], (json \ "author").as[String])
      Json.toJson(tweetInfo)
    }
}

private def streamResponse(request: WSRequest) = Source.fromFuture(request.stream()).flatMapConcat(_.body)

Play WS Client stream method returns an Akka Streams Source of Akka ByteString via the body value of the response. The response itself is wrapped in a Future. We can convert a Future object into a Source using Source.fromFuture. As we get a Source of Source of WebService response, we can use flatMapConcat(_.body) to keep only the response body in a single source.

The Twitter service may send several messages in a single chunk, so we need to split them on line breaks. We can also imagine that the Twitter service could send chunks with incomplete messages. In this case the messages need to be saved in a buffer until we reach a line break. Fortunately, the Framing object does all the job for us. We just need to provide a separator (line break) and a max frame length for the source elements. This is a great improvement from Akka Streams 1, which was forcing developers to write a custom line parser. The result of this operation is a new source that can be transformed into the new desired format. In this case we’re just adding the search query in the response to ease filtering on the client side (c.f. TweetInfo case class). Then each source element can be transformed into Json thanks to the TweetInfo implicit Json formatter.

Finally, let’s go back to our mixedStream action. Play’s EventSource.flow method helps us to format the messages into the Server Sent Events format… and the stream can flow! Quite easy isn’t it?

You can find all the sources of this example (including the frontend code) here.
Now let’s continue with Part 2 - Akka Streams, Play Framework and queues!

Play2-HTML5-Tags 1.2.2 for Play 2.4

2015-12-03T00:00:00+00:00

Just a quick note to say that I’ve release Play2-HTML5Tags 1.2.2. This version is compatible with Play 2.4.x.
The aim of this plugin is to provide client side validation tags for Play Framework templates.

Playing with Akka Streams and Twitter

2015-06-19T00:00:00+00:00

Disclaimer : Since Play 2.5/Akka Streams 2, things shown in this post can be achieved more easily, for example without relying directly on the actor API. This this post for more information.

You may have heard of Reactive-Streams, a specification inspired by the Reactive Manifesto.
A lot of big actors are involved in this initiative, like Typesafe, Netflix, Pivotal, RedHat and Twitter.
Its goal is to “provide a standard for asynchronous stream processing with non-blocking back pressure”.
In a few words, back pressure is the ability for a data producer to adapt the data transmission rate depending on the speed of the consumer, to avoid overwhelming it.

This article will focus on Akka Streams, which is an implementation of Reactive-Streams that relies on Akka actors and provides an higher level API on top of them.

Play 2.5 will rely on Akka-Streams under the hood, and it will also allow to benefit from it to to handle data streams reactively, like we are doing today with Play 2 and the Iteratee API.
Akka Streams is still tagged as “experimental”, but the 1.0 version is close. With Play 2.4, it’s already possible to use it, via Reactive-Streams to Iteratee and Iteratee to Reactive-Streams conversions.

MixTweets, updated

In a previous post, we’ve seen how to handle streams of Twitter messages reactively, how to merge several Twitter searches in a unique stream, transform them into JSON and push this into an EventSource (Server Sent Event) output with the Iteratee API.
Now, we will do the same with Akka Streams and compare this two solutions.

How to define a source

In Akka Streams, data producers are called sources. It’s the equivalent of the Publisher interface in the Reactive-Streams specification.
We are using twitter4j and Twitter Streaming API to receive messages. For each new message, we need to push data into a source.
To achieve this, we need a reference to an actor implementing the ActorPublisher trait. We also need to bind the source to this actor.
Then, we can push some data into the actor and retrieve this data from the source.

def listenAndStream = {

    val (actorRef, publisher) =  Source.actorRef[TweetInfo](1000, OverflowStrategy.fail).toMat(Sink.publisher)(Keep.both).run()

    Logger.info(s"#start listener for $searchQuery")
 
    val statusListener = new StatusListener() {
 
      override def onStatus(status: TwitterStatus) = {   
       Logger.debug(status.getText)
       //push elements into a publisher
       actorRef ! TweetInfo(searchQuery, status.getText, status.getUser.getName)
      }
 
    }
 
    twitterStream.addListener(statusListener)
    twitterStream.filter(query)
    
    Source(publisher)
}

The first line creates an actor reference and a publisher. Now we can keep references to the actor and to the source. When we send elements to the actor (with the ! method), it will dynamically add elements in the source. We must choose a size for the actor message queue (1000) and an OverflowStrategy to handle cases where this size is exceeded.

Note that we need to materialize the source before being able to use the actor behind it.
The source is materialized into a publisher, and then we need to redefine a source from this publisher.
This trick helps us to solve a problem that is specific to our use case : we don’t want to consume the source right now because we must keep it for later, and merge it with other sources. Fortunately, this kind of use case should be simplified in future versions of Akka streams (an issue is opened in Github to address this).

In simpler cases, you can just declare a Source, a Flow (i.e. a bridge that takes an input a produces an output) and define a way to consume the data in the same time. Using directly a Sink, the source will be materialized so you can get a reference to an actor.

Here is an example that just prints messages matching a pattern :

val source = Source.actorRef[Message](1000, OverflowStrategy.fail)

val helloSource = source.filter(message => message.text.startsWith("Hello"))

val ref = Flow[Message].to(Sink.foreach(println)).runWith(helloSource)

ref ! Message("Hi there!")
ref ! Message("Hello there!")

If you need more customization, it’s also possible to create a custom ActorPublisher to do this.

In comparison, with Iteratee we would do things like that :

val (enum, channel) = Concurrent.broadcast[TweetInfo]

override def onStatus(status: TwitterStatus) = {  
  channel push TweetInfo(searchQuery, status.getText, status.getUser.getName)
}

The (enum, channel) tuple is quite similar to (actorRef, publisher).
We would also push new elements into the channel to feed the enumerator (i.e the data producer).

How to transform and merge sources

As we’re doing several Twitter searches, we have in result several sources. We will use a merge to have a single stream, containing all messages, so we can consume them more easily.

// get an Array of Source from an array of Twitter search queries
val streams = queries.map { query => 
    val twitterStreamListener = new TwitterStreamListener(query, config)
    twitterStreamListener.listenAndStream 
}

//merge streams in a single stream
val mergedStream = Source[TweetInfo]() { implicit builder =>

  val merge = builder.add(Merge[TweetInfo](streams.length))

  for (i <- 0 until streams.length) {
    builder.addEdge(builder.add(streams(i)), merge.in(i))
  }

  merge.out
}

//transform to JSON
val toJson = (tweet: TweetInfo) => Json.obj("message" -> s"${tweet.searchQuery} : ${tweet.message}", "author" -> s"${tweet.author}")

val jsonStream = mergedStream.map(tweets => toJson(tweets))

Note : At this stage, we could consume messages directly using a sink (an Akka Stream data consumer), for example source.runWith(Sink.foreach(println))

In comparison, with Iteratee would transform and merge streams like that :

// get an Array of Enumerator from an array of Twitter search query
val streams = queries.map { query => 
  val twitterListener = new TwitterStreamListener(query, config)
  twitterListener.listenAndStream
}

//merge streams in a single stream
val mixStreams = streams.reduce((s1,s2) => s1 interleave s2)

//transform to JSON
val toJson : Enumeratee[TweetInfo, JsValue] = Enumeratee.map[TweetInfo] { case tweet =>
  Json.obj("message" -> s"${tweet.searchQuery} : ${tweet.message}", "author" -> s"${tweet.author}")
}

val jsonMixStreams = mixStreams through toJson

The advantage of Akka Streams here is that we don’t need to use an intermediate structure like Enumeratee, we can just use a good old map function. Right now, the merge is really simpler with Iteratee. But there is already an issue opened in github (yes, another one) to be able to merge a stream of streams in a simple line, like streams.flatten(FlattenStrategy.concat).

Push to EventSource

To be compatible with Play controllers, we still need to use Enumerator and Iteratee objects.
Then we can use conversions included in Play 2.4 :

val jsonEumerator : Enumerator[JsValue] = sourceToEnumerator(jsonStream)
Ok.chunked(jsonEumerator through EventSource()).as("text/event-stream")

That’all!
You can see all the code (including details of the sourceToEnumerator method) in this github project.

You can create an application account to configure the Twitter API client on the Twitter apps page.
Then open your browser on http://localhost:9000/liveTweets?query=java&query=ruby (for example) to see the stream live!

In conclusion, I’ve tried to note a few good/bad points for each library :

Akka-stream

++ Very flexible, while the API is quite high level, we can easily go to a lower level with actors for more specific needs
++ It’s easy to defines flow and transformations on sources (need to use an Enumerattee to transform an input (Enumerator) in Iteratee API)
– A little more verbose / needs more boilerplate code for some basic stuffs. But Akka Streams is very young and will continue to provide new features quickly

Iteratee

++ Very good for this use case (broadcast and channel are very handy, merge is very easy) – Low level / complex cases can be more difficult

Both are very good stream processing libraries, with high level API and good capabilities to handle back pressure.

Note that I’m just beginning with Akka Streams, if you find better ideas to achieve this, please tell me :)

Compile Time Dependency Injection with Play 2.4

2015-06-03T00:00:00+00:00

Update 23/07/2015 : How to add stop hooks for your modules, how to inject play modules (WSClient, I18n, Cache, …)
Update 15/12/2015 : How to configure Play logs and Http filters

Play Framework, in its new version, provides a lot of new stuff to handle dependency injection at runtime, using Guice.
With Scala, I always prefer using compile time dependency injection, as it allows to see errors as soon as possible. I must also admit that I find compile time DI a lot more elegant as it needs no container or proxy at runtime!

Fortunately, Play team also added the ability to control the way routes, controllers and other components are binded together at compile time.

Controller and service

In this example, we have a controller that takes a LinkService dependency.
We want to be able to mock this service in our controller tests.

class Application(linkService: LinkService) extends Controller {

  def findLinks(query: String) = Action.async {
    val links = linkService.findLinks(query)
    links.map(response => Ok(views.html.index(response)))
  }

}

The routes file is defined as follows :

# Home page
GET     /                           controllers.Application.findLinks(query: String)
GET     /assets/*file               controllers.Assets.at(path="/public", file)

How to wire this

To have a fully functional application we need to tell the Play router how to find our LinkService dependency and how to wire it into our controller.

This can be done by defining a custom ApplicationLoader :

class SimpleApplicationLoader extends ApplicationLoader {
  def load(context: Context) = {
    new ApplicationComponents(context).application
  }
}

class ApplicationComponents(context: Context) extends BuiltInComponentsFromContext(context) {  
  lazy val logService = new LogService
  lazy val linkService = new LinkService(logService)

  lazy val applicationController = new controllers.Application(linkService)
  lazy val assets = new controllers.Assets(httpErrorHandler)
  override lazy val router = new Routes(httpErrorHandler, applicationController, assets)
}

Note : the Routes constructor is generated by Play at compile time, depending on the controllers referenced in the routes file.

To enable this configuration we need to add this line in application.conf :

play.application.loader=SimpleApplicationLoader

And this line in build.sbt :

routesGenerator := InjectedRoutesGenerator

How to test it

In this example, the tests will be written using Specs2.
To test our controller, we can easily wire a mocked linkService :

class ControllerSpec extends Specification with Mockito {

  val mockLinkService = mock[LinkService]
  mockLinkService.findLinks("hello") returns Future("http://coucou.com")

  "Application" should {

    "display query results" in {

      val app = new Application(mockLinkService)
      val response = app.findLinks("hello")(FakeRequest())
      
      status(response) must equalTo(OK)
      contentType(response) must beSome("text/html")
      contentAsString(response) must contain("http://coucou.com")

    }
  }
}

To be able to test the routes directly with our real services, e.g. for integration tests, we need to define our own ‘WithApplication’ helper :

class WithDepsApplication extends WithApplicationLoader(new SimpleApplicationLoader)

Of course if you want your router to use controllers with mocked services, you can define a new ApplicationLoader for tests instead of our SimpleApplicationLoader.

Then, we can use WithDepsApplication in our tests :

@RunWith(classOf[JUnitRunner])
class ApplicationSpec extends Specification {

  "Application" should {

    "render the index page" in new WithDepsApplication{
      val home = route(FakeRequest(GET, "/?query=hello")).get

      status(home) must equalTo(OK)
      contentType(home) must beSome.which(_ == "text/html")
      contentAsString(home) must contain ("Results :")
    }
  }
  
}

Note that you can also use macwire macros to wire automatically services, controllers and routes. A good example can be found here.

How to inject Play modules

As Ryan pointed out in the comments, it is possible to inject modules provided by Play APIs. The BuiltInComponentsFromContext can be mixed with several traits, provided by the framework.
Let’s consider that our link service needs to use the Web service client API :

class LinkService(logService: LogService, wsClient: WSClient){

  def findLinks(query: String) = {
    val duckDuckUrl = Play.current.configuration.getString("duckduck.url").getOrElse("http://api.duckduckgo.com/?format=json&q=") + query
    wsClient.url(duckDuckUrl).get().map{ response =>
      //...
    }
  }

  def cleanup = Logger.info("cleanup") // we will use this later
}

To inject an instance of WS client in your controller, you have to mix BuiltInComponentsFromContext with the NingWSComponents trait and then pass the wsClient to your service :

class ApplicationComponents(context: Context) extends BuiltInComponentsFromContext(context) with NingWSComponents {  
  lazy val logService = new LogService
  lazy val linkService = new LinkService(logService, wsClient)  

  lazy val applicationController = new controllers.Application(linkService)
  lazy val assets = new controllers.Assets(httpErrorHandler)
  override lazy val router = new Routes(httpErrorHandler, applicationController, assets)
}

You can use other traits like the Cache API, the I18n API, etc.

How to add stop hooks for your modules

You may need to add some behavior to your dependencies when your application is stopped.
BuiltInComponentsFromContext contains an applicationLifecycle value, which is a DefaultApplicationLifecycle. This class provides an addStopHook method. You can pass an asynchronous function to this method, that will be executed when your application will be stopped.

In this example, we are using LinkService cleanup method in the stop hook :

class ApplicationComponents(context: Context) extends BuiltInComponentsFromContext(context) with NingWSComponents {  
  lazy val logService = new LogService
  lazy val linkService = new LinkService(logService, wsClient)
  lazy val applicationController = new controllers.Application(linkService)  

  applicationLifecycle.addStopHook(() => Future.successful(linkService.cleanup)) // <-- stop hook

  lazy val assets = new controllers.Assets(httpErrorHandler)
  override lazy val router = new Routes(httpErrorHandler, applicationController, assets)
}

Important note : stop hooks are called in the reverse order they have been added in applicationLifecycle.

You can find the sources of this examples in this Github project. Next time we will see another new feature of Play 2.4, its (experimental) integration with Akka-Streams!

More tips

When you’re using compile time dependency injection with Play, some basic stuff that used to work out of the box don’t work anymore.

To activate Play logs, you need to add this line in the ApplicationLoader :

Logger.configure(context.environment)

So the SimpleApplicationLoader definition looks like :

class SimpleApplicationLoader extends ApplicationLoader {
  def load(context: Context) = {
    Logger.configure(context.environment)
    new ApplicationComponents(context).application
  }
}

To activate Http filters (or essential filters), you need to override a value from BuiltInComponentsFromContext :

val myFilter = new MyFilter(configuration)
override lazy val httpFilters = Seq(myFilter)

So the ApplicationComponents definition looks like :

class ApplicationComponents(context: Context) extends BuiltInComponentsFromContext(context) with NingWSComponents {  
  lazy val logService = new LogService
  lazy val linkService = new LinkService(logService, wsClient)
  lazy val applicationController = new controllers.Application(linkService)  

  applicationLifecycle.addStopHook(() => Future.successful(linkService.cleanup))

  lazy val assets = new controllers.Assets(httpErrorHandler)
  override lazy val router = new Routes(httpErrorHandler, applicationController, assets)
  val myFilter = new MyFilter(configuration)
  override lazy val httpFilters = Seq(myFilter)
}

How to compose Future and Option in Scala

2015-01-23T00:00:00+00:00

Update (June 24, 2016) : New example using Hamsters lib

In Scala, it’s easy to compose options :

val oa: Option[String] = Some("a")
val ob: Option[String] = Some("b")

val oab : Option[String] = for{
  a <- oa
  b <- ob
} yield a+b

Note : you can read my helper about for comprehension translation to map/flatMap.

It’s also easy to compose futures, in a non blocking way :

def fa: Future[String] = Future("a")
def fb(a: String): Future[String] = Future(a+"b")

val fab : Future[String] = for{
  a <- fa
  ab <- fb(a)
} yield ab

But is it easy to compose several Future[Option[A]]? It’s a common problem if you call sequentially several external resources, that can give you one or zero value.

def foa: Future[Option[String]] = Future(Some("a"))
def fob(a: String): Future[Option[String]] = Future(Some(a+"b"))

/* can't compose like this : type mismatch
val composedAB = for {
  optA <-foa
  a <- optA // option instead of Future
  ab <- fob(a)
}yield ab
*/

We can’t compose easily fob and foa: types will mismatch because futures and options don’t compose naturally.

Fortunately, there are some solutions.

Handmade binding

You can solve this with pattern matching :

val composedAB: Future[Option[String]] = foa.flatMap {
  case Some(a) => fob(a) //(*)
  case None => Future.successful(None)
}

(*) : In the for, call fob with the value inside the option and get a Future[Option[String]]. Doing this, you can compose result from foa and fob and keep consistent types.

Using a custom FutureO monad

This is an idea from Edofic’s blog. It allows more natural composition of Option and Future types.

You can create a custom FutureO monad that will compose well with other FutureO instances.

You just need to define a constructor and flatMap (plus map as it’s used by for comprehension syntactic sugar) :

case class FutureO[+A](future: Future[Option[A]]) extends AnyVal {
  def flatMap[B](f: A => FutureO[B])(implicit ec: ExecutionContext): FutureO[B] = {
    val newFuture = future.flatMap{
      case Some(a) => f(a).future
      case None => Future.successful(None)
    }
    FutureO(newFuture)
  }

  def map[B](f: A => B)(implicit ec: ExecutionContext): FutureO[B] = {
    FutureO(future.map(option => option map f))
  }
}

Then you can use a simple for comprehension form :

val composedAB2: Future[Option[String]] = (for {
  a <- FutureO(foa)
  ab <- FutureO(fob(a))
}yield ab).future

Or, with flatMap :

val noSugarComposedAB2 = FutureO(foa).flatMap(a => FutureO(fob(a))).future

Using a Scalaz monad transformer

Finally, if you want to generalize this kind of usage, you should try Scalaz monad transformers, which work for a larger kind of types than options and futures.

import scalaz._
import Scalaz._
import scalaz.OptionT._

// We are importing a scalaz.Monad[scala.concurrent.Future] implicit transformer

val composedAB3: Future[Option[String]] = (for {
  a <- optionT(foa)
  ab <- optionT(fob(a))
}yield ab).run

Update : Using Hamsters monad transformers

Since the moment I wrote this post, I’ve made a small library that provides (among other things) some simple monad transformers :

val composedAB: Future[Option[String]] = for {
  a <- FutureOption(foa)
  ab <- FutureOption(fob(a))
} yield ab

You can find more information about Hamsters here.

Scala for comprehension translation helper

2015-01-22T00:00:00+00:00

“For comprehension” is a another syntaxe to use map, flatMap and withFilter (or filter) methods.

yield keyword is used to aggregate values in the resulting structure.

This composition can be used on any “monadic” type implementing this methods, like List, Option, Future, etc.

Case 1

case class Book(author: String, title: String)

for {
  book <- books
  if book.author startsWith "bob" 
} yield book.title

is the same as

books.withFilter(book => book.author startsWith "bob").map(book => book.title)

Case 2

case class Book(authors: List[String], title: String)

for {
  book <- books
  author <- book.authors 
  if author startsWith "bob" 
} yield book.title

is the same as

books.flatMap(book => book.authors.withFilter(author => author startsWith "bob").map(author => book.title))

Case 3

val optA : Option[String] = Some("a value")
val optB : Option[String] = Some("b value")

I want an option tuple : Option[(String, String) composing this two options, i.e. Some(("a value","b value")) :

for {
     a <- optA
     b <- optB
} yield (a,b)

is the same as

optA.flatMap(a => optB.map(b => (a, b)))

You can also filter options with if/withFilter :

for {
  a <- optA
  if(a startsWith "a")
  b <- optB
} yield (a,b)

optA.withFilter(a=> a startsWith "a").flatMap(a => optB.map(b => (a, b)))

If you change “a” to “c” in the condition, the resulting value will be None instead of Some(("a value","b value")).

Case 4

You can mix options and list types in map/flatMap (Option can be seen as a simple collection) :

val optNumbers = List(Some(1), Some(2), None, Some(3))

We want to remove empty values and increment other values : List(2, 3, 4) :

for {
  optNumber <- optNumbers
  value <- optNumber
} yield value +1

is the same as

optNumbers.flatMap(optNumber => optNumber.map(value => value+1))

Scala and generic types erasure

2015-01-14T00:00:00+00:00

Just a little tip or reminder about generic types erasure in Scala.

scala> def displayGeneric[T](arg: T)(implicit m: reflect.Manifest[T]) = println(s"${arg.toString} : ${m.toString}")
displayGeneric: [T](arg: T)(implicit m: scala.reflect.Manifest[T])Unit

scala> displayGeneric(1)
1 : Int

You can retrieve type information and get over erasure using an implicit Manifest. It is also very useful if you need to access the T type in your method code. Example from the Scaladoc :

def arr[T] = new Array[T](0)                          // does not compile
def arr[T](implicit m: Manifest[T]) = new Array[T](0) // compiles
def arr[T: Manifest] = new Array[T](0)                // shorthand for the preceding

Playing with ES7 Object.observe

2014-11-26T00:00:00+00:00

I’m really not a JavaScript expert but today I wanted to try the new Object.observe API that will be part of ECMAScript 7 and that is already supported in Chrome.
In my example I want to observe an array of user names, and update an HTML list according to the events occurring on this array (add, delete or update). To achieve this, the appropriate function is Array.observe.

Model to View

var users = [{name: 'joe',id: 0}, {name: 'bob',id: 1}, {name: 'loic',id: 2}];

Array.observe(users, function(changes){
    changes.forEach(function(change) {  
        if(change.type == "splice"){
            if(change.addedCount){
                var index = change.index;
                var user = users[index];
                $('#list ul').append('<li>...</li>') //add element in markup
            }
            else{
                var id = change.removed[0].id;
                $('li[id='+id+']').remove();
            }
        }
    });  
});

This is very simple and handy, even for a backend developper like me, to automatically udpate the view from the model.

View to Model

Of course, it is also possible to have 2 ways databinding, for example we can add an element in the array using an input field. Then the list will be automatically updated, thanks to the array observer.

$("#ok").click(function(){
    lastId +=1;
    var name = $("#in").val();
    var user = {name: name,id: lastId};
    users.push(user);
});

We can also delete elements in the HTML list :

$("#list").on("click", ".del", function(event) {   
    var id = parseInt($(this).parent().attr('id'),10);
    var index = users.map(function(x) {return x.id; }).indexOf(id);
    users.splice(index, 1);
});

See this JsFiddle for the full example

Playing with Twitter streams

2014-09-23T00:00:00+00:00

I’ve updated a POC I made two years ago, about mixing and streaming some Twitter searches with Play Framework.

The new version handles Twitter authentication through OAuth. The number of parallel Twitter search queries is dynamic and the results are pushed to the browser in real time through Server Sent Events. Instead of using the Play Web Service API to call the Twitter REST API, this version uses Twitter4J and the Twitter Streaming API (the connection stays open to retrieve new arriving Tweets).

MixedTweets github project

All the code is available in this mini project. The old version using Play WS API is available here.

Let’s see what the code looks like.

Controller

We define a stream method in our Controller :

def stream(query: String) = Action {
  val queries = query.split(",")

  val streams = queries.map { query => 
    val twitterListener = new TwitterStreamListener(query, config)
    twitterListener.listenAndStream
  }

  val mixStreams = streams.reduce((s1,s2) => s1 interleave s2)

  val jsonMixStreams = mixStreams through toJson
  Ok.chunked(jsonMixStreams through EventSource()).as("text/event-stream")  
}

For each query, we create a stream (an Enumerator in the Iteratee API). Then we can reduce the streams into only one mixed stream using the interleave method. The last two lines will be explained later.

In the TwitterStreamListener class, we create an Enumerator from a Twitter search. We use Twitter4J to handle authentication and Twitter searches (via the Twitter streaming API) :

class TwitterStreamListener(searchQuery: String, config: Configuration) {
 
  val query = new FilterQuery(0, Array(), Array(searchQuery))
 
  val twitterStream = new TwitterStreamFactory(config).getInstance
 
  def listenAndStream = {
    Logger.info(s"#start listener for $searchQuery")
 
    val (enum, channel) = Concurrent.broadcast[(String, TwitterStatus)]
 
    val statusListener = new StatusListener() {
 
      override def onStatus(status: TwitterStatus) = {      
        Logger.debug(status.getText)  
        channel push (searchQuery, status)
      }

    //...
 
    }
 
    twitterStream.addListener(statusListener)
    twitterStream.filter(query)
    enum
  }
 
}

Concurrent.broadcast is useful to feed an Enumerator via an input channel. When a new message arrives, we push the Twitter status into the channel (see onStatus method). The resulting Enumerator contains tuples of search queries and Twitter status.

Adapt the content with an Enumeratee

An Enumeratee is a kind of adapter in the Iteratee API. We will use it to transform the results sent to the browser. The results will be converted into JSON values :

val toJson : Enumeratee[(String, TwitterStatus), JsValue] = Enumeratee.map[(String,TwitterStatus)] { case (searchQuery, status) =>
  Json.obj("message" -> s"$searchQuery : ${status.getText}", "author" -> status.getUser.getName)
}

Finally we can stream the result over SSE using an EventSource Enumeratee, with a text/event-stream Content-Type :

val jsonMixStreams = mixStreams through toJson
Ok.chunked(jsonMixStreams through EventSource()).as("text/event-stream")

Now we just have to make the stream alive on the browser (see index.scala.html).

Note : To configure the Twitter API client you need to declare an application on the Twitter apps page. Then use your credentials as follows :

val cb = new ConfigurationBuilder()
cb.setDebugEnabled(true)
  .setOAuthConsumerKey("xxx")
  .setOAuthConsumerSecret("xxx")
  .setOAuthAccessToken("xxx")
  .setOAuthAccessTokenSecret("xxx")

val config = cb.build

Enjoy!!

Streaming and proxying with Play part 2

2014-09-12T00:00:00+00:00

In a last post, we’ve seen some streaming and proxying examples with Play Framework. Let’s see how we can simplify and improve the last example with an implicit class.

Update : As Martin said in the comments, since Play 2.3, WS provides a getStream method returning what we’re trying to achieve, i.e. a Future[(WSResponseHeaders, Enumerator[Array[Byte]])].

But you can still read the following post for pedagogical purpose :)

Define a new function

First, let’s create a new case class, just for readability purpose.

case class WSStream(wsr: WSResponseHeaders, stream: Enumerator[Array[Byte]])

Then we create our new function.

def getAndStream(wsr: WSRequestHolder) = { 
    
  def consumer(promiseToFeed: Promise[WSStream]) = { rs: WSResponseHeaders =>
    val (wsConsumer, stream) = joined[Array[Byte]]
    promiseToFeed.success(WSStream(rs, stream))
    wsConsumer
  }
  
  val resultPromise = Promise[WSStream]
  wsr.get(consumer(resultPromise)).map(_.run)
  resultPromise.future
}

This is quite similar as the code seen in the previous example, except that all the tricky parts are in a separated function.

We can call this function to get a stream enumerator (and the response headers) from the WS response.

  val headersAndStream = getAndStream(WS.url("http://dumps.wikimedia.org/simplewiki/latest/simplewiki-latest-pages-articles.xml.bz2"))
  val streamEnumerator = headersAndStream.map(_.stream)

This is a little easier to read but we can do better!

Define an implicit class

Let’s hide all the complexity in an implicit class.

implicit class RichWSRequestHolder(val self: WSRequestHolder) {
    def getAndStream = {    

        def consumer(promiseToFeed: Promise[WSStream]) = { rs: WSResponseHeaders =>        
          val (wsConsumer, stream) = joined[Array[Byte]]
          promiseToFeed.success(WSStream(rs, stream))
          wsConsumer
        }

        val resultPromise = Promise[WSStream]
        self.get(consumer(resultPromise)).map(_.run)
        resultPromise.future
    }
  }

RichWSRequestHolder will implicitly add a new getAndStream method on the WSRequestHolder class. Don’t worry, everything is resolved at compile time so it’s safe.

Now you can import the RichWSRequestHolder class anywhere, and you will be able to call a Web Service like this to get a WSStream :

  val headersAndStream = WS.url("http://dumps.wikimedia.org/simplewiki/latest/simplewiki-latest-pages-articles.xml.bz2").getAndStream

Really easier isn’t it?

N.B : Of course you can do the same with other HTTP methods (POST, PUT, etc.).

Streaming and proxying with Play

2014-09-10T00:00:00+00:00

You’ll find bellow a few streaming examples with Play Framework, a kind of small streaming/proxying cheat sheet :)

This examples use the Iteratee API. The main elements from this API are the iteratees (data consumers) and the enumerators (data producers). If you want to learn more about this API I recommend to read this post.

Stream a file (and transform)

Chunck by chunk

 
def  transform = Action {

     import play.api.libs.iteratee._
    
     val fileStream: Enumerator[Array[Byte]] = {
         Enumerator.fromFile(new File("data.txt"))
     }
     
     val transfo = Enumeratee.map[Array[Byte]]{byteArray =>  
         val chunckedString = new String(byteArray)
         //add blablabla on each line
         val newChunk = chunckedString.replaceAll("\n", " blablabla\n")         
         //newChunk <-- stream text
         newChunk.getBytes //<-- stream file
     }
               
     Ok.chunked(fileStream.through(transfo))            
}

Line by line

 
def  transform = Action {

    import play.api.libs.iteratee._

    lazy val bufferedReader =  new BufferedReader(new InputStreamReader(new FileInputStream("data.txt")))
     
     val fileStream : Enumerator[String] = Enumerator.generateM[String] {
        scala.concurrent.Future{
          val line: String = bufferedReader.readLine()
          Option(line)  
        }
      }
     
     val transfo = Enumeratee.map[String]{line =>  
         val newLine = line + " blablabla" + "\n"
         newLine.getBytes
     }
               
    Ok.chunked(fileStream.through(transfo)) 
}

HTTP Proxying

In this proxying example we will call an external service in streaming from a Play server, and stream the response to our HTTP clients.

Update : Since Play 2.3, WS provides a getStream method returning a Future[(WSResponseHeaders, Enumerator[Array[Byte]])].

So it’s a little easier to use :

 
val resultFuture = WS.url("http://dumps.wikimedia.org/simplewiki/latest/simplewiki-latest-pages-articles.xml.bz2").getStream

//val dataEnumeratorFuture = resultFuture.map(stream => stream._2)
//dataEnumeratorFuture.map(Ok.chunked(_))

//OR   
resultFuture.map {
  case (rs, stream) =>
    Result(
      header = ResponseHeader(
        status = OK,
        headers = Map(
          CONTENT_LENGTH -> rs.headers.get("Content-Length").map(_.head).get,
          CONTENT_DISPOSITION -> s"""attachment; filename="wikipedia.xml.bz2"""",
          CONTENT_TYPE -> rs.headers.get("Content-Type").map(_.head).getOrElse("binary/octet-stream"))
      ),
      body = stream
    )
}

(Thanks Martin for the comment)

Previous solution :

def streamFromWS = Action.async { request =>

    import play.api.libs.iteratee._
    import scala.concurrent.Promise
    import play.api.mvc.ResponseHeader
    import play.api.mvc.Result
    import play.api.libs.ws.{ WSResponseHeaders, WS }
    import play.api.libs.iteratee.Concurrent.joined

    def consumer(promiseToFeed: Promise[(WSResponseHeaders, Enumerator[Array[Byte]])]) = { rs: WSResponseHeaders =>
      val (wsConsumer, stream) = joined[Array[Byte]]
      promiseToFeed.success((rs, stream))
      wsConsumer
    }

    val resultPromise = Promise[(WSResponseHeaders, Enumerator[Array[Byte]])]
    WS.url("http://dumps.wikimedia.org/simplewiki/latest/simplewiki-latest-pages-articles.xml.bz2").get(consumer(resultPromise)).map(_.run)
    
    //now you can do whatever you want with your enumerator : combine with another one, transform with an enumeratee, etc : 
    //val dataEnumeratorFuture = resultPromise.future.map(stream => stream._2)
    //dataEnumeratorFuture.map(Ok.chunked(_))
    
    //OR if you need to setup response headers
    resultPromise.future.map {
      case (rs, stream) =>
        Result(
          header = ResponseHeader(
            status = OK,
            headers = Map(
              CONTENT_LENGTH -> rs.headers.get("Content-Length").map(_.head).get,
              CONTENT_DISPOSITION -> s"""attachment; filename="wikipedia.xml.bz2"""",
              CONTENT_TYPE -> rs.headers.get("Content-Type").map(_.head).getOrElse("binary/octet-stream"))
          ),
          body = stream
        )
    }
}

This one deserves an explanation. Concurrent.joined creates an iteratee/enumerator tuple. According to the documentation, “When the enumerator is applied to an iteratee, the iteratee subsequently consumes whatever the iteratee in the pair is applied to. Consequently the enumerator is “one shot”, applying it to subsequent iteratees will throw an exception.”.

WS.url accepts a WSResponseHeaders => Iteratee function to consume the data from the remote service. In the consumer method we feed the promise with the enumerator (stream) created from the WS result. Then we can stream a response from the enumerator.

(*) This example is inspired by an example from Yann Simon on Github

If you want to mix several sources, you can read the following post : Playing with Twitter streams

Playing with Sphero

2014-09-10T00:00:00+00:00

Lately, I’ve acquired a Sphero, a little ball described as “The Ball. Evolved”.

Sphero can be controlled by Bluetooth. SDKs are available for Android and iOS, as well as many applications and games. The ball can also work with Cylon.js, a javascript library for robotics, GoBot (for the Go langage) or Artoo (for Ruby).

Let’s see is a very simple example running with Cylon.js over node.js. In this example we will ask Sphero to roll and to make a u-turn if there is a collision (e.g. a wall). We will also change the color at each collision.

var Cylon = require('cylon');
var run = false;
var direction = 0;
var speed = 100;
Cylon.robot({
  // connection port working on OSX
  connection: { name: 'sphero', adaptor: 'sphero', port: '/dev/tty.Sphero-RPY-AMP-SPP' },
  device: { name: 'sphero', driver: 'sphero' },

  work: function(me) {
    var color = 0x00FF00,
        bitFilter = 0xFFFF00;

    var shuffle = function(){
        direction = Math.floor(Math.random() * 360);
        me.sphero.roll(speed, direction);
        run = true;
    }

    var uTurn = function(){
      direction = (direction + 180) % 360;
      console.log("Direction: " + direction);
      me.sphero.roll(90, direction);
      run = true;
    }
    
    after((1).second(), function() {
      console.log("Setting up Collision Detection...");
      me.sphero.detectCollisions();
      me.sphero.setRGB(color);
      me.sphero.stop();
    });
    
    me.sphero.on('collision', function(data) {
      console.log("Collision:");
      color = color ^ bitFilter;
      console.log("Color: " + (color.toString(16)) + " ");
      me.sphero.setRGB(color);
      uTurn();

    });   

    shuffle();
  }

}).start();

We can improve the script to make the ball run, do a u-turn, change the direction or the speed using the keyboard.

var keypress = require('keypress');
process.stdin.setRawMode(true);
process.stdin.resume();   
keypress(process.stdin);

process.stdin.on('keypress', function (ch, key) {
  //console.log('got "keypress"', key.name);
  if (key.name == 'r') {
    if(!run){
      me.sphero.roll(speed, direction);
      run = true;
    }
    else {
      me.sphero.stop();
      run = false;
    }
  }
  else if (key.name == 'u') {
    uTurn();
  }
  else if (key.name == 's') {
    shuffle();
  }
  //boost
  else if (key.name == 'b') {
    if(speed == 100){
      speed = 200;
    }
    else {
      speed = 100;
    }
    console.log("speed:", speed)
    me.sphero.roll(speed, direction);
  }
  //quit
  else if (key.name == 'q') {
    process.exit(0);
  }
  
});

Finally, we can add a function to ask Sphero to change its drirection automatically.

//auto-change direction
every((5).second(), function() {
  if(run){
    shuffle();
  }
});

In conclusion, I would say that Sphero is a very fun toy (touch controls on smartphone are really amusing). But it is also very good tool to learn programing, in particular for children.

In fact it could be renamed “Turtle graphics. Evolved.” :)

No generics in Golang, but...

2014-08-28T00:00:00+00:00

Everyone knows that Go is a great language but that it does not support generics (yet?).

However in a language that supports first-class functions, generic types can be very handy. Fortunately, the gen package helps to fill this gap. Gen is a kind of macro tool that generate new structs from your models, with collections-like functions to filter, sort, group, iterate over your data.

You just need to add a +gen annotation comment on your models :

 
// +gen
type Message struct {
	author, text string
}

Then, after running go get github.com/clipperhouse/gen, you will be able to generate some Go code with the gen command. Finally, the new generated types can be used as follow, for example to filter a list of messages :

 
func main() {

	messages := make(Messages, 3)
	messages[0] = Message{"loic", "hi"}
	messages[1] = Message{"loic", "hello"}
	messages[2] = Message{"bob", "hi"}

	myMessages := messages.Where(func(m Message) bool {
		return m.author == "loic"
	})

	for _, message := range myMessages {
		fmt.Printf("author : %s - message : %s \n", message.author, message.text)
	}

}

It only remains to integrate the generation step in your build tool.

Read older posts

2014-06-08T00:00:00+00:00

A new blog

Welcome, this is a new place for the former coffeebean blog.

val message = "This new blog is based on Jekyll 3 and hosted on Github pages."

Older posts

You can see my previous posts on coffeebean.

Loïc's Blog

A Scala REST API in 2026, and beyond

Direct-Style REST in Scala 3: Virtual Threads, sttp, and the Road to Caprese

The Stack

Why Virtual Threads Change the Signature

Architecture

Error Handling with Either

MacWire: Zero-Reflection DI

Testing

The Road Ahead: Caprese and saferExceptions

What the current service looks like with Either

The same code with saferExceptions (Scala 3.x experimental)

Either vs saferExceptions — a practical comparison

The bigger picture: captureChecking

Summary

Some Scala & ZIO tips

How to cache queries with ZIO and ZQuery

How to write a generic micro service with Scala 3

How to define a flexible String decoder with ZIO JSON

All We Leave Behind (my doom metal band)

Type classes and extension methods in Scala 3

Type classes in Scala

Type class definition

Instances

More powerful functions

How to traverse optional values in Scala, Kotlin and Java

Scala version

Kotlin version

Java version

Memoization with ZIO

ZIO tip : How to memoize the result of an effectful computation :

ZIO HTTP basic app example

Data Validation in Scala with ZIO Prelude

Resources to run Spark and HDFS in Kubernetes

HDFS

Spark

Submit apps

A ZIO Akka HTTP giter8 template

How to add elements dynamically to a ZIO stream using a queue

Scala snippets (ZIO, Cats, Slinky, Akka Streams)

Change delimiter in a ZIO Stream

Accumulate errors with Either and Cats

Slinky React Hooks example

Akka Streams : Consume a source with two sinks and get the result of one sink

Akka Streams and Play Framework : add elements to a source dynamically, using a queue

HTTP streaming in Scala with STTP and ZIO

How to manage UI state with immutable data structures in Scala

Playing with ReasonML and ReasonReact

Introduction to functional programming and error management in Kotlin and TypeScript (french)

Lightbend VS Typelevel at ScalaIO - code samples and slides

Smart function application in Scala using currying and pipe operator

How to build a purely functional webapp in Scala

Small update about Scala Hamsters

Scala Hamsters 1.4.0 is out with Scala.js support

Scala Hamsters 1.3.1 is out

Creating user interfaces with Scala.js and Binding.scala

JavaScript for Scala Developers

Basic examples

Define a variable

Define a constant (a value)

Functions

Classes

Default parameters

Generators

Destructuring

Spread operator

Rest operator

Modules and imports

Promises

Where are my types ?

Type aliases with TypeScript

Immutable data

ES proposal : rest/spread properties

More advanced examples

Extensions methods

Other functional goodness

Spark Datasets and type-safety

A small update on Scala Hamsters

How to compose Future and Nullable in Kotlin

Scala Hamsters 1.0

Error Handling with `Either`

The Road Ahead: Caprese and `saferExceptions`

What the current service looks like with `Either`

The same code with `saferExceptions` (Scala 3.x experimental)

`Either` vs `saferExceptions` — a practical comparison

The bigger picture: `captureChecking`