Checkpoints: Creating Conditional Suspensions in Transmission • Yigit Ozgumus

Transmission Library aims to offer a flexible API to users to build their asynchronous communication network inside their applications. It is built on top of Kotlin Coroutines, which provides a versatile and extensive API to implement all kinds of concurrent operations.

This post will explain an edge case we found while using Transmission and how we implemented the solution. Get your hot beverage of choice, and let’s explore this edge case 👇🏻

Setting the Stage

In Transmission, effects, and signals are handled concurrently. We can pause the current signal handler process if we get data from another transformer or if the result is from a computation. These cases are now stable parts of the Communication API.

// Inside a Transformer
onEffect<InputEffect.InputUpdate> { effect ->
    holder.update { it.copy(writtenUppercaseText = effect.value.uppercase()) }
  // We are suspending on compute
    val result = compute(OutputTransformer.outputCalculationContract)
    holder.update {
        it.copy(writtenUppercaseText = it.writtenUppercaseText + " ${result?.result}")
    }
}
// ... Another Transformer Code
onEffect<InputEffect.InputUpdate> { effect ->
    holder.update { it.copy(outputText = effect.value) }
    delay(3.seconds)
  // Here getData is a suspending function
    val selectedColor = getData(ColorPickerTransformer.holderContract)
    selectedColor ?: return@onEffect
  // ... more code
}

We can expect that the function we use will suspend the current execution if it takes a while to get the data. Making all functions support suspension gives us soft synchronization points and allows us to mix network calls and services if necessary.

In these cases, We know what to execute in advance to get the desired result.

What if we have a condition that we need for the rest of the execution, but we don’t know what to execute?

Coupling Logic

There are different scenarios where this might happen. The condition we want to wait for could result from a user Interaction or a suspending API call. We will focus on user interaction, but let’s briefly talk about the coupling.

Imagine a function that does some work and has a suspending function that returns a value.

interface SomeUseCase {
  suspend fun execute(): String
}
// ...
private suspend fun startWorkX() {
  // ...
  val value = someUseCase.execute()
  executeRemainingLogic(value)
}

Let’s build up on this case. We need to refactor executeRemainingLogic. It needs to take a value from some other use case. This someOtherUseCase is executed from somewhere else, and we need to get its result value.

// ...
interface SomeOtherUseCase {
  suspend fun executeOtherThing(value: String): Flow<Int>
}
// ...
private suspend fun startWorkY() {
  someOtherUseCase.executeOtherThing("otherThing")
    .onEach {
      processValue(it)
      // We need to pass this value to [executeRemainingLogic]
    }.collect()
}

Functions that return Flows or functions that return a value with suspension can be coupled in various ways. A straightforward solution would be to merge these executions so that we can supply the data.

private suspend fun startWorkXAndY() {
  // more logic
  val value = someUseCase.execute()
  someOTherUseCase.executeOtherThing("otherThing")
    .onEach { targetValue ->
      processValue(targetValue)
      // we can pass the value
      executeRemainingLogic(value, targetValue)
    }
}

Doing this means more glue work 🚧:

Depending on the remaining logic, we might end up with an unnecessary refactor.
All the remaining logic should be tested. We also need to refactor the tests as well.
Flow-based operations are reactive, so we must match the flow’s emit frequency of the other use case.
Extra: It might not be possible to refactor and merge the logic due to different modules, etc.

Sometimes, we might not need a result. We need a check mark to verify that our prerequisite condition is executed. Now that we understand the problem at hand let’s move over to the sample application of Transmission 👉🏻

Coupling User Interactions

Let’s consider the InputComponent in the sample application. It has two handlers.

When the input text is updated, it publishes the value to all other Transformers.
When we change the selected color using ColorPickerComponent, its background color is updated.

override val handlers: Handlers = createHandlers {
    onSignal<InputSignal.InputUpdate> { signal ->
        holder.update { it.copy(writtenText = signal.value) }
        publish(effect = InputEffect.InputUpdate(signal.value))
    }
    onEffect<ColorPickerEffect.BackgroundColorUpdate> { effect ->
        holder.update { it.copy(backgroundColor = effect.color) }
    }
}

Checkpoints

Let’s continue imagining a scenario where we can only continue input update execution once the component’s background color is updated.

Background updates before changing text should not have any effect.
This condition should run only when there is new input coming in from the UI

Our use cases in the Trendyol Android App were somewhat similar. We can now use checkpoints for these types of interactions.

Here is the updated UI logic and how it behaves under the same user interaction:

Old Execution Flow	New Execution Flow

And here is the updated code using checkpoints API:

@OptIn(ExperimentalTransmissionApi::class)
override val handlers: Handlers = createHandlers {
    onSignal<InputSignal.InputUpdate> { signal ->
        holder.update { it.copy(writtenText = signal.value) }
        // We wait for the color checkpoint
        val color = pauseOn(colorCheckpoint)
        // We also send the color to another transformer
        // (Simulating Business Logic Update)
        send(
            effect = ColorPickerEffect.SelectedColorUpdate(color),
            identity = multiOutputTransformerIdentity
        )
        publish(effect = InputEffect.InputUpdate(signal.value))
    }
    onEffect<ColorPickerEffect.BackgroundColorUpdate> { effect ->
    // Once the background update arrives, we validate the checkpoint
        validate(colorCheckpoint, effect.color)
        holder.update { it.copy(backgroundColor = effect.color) }
    }
}

Let’s talk about how this API actually works.

API Internals

First of all, checkpoint API is highly experimental. Additionally, we added ExperimentalTranmissionApi annotation to our API this time because we still need to implement the corresponding test utility functions.

Here are the current API methods which are subject to change:

In Coroutines, the compiler converts suspending functions that have other suspended functions inside into State Machines to control the suspending executions and potential thread switching. We want to leverage this machinery.

Considering our Signal and Effect transmissions are just suspending functions, we can divide the execution into multiple parts by creating anonymous suspending functions that we can call from the codebase.

Under the hood, checkpoint API allows us to create anonymous suspending functions. It uses SuspendCancellableCoroutine to create a Checkpoint.

@ExperimentalTransmissionApi
override suspend fun CommunicationScope.pauseOn(
  contract: Contract.Checkpoint.Default
) {
  val queryIdentifier = IdentifierGenerator.generateIdentifier()
  suspendCancellableCoroutine<Unit> { continuation ->
      val validator =
          object : CheckpointValidator<Contract.Checkpoint.Default, Unit> {
              override suspend fun validate(
                  contract: Contract.Checkpoint.Default,
                  args: Unit
              ): Boolean {
                  continuation.resume(Unit)
                  return true
              }
          }
      checkpointTrackerProvider()?.run {
          registerContract(contract, queryIdentifier)
          putOrCreate(queryIdentifier, validator)
      }
}

When we need execution to continue, target interaction might trigger the checkpoint using its Contract. If the conditions of the checkpoints are validated, we remove them from the tracker and continue the execution.

override suspend fun validate(contract: Contract.Checkpoint.Default) {
    val validator = checkpointTrackerProvider()
        ?.useValidator<Contract.Checkpoint.Default, Unit>(contract)
    if (validator?.validate(contract, Unit) == true) {
        checkpointTrackerProvider()
            ?.removeValidator(contract)
    }
}

In the early API versions, checkpoints also supported frequency, which is the number of times it can create a checkpoint inside the handler lambda. But that made the implementation more verbose than it needed to, so we decided to remove it and delegate that functionality to the actual implementation of the Signal and Effect Handling.

Conclusion

We are now adding features to the Transmission Library based on our needs. However, we also want to improve the stability of the API and offer equally stable testing utilities to make decoupling easier on complex business code. See you in the next post!

Photo by Sean Z on Unsplash
You can check out the Transmission Library here and read the series here