Pipes and Filters in Kotlin
Tags: kotlin, integration, patterns
Published: 2019-03-12 18:31:36

---
title: Pipes and Filters in Kotlin
published: true
description: Using kotlin's builtins to implement a traditional EIP
tags: kotlin, integration, patterns
---

> Disclaimer: coroutines are still a new topic to me, many of the implementation ideas presented in this post could not be the best usage of it. You have been warned :)


#Pipes and Filters

<img src="http://i.imgur.com/DzG5JRs.png">

[Pipes and Filters](https://www.enterpriseintegrationpatterns.com/patterns/messaging/PipesAndFilters.html) is a very common Integration Pattern. Entire frameworks such as the awesome [Spring Integration](https://spring.io/projects/spring-integration) have been built around it.

It's a simple concept, pipes connects endpoints much like you would use in a unix command `ps | grep`, this model allows endpoints to be independent of each other. 

If you look into how Spring Integration achieves this, is by using a [Channel abstraction](https://docs.spring.io/spring-integration/docs/5.1.2.RELEASE/reference/html/overview.html#overview-components-channel)

Well, it does so happens that kotlin coroutines also have a [Channel](https://kotlinlang.org/docs/reference/coroutines/channels.html) implementation used to allow communication across coroutines.

So I was wondering if we could use it to create a very simple pipes and filters flow without leveraging any other external dependency.

#The idea

* Write something that allows leveraging channels as pipes to connect coroutine functions.

* Leverage coroutine suspend features for concurrent execution, sometimes parallel withContext too :)

* Write a simple DSL(ish) like `source via fn via fn2 into fn4` 

It's not on the scope of this blog post to attempt to replicate all the features of a complete integration framework. No support for fanout via a publisher channel for example. No error handling or ack support.
 
#How

So before diving into some code, I must confess that so far I have been always using `GlobalScope.launch` as the only way to launch my coroutines. And then I read the following blog posts below. If anything useful could come out of my blog today is those links for you:

* [Explicit Concurrency](https://medium.com/@elizarov/explicit-concurrency-67a8e8fd9b25)
* [Coroutines Context and Scope](https://medium.com/@elizarov/coroutine-context-and-scope-c8b255d59055)
* [Avoid using Global Context](https://medium.com/@elizarov/the-reason-to-avoid-globalscope-835337445abc)

We are starting with an assumption that the source will always be a `Channel<T>`, this is just to simplify things.

We have two pipes [via || into] and two top level endpoints: `ChannelSink<T>, ChannelProcessor<T, R>`

Let's look into our Sink first

```kotlin
class ChannelSink<T>(private val channel: Channel<T>, block:  (T) -> Unit) : CoroutineScope {
    private val job: Job = Job()
    override val coroutineContext: CoroutineContext
        get() = Dispatchers.Default + job
    init {
        launch {
            for(message in channel){
                block(message)
            }
        }
    }
}
```

Not too much going on here, our Sink reads messages from a channel (in a non blocking way using the `launch` builder). And then invokes the `block` function that would process the sink (invoke a rest endpoint, call a database, etc)

Our processor is a map style of function `(T) -> (R)`
```kotlin
class ChannelProcessor<T, R>(private val inputChannel: Channel<T>, block: (T) -> (R)) : CoroutineScope {
    private val job = Job()
    override val coroutineContext: CoroutineContext
        get() = Dispatchers.Default + job
    val outputChannel = Channel<R>()
    init {
        launch {
            for(message in inputChannel){
                outputChannel.send(block(message))
            }
        }
    }
}

```
Very similar to our Sink, but now we create an `outputChannel` that we write transformed messages using `block(message)`

And that's all the types we need. Now we need some plumbing code around our pipes [No pun intended]

##Plumbing code (literally)

Ok so our "DSL" should be connecting the functions using channels as Pipes. So at the very beginning we said we want our `sources` to be `Channels`. So let's write two possible scenarios:

Source -> Processor
Source -> Sink

```kotlin
//Source -> Sink
infix fun <T> Channel<T>.into(block:  (T) -> Unit) : ChannelSink<T> {
    return ChannelSink(this, block)
}
//Source -> Processor
infix fun <T, R> Channel<T>.via(block: (T) -> R) : ChannelProcessor<T, R> {
    return ChannelProcessor(this, block)
}
```

our `via/into` functions pass the function receiver here (Channel) to the constructor of the ChannelSink/ChannelProcessor.

To daisy chain our processors now we need two extra functions:

```kotlin
infix fun <T, R> ChannelProcessor<T, R>.into(block: (R) -> Unit) : ChannelSink<R> {
    return ChannelSink(this.outputChannel, block)
}
infix fun <T, R, S> ChannelProcessor<T, R>.via( block: (R) -> S) : ChannelProcessor<R, S> {
    return ChannelProcessor(this.outputChannel, block)
}

```
Note that we daisy chain the channels by getting the outputChannel of a Processor and passing it as the inputChannel of the next Processor or the Sink.

#Running it

First let's define some functions

```kotlin
//sink function
val fn: (Double) -> Unit = { i ->
   log("Received $i")
}
//first processor
val tn: (Int) -> (Int) = { i ->
       i * 2
}
//second processor
val tn2: (Int) -> Double = { i -> i * 3.0 } 

```

Now let's get our producer channel ready

```kotlin
val source = Channel<Int>()
launch {
    with(source) {
        for (i in 0..size) {
            send(i)
        }
        close()
    }
}
```

And finally connect the pipes:

```kotlin
source via tn via tn2 into fn
```

And that's pretty much it, we can now use our channels with a pipes and filters syntax. 

# WAIT A MINUTE
<img src="https://media2.giphy.com/media/uLMxqxVvVtuVO/giphy.gif?cid=3640f6095c87f70e464e447236be533e">

Kotlin has support for Collection Streaming, why not just create a collection or a Sequence and use `map` and `collect` functions?

Looking at this code, you are absolutely right to spot that, and I'm truly sorry if I mad you read all that so far, but I wanted to make a point :)

Say we change our Processor and Sink code a little bit:

```kotlin
class ChannelProcessor<T, R>(private val inputChannel: Channel<T>, block: (T) -> (R)) : CoroutineScope {
    private val job = Job()
    override val coroutineContext: CoroutineContext
        get() = Dispatchers.Default + job
    val outputChannel = Channel<R>()
    init {
        launch {
            for(message in inputChannel){
                launch {  outputChannel.send(block(message)) }
            }
        }
    }
}

class ChannelSink<T>(private val channel: Channel<T>, block:  (T) -> Unit) : CoroutineScope {
    private val job: Job = Job()
    override val coroutineContext: CoroutineContext
        get() = Dispatchers.Default + job
    init {
        launch {
            for(message in channel){
                launch {
                    block(message)
                }
            }
        }
    }
}

```

Note that now, we `launch` a child coroutine for every function call, what this enables is that we could run all the steps in parallel. 

Kotlin currently does not offer a standard way to run streams in parallel (unless you rely on java version). With this last change, the code runs concurrently. *note this means sequencing of messages is affected*.

To test this just try the updated code with this new scenario:

```kotlin
@ExperimentalCoroutinesApi
    @Test
    fun testAsyncProduce() {
        val size = 100
        val latch = CountDownLatch(size)
        runBlocking<Unit> {

            val source = Channel<Int>()
            launch {
                with(source) {
                    for (i in 0..size) {
                        send(i)
                    }
                    close()
                }
            }
            val fn: (Double) -> Unit = { i ->
                log("Received $i")
                latch.countDown()
            }
            val tn: (Int) -> (Int) = { i ->
                runBlocking {
                    if (i % 2 == 0) {
                        delay(10)
                    }
                    i * 2
                }
            }
            val tn2: (Int) -> Double = { i -> i * 3.0 }

            source via tn via tn2 into fn
        }
        latch.await()
    }
```

In this test, one of our functions will cause a delay on even numbers, but that will not cause a sequencial pause on the execution.

Those were just some thoughts I had while writing a code this week that had to deal with a simple pipeline and a sink to invoke a REST service. Put some of this in practice, but it's not to be taken as too serious.

Happy coding