17. Sequence & Flow

Written by Massimo Carli

In Chapter 16, “Handling Side Effects”, you learned how to use the IO<T> monad as a special case of the State<S, T> data type. You also learned that Kotlin provides coroutines to handle side effects as pure functions in a more idiomatic way. In this chapter, you’ll learn everything you need to know about the following special Kotlin data types:

• Sequence<T>
• Flow<T>

You’ll also have a quick overview of SharedFlow<T> and StateFlow<T>.

You’ll learn how these data types work from a functional programming point of view. In particular, you’ll answer the following questions for each of these:

• What is the context they provide?
• Are they a functor?
• Are they an applicative functor?
• Are they a monad?

Note: If you don’t know coroutines yet, Kotlin Coroutines by Tutorials is the book for you.

This chapter is a big exercise that helps you take the concepts you’ve learned so far and apply them to the types you use every day in your job. It’s time to have fun!

The Sequence<T> data type

In Chapter 9, “Data Types”, you learned that List<T> is a data type with the functor and monad superpowers with the map and flatMap functions. In Chapter 4, “Expression Evaluation, Laziness & More About Functions”, you also learned that laziness is one of the main characteristics of functional programming. To remind you what this means, open ListDataType.kt in this chapter’s material and write the following code:

fun main() {
  listOf(1, 2, 3, 4, 5) // 1
    .filter(filterOdd.logged("filterOdd")) // 2
    .map(double.logged("double")) // 3
}

In this code, you:

1. Use listOf as a builder for a List<Int> with five elements of type Int.
2. Invoke filter, passing the reference to the logged version of filterOdd.
3. Use map to transform the filter’s values using a logged version of double.

Note: filterOdd and double are two very simple functions you find in Util.kt in the lib sub-package. logged is a utility higher-order function that decorates another function with a log message. Take a look at their simple implementation, if you want.

The interesting fact about the previous code happens when you run it, getting the following output:

filterOdd(1) = false
filterOdd(2) = true
filterOdd(3) = false
filterOdd(4) = true
filterOdd(5) = false
double(2) = 4
double(4) = 8

This happens because, in each line, you:

1. Create a List<Int> with five elements.
2. Invoke filter, which returns another List<Int> containing only the even values. It’s crucial to see that filterOdd has been invoked for all the elements of the original List<Int>.
3. Use map, getting a new List<Int> with the double of the values in the previous List<Int>.

With this code, you basically created three lists without using any of the individual lists’ values. What happens if you don’t really need the values in the List<Int>? In this case, you started with a List<Int> of five elements. What if the list has a lot more elements? What if the elements in the List<T> are infinite?

Well, you don’t have to blame the List<T> data type because its job is to contain an ordered collection of elements of type T. That’s why it’s been created that way. That’s its context, or purpose, if you will. Another way to say it is that List<T> is eager.

If you don’t want to keep all the possible values in a List<T>, Kotlin provides the Sequence<T> data type.

Open SequenceDataType.kt and write the following code:

fun main() {
  sequenceOf(1, 2, 3, 4, 5) // HERE
    .filter(filterOdd.logged("filterOddSeq"))
    .map(double.logged("doubleSeq"))
}

This code differs from the previous one because of the use of sequenceOf instead of listOf. More importantly, if you run the code, you’ll get nothing as output. This is because Sequence<T> is lazy. If you want to actually consume the values in the Sequence<Int> you just created, you need to consume them using a terminal operator. To see how, add .count() to the end of the method chain. It should now look like this:

fun main() {
  sequenceOf(1, 2, 3, 4, 5)
    .filter(filterOdd.logged("filterOddSeq"))
    .map(double.logged("doubleSeq"))
    .count() // HERE
}

Here, you’re just counting the elements in the sequence and, to do it, you need to consume all of them. This time, running the code, you’ll get the following:

filterOddSeq(1) = false
filterOddSeq(2) = true
doubleSeq(2) = 4
filterOddSeq(3) = false
filterOddSeq(4) = true
doubleSeq(4) = 8
filterOddSeq(5) = false

Note how the order of the log messages is different from the one you got from the List<T>. In that case, each operator read the values from the input List<T>. Now, the chain of operators is called for each value you consume.

Note: If you’re curious and want to look at the definition of Sequence<T>, you’ll find that it differs from the Iterable<T> interface in the use of the operator keyword, which allows its use in an enhanced form.

