Forem: Pin-Sho Feng

A story of building a custom Flow operator - bufferTimeout

Pin-Sho Feng — Sun, 17 May 2020 20:19:45 +0000

I have this particular use case where I need to process a live stream of elements to send them to a distributed queue (e.g. Kafka, NATS...). I'd like to consume this stream in batches that can be up to N elements in size and I want to buffer them for at most t duration. In other words, a batch should be processed when either of the following conditions is met:

Number of buffered elements reaches a certain size N
Timeout after t

Another use case would be consuming a live stream that we want to process and store to the DB, and in this stream there could be duplicated events or events that could be merged, which we'd like to process before storing.

1st attempt: Flow + Reactor

In this particular project I wanted it to be pure Kotlin with as few dependencies as possible so initially I didn't look at either RxJava or Reactor, and Kotlin coroutines (in particular Flow) came to mind. Looking at the available operators (using version 1.3.6 at the time of writing), I didn't find one for that functionality so I went to coroutine's Github issues to see if anyone had run into a similar problem and found an open PR for chunked and window. Still, this is missing the timeout functionality.

The next thing I thought was that either Reactor or RxJava would probably have something like that, and I was lucky to find bufferTimeout! Now, I wanted to avoid having an extra dependency but building an operator and doing it right must be quite difficult, right? I still want to use coroutines as much as possible, so maybe I could just use the interop library and use Reactor to process a Flow.

// some channel where events will be produced
val channel = Channel<String>()

