Channels & Flows

In this talk, I will presented on how to use constructs such as actors, channels and flows in use cases. These use cases are modeling streams with channels and flows and using actors to ensure one concurrent job is running. Channels and Flows are experimental features that allow us to represent a stream of data. Flows represent a cold stream whereas a Channel is like a hot observable. We’ll dive into these topics.

Transcript

1. Channels & Flows Mohit Sarveiya /heyitsmohit twitter.com
2. Channels & Flows • Channels • Actors • Flows
3. Coroutine A Passing Data Coroutine B
4. Coroutine A Passing Data Coroutine B
5. Passing Data • Deferred • Channels • Flows
6. Channels
7. What is a Channel? • Similar to BlockingQueue.
 interface Channel<E> { suspend fun send(element: E) suspend fun receive(): E }
8. Using Channels runBlocking {
 val channel = Channel<Int>() launch { channel.send(1) } val num = channel.receive() }
9. Using Channels runBlocking {
 val channel = Channel<Int>() launch { channel.send(1) } val num = channel.receive() } Parent Coroutine
10. Using Channels runBlocking {
 val channel = Channel<Int>() launch { channel.send(1) } val num = channel.receive() } Child Coroutine
11. Using Channels runBlocking {
 val channel = Channel<Int>() launch { channel.send(1) } val num = channel.receive() }
12. Using Channels runBlocking {
 val channel = Channel<Int>() launch { channel.send(1) } val num = channel.receive() }
13. Using Channels runBlocking {
 val channel = Channel<Int>() launch { channel.send(1) } val num = channel.receive() }
14. Channel Types • Rendezvous • Buffered • Unlimited • Conflated
15. Creating Channel fun <E> Channel(capacity: Int = RENDEZVOUS) = when (capacity) { RENDEZVOUS !-> RendezvousChannel() UNLIMITED !-> LinkedListChannel() CONFLATED !-> ConflatedChannel() BUFFERED !-> ArrayChannel(CHANNEL_DEFAULT_CAPACITY) else !-> ArrayChannel(capacity) }
16. Rendezvous Channel • No Buffer • Waits for receiver to send data
17. Rendezvous Channel runBlocking { val channel = Channel<Int>() launch { channel.send(1) channel.send(2) } }
18. Rendezvous Channel runBlocking { val channel = Channel<Int>() launch { channel.send(1) channel.send(2) } } SUSPEND
19. Rendezvous Channel runBlocking { val channel = Channel<Int>() launch { channel.send(1) channel.send(2) }
 channel.receive() } Receiver
20. runBlocking { val channel = Channel<Int>() launch { channel.send(1)
 channel.send(2) }
 channel.receive() } Rendezvous Channel RESUME
21. runBlocking { val channel = Channel<Int>() launch { channel.send(1)
 channel.send(2) }
 channel.receive() } Rendezvous Channel SUSPEND
22. Channel State • Queue • Buffer (Optional)
23. Rendezvous Channel val channel = Channel<Int>() EmptyQueue Queue
24. Channel States • Empty Queue • Send Queued • Send Buffered • Closed
25. Channel States EmptyQueue LockFreeLinkedList SendBuffered SendQueued Closed
26. Rendezvous Channel val channel = Channel<Int>() channel.send(1)
27. Rendezvous Channel class SendElement( val data: Any?, val cont: CancellableContinuation<Unit> ) : LockFreeLinkedListNode()
28. Rendezvous Channel val channel = Channel<Int>() channel.send(1) SendElement(1) Queue
29. Rendezvous Channel val channel = Channel<Int>() channel.send(1)
 channel.receive() EmptyQueue Queue
30. class AbstractSendChannel { fun toString() = {$queueDebugStateString}$bufferDebugString” } Query Channel State Current Channel State Buffer Size
31. Query Channel State • Get most recent state • Size of buffer
32. 
 
 $queueDebugStateString get() {
 if (head !!=== queue) return "EmptyQueue" } Query Channel State EmptyQueue