This clarifies the context for a Sequence<T> as a container that produces the values it contains only when required. That means it’s lazy. But is Sequence<T> a functor?

Sequence<T> as a functor

Looking at the Sequence<T> documentation, you see the definition of map with the following signature:

public fun <T, R> Sequence<T>.map(transform: (T) -> R): Sequence<R>

You used map in the example in the previous section. In this case, you want to do something more and use property-based testing to prove the functor laws for Sequence<T>.

Note: You already used property-based testing in Chapter 12, “Monoids & Semigroups”.

You want to prove that, given the two functions f and g and the identity i:

• map(i) == i
• map(f compose g) == map(f) compose map(g)

If you want to use property-based testing, you should define a way to generate random functions using a specific implementation of the following interface you find in PropertyTest.kt in the lib sub-package in this chapter’s material:

fun interface Generator<T> {
  fun generate(n: Int): List<T>
}

As the first step, add the following to the PropertyTestFun.kt file:

fun <T, R> Generator<T>.map(fn: (T) -> R): Generator<R> = object : Generator<R> {
  override fun generate(n: Int): List<R> = this@map.generate(n).map(fn)
}

This describes the map function for a Generator<T>. It’s just a simple way to create a Generator<R> from a Generator<T> using a function of type (T) -> R or, using the typealiases in Definitions.kt, Fun<T, R>.

Note: Be careful that Generator<T> isn’t a functor because it’s not even pure: It generates random values.

Now, you can get a Generator<Fun<A, B>> from a Generator<A> and Generator<B> with the following code you can add to the same PropertyTestFun.kt:

fun <A, B> funGenerator(bGen: Generator<B>): Generator<Fun<A, B>> =
  bGen.map { b: B -> { b } }

Every Fun<A, B> describes a way to map values of type A into values of type B. The Fun<A, B> you’ll get from funGenerator is a function that maps the same random value of type B to any values of type A you’ll pass as input to the generated function. Because that return value is the same for all the input values, you can assume that the same would happen for a specific value you might generate during testing.

Open SequenceDataTypeTest.kt in the test build type, and add the following code:

  @Test
  fun `Identity Functor Law for Sequences`() {
    val intToStringFunGenerator =
      funGenerator<Int, String>(StringGenerator(5)) // 1
    val i = { s: String -> s } // 2    
    100.times {  // 3
      val f = intToStringFunGenerator.one() // 4
      val seq = IntGenerator.generate(5).asSequence() // 5
      val list1 = seq.map(f compose i).toList() // 6
      val list2 = seq.map(f).toList() // 7
      Truth.assertThat(list1).isEqualTo(list2) // 8
    }
  }

Here, you:

1. Use funGenerator to generate a random function of type Fun<Int, String>. This function maps Ints to Strings of length 5.
2. Provide the identity function i.
3. Iterate 100 times over the following commands.
4. Get a Fun<Int, String> from intToStringFunGenerator and store it in f.
5. Generate a sequence of 5 random elements in seq.
6. Apply the composition between f and i to seq using map.
7. In the same way, you apply only f.
8. Verify that the results are the same.

Run the test, and check that everything is successful. Now, you can be confident that the first law of Sequence<T> as a functor is valid.

The second law is very simple. Just add this code to the same file:

  @Test
  fun `Composition Functor Law for Sequences`() {
    val intToStringFunGenerator =
      funGenerator<Int, String>(StringGenerator(5))
    val stringToLongFunGenerator =
      funGenerator<String, Long>(LongGenerator) // 1
    100.times {
      val f = intToStringFunGenerator.one()
      val g = stringToLongFunGenerator.one() // 2
      val seq = IntGenerator.generate(5).asSequence()
      val list1 = seq.map(f compose g).toList() // 3
      val list2 = seq.map(f).map(g).toList() // 4
      Truth.assertThat(list1).isEqualTo(list2) // 5
    }
  }

The only difference here is that you:

1. Create a new Generator<Fun<String, Long>> in stringToLongFunGenerator.
2. Use intToStringFunGenerator and stringToLongFunGenerator to generate two different functions: f and g of type Fun<Int, String> and Fun<String, Long>, respectively.
3. Invoke map, passing the composition of f and g as a parameter.
4. Invoke map, first with f and then with g.
5. Compare the results of the two cases.

Now, just run the test and see that all the tests are successful. Great job!