val disposable = channel.consumeAsFlow()  
    .asFlux()  
    .bufferTimeout(500, Duration.ofMillis(500))  
    .concatMap { events ->  
        events  
            .map { event ->  
                mono {  
                    println("Received $event, ts: ${System.currentTimeMillis()}")
                    delay(100) // simultate suspending operation like sending to a queue  
                }
            .let { Flux.concat(it) }  
    }.subscribe()

Not bad, right? The problem is that when you start testing you'll find out that bufferTimeout doesn't implement backpressure (see this and this) and if your producer goes faster than you can consume, the flux will crash. The issue was first reported early 2018 and it doesn't seem easy to fix, so now what?

Maybe we could implement our own operator... Trying to fix bufferTimeout is out of the question so it has to be for Flow... but where do we start? 🤔 It's my first time using Flow, so I have no clue, let's see.

2nd try: imperative approach

If we can produce from a channel and can read from it, we should be able to just whip some naive code with a timeout to create our solution right? So I tried this path first. We'll need:

Something that will provide clock ticks for implementing the timeout
Accumulate values in a buffer

// some channel where events will be produced
val channel = Channel<String>()

val receiverJob = launch {
    // This launches a coroutine that will send ticks to a channel with a given frequency
    val tickerChannel = ticker(1000)
    val events = mutableListOf<String>()  

    try {
        while (true) {
            var hasTimedOut = false  
            val tickerJob = launch {  
                tickerChannel.receive()
                hasTimedOut = true
            }

            while (events.size < 500 && !hasTimedOut) {  
                withTimeoutOrNull(10) { channel.receive() }  
                    ?.let { events.add(it) }
            }

            events.forEach { event ->  
                println("Received $event, ts: ${System.currentTimeMillis()}")
            }

            events.clear()  
            tickerJob.cancel()
            hasTimedOut = false
        }
    } finally {
        tickerChannel.cancel()
    }
}

And voilà! Let's break this down a bit:

We launch a coroutine that will run the consumer logic
We have a mutable list that we use as the buffer of elements
There's an infinite loop to continuously read events from a channel
- Inside the loop, we launch a coroutine to check if we reached the timeout. This is necessary because a channel's receive is suspending.
- We read from the events channel while we haven't timed out or reached the desired buffer size. This is done again via receive and since it's suspending, we need to use withTimeoutOrNull to ensure it doesn't suspend forever if there are no items in the channel, as we might have elements in the buffer that we have to emit after the timeout.
Finally, since the current implementation of ticker doesn't respect structured concurrency and it's launched in the global scope, we need to make sure to cancel it to avoid a resource leak.
Note that backpressure is not a problem because by design Channel will suspend the producer if the consumer is not available to receive.

Now, this is a lot of code with a few undesirable properties:

Mutable variables
Manual resource management
Does not easily compose
Code will probably grow spaghetti over time

How can we improve it? There's probably already a bug that I haven't seen and over time that bug might be buried deep in more code. While some of the points can't be helped, can we encapsulate the logic so that it's easy to use? Sounds like it'd be awesome as a Flow operator!

Flow operator: bufferTimeout

Wouldn't it be great if we could just use it in the same way as with Flux so that the code looks like this?

channel.consumeAsFlow()
    .bufferTimeout(500, Duration.ofMillis(1000))
    .collect { println("Received $it") }

Taking the previous imperative implementation as a basis, we should be able to encapsulate it as an operator. With a few tweaks, this what I got initially:

@ExperimentalCoroutinesApi  
@ObsoleteCoroutinesApi  
fun <T> Flow<T>.bufferTimeout(size: Int, duration: Duration): Flow<T> {  
    require(size > 0) { "Window size should be greater than 0" }  
    require(duration.toMillis() > 0) { "Duration should be greater than 0" }

    return flow {  
        coroutineScope {  
            val events = ArrayList<T>(size)
            val tickerChannel = ticker(duration.toMillis())
            try {
                var hasTimedOut = false  
                val upstreamValues = produce { collect { send(it) } }  

                while (isActive) {  
                    val tickerJob = launch {  
                        tickerChannel.receive()
                        hasTimedOut = true  
                    }  

                    withTimeoutOrNull(10) { upstreamValues.receive() }  
                        ?.let { events.add(it) }

                    if (events.size == size || (hasTimedOut && events.isNotEmpty())) {  
                        emit(events.toList())  
                        events.clear()
                        tickerJob.cancel()
                        hasTimedOut = false
                    }
                }
            } finally {
                tickerChannel.cancel()
            }
        }
    }
}

As you can see, the code is pretty much the same as before but there are a few things worth noting:

Conceptually our operator creates a new Flow that consumes from the upstream Flow and emits for downstream consumption.
As we're going to call suspending functions, we need to be in a CoroutineScope.
Instead of receiving from a channel, we need to collect from the upstream Flow. As collect is a terminal operation that will not finish until the upstream is finished, we need to put it in a coroutine, which is what produce { collect { send(it) } } does while also creating a channel. All the collected values are sent to this channel that we'll drain in the infinite loop.
The rest is pretty much the same. Instead of while (true) I'm using while (isActive) but it doesn't really matter as all the suspending functions that are called support cancellation.
Some of the APIs used are either obsolete or experimental, but with no current alternatives AFAIK (coroutines 1.3.6), so this might change in a few months. I suspect select could make the code prettier but they're yet another experimental functionality so I thought what we had was enough for now.

That should be it, right? I proceed to write a simple test and run into an issue straight away:

@ExperimentalCoroutinesApi  
@ObsoleteCoroutinesApi  
internal class OperatorsTest : StringSpec({  
    "no emissions with empty flow" {  
        val actualCount = flowOf<Int>().bufferTimeout(5, Duration.ofMillis(10)).count()  
        actualCount shouldBe 0  
    }
})

// throws ClosedReceiveChannelException: Channel was closed

What the... ? It turns out withTimeoutOrNull(10) { values.receive() } is not accounting for the case when the channel is closed (upstream is done emitting). There's a receiveOrClosed() function but it's marked as internal so we can't use that. I decided to catch the exception and drain any event that might be left in the buffer:

try {
    // previous loop here
} catch {
    emit(events)
} finally {
    tickerChannel.cancel()
}

Now the previous test passes and I continue writing a test to check that the timeout is respected:

"buffers correctly by timeout" {
    var firstEmissionTs = 0L
    var firstItemReceiveTs = 0L
    var secondItemReceiveTs = 0L

    flow {
        firstEmissionTs = System.currentTimeMillis()
        emit(1)
        delay(50)

        emit(2)
        delay(50)

        emit(3)
    }
        .bufferTimeout(5, Duration.ofMillis(50))
        .collectIndexed { index, value ->
            when (index) {
                0 -> {
                    firstItemReceiveTs = System.currentTimeMillis()
                }
                1 -> {
                    secondItemReceiveTs = System.currentTimeMillis()
                }
                2 -> {}
                else -> {
                    throw RuntimeException("Emitted inexistent item")
                }
            }
        }

    (firstItemReceiveTs - firstEmissionTs) shouldBeGreaterThanOrEqual 50
    (secondItemReceiveTs - firstEmissionTs) shouldBeGreaterThan 100
}

// test fails randomly for the second assertion:
//   java.lang.AssertionError: 53 should be > 100

Something's off because the second item should be received after the second timeout, which is 50 + 50 = 100. What's going on? It turns out the delay that we specify in withTimeoutOrNull(10) is important because due to concurrency it could happen we're waiting on withTimeoutOrNull when the timeout ticker has already passed, meaning that a batch should be emitted. If during this timeout we receive something, a batch is going to be sent with this new item which should actually belong to the next batch. If we add some logs, it may look like this:

withTimeoutOrNull Got 1
withTimeoutOrNull timedout
withTimeoutOrNull timedout
withTimeoutOrNull timedout
withTimeoutOrNull timedout
timeout tick                 -- happens concurrently
withTimeoutOrNull Got 2      -- we were waiting due to delay
start emitting batch         -- batch has 1 and 2

Alright, let's take a step back. How do we decide when to emit a batch? When either we got the buffer filled or timeout has passed, whichever condition happens first but not concurrently. Can we somehow select between suspending functions and get the first one that is continued? Yes, we'll finally use the experimental select expression we mentioned before! Using that, the code is actually simplified:

@ExperimentalCoroutinesApi
@ObsoleteCoroutinesApi
fun <T> Flow<T>.bufferTimeout(size: Int, duration: Duration): Flow<List<T>> {
    require(size > 0) { "Window size should be greater than 0" }
    require(duration.toMillis() > 0) { "Duration should be greater than 0" }

    return flow {
        coroutineScope {
            val events = ArrayList<T>(size)
            val tickerChannel = ticker(duration.toMillis())
            try {
                val upstreamValues = produce { collect { send(it) } }

                while (isActive) {
                    var hasTimedOut = false

                    select<Unit> {
                        upstreamValues.onReceive {
                            events.add(it)
                        }

                        tickerChannel.onReceive {
                            hasTimedOut = true
                        }
                    }

                    if (events.size == size || (hasTimedOut && events.isNotEmpty())) {
                        emit(events.toList())
                        events.clear()
                    }
                }
            } catch (e: ClosedReceiveChannelException) {
                // drain remaining events
                if (events.isNotEmpty()) emit(events.toList())
            } finally {
                tickerChannel.cancel()
            }
        }
    }
}

And there goes my first attempt at making an operator out of an actual need! To save you from scrolling up again, the final usage is

someFlow
    .bufferTimeout(size = 500, duration = Duration.ofMillis(1000))
    .collect { println("Received batch: $it") }

Final remarks

This wasn't exactly easy, concurrency is hard, but it was definitely less scary than attempting to do the same with Rx as suspending channels make the backpressure problem a lot easier. There may be bugs or improvements that I'm not seeing so if you spot anything, I'd really appreciate a comment!

Thanks for reading and see you next time.

Make impossible states impossible - Kotlin edition

Pin-Sho Feng — Mon, 16 Mar 2020 18:31:36 +0000

Sometimes called "functional domain modelling", there are a bunch of articles and videos titled like that but I haven't found a Kotlin edition, so took the chance and hope it's useful :)

Let's state our goal clearly again: make impossible states impossible.

What does that mean? In a statically typed language like Kotlin, it means we want to leverage the compiler to ensure that we can only have valid models so that it's impossible to have incorrect cases. In other words:

Use the type system to describe what is possible or not. If something is not possible, it should not compile.
Another way to look at it is: use types as if they were tests that you won't need to write. Think about nullable types in Kotlin, for example... if it's not marked with ? you know it's not null (unless someone cheated along the way... but you can't control who uses reflection either).

Let's start with an apparently normal example of what could be a user profile in some platform:

const val ACCOUNT_TYPE_BASIC = 0
const val ACCOUNT_TYPE_PREMIUM = 1
const val ACCOUNT_TYPE_PRIVILEGE = 2

data class UserProfile(
    val name: String,
    val age: Long,
    val accountType: Int,
    val email: String?,
    val phoneNumber: String?,
    val address: String?
)

What problems can you spot?