33. 
 
 
 $queueDebugStateString get() {
 when (head) {
 is Closed!!<*> !-> head.toString()
 is Receive!!<*> !-> "ReceiveQueued"
 is Send !-> "SendQueued" else !-> "UNEXPECTED:$head" } } Query Channel State SendElement
34. 
 
 
 $queueDebugStateString get() {
 if (tail !!!== head) { result += ",queueSize=${countQueueSize()}" } } Query Channel State SendElement Buffer SendElement SendElement
35. Query Channel State val channel = Channel<Int>() channel.send(1)
 println(channel)
 
 Output: 
 RendezvousChannel{SendQueued}
36. Idiomatic Channel Creation runBlocking {
 val channel = Channel<Int>() launch { for (i in 1!..10) { channel.send(i) } } }
37. Idiomatic Channel Creation runBlocking {
 val channel = Channel<Int>() launch { for (i in 1!..10) { channel.send(i) } } } Producer
38. Idiomatic Channel Creation 
 fun CoroutineScope.producer(): ReceiveChannel<Int> = 
 produce { for (i in 1!..5) { send(i) } }
39. Idiomatic Channel Creation runBlocking {
 val channel: ReceiveChannel<Int> = producer() }
40. Producer Builder • Creates Coroutine with Channel • DSL for creating channels
41. Produce Builder fun <E> CoroutineScope.produce( context: CoroutineContext, capacity: Int = 0, block: suspend ProducerScope<E>.() !-> Unit ): ReceiveChannel<E>
42. Produce Builder fun <E> CoroutineScope.produce(…) { val channel = Channel<E>(capacity) val newContext = newCoroutineContext(context) val coroutine = ProducerCoroutine(ctx, channel) coroutine.start(…, coroutine, block) return coroutine }
43. Idiomatic Channel Consumption runBlocking {
 val channel: ReceiveChannel<Int> = producer() channel.close() }
44. Idiomatic Channel Consumption runBlocking {
 val channel: ReceiveChannel<Int> = producer() channel.close() for (num in channel) { calculate(num) } }
45. Channel Iterator interface ChannelIterator<out E> { suspend fun hasNextSuspend(): E fun next(): E }
46. Coroutine A Channel Iterator Coroutine B 1 2 3 x Closed token
47. Coroutine A Channel Iterator Coroutine B EmptyQueue Channel Queue
48. Coroutine A Channel Iterator Coroutine B 1 SendElement(1) Channel Queue hasNext()
49. Coroutine A Channel Iterator Coroutine B 1 SendElement(1) Channel Queue next()
50. Coroutine A Channel Iterator Coroutine B 1 SendElement(1) Channel Queue next()
51. Coroutine A Channel Iterator Coroutine B 1 SendElement(3) Channel Queue next() 2 3
52. Coroutine A Channel Iterator Coroutine B 1 Closed Channel Queue 2 3 x Closed token
53. Coroutine A Channel Iterator Coroutine B 1 2 3 Suspended No Closed token
54. Channel Iterator fun hasNextSuspend() = suspendCancellable { cont !-> while (true) { val result = channel.pollInternal() this.result = result if (result is Closed!!<*>) {
 cont.resume(false) } else { cont.resume(true) } …
 } }
55. Channel Types • Rendezvous • Buffered • Unlimited • Conflated
56. Buffered Channel • Buffer with specific capacity. • Caches sent items in Array
57. Buffered Channel runBlocking { val channel = Channel<Int>(capacity = 3) }
58. Buffered Channel class ArrayChannel(val capacity: Int) { val buffer = arrayOfNulls<Any?>(capacity) … }
59. runBlocking { val channel = Channel<Int>(capacity = 3) launch { channel.send(1) channel.send(2) channel.send(3) } } Buffered Channel 1 2 3 Buffer
60. runBlocking { val channel = Channel<Int>(capacity = 3) launch { channel.send(1) channel.send(2) channel.send(3)
 channel.send(4) } } Buffered Channel SUSPEND 1 2 3 Buffer