Sequence<T> as an applicative functor

You just proved that the Sequence<T> data type is a functor because of the existing implementation of map. But what about applicative functors? Looking at the Kotlin documentation, you don’t see any higher-order functions like your ap and app. No problem — you can do this!

Open SequenceDataType.kt, and add the following code:

fun <A, B> Sequence<A>.ap(
  fn: Sequence<(A) -> B>
): Sequence<B> = TODO()

This is the signature for the ap function for a Sequence<T>. It’s an extension function on the type Sequence<A> and accepts an input parameter of type Sequence<(A) -> B> or Sequence<Fun<A, B>> if you use the type alias. The return type is Sequence<B>. Basically, you have two sequences. The first generates values of type A, and the second generates functions of type Fun<A, B>. The result, then, is a Sequence<B> of the value you get by applying the function in Sequence<Fun<A, B>> to the values in Sequence<A>. How would you implement ap?

Note: Feel free to provide your implementation as a fun exercise if you want.

Now, replace the previous code with the following:

fun <A, B> Sequence<A>.ap(fn: Sequence<(A) -> B>): Sequence<B> =
  sequence { // 1
    val iterator = iterator() // 2
    while (iterator.hasNext()) {
      val fnIterator = fn.iterator() // 3
      val item = iterator.next()
      while (fnIterator.hasNext()) {
        yield(fnIterator.next().invoke(item)) // 4
      }
    }
  }

In this code, you:

1. Generate a Sequence<B> using the sequence builder.
2. Get the reference to the Iterator<A> from the Sequence<A> and iterate over it.
3. Get the reference to the Iterator<Fun<A, B>> from the Sequence<Fun<A, B>> you get as an input parameter and iterate over them.
4. Use yield to produce the value you get by applying the current function Fun<A, B> to the current value A.

To see how it works, start by adding the following utility function. It allows you to use ap as an infix operator, as you did with the other applicative functor implementations:

infix fun <A, B> Sequence<(A) -> B>.appl(a: Sequence<A>) = a.ap(this)

To see it working, replace the main implementation in SequenceDataType.kt with the following:

fun main() {
  data class User(  // 1
    val id: Int,
    val name: String,
    val email: String
  )

  val userBuilder = ::User.curry() // 2
  val userBuilderSeq = sequenceOf(userBuilder) // 3
  val idSeq = sequenceOf(10, 20, 30) // 4
  val nameSeq = sequenceOf("Minnie", "Donald", "Mickey") // 4
  val emailSeq =
    sequenceOf("aaaaaa@aaaaa.com", "bbbbb@bbbbbb.com") // 4

  val userSeq =
    userBuilderSeq appl idSeq appl nameSeq appl emailSeq // 5

  userSeq.forEach(::println) // 6
}

In this example, you:

1. Create a User data class representing a user with id, name and email properties.
2. Use curry to get the User constructor as a function of type (Int) -> (String) -> (String) -> User.
3. Create a Sequence<(Int) -> (String) -> (String) -> User> using the sequenceOf builder.
4. Use sequenceOf for creating a Sequence<Int> for the ids, Sequence<String> for names and Sequence<String> for the emails.
5. Use appl to create a Sequence<User>.
6. Print all the values in the Sequence<User>.

When you run this code, you’ll get 3 * 3 * 2 = 18 different values, like the following:

User(id=10, name=Minnie, email=aaaaaa@aaaaa.com)
User(id=20, name=Minnie, email=aaaaaa@aaaaa.com)
// ...
User(id=20, name=Mickey, email=bbbbb@bbbbbb.com)
User(id=30, name=Mickey, email=bbbbb@bbbbbb.com)

Considering that the context of a Sequence<T> is to provide values of type T in a lazy way, applying a function with multiple parameters leads to a number of values, like in the previous example.

Sequence<T> as a monad

Is Sequence<T> finally a monad? Of course it is, because of the flatMap operation that Kotlin APIs provide with the following signature, similar to the Kleisli category:

fun <T, R> Sequence<T>.flatMap(
  transform: (T) -> Sequence<R>
): Sequence<R>

You can test how this works by running the following example:

fun main() {
  // ...
  val seqTo = { n: Int -> (1..n).toList().asSequence() }
  val seqOfSeq = sequenceOf(1, 2, 3, 4, 5).flatMap(seqTo)
  seqOfSeq.forEach { print("$it ") }
}