name: there could be a number within the string. What about the surname? Special characters? What if it's empty?
age: can anybody live so long to fill a Long? Can it be 0? Negative?
accountType: what if someone puts a number that's not in our constants list?
email: an email needs a specific format, but we could write anything in there and it'd compile.
phoneNumber: same...
address: same again...

How could we use the type system to make these cases impossible?

Let's start with name.

Name

We'd normally have some sort of function that validates the input before putting it into the model, but a String doesn't reflect any of those validations. What can we do? How about a type that represents a validated name? Let's explore that.

data class UserName(val name: String) {
    companion object {
        fun createUsername(name: String): UserName {
            // perform some validations on name
            if (name.isEmpty()) throw IllegalArgumentException()
            return name
        }
    }
}

So with a specific type we can represent a validated model. But still, there are problems in the proposed solution...

Anyone can just create a UserName, bypassing createUserName.
It blows in runtime if there's any problem

For the first case, an attempt could be to make the constructor private:

data class UserName private constructor(val name: String)

But then we get a warning telling us that "private data class constructor is exposed via the generated 'copy' method". Bummer... it's a reported bug and it doesn't seem to be a priority for JetBrains. At this point for simplicity I think we could just ignore the warning, rely on convention and call it a day...

Now, if it still doesn't feel right, we could hack it using some interface magic:

interface IUserName {
    val name: String

    companion object {
        fun create(name: String): IUserName {
            // perform some validations on name
            if (name.isEmpty()) throw IllegalArgumentException()
            return UserName(name)
        }
    }
}

private data class UserName(override val name: String) : IUserName

More verbose than desirable... but it works. Now, there's still the problem of the IllegalArgumentException. Ideally we want to handle all our cases and make it impossible to blow up. We could use something like Either, or if we don't want to add Arrow we can just use Java's Optional.

fun create(name: String): Optional<IUserName> {
    // perform some validations on name
    if (name.isEmpty()) return Optional.empty()
    return Optional.of(UserName(name))
}

Or since this is Kotlin, you could just make it nullable:

fun create(name: String): IUserName? {
    // perform some validations on name
    if (name.isEmpty()) return null
    return UserName(name)
}

For the rest of this post I'll just use the non-interface version for simplicity, but hopefully the point has come across.

age

Pretty much the same as for name, we can create a validated type for it.

accountType

For this one we can just use a common construct available in many languages: Enum.

enum class AccountType {
    ACCOUNT_TYPE_BASIC,
    ACCOUNT_TYPE_PREMIUM,
    ACCOUNT_TYPE_PRIVILEGE
}

email, phoneNumber, address

At this point it should be obvious that these fields share the same problem, so we can create validated types for each of them.

But is that the only problem? Is it possible to have a user that we can't contact in any way? According to the model, this is perfectly valid:

val profile = UserProfile(
    ...,
    email = null,
    phoneNumber = null,
    address = null
)

We probably want to be able to contact the user right? So maybe we can generalize all of them as a ContactInfo. Is there any way to express that a ContactInfo can be "an email, a phone number, or an address"? How would you do that with GraphQL? Hmm... union types? In Kotlin we can represent these with sealed classes.

// for simplicity, assume that we have factory methods for those data classes and the constructors are private... 
sealed class ContactInfo
data class Email(val email: String) : ContactInfo
data class PhoneNumber(val number: String) : ContactInfo
data class Address(val address: String) : ContactInfo

In GraphQL syntax this would be: union ContactInfo = Email | PhoneNumber | Address.

So everything's validated... is that enough?

Our UserProfile might look like this now:

data class UserProfile(
    val name: UserName,
    val age: Age,
    val accountType: AccountType,
    val contactInfo: List<ContactInfo>
)

Is that OK? Can contactInfo be empty? We did say 'no' before, didn't we? We could create a special NonEmptyList type (or use Arrow):

data class UserProfile(
    val name: UserName,
    val age: Age,
    val accountType: AccountType,
    val contactInfo: NonEmptyList<ContactInfo>
)

Now? Hmm... are duplicate ContactInfos allowed? 🤔 What's a data structure that can contain only unique elements?

data class UserProfile(
    val name: UserName,
    val age: Age,
    val accountType: AccountType,
    val contactInfo: NonEmptySet<ContactInfo>
)

And that's it! (NonEmptySet is non-standard Java/Kotlin, but should be easy to create)

Conclusion

Making impossible states impossible is about using our data types to represent valid things in our domain, which will ultimately lead to more robust software with less bugs. It's often called "functional domain modelling", probably not because it has anything to do with functional programming per se, but most likely because in the statically-typed functional world we strive for "total functions", which are those that consider all possible inputs and outputs.

Just asking yourself the question of "does my model allow any illegal state?" will get you a long way!

Some resources you can check:

Bonus

When Kotlin finally gets inline classes, we'll be able to have zero-cost abstractions.

// this needs a data class to wrap a String
data class Name(val name: String)

// inline classes are basically the underlying primitive,
// verified by the compiler
inline class Name(val name: String)

Setting up Jenkins on MicroK8s

Pin-Sho Feng — Tue, 16 Apr 2019 15:59:15 +0000

I recently bought a mini-pc (MSI Cubi 3 Silent) for home that I use as a server for various purposes and something that I wanted was to set up Jenkins for automatically building and deploying my side projects to my local MicroK8s cluster (Ubuntu Server 18.04).

It turns out you can install Jenkins on Kubernetes and thanks to kubernetes-plugin Jenkins agents will run in ephimeral pods created just for building your task.

If you know what you have to do, it's actually pretty easy to configure but that's very often the case in hindsight, isn't it? I battled it out for a whole day and here are my findings, should probably take 30 mins to set up.

Prerequisites

MicroK8s cluster (should work with other Kubernetes clusters too), with storage enabled.
Docker service running on the host (outside Kubernetes)
Helm installed and deployed in Kubernetes

Installing Jenkins on Kubernetes

# Download Jenkins chart values for customization. I'm linking the version I used, but the newest should work.
$ wget -O jenkins-chart.yml https://github.com/helm/charts/blob/b22774e2f2019c0e5d8182aab6111a84e95fa8ca/stable/jenkins/values.yaml

# Edit chart options to use NodePort instead of LoadBalancer, as MicroK8s doesn't
# ship with any: https://github.com/ubuntu/microk8s/issues/200#issuecomment-441180273
# 
# ServiceType: NodePort
# NodePort: 32000 (optional, if you want to have a fixed one for exposing it externally)
#
# Alternatively, if you have Ingress enabled on MicroK8s you could 
# use ServiceType: ClusterIP and add a rule in your Ingress configuration to
# direct traffic to Jenkins.
$ vim jenkins-chart.yml

$ helm install --name jenkins -f jenkins-chart.yaml stable/jenkins

# Follow the instructions to retrieve the generated password.
# If you need to uninstall Jenkins, then
$ helm del --purge jenkins

At this point, Jenkins should be accessible at http://your-server:32000. I then installed the Blue Ocean plugin as I find it more aesthetically pleasant plus it's designed for the Jenkins Pipeline. Installing it is as easy as going to Manage Plugins and checking the Blue Ocean box. You don't need to check the other ones, they'll be added automatically as dependencies.

Now we're ready to create our first pipeline!

Creating a Github pipeline

Let's assume we want to create a pipeline for a private repository we've got at Github.

Follow the instructions here but skip the final pipeline creation wizard, as we're going to use a Jenkinsfile with the declarative pipeline syntax instead.
Create a webhook in your Github repo pointing to http://your-server:32000/github-webhook/ (customize with your domain/port/https, and keep the trailing slash) and verify that it's working (the secret doesn't matter). Theoretically this should be done automatically by Blue Ocean in step 1, but it didn't work for me. Note that if you're using a firewall or you're in a private network, you'll have to open the port and/or forward it to your server appropriately.
Create a file named Jenkinsfile at the root of the repository. I'm going to assume you've got a dockerized app and you've got the Kubernetes deployment files well configured.

// Jenkinsfile

pipeline {
  agent {
    kubernetes {
      // this label will be the prefix of the generated pod's name
      label 'jenkins-agent-my-app'
      yaml """
apiVersion: v1
kind: Pod
metadata:
  labels:
    component: ci
spec:
  containers:
    - name: docker
      image: docker
      command:
        - cat
      tty: true
      volumeMounts:
        - mountPath: /var/run/docker.sock
          name: docker-sock
    - name: kubectl
      image: lachlanevenson/k8s-kubectl:v1.14.0 # use a version that matches your K8s version
      command:
        - cat
      tty: true
  volumes:
    - name: docker-sock
      hostPath:
        path: /var/run/docker.sock
"""
    }
  }

  stages {
    stage('Build image') {
      steps {
        container('docker') {
          sh "docker build -t your-registry/my-app:latest ."
          sh "docker push your-registry/my-app:latest"
        }
      }
    }

    stage('Deploy') {
      steps {
        container('kubectl') {
          sh "kubectl delete -f ./kubernetes/deployment.yaml"
          sh "kubectl apply -f ./kubernetes/deployment.yaml"
          sh "kubectl apply -f ./kubernetes/service.yaml"
        }
      }
    }
  }
}

Commit and push it to your repository. Jenkins should be notified and a build should be starting!

A few things to note from the Jenkinsfile

The docker and kubectl containers declared are so that we can run these commands in the pipeline without having them installed in the host, which is typically what you'd do with Jenkins outside Kubernetes.
We're using the host's Docker daemon, which is why it was a prerequisite and the reason we defined docker-sock with path: /var/run/docker.sock. If you don't want to need to have Docker in your host, you could try Docker in Docker.
We're using lachlanevenson/k8s-kubectl, which is a very simple image with just kubectl. You could use any other image with kubectl, but make sure the image doesn't run with another user id or you'll have permission issues with Jenkins. I know it from the experience of trying to use bitnami/kubectl.
You might wonder why we've got command: cat and tty: true. They avoid containers exiting early. To be honest, I don't know exactly how it works.
There's no specific step for git. If you use webhooks, the git step is automagically added when a push event is received.

Other things of interest

If you have submodules, make sure you check Recursively update submodules in your Jenkins repo build configuration, under Advanced sub-modules behaviours.
You might notice 'zombie' processes (see top command) as builds are executed. I don't know why this happens but it could be a bug in Jenkins or Docker. It doesn't happen consistently though.

Final words

Configuring Jenkins might seem a very daunting task and it's certainly not obvious. With Kubernetes and Helm I believe it's much simpler but still it's difficult to find easily reproducible instructions.

If you're setting a Jenkins environment without an existing Kubernetes cluster completely from scratch, you might want to have a look at Jenkins X. Otherwise, I hope this guide works for you.

See you next time!

Dependency injection in functional programming

Pin-Sho Feng — Sun, 13 Jan 2019 19:54:44 +0000

Dependency injection in functional programming (Kotlin)

tl;dr

DI in FP can be achieved using partial function application or the Reader monad
But DI may not make sense in FP
An OO-FP-OO sandwich where the domain core is purely functional but the outer layers are OO and use a dependency injector could be a good approach for applications which have a lot of side-effects.
All solutions have pros and cons, final choice depends on the task at hand and personal preference

When I started looking into functional programming, I kept wondering how DI works in this paradigm and was surprised to find that there are not many articles written about it, especially for beginners, so I decided I could try to help in filling that void by sharing what I've learnt. Therefore, the target of this article are the developers that have started looking into the functional paradigm and wonder how dependency injection works there, so I'll try to introduce every topic without assuming much prior knowledge. The language I use is Kotlin, though the principles are universal. The article should preferrably be read from top to bottom, as the explanation of the concepts follow a logical order. Let's start!

Why functional programming?

Like many developers, the past couple of years I started gaining a lot of interest in functional programming due to all the problems I've seen in practice with mutable state. As an Android developer for the past 5 years, I've experimented these problems with the wide adoption of the MVP architecture. At the time it was all the rage as it allowed us to move away from untestable God-Activities/Fragments and have the presentation logic delegated to testable Presenters. As new features were implemented and requirements changed, our presenters would start to look like the old huge activities/fragments, and making any change would become increasingly dangerous, as presenters modify the view in imperative style (e.g. view.showLoading()) and multiple view-changing functions were in place. We'd split the presenters further but still, the complexity of not knowing if a change would affect something else was there. There had to be a better way!

One day I was still in the office (XING Barcelona) while a meetup was held in which the speaker was introducing Redux to the audience. I'm glad I decided to stay as it ultimately inspired me and got me to learn about functional programming. I started digging and discovered that Redux was partly inspired by Elm and then I started checking Elm.

Elm is an awesome pure functional language that tries very hard to be beginner friendly and while I think it does a great job, the paradigm shift is still enourmous. It's not so much the syntax itself, but the mindset change from an imperative to a functional one (to the Android folks, do you remember when you started with RxJava?). After some practice I began feeling comfortable with Elm, but I still remembered a conversation with a colleague in which he was trying to explain "monads" to me (Elm doesn't have them). No matter how many articles I'd read (including this one with pictures) I still wouldn't get it. The thing is that most, if not all articles focus on the "what", but not "why" we need monads. That's when I decided to go all-in and teach myself Haskell.

After reading LYAH and chunks of Haskell Programming (amazing book btw) I decided to do some practice by building an ArnoldC compiler for fun. The experience was very rewarding, even if I end up never using Haskell professionally. It's forever changed the way I look at code but while I do know what a monad is now, I'm also convinced that type classes (functors, applicatives, monads...) are not essential to functional programming, as Elm or Elixir show. Abstraction is a tool, not a goal! I'm aware this is opinionated and others may have a different opinion, which is fine. I think that type classes are especially useful for library writers when they strive to provide generic solutions but end-user application features rarely need such level of abstractions.

Joe Armstrong said "state is the root of all evil" and I subscribe to it. I've concluded that with these 2 principles you can get very far:

Immutability
Referential transparency (for a given input, you get the same output always with no side-effects)

As long as your programming language has first-class/top-level functions, just applying those 2 principles can get you a long way.

Pure functions excel at transforming data. For any given input, it'll return the same output, without side effects. This makes functional programming suitable for applications that resemble pipelines in which the input data is transformed through various stages. Examples of such applications are parsers, compilers, data analysis, etc.

But what about UI or CRUD applications? These sort of applications are basically full of side-effects. How can we fit a functional programming paradigm here? Doesn't this go against the 2 principles mentioned before?

So in the quest for trying to use functional programming to solve the same problems I've been running into the past few years, I've tried to force myself to strictly apply functional programming and see how it goes.

I built a sample server application in Kotlin to test the waters, having the 2 principles above in mind. It's a very simple Spring Boot application that exposes an endpoint that returns some data. Internally, the app will call another endpoint and process the results before returning a response. Fairly standard.

To be functional I followed these guidelines:

Use only functions, avoiding classes except for the glue initialization code. Having no classes and just functions encourages you to think in terms of data transformation and avoid state.
Use val and immutable data structures (Map, List...).
Avoid side effects.

I followed Clean Architecture and separated my code in 3 logical layers: presentation, domain and data. When the server receives a request, the logic will flow like

request handler -> interactor -> external services

All good. But then I ran into a problem... How do I achieve DI? My interactors (domain) are no longer classes so I can't inject them the repositories that they should be calling.

If you look up "functional programming dependency injection" in Google, you'll find that there's (perhaps) surprisingly very little literature written on the topic. I may have an idea why that's the case but bear with me and keep reading! I'll be covering two of the most typical approaches:

Partial function application
Reader monad

There are other styles that can help you achieve similar goals, such as the tagless-final pattern or free monads, but I'm not going to cover them in this article.

Dependency injection goals

Before we look at functional programming's solutions to DI, let's take a step back and reflect on the problem that we're trying to solve. Without going into too much detail, we could summarize the DI goals as:

Inversion of Control and loose coupling. Implementations don't choose their dependencies. This allows the caller to supply the specific implementations of these dependencies, allowing the behavior of the code to be configured externally. Typically, this is done by making classes depend on interfaces, the implementations of which are passed-in via constructor.
Unit testing. This derives from the previous goal, as it allows us to inject mock implementations into the class under test.

Let's take a look at how we might achieve these goals with functional programming.

Partial function application

Partial function application is a very cool concept in functional programming that consists in passing some of the parameters to a function and getting a function that accepts the rest of the parameters in return. This concept derives from currying, which is a technique by which any function of N arguments can be translated into a sequence of N functions of 1 argument. For example:

fun sum(a: Int, b: Int): Int = a + b
sum(1, 2) // returns 3

// can be converted into a function that accepts 1 argument (a)
// and returns another function that accepts 1 argument (b)
fun sum(a: Int): (Int) -> Int {  
    return { b -> a + b }  
}

val sum1 = sum(1)
sum1(2) // returns 3

How can this help us? Imagine that we want to be able to inject different loggers into a function. We might want one that simply prints its output to the console or we might want another one that stores the log in a file. We can partially apply the function with the logger and the resulting function won't need the logger parameter anymore.

typealias ItemFetcherFn = () -> List<Item>

fun fetchItemsLogged(log: (String) -> Unit): ItemFetcherFn  {
    return {
        log("Fetching items")
        // fetch and return items...
    }
}

// let's use partial application to "inject" some logger
val consoleLogger: (String) -> Unit = { str -> println(str) }
val fetchItems: ItemFetcherFn  = fetchItemsLogged(consoleLogger)

// we can now use fetchItems anywhere, without knowing it has a logger
fun onStart(fetchItems: ItemFetcherFn ) { ... }

onStart(fetchItems)

This technique was the one I used in the master branch of my sample project. As I mentioned before, I separated the code in 3 logical layers (clean architecture style): presentation (or controllers), domain and data. If you remember the dependency rule, the inner layer cannot access the outer layer. Our inner layer is the domain layer, so it shouldn't have references to the presentation and the data layer (note that a Repository is considered an interface to the data layer, and the domain can reference interfaces). With this in mind, let's analyze our example starting with the data layer's Services.kt:

fun getUserTodos(
    baseUrl: String,
    userId: Int
): Mono<List<Todo>> {
    return WebClient.create(baseUrl).get()
        .uri { uriBuilder ->
            uriBuilder.path("/users/$userId/todos")
                .build()
        }
        .retrieve()
        .bodyToMono(object : ParameterizedTypeReference<List<DataTodo>>() {})
        .map { dataTodos -> dataTodos.map { it.toDomain() } }
}

From the domain layer's perspective, this function has a baseUrl, which is clearly a data layer detail. From the domain we shouldn't be concerned about where we get the data from. If we were using OO, this function would be encapsulated in a Repository object and baseUrl could be injected via constructor, so the function signature that the domain layer would end up using is fun getUserTodos(userId: Int). How can we achieve this in functional programming style? Using partial function application! We could partially apply baseUrl. Let's see how this looks in practice for Interactors.kt:

typealias TodoFetcherFn = (userId: Int) -> Mono<List<Todo>>

fun getCompletedTodos(fetchTodos: TodoFetcherFn, userId: Int): Mono<List<Todo>> {
    return fetchTodos(userId).map { todos -> todos.filter { it.isCompleted } }
}

Our domain function depends on a function of type TodoFetcherFn that takes a userId and returns a Mono<List<Todo>>. We're not depending on any specific implementation and as such the function is unit testable, so we achieved our DI goals. As long as we pass in a function that complies with this signature, we're good to go!

Let's see how this function is partially applied and "injected". This is Todos.kt from the controller layer:

fun getTodos(request: ServerRequest, fetchTodos: TodoFetcherFn): Mono<ServerResponse> {
    val userId = request.pathVariable("id").toIntOrNull()

    return if (userId != null) {
        val todosResponseMono =
            getCompletedTodos(fetchTodos, userId)
                .map { it.toResponse(userId) }
                .onErrorResume { e -> internalServerError(e) }

        ServerResponse.ok().contentType(MediaType.APPLICATION_JSON)
            .body(todosResponseMono, GetTodosResponse::class.java)
    } else {
        ServerResponse.badRequest().syncBody("id is not an integer")
    }
}

And finally this is the injection root (SampleApplication.kt):

@Configuration
class Router @Autowired constructor(
    @Qualifier(TODOS_BASE_URL) private val todosBaseUrl: String
) {

    @Bean
    fun routes(): RouterFunction<ServerResponse> = router {
        accept(MediaType.APPLICATION_JSON).nest {
            GET("/user/{id}/completedtodos") {
                getTodos(it, getTodosFetcherFn())
            }
        }
    }

    private fun getTodosFetcherFn(): TodoFetcherFn {
        return ::getUserTodos.partially1(todosBaseUrl)
    }
}

The sample is using Arrow to partially apply the first argument of getUserTodos (baseUrl). This function is then passed to the route handler, which then calls the interactor function with it.

Using partial function application we've successfully achieved FP's version of dependency injection. However, you might have some questions.

How do you test the route handler (`getTodos`) in isolation?

Fair question. If you want to test getTodos, you'd inevitably test getCompletedTodos as well as it makes a direct call to it. My advice is to not test it in isolation, test the whole feature. Just use a mock web server and test the whole feature end-to-end. There are many benefits to this approach.

Still, this is my opinion and you may not agree. If you really want to test it separately, you could make getTodos depend on a function that you'd pass in instead of having it call getCompletedTodos directly. I accidentally did that in a previous commit, so you could check how it'd work. Note that that implementation is wrong, the fetchTodos function that's passed in should have a signature separate from TodoFetcherFn from the domain layer, even if they look the same, so that we keep the functions independent.

If you have 5 dependencies, do you pass 5 arguments down?

My example is unrealistic as any real application will likely have more dependencies than that. For simplicity I'm just passing the function that I know I'll use, but you could construct a composition root in which you partially apply the functions that will have to be called as we go through the layers. Then you only have to pass the functions that will have to be called in the immediately next layer. Check how this works here. Another approach would be to pass a single argument with all the dependencies, which is what you'd do with the Reader monad.

The dependencies appear as arguments in the function signature of all functions in the call chain

Indeed. Whether that's cool or not is subjective, but there's a technique in functional programming that allows us to pass an argument implicitly so that it disappears from the argument list of your functions. Meet the Reader monad in the next section!

Code navigation is more difficult as cmd + click doesn't work on partial function signatures

Yes. If you think about it, the key idea of using partial application for DI is loose coupling, so you're not supposed to know which function was originally partially applied. If we organize the code in a way that we don't have deeply nested calls of partially applied functions, it should be simple to find the root.

I hope I managed to explain the technique in an understandable way, but if it's not enough or if you simply would like to see another take, there's an awesome article by Scott Wlaschin explaining it with a more complex example.

The Reader monad

We've found that passing the same argument down a chain of functions can be (subjectively) annoying. I mentioned that the Reader monad can allow us to pass that argument implicitly... how's that possible? We could put it in some global value and access it directly where needed. We no longer need to pass it to any function! But then this is just hardcoding and beats completely the purpose of dependency injection, in case we lost sight of our initial goals.

Alright, so what is the Reader monad about? To properly understand it we'd first need to understand what's a monad and why do we need it. Before that, you'd need to be introduced to functors and applicatives. It personally took me a lot of time to figure it out, even after reading many explanations and seeing them represented as pictures. I actually never got it until I actually used them in practice, so I'm not going to attempt to explain all the concepts here.

Instead, let's see if we can build an intuition about it. We can start by stating our goals clearly:

Pass an argument to all functions in a call chain
Do so without explicitly having it in the parameters list of these functions

If we call a function that returns a value, we get that value straight away. However, if we want to defer this computation, we can wrap that value in a function and return it. Then we won't get the value until we call this returned function.

fun giveMe5(): Int = 5

fun deferGiveMe5(): () -> Int = { 5 }

giveMe5() // returns 5
deferGiveMe5()() // returns 5

What if we connected a bunch of functions that return functions that take the same input? The result would be a function chain that wouldn't be evaluated until we actually call it and then we can pass some input down the chain.

fun delegateNum(): (Int) -> Unit {
    return { num -> furtherDelegateNum()(num) }
}

fun furtherDelegateNum(): (Int) -> Unit {
    return { num -> printNum()(num) }
}

fun printNum(): (Int) -> Unit {
    return { num -> println(num) }
}

val chain = delegateNum()
chain(5) // prints 5

5 in this case would be the "thing" that we want to inject down the chain. It could be the dependency that we need to make available to all the functions in it and it's implicit! So we achieved the goals we set before.

Now, notice how delegateNum(), furtherDelegateNum() and printNum() have the same structure. It turns out we can abstract this and this is loosely speaking what the Reader monad is about! Create a lazy (as in deferred evaluation) chain of functions that can read a common value, and then run this chain passing the value that we want.

Hopefully at this point I managed to give you an idea of the Reader monad, so let's see it with a real example. I didn't create a version of our example but we can use one from Jorge Castillo that shares a similar architecture to the one I used previously, although his example is an Android app. In his example he's fetching a list of superheroes from some endpoint. Let's check the data source:

// MarvelNetworkDataSource.kt

/*
 * This is the network data source. Calls are made using Karumi's MarvelApiClient.
 * @see "https://github.com/Karumi/MarvelApiClientAndroid"
 *
 * Both requests return a new Reader enclosing an action to resolve when you provide them with the
 * required execution context.
 */
fun fetchAllHeroes(): Reader<GetHeroesContext, IO<List<CharacterDto>>> =
    ReaderApi.ask<GetHeroesContext>().map({ ctx ->
      IO.monadDefer().binding {
        val result = runInAsyncContext(
            f = { queryForHeroes(ctx) },
            onError = { IO.raiseError<List<CharacterDto>>(it) },
            onSuccess = { IO.just(it) },
            AC = ctx.threading
        ).bind()
        result.bind()
      }.fix()
    })

The value that's passed down through his function chain is ctx, which is of type GetHeroesContext, which is a data class with all the dependencies that this chain needs (see SuperHeroesContext.kt).

This function is called from getHeroes(HeroesRepository), which is called from HeroesUseCases.kt:

fun getHeroesUseCase() = getHeroes(NetworkOnly).map { io -> io.map { discardNonValidHeroes(it) } }

which is then called from getSuperHeroes(SuperHeroesPresentation.kt), which is finally called from SuperHeroListActivity.kt:

override fun onResume() {
    super.onResume()
    getSuperHeroes().run(heroesContext)
}

Notice how heroesContext is passed down the function chain through the run function and how there's no trace of it in the parameters of the functions in the chain.

So this is how you do DI with the Reader monad! If you feel the need for a more technical explanation, check this post from the author himself.

Now, you might again have some questions...

How do you decide between the Reader monad and partial application?

This is a common question. The main reason people usually use the Reader monad is to avoid passing the same argument over and over again. Still, this may be subjective.

I said in the introduction that in my opinion these sort of abstractions (typeclasses) are not strictly necessary and we have an alternative to it, but whether you prefer it or not is subjective to a certain point. I personally don't currently feel I need them with the approach that I'll suggest later (I will tell you about it below, keep reading!).

Aren't we coupling all the functions in the Reader chain to the value that we're passing?

I think so... any function in the chain will be able to access the value that we're passing. In the example above, any of these functions has access to GetHeroesContext. I'll go in more detail in the next section.

A critical view

We've seen 2 ways of achieving DI in functional programming, but while we did achieve the goals that we set in the beginning, is this way of doing things better than having a dependency injection framework in OO style?

Regarding partial function application, a user named Onur left a thought-provoking comment in the Scott Wlaschin's article that I already mentioned before:

This is a much needed article unfortunately it misses the point. Normally, objects are not composable. Dependency Injection is a cure for that. Just like functional programming uses functions to build larger programs, dependency injection is used to build larger object programs. In your examples, dependencies are only represented in 1 level but there can be far more deep hierarchies. Secondly, dependency injection registration is a way to achieve inversion of control, such that the components don't create themselves, but registrations are like a recipe that describes the dependencies and hierarchy and is external to the application. Thirdly, dependency injection also incorporates to the life times of the components. Finally, all of these imply impurity.

This is indeed very true. The composition root in our case is arguably not external to our application. We've got it in our router function. We could tuck it away by putting it into some other file, but we still need to manually define the partial function applications that we want to pass to our specific entry function, whereas with a DI framework we'd simply define the provider functions somewhere and then simply declare what we want to use in the constructors.

Lifetime is another thing... lifetime as such doesn't make sense in a pure context but things such as a database connection have a lifetime. So DI implies impurity, does it make sense to try to do it in the functional world?

Let's come back to the last concern of the previous section. Isn't using the Reader monad in our example implying that we're coupling all our functions in the chain to the dependency context that we're passing? Let's check SuperHeroesContext:

sealed class SuperHeroesContext(ctx: Context) {

  val heroDetailsPage = HeroDetailsPage()
  val apiClient
    get() = CharacterApiClient(MarvelApiConfig.Builder(
        BuildConfig.MARVEL_PUBLIC_KEY,
        BuildConfig.MARVEL_PRIVATE_KEY).debug().build())
  val threading = IO.async()

  data class GetHeroesContext(val ctx: Context, val view: SuperHeroesListView) : SuperHeroesContext(ctx)
  data class GetHeroDetailsContext(val ctx: Context, val view: SuperHeroDetailView) : SuperHeroesContext(ctx)
}

This context has 5 dependencies. Let's see MarvelNetworkDataSource.kt again:

fun fetchAllHeroes(): Reader<GetHeroesContext, IO<List<CharacterDto>>> =
    ReaderApi.ask<GetHeroesContext>().map({ ctx ->
      IO.monadDefer().binding {
        val result = runInAsyncContext(
            f = { queryForHeroes(ctx) },
            onError = { IO.raiseError<List<CharacterDto>>(it) },
            onSuccess = { IO.just(it) },
            AC = ctx.threading
        ).bind()
        result.bind()
      }.fix()
    })

We can see that we're accessing ctx.threading and ctx is passed to queryForHeroes, which internally accesses apiClient. But what's impeding us to also access ctx.view or ctx.heroesDetailPage? Should those be available to the data layer? I think we're violating the dependency rule. Besides that, generally speaking it's a good idea to strive for "making impossible states impossible" (see talks by Richard Feldman and Patrick Stapfer, or this article). If we can't access ctx.view, there's no way we can do anything about it!

We've showed that we can achieve DI with partial function application or the Reader monad, but the question is, should we?

OO-FP-OO, a hybrid approach

Let's remind ourselves of the things we like most about functional programming:

Immutability. Once a data value is created, we can never modify it.
Referential transparency. Given a function and an input value, you receive the same output, without side effects, every single time.

These properties make it easier to reason about programs, as you can be sure about the behavior being consistent. If the language is statically typed, equational reasoning becomes possible. This is going to make us have less bugs and less cognitive load, as your thought context should just be the function you have in front.

Typical Android applications or backend CRUD services are full of side effects. You need to access the network or a database to retrieve some data, and then you need to modify the UI to display this data or return some Json response. On top of that, like most languages Kotlin does not have a way to enforce referential transparency, so you can use any typeclass that Arrow provides but there's nothing preventing you from doing side effectful code that can be something as apparently harmless as a println or you could just call Thread.sleep(...) and stop the world. In practice, this means we need to follow some guidelines and make sure everybody in our team is aware of them.

If we need to access the network or the database and we can't enforce purity anyway, why aren't we even more pragmatic and simply do these in OO style? Our domain logic is what can be isolated from external factors such as UI or the data layer, so we could write it in FP style. If you view our business logic as a deterministic black box, you can just pass some data in (loaded from a DB), get some data out and render it to the UI.

data layer input -> domain logic -> output UI

As you can see, the domain logic is in the middle. If the rest is OO, we get a OO-FP-OO sandwich (I was inspired by Dax Fohl for the concept, although he's suggesting an FP-OO-FP approach).

Going back to what this article is about and the open question left at the end of the previous section, we could continue to leverage our DI framework and just keep our business logic core functional. Note that this doesn't mean that we should start having mutable state everywhere in the OO part. My approach would be "prefer FP over OO", which means to generally have an FP mindset but be pragmatic and use OO sparsely when it's much more convenient and it's not dangerous.

I tried this approach in another branch of the example. The domain logic stays exactly the same as in the partial application case:

// Interactors.kt
fun getCompletedTodos(fetchTodos: TodoFetcherFn, userId: Int): Mono<List<Todo>> {
    return fetchTodos(userId).map { todos -> todos.filter { it.isCompleted } }
}

However, now Repositories.kt takes advantage of Spring Boot's dependency injection system to inject the base url:

@Component
class UserTodosRepository @Autowired constructor(
    @Qualifier(ExternalServices.TODOS_BASE_URL) private val baseUrl: String
) {
    fun getUserTodos(userId: Int): Mono<List<Todo>> {
        return WebClient.create(baseUrl).get()
            .uri { uriBuilder ->
                uriBuilder.path("/users/$userId/todos")
                    .build()
            }
            .retrieve()
            .bodyToMono(object : ParameterizedTypeReference<List<DataTodo>>() {})
            .map { dataTodos -> dataTodos.map { it.toDomain() } }
    }
}

And it's injected into the route handler, who only passes the UserTodosRepository::getUserTodos function to the domain function.

@Component
class TodosHandler @Autowired constructor(
    private val repository: UserTodosRepository
) {
    fun getTodos(request: ServerRequest): Mono<ServerResponse> {
        val userId = request.pathVariable("id").toIntOrNull()

        return if (userId != null) {
            val todosResponseMono =
                getCompletedTodos(repository::getUserTodos, userId)
                    .map { it.toResponse(userId) }
                    .onErrorResume { e -> internalServerError(e) }

            ServerResponse.ok().contentType(MediaType.APPLICATION_JSON)
                .body(todosResponseMono, GetTodosResponse::class.java)
        } else {
            ServerResponse.badRequest().syncBody("id is not an integer")
        }
    }
}

As we're using the DI framework, we no longer need to have the partial application boilerplate in the outer layers. Also, our data layer could just request the dependencies it requires, instead of having a context injected with the dependencies required across all the layers involved in a feature, like in the Reader monad case.

A purist could say that the functional core shouldn't know anything about a getUserTodos function, but should be fed the user todos directly. Our domain function would simply filter by isCompleted and do nothing else. Essentially, side-effects are pushed to the outer layers. This style is similar to the "Ports and Adapters" architecture and is well described by Mark Seemann here (warning: F# and Haskell ahead!). Fair enough, though you could also argue that we'd do that at the expense of having the business logic pushed to the outer layers just so we can have a pure functional core.

In any case, you can see that we can successfully mix OO and FP and use them for what they're best at. DI frameworks are convenient and as long as we keep the side-effects well encapsulated in our OO parts and the pure data processing parts in FP style, we can have the best of both worlds!

Conclusions

We've seen that using partial function application or the Reader monad we can have similar results to OO-style dependency injection. However, they may not be the best approaches as they seem more like an almost direct translation of the OO pattern into the functional world.

OO and FP are not mutually exclusive, we can use both of them where they work best. An OO-FP-OO sandwich in which the business core is functional and outer layers use OO may be a pragmatic approach forward.

In any case, each approach has its pros and cons and which one you choose will depend on your task at hand and your own preferences.

Partial function application

+ Functional approach (no classes)
+ Explicit, no magic
+ Simple to implement
- Verbose (in Kotlin + Arrow)
- Lots of manual setup in the composition root.

Reader monad

+ Functional approach (no classes)
+ Relatively explicit
- Single parameter passed down through all layers (if used as in the example we used)
- Manual setup in the composition root
- Different layer dependencies may be accessible to all functions in the chain

OO-FP-OO, hybrid approach

+ Concise code
+ Easy to specify dependencies
+ DI framework takes care of the composition boilerplate
- DI is magical, especially when you're not familiar with the framework
- Uses classes
- Classes may encourage stateful code
- Developers need to be strict in trying to keep the functions pure and making sure that classes are only used for DI and not to hold state

I personally find the hybrid approach a good pragmatic one in Kotlin, but be mindful about the cons, especially the last one.

This was a long post! I hope you enjoyed the read and see you next time!

Forem: Pin-Sho Feng

A story of building a custom Flow operator - bufferTimeout

1st attempt: Flow + Reactor

2nd try: imperative approach

Flow operator: bufferTimeout

Final remarks

Make impossible states impossible - Kotlin edition

Name

age

accountType

email, phoneNumber, address

So everything's validated... is that enough?

Conclusion

Bonus

Setting up Jenkins on MicroK8s

Prerequisites

Installing Jenkins on Kubernetes

Creating a Github pipeline

A few things to note from the Jenkinsfile

Other things of interest

Final words

Dependency injection in functional programming

Dependency injection in functional programming (Kotlin)

tl;dr

Why functional programming?

Dependency injection goals

Partial function application

How do you test the route handler (getTodos) in isolation?

If you have 5 dependencies, do you pass 5 arguments down?

The dependencies appear as arguments in the function signature of all functions in the call chain

Code navigation is more difficult as cmd + click doesn't work on partial function signatures

The Reader monad

How do you decide between the Reader monad and partial application?

Aren't we coupling all the functions in the Reader chain to the value that we're passing?

A critical view

OO-FP-OO, a hybrid approach

Conclusions

Partial function application

Reader monad

OO-FP-OO, hybrid approach

How do you test the route handler (`getTodos`) in isolation?