61. Unlimited Channel • Buffer with unlimited capacity. • LinkedList Buffer.
62. Unlimited Channel runBlocking { val channel = Channel<Int>(Channel.UNLIMITED) } Empty Queue
63. Unlimited Channel runBlocking { val channel = Channel<Int>(Channel.UNLIMITED) launch { channel.send(1) channel.send(2) channel.send(3) } } SendBuffered(1) SendBuffered(2) SendBuffered(3)
64. Conflated Channel • Get most recent sent element • Similar to ReplaySubject (RxJava)
65. Conflated Channel 
 val channel = Channel<Int>(Channel.CONFLATED) channel.send(1) channel.send(2) !// Conflate previously sent 
 println(channel.receive()) Output:
 2
66. Conflated Channel 
 val channel = Channel<Int>(Channel.CONFLATED) channel.send(1) SendBuffered(1)
67. Conflated Channel 
 val channel = Channel<Int>(Channel.CONFLATED) channel.send(1) channel.send(2) SendBuffered(2)
68. Conflated Channel 
 val channel = Channel<Int>(Channel.CONFLATED) channel.send(1) channel.send(2) SendBuffered(2) Never Suspend
69. Channel Types • Rendezvous • Buffered • Unlimited • Conflated
70. Channels Rendezvous • Queue size = 1 • Buffer: None Array Channel • Queue size = 1 • Buffer: Array Linked List Channel • Queue size = Unlimited • Buffer: Queue Conflated • Queue size = 1 • Buffer: Queue
71. Tips for using Channels • Query channel queue for debugging. • Channel type determines back pressure strategy to use.
72. Actors
73. Problems they solve • Shared mutable state • Event conflation
74. Problem class MyViewModel(val interactor: Interactor) { val scope = CoroutineScope(Dispatchers.Main) fun refresh() { scope.launch { val response = interactor.getData() handleResponse(response) } } }
75. Problem class MyViewModel(val interactor: Interactor) { val scope = CoroutineScope(Dispatchers.Main) fun refresh() { scope.launch { val response = interactor.getData() handleResponse(response) } } } Ensure 1 concurrent job runs
76. Solution • Check status of Job • Use Actors
77. Check Job State class MyViewModel(val interactor: Interactor) { val job: Job? = null fun refresh() { if (job!?.isCompleted !== false) return
 val job = scope.launch { val response = interactor.getData() handleResponse(response) } } }
78. Actor State Actor Channel • Coroutine with mailbox (Channel)
79. Actor State Actor Channel Coroutine Event
80. Actor State Actor Channel Coroutine Event
81. Events Channel Coroutine LoadData sealed class ActorEvents object LoadData : ActorEvents() …
82. fun <E> CoroutineScope.actor( capacity: Int = 0, block: suspend ActorScope<E>.() !-> Unit ): SendChannel<E> { val channel = Channel<E>(capacity) val coroutine = ActorCoroutine(ctx, channel, true) coroutine.start(start, coroutine, block) return coroutine } Inside Actor
83. fun CoroutineScope.modelActor() = 
 
 actor<ActorEvents>(capacity = Channel.CONFLATED) { for (event in channel) { when(event) { is ActorEvents.LoadData !-> apiResponse = interactor.getData() }
 } } Using Actor