Getting an output like the following:

1 1 2 1 2 3 1 2 3 4 1 2 3 4 5

The Flow<T> data type

In Chapter 16, “Handling Side Effects”, you learned that Kotlin allows you to achieve with suspendable functions what you can do with the IO<T> monad. In this chapter, you’ve already learned how to produce a theoretically infinite sequence of values in a lazy way. If the values you want to generate are the result of a suspendable function, the Flow<T> data type is what you need.

You can then say that the context of the Flow<T> data type is the generation of a sequence of values you create using a suspendable function which, as you know, allows you to handle side effects in a pure fashion.

As you’ll see, a Flow<T> is very similar to a Sequence<T> in terms of functional programming concepts. So, the following sections are basically a good review of things you’ve already learned: repetita juvant, as the Romans used to say! :]

Flow<T> as a functor

To prove that Flow<T> is a functor, you could repeat the same process you did for Sequence<T> using property-based testing. In this case, you’ll keep things easier, implementing some practical examples.

Note: As an interesting exercise, you could use property-based testing for Flow<T> as well.

Open FlowDataType.kt and add the following code:

fun inputStringFlow(question: String = "") = flow { // 1
    val scanner = java.util.Scanner(System.`in`) // 2
    print(question) // 3
    while (scanner.hasNextLine()) { // 4
        val line = scanner.nextLine() // 4
        if (line.isNullOrEmpty()) { // 5
            break
        }
        emit(line) // 6
        print(question) // 3
    }
    scanner.close() // 7
}

In this code, you:

1. Define inputStringFlow as a function that returns a Flow<String> of the text you write as input using a Scanner reading from the standard input. This is a flow version of the effect you used in Chapter 16, “Handling Side Effects”. You have a question parameter that allows you to print some text before the user enters anything.
2. Initialize the Scanner reading from the standard input.
3. Print the question.
4. Read all the input one line at a time.
5. Exit the cycle if the user enters an empty line.
6. Emit the value from the user as a value from the Flow<String>.
7. Close the Scanner.

As an example of a functor, add the following code to the same file:

fun main() {
  val strLengthFlow = inputStringFlow("Insert a word: ") // 1
    .map { str -> // 2
      str to str.length
    }
  runBlocking { // 3
    strLengthFlow.collect { strInfo -> // 4
      println("${strInfo.first} has length ${strInfo.second}")
    }
  }
}

In this code, you:

1. Use inputStringFlow to get a Flow<String> for the user input. Note that you run this outside any specific CoroutineScope. You’re not executing anything — you’re just stating you eventually might.
2. Invoke map, passing a lambda that returns a Pair<String, Int> of the input String and its length. It’s important to note that the lambda here is executed as a suspendable block. This means that it has a scope, and it can contain invocations to other suspendable functions. In other words, using map(String::length) would give a compilation error because String::length isn’t a suspendable function. In your context, this also means you can apply a transformation, Fun<A, B>, to the values you get from a Flow<A, B>, which is the consequence of a side effect.
3. Define a runBlocking block. This is because you’ll consume, or better, collect what the Flow<Pair<String, Int>> produces, so the side effects you described will actually run.
4. Collect and print the output values of type Pair<String, Int> you get from the flow.

Now, you can run the code and get an output like the following:

To answer the initial question, yes, Flow<T> is a functor, but remember that the transformation Fun<A, B> must be a suspendable function.

Flow<T> as an applicative functor

To see if the Flow<T> also behaves as an applicative functor, either repeat what you did for the Sequence<T> or just follow along. In FlowDataType.kt, add the following code:

fun <A, B> Flow<A>.ap(fn: Flow<(A) -> B>): Flow<B> = flow { // 1
  collect { a -> // 2
    fn.collect { f -> // 3
      emit(f(a)) // 4
    }
  }
}

infix fun <A, B> Flow<(A) -> B>.appl(
  a: Flow<A>
) = a.ap(this) // 5

Here, you:

1. Define ap with the usual signature as an extension function for Flow<A>.
2. Collect the value of type A from the Flow<A> you have as the receiver.
3. Collect the function f of type Fun<A, B> from the Flow<Fun<A, B>> or Flow<(A) -> B> you pass as the input parameter fn.
4. Apply the function f to the value a and emit the result.
5. As usual, define appl as an infix version of ap.

You can then test everything by adding the following code to the same file:

fun main() {
  val userBuilder = { id: Int ->
    { name: String ->
      { email: String -> User(id, name, email) }
    }
  }

  val userBuilderFlow = flowOf(userBuilder)
  val idFlow = listOf(10, 20, 30).asFlow()
  val nameFlow = listOf("Pippo", "Pippo2", "Pippo3").asFlow()
  val emailFlow = listOf(
    "pippo@pippo.com", "pippo2@pippo.com", "pippo3@pippo.com"
  ).asFlow()

  val userFlow =
    userBuilderFlow appl idFlow appl nameFlow appl emailFlow
  runBlocking {
    userFlow.collect(::println)
  }
}

This code shouldn’t be a surprise anymore. When you run it, you’ll get:

User(id=10, name=Pippo, email=pippo@pippo.com)
User(id=20, name=Pippo, email=pippo@pippo.com)
// ...
User(id=20, name=Pippo3, email=pippo3@pippo.com)
User(id=30, name=Pippo3, email=pippo3@pippo.com)

The same notes you learned for Sequence<T> are valid here.

Flow<T> as a monad

To answer the last question, you’ll implement a more complex example using some of the code you already implemented in Chapter 14, “Error Handling With Functional Programming”, that you can find in the material for this project. You basically want to use inputStringFlow to allow a user to insert some text to search for in the TV show database using the TVmaze API. Now, you’re in the world of coroutines, so you should use their power. It’s time for an interesting exercise to improve your functional thinking.

Imagine you have a basic function, like the following you can write in Basic.kt:

fun doSomeWork(name: String): Int = 10

It doesn’t really matter what doSomeWork does. What’s important is the type, which is (String) -> Int, and how the function achieves its goal. If it needs to do some hard work, you probably want to run it in the background, so in the context of a coroutine. You can do this with the following code, which you should add to the same file:

suspend fun doSomeBgWork(
  ctx: CoroutineContext,
  name: String
): Int = withContext(ctx) {
    doSomeWork(name)
}

This simple code has a few interesting things to note. doSomeBgWork:

• Is a suspend function.
• Accepts two parameters now. The first is of type CoroutineContext, and the second is the input for doSomeWork.
• Uses withContext to run doSomeWork in the CoroutineContext you provide as input.
• Has the return type Int.

You already know that in functional programming, you don’t like functions with multiple parameters. No problem — you also know you can curry them, but with doSomeBgWork, there’s a problem. You can see the problem by adding the following code:

fun main() {
  doSomeBgWork.curry()
}

You’ll get the following error:

The reason is very simple. When you set a function as suspend, you’re basically changing its type. The Kotlin compiler adds implicit parameters of type Continuation that keep track of the state of the coroutine.

Note: Look at the decompiled code, and you can see how the Continuation is used in a way that reminds you how you implemented the State<S, T> and IO<T> data types.

How can you then implement curry for a suspendable function? You already know that. In the same Basic.kt file, add the following definitions:

typealias SuspendFun<A, B> = suspend (A) -> B // 1
typealias SuspendFun2<A, B, C> = suspend (A, B) -> C // 2
typealias SuspendChain2<A, B, C> =
  suspend (A) -> suspend (B) -> C // 3

In this simple code, you define an alias for suspendable functions of:

1. One input parameter of type A and one output parameter of type B.
2. Two input parameters of type A and B and one output parameter of type C.
3. One input parameter of type A and a suspendable function of one input parameter of type B and output of type C.

These allow you to define curry for suspendable functions by adding the following code:

fun <A, B, C> SuspendFun2<A, B, C>.curry(): SuspendChain2<A, B, C> =
  { a: A ->
    { b: B ->
      this(a, b)
    }
  }

Now, the previous code compiles successfully:

Implementing curry for suspendable functions is a kind of a warm-up. Going back to doSomeBgWork, you can say that the type of ::doSomeBgWork.curry() is now (String) -> (CoroutineContext) -> Int. Umm, that’s starting to ring a bell.

What really matters in functional programming is composition. What you did earlier is say that you can create a suspendable function from a non-suspendable one, providing a CoroutineContext, and this carrying on reminds you of something you already learned: the State<S, T> monad.

CoroutineContext as a state

To relate the State<S, T> monad to what you saw about suspendable functions, look again at doSomeBgWork, which you wrote in Basic.kt:

suspend fun doSomeBgWork(ctx: CoroutineContext, name: String): Int =
  withContext(ctx) {
    doSomeWork(name)
  }

It has the type (CoroutineContext, String) -> Int. If the CoroutineContext is something you want to carry on, you can create doSomeMoreBgWork like the following:

suspend fun doSomeMoreBgWork(
  ctx: CoroutineContext,
  name: String
): Pair<CoroutineContext, Int> = withContext(ctx) {
  ctx to doSomeWork(name)
}

doSomeMoreBgWork has the type (CoroutineContext, String) -> Pair<CoroutineContext, Int>. Applying curry, you can get a function of type (String) -> (CoroutineContext) -> Pair<CoroutineContex, Int>.

Note: Yes, the CoroutineContext and String types have different orders, but you already know how to implement flip, right?

Now, the analogy with State<S, T> is almost obvious. You just need to fix a problem caused by the presence of the suspend modifier. No problem.

Open SuspendState.kt and add the following first definition:

typealias SuspendStateTransformer<S, T> =
  suspend (S) -> Pair<S, T>

Ring a bell now? Using this definition, the type of doSomeMoreBgWork is suspend (String) -> SuspendStateTransformer<CoroutineContext, Int>.

You can now follow the same process you did for State<S, T>, keeping in mind you’re working with suspendable functions, and the state is a CoroutineContext.

In SuspendState.kt, add the following code:

data class SuspendableState<S, T>(
  val sst: SuspendStateTransformer<S, T>
) {

  companion object {
    @JvmStatic
    fun <S, T> lift(
      value: T
    ): SuspendableState<S, T> =
      SuspendableState { state -> state to value }
  }
}

Here, you define the SuspendableState data type using SuspendStateTransformer.

You can now add the implementation for map:

fun <S, A, B> SuspendableState<S, A>.map(
  fn: SuspendFun<A, B>
): SuspendableState<S, B> =
  SuspendableState { s0: S ->
    val (s1, a) = this.sst(s0)
    s1 to fn(a)
  }

Finally, add the implementation for flatMap like this:

fun <S, A, B> SuspendableState<S, A>.flatMap(
  fn: suspend (A) -> SuspendableState<S, B>
): SuspendableState<S, B> =
  SuspendableState { s0: S ->
    val (s1, a) = this.sst(s0)
    fn(a).sst(s1)
  }

Note: This time, you can’t override the invoke operator as you did in a non-coroutine environment. To be useful, it should also be suspendable, and this wouldn’t work.

How can you use this for your initial TV show problem? There’s actually quite a bit more fun in store for you. :]

Back to the TV show

In the previous section, you created the SuspendableState<S, T> data type and implemented lift, map and flatMap. How can you use these for getting data about a TV show? In the tools sub-package in this chapter’s material, you find TvShowFetcher and TvShowParser for, respectively, fetching and parsing data using the TVmaze API.

Look at the existing code, and you’ll see that TvShowFetcher and TvShowParser don’t actually handle exceptions. This is also why you used those objects in many different ways in Chapter 14, “Error Handling With Functional Programming”.

Now, you want to run them as side effects in a suspendable function and handle errors. How can you do that?

Open ShowSearchService.kt and add the following code:

suspend fun fetchTvShowResult( // 1
  ctx: CoroutineContext,
  query: String
): Result<String> = // 2
  withContext(ctx) { // 3
    try {
      Result.success(TvShowFetcher.fetch(query)) // 4
    } catch (ioe: IOException) {
      Result.failure(ioe) // 5
    }
  }

This code should look familiar, even if it combines a few concepts. Here, you:

1. Define fetchTvShowResult as a suspendable function with a CoroutineContext as the first parameter and a String as the second. Note how the structure of this function is very similar to the one of doSomeMoreBgWork.
2. Set Result<String> as the return type. This is a little bit more complicated than a simple String. You’ll need to do some more work because of this, as you’ll see later.
3. Use the CoroutineContext you receive in ctx to create a coroutine.
4. Invoke TvShowFetcher.fetch, passing the query as input. In the case of success, you return the String wrapped in a Result<String>.
5. In the case of error, you encapsulate the IOException in a Result<String>. Yes, the value for the type parameter is String.

For TvShowParser.parse, you can follow the same pattern, adding this to the same file:

suspend fun parseTvShowResult(
  ctx: CoroutineContext,
  json: String
): Result<List<ScoredShow>> =
  withContext(ctx) {
    try {
      Result.success(TvShowParser.parse(json))
    } catch (e: Exception) {
      Result.failure(e)
    }
  }

This time, the resulting type is Result<List<ScoredShow>>, but the structure is the same.

So far, so good. Now, you have two functions:

1. fetchTvShowResult of type suspend (CoroutineContext, String) -> Result<String>.
2. parseTvShowResult of type suspend (CoroutineContext, String) -> Result<List<ScoredShow>>.

They look like the functions of type suspend (A) -> SuspendStateTransformer<CoroutineContext, B> you can compose using the flatMap implementation you created earlier for SuspendableState.

You can try to solve this problem by adding the following code:

val fetchSuspend: (String) -> SuspendableState<
  CoroutineContext, Result<String>> = { query ->
    SuspendableState { ctx: CoroutineContext ->
      ctx to fetchTvShowResult(ctx, query)
    }
  }

val parseSuspend: (String) -> SuspendableState<
  CoroutineContext, Result<List<ScoredShow>>> = { json ->
    SuspendableState { ctx: CoroutineContext ->
      ctx to parseTvShowResult(ctx, json)
    }
  }

Now:

1. fetchSuspend has type (String) -> SuspendableState<CoroutineContext, Result<String>>.
2. parseSuspend has type (String) -> SuspendableState<CoroutineContext, Result<List<ScoredShow>>>.

This is a problem because, in both cases, you have a SuspendableState<CoroutineContext, Result<T>>. This means a Result<T> data type encapsulates into a SuspendableState<S, T> data type. Composition, as defined in flatMap for SuspendableState, doesn’t work. How can you fix it?

Composing SuspendableState<CoroutineContext, Result<T>>

To implement composition now is simpler than it seems. Open SuspendableStateResult.kt, and add the following code:

typealias SuspendStateResultTransformer<S, T> =
  suspend (S) -> Pair<S, Result<T>> // 1

data class SuspendableStateResult<S, T>( // 2
  val sst: SuspendStateResultTransformer<S, T>
) {

  companion object {
    @JvmStatic
    fun <S, T> lift( // 3
      value: T
    ): SuspendableStateResult<S, T> =
      SuspendableStateResult { state ->
        state to Result.success(value)
      }
  }
}

fun <S, A, B> SuspendableStateResult<S, A>.map( // 4
  fn: SuspendFun<A, B>
): SuspendableStateResult<S, B> =
  SuspendableStateResult { s0: S ->
    val (s1, a) = this.sst(s0)
    s1 to a.fold(
      onSuccess = { Result.success(fn(it)) },
      onFailure = { Result.failure(it) }
    )
  }

fun <S, A, B> SuspendableStateResult<S, A>.flatMap( // 5
  fn: suspend (A) -> SuspendableStateResult<S, B>
): SuspendableStateResult<S, B> = SuspendableStateResult { s0 ->
  val (s1, res) = sst(s0)
  res.fold(onSuccess = { a: A ->
    fn(a).sst(s1)
  }, onFailure = { thowable ->
    s1 to Result.failure(thowable)
  })
}

It’s a lot of code, but everything should be clear. In particular, you define:

1. SuspendStateResultTransformer<S, T> as a type of suspendable functions returning a Pair<S, Result<T>>.
2. SuspendableStateResult<S, T> as a data type encapsulating a SuspendStateResultTransformer<S, T>.
3. lift to create a SuspendableStateResult<S, T> from a value of type T.
4. map to apply a function of type SuspendFun<A, B> to a SuspendableStateResult<S, A> to get a SuspendableStateResult<S, B>.
5. flatMap to finally be able to compose functions returning SuspendableStateResult<S, T>.

Now, you’re finally ready to access your TV show information.

Finally flatMap

It’s finally time to put everything together so you can access the TVmaze database. Open ShowSearchService.kt, and add the following code:

val fetchSuspendResult: (String) -> SuspendableStateResult<
  CoroutineContext, String> = { query ->
    SuspendableStateResult { ctx: CoroutineContext ->
      ctx to fetchTvShowResult(ctx, query)
    }
  }

val parseSuspendResult: (String) -> SuspendableStateResult<
  CoroutineContext, List<ScoredShow>> = { json ->
    SuspendableStateResult { ctx: CoroutineContext ->
      ctx to parseTvShowResult(ctx, json)
    }
  }

Here, you create:

1. fetchSuspendResult of type (String) -> SuspendableStateResult<CoroutineContext, String>.
2. parseSuspendResult of type (String) -> SuspendableStateResult<CoroutineContext, List<ScoredShow>>.

You can now compose these functions. Just add the following code to the same file:

@OptIn(FlowPreview::class) // 1
suspend fun searchTvShow(ctx: CoroutineContext) = // 2
  withContext(ctx) {
    inputStringFlow("Search Your Show: ") // 3
      .flatMapConcat { query ->  // 4
        fetchSuspendResult(query)
          .flatMap(parseSuspendResult).sst(ctx) // 5
          .second.fold(
            onSuccess = { it.asFlow() }, // 6
            onFailure = { emptyFlow() }) // 7
      }
}

In this code, you:

1. Opt-in to the experimental Flow<T> API.
2. Define searchTvShow as a function accepting a CoroutineContext in input.
3. Invoke inputStringFlow, passing a String to use as a message.
4. Use the predefined flatMapConcat to compose inputStringFlow with the suspendable function you get by composing fetchSuspendResult and parseSuspendResult.
5. Pass the CoroutineContext you get as input to the function you get from the composition of fetchSuspendResult and parseSuspendResult.
6. Return the value in Result<List<ScoredShow>> as a Flow<ScoredShow> in case of success.
7. Return an empty Flow<ScoredShow> in case of error.

To test this, add the following code:

@OptIn(FlowPreview::class)
fun main() {
  runBlocking { // 1
    searchTvShow(Dispatchers.IO) // 2
      .collect { // 3
        println("Score: ${it.score}  " +
          "Name: ${it.show.name} " +
          "Genres: ${it.show.genres}") // 4
        println(it.show.summary)
        println("--------------------------")
      }
  }
}

Here, you:

1. Use runBlocking to give some scope to searchTvShow.
2. Invoke searchTvShow, passing Dispatchers.IO as CoroutineContext. This allows your code to run in the background.
3. Collect all the results.
4. Print its content, if any.

Now, you can run the previous code and have some fun, like this:

Everything works as expected, and this has been a great exercise to understand how to:

• Define the right abstraction.
• Reuse what you learned in the previous chapters of the book.
• Implement composition for the previous abstractions.
• Think in a functional way.

Great job!

The SharedFlow<T> & StateFlow<T> data types

SharedFlow<T> and StateFlow<T> are two additional flavors the coroutines API provides for flows. In terms of data types and the functions they provide, you can think of SharedState<T> and StateFlow<T> as implementations of Flow<T> with specific behavior when collected by multiple collectors.

For this reason, all the concepts you’ve seen so far are also valid for SharedState<T> and StateFlow<T>.

Key points

• The List<T> data type allows you to store an ordered collection of elements of type T in an eager way.
• All the elements of a List<T>, which is immutable, are present at the moment you create it.
• The List<T> data type is a functor and monad because of the presence of map and flatMap. You can also make it an applicative functor by implementing ap.
• The Sequence<T> data type allows you to generate a sequence of values of type T in a lazy way.
• In a Sequence<T>, map and flatMapConcat are invoked when the values need to be collected and consumed.
• A Sequence<T> can work as a functor, applicative functor and monad.
• The Flow<T> data type is similar to Sequence<T> but in the context of a coroutine.
• Suspendable functions are an idiomatic and powerful tool to handle side effects in Kotlin.
• A Flow<T> allows you to generate a sequence, or flow, of values of type T that can be generated from suspendable functions.
• You can implement curry and composition for suspendable functions as you did for non-suspendable ones, just following the functional programming principles you learned in the previous chapters.
• You can repeat for SharedFlow<T> and StateFlow<T> the same process you followed for a Flow<T>.

Where to go from here?

Congratulations! In this chapter, you had the opportunity to apply concepts you learned in the previous chapter in a concrete example that allowed you to fetch information about your favorite TV shows. You’ve learned how to create Sequence<T> and how to use Flow<T> in an environment of concurrency. Finally, you’ve empowered your functional thinking by implementing abstractions for composing suspendable functions, returning a Result<T> monad. It’s been a lot of work and also a lot of fun!

As mentioned previously, you can take a look at Kotlin Coroutines by Tutorials to learn more about coroutines, SharedFlow<T> and StateFlow<T>. The following chapters will talk about a couple more libraries that embody functional programming principles.