Programming Android with Kotlin Achieving Structured Concurrency with Coroutines

fun main() = runBlocking {
  val channel = Channel<Int>()
  launch {
    for (x in 1..5) channel.send(x * x)
    channel.close()
  }

  for (y in channel) println(y)
}

fun messageFlow() = callbackFlow<Message> {
  // 1. Instantiate the "callback." In this case, it's an observer.
  val cb = object : Callback {
    override fun onMessage(message: Message) {
      trySend(message)
    }

    override fun onClosed() {
      channel.close()
    }

    override fun onFailure(t: Throwable) {
      cancel(CancellationException("Failure", t))
    }
  }
  // 2. Register that callback using the available api.
  addCallback(cb)
  // 3. Listen for close event using `awaitClose`, and provide a relevant action
  // to take in this case. Most probably, you'll have to unregister the callback.
  awaitClose { removeCallback(cb) }
}

suspend fun transform(loc: Location): Content = withContext(Dispatchers.IO) {
  // do some work
}

fun main() = runBlocking {
  val contentFlow = locationsFlow.map {
    flow {
      emit(transform(it))
    }
  }.flattenMerge(4)

  val contents = contentFlow.toList()
}

fun <T> Flow<T>.bufferTimeout(timeoutMillis: Long, capacity: Int = 16): Flow<List<T>> = flow {
  require(capacity > 0) { "Buffer capacity must be positive" }
  require(timeoutMillis > 0) { "Timeout must be positive" }

  coroutineScope {
    // Consume the upstram flow and send them over a channel.
    val channel = produceIn(this)
    val buffer = mutableListOf<T>()
    val ticker = ticker(timeoutMillis)

    suspend fun flush() {
      if (buffer.isNotEmpty()) {
        emit(buffer.toList())
        buffer.clear()
      }
    }

    try {
      whileSelect {
        channel.onReceive {
          buffer.add(it)
          if (buffer.size == capacity) {
            flush()
          }
          true
        }
        ticker.onReceive {
          flush()
          true
        }
      } catch (e: ClosedReceiveChannelException) {
        // When  the  upstream  flow  collection  completes,  the  coroutine  started  with `produceIn`
        // will still attempt to read from that flow, and a `ClosedReceiveChannelException`  will
         // be  raised.  So  we  catch  that  exception, and we know that we should flush the content
        // of the buffer.
        flush()
      } finnaly {
        channel.cancel()
        ticker.cancel()
      }
    }
  }
}

suspend fun main() {
  val flow = (1..100).asFlow().onEach { delay(10) }
  val startTime = System.currentTimeMillis()
  flow.bufferTimeout(50, 10).collect {
    println("${System.currentTimeMillis() - startTime} ms: $it")
  }
}

try {
  aFlow.collect {
    // ...
  }
} catch (e: Exception) {
  // ...
}

data class Image(val url: String)

sealed calss Result
data class Success(val image: Image) : Result()
data class Error(val url: String) : Result()

suspend fun fetchImage(url: String): Image {
  // simulate network request
  delay(10)

  // simulate an exception thrown by the network request
  if (url.contains("error")) {
    throw IOException("Network error")
  }

  return Image(url)
}

suspend fun fetchImageWithResult(url: String): Result {
  println("fetchImageWithResult: $url")
  return try {
    Success(fetchImage(url))
  } catch (e: IOException) {
    Error(url)
  }
}

fun main() = runBlocking {
  val urls = flowOf(
    "https://www.google.com",
    "https://www.google.com/error",
    "https://www.google.com"
  )

  urls.map { url ->
    fetchImageWithResult(url)
  }.collect {
    println("Result: $it")
  }
}

fun <T, R: Any> Flow<T>.mapWithRetry(
  action: suspend (T) -> R,
  predicate: suspend (R, attempt: Int) -> Boolean
) = map { data ->
  var attempt = 0
  var shallRetry: Boolean
  var lastValue: R? = null

  do {
    val tr = action(data)
    shallRetry = predicate(tr, ++attempt)
    if (!shallRetry) {
      lastValue = tr
    }
  } while (shallRetry)
  return@map lastValue
}

fun main = runBlocking {
  val flow = flowOf(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
  flow.mapWithRetry(
    action = { value ->
      println("action: $value")
      value
    },
    predicate = { value, attempt ->
      value % 2 == 0 && attempt < 3
    }
  ).collect {
    println("collect: $it")
  }
}

MutableSharedFlow(replay = 3, onBufferOverflow = BufferOverflow.DROP_OLDEST)