84. Actor Use Cases • UI Event Conflation • Implement Redux-like architecture
85. Flows
86. What is a Flow? • Cold stream interface Flow<out T> { suspend fun collect(collector: FlowCollector<T>) }
87. What is a Flow? interface Flow<out T> { suspend fun collect(collector: FlowCollector<T>) } Consume stream
88. What is a Flow? interface Flow<out T> { suspend fun collect(collector: FlowCollector<T>) } What to consume?
89. What is a Flow? interface FlowCollector<in T> { suspend fun emit(value: T) }
90. Using Flows runBlocking { val flow: Flow<Int> = flowOf(1, 2, 3) flow.collect { println(it) } }
91. Creating Flows • flowOf(…) • flow { … }
92. flowOf Builder flowOf(1, 2, 3) fun <T> flowOf(vararg elements: T): Flow<T>
93. flowOf Builder fun <T> flowOf(vararg elements: T): Flow<T> { object : Flow<T> { override suspend fun collect(collector: FlowCollector<T>) { for (element in elements) { collector.emit(element) } } } }
94. Flow Builder runBlocking { flow { emit(1) emit(2) emit(3) } }
95. Flow Builder fun <T> flow( block: suspend FlowCollector<T>.() !-> Unit ): Flow<T>
96. Terminal Operators • Collect • Single • Reduce • toList
97. Single Operator Collects only one value. val flow = flowOf(1) val num: Int = flow.single()
98. Single Operator Collects only one value. val flow = flowOf(1,2,3) val num: Int = flow.single() java.lang.IllegalStateException: Expected only one element
99. Flow Operators • Map • Filter • Take • Zip • delayEach • delayFlow
100. DelayEach val flow = flowOf(1,2,3) flow.delayEach(1000) .collect { println(it) }
101. DelayEach val flow = flowOf(1,2,3) flow.delayEach(1000) .collect { println(it) } Operators are not suspending functions
102. DelayEach runBlocking { val flow = getFlow() flow.collect { println(it) } } fun getFlow() = flowOf(1,2,3).delayEach(1000)
103. Flow Constraints • Context Preservation • Exception Transparency
104. Flow Constraints Emit from the same coroutine val flow = flow { launch(Dispatchers.IO) { emit(1) } } Flow invariant is violated: emission from another coroutine is detected
105. Channel Flow Builder val flow = channelFlow { launch(Dispatchers.IO) { send(1) } launch(Dispatchers.IO) { send(2) } } Flow Channel
106. Channel Flow Builder class ChannelFlowBuilder<T>( val block: suspend ProducerScope<T>.() !-> Unit, capacity: Int = BUFFERED ) : ChannelFlow<T>(context, capacity) Uses Buffered Channel
107. Flow Context runBlocking { val flow = flow { emit(1) emit(2) emit(3) } flow .map { it + 1 } .collect { println(it) } }
108. Flow Context runBlocking { val flow = flow { emit(1) emit(2) emit(3) } flow .map { it + 1 } .collect { println(it) } } Dispatchers.Main Dispatchers.Main
109. Flow Context runBlocking { val flow = flow { emit(1) emit(2) emit(3) } flow .map { it + 1 } .flowOn(Dispatchers.IO) .collect { println(it) } } Dispatchers.IO Dispatchers.IO Dispatchers.Main
110. Exception Transparency val flow = flow { try { makeRequest() } catch (e: IOException) { } }
111. Exception Transparency fun Flow<T>.catch(action: suspend FlowCollector<T>.(cause: Throwable) !-> Unit ): Flow<T> Emit exception in a wrapper
112. Exception Transparency sealed class ApiResponse<T> { data class Success<T>(val data: T): ApiResponse<T>() data class Error<T>(val t: Throwable): ApiResponse<T>() }
113. Exception Transparency val flow = flow<ApiResponse<Data!>> { val data = makeRequest() emit(ApiResponse.Success(data)) }.catch { emit(ApiResponse.Error(it)) }
114. Resources
115. Debugging (JVM) runBlocking<Unit> { DebugProbes.install() val channel = Channel<Int>() launch { channel.send(2) }
 DebugProbes.dumpCoroutines() } Coroutine StandaloneCoroutine{Active}, state: SUSPENDED Install Debugger View Coroutines State
116. Resources (Inside Channels)
117. Thank you! twitter.com/heyitsmohit