Introduce Flow.last and Flow.lastOrNull operators (Kotlin#2662)

Fixes Kotlin#2246

kotlinx-coroutines-core/api/kotlinx-coroutines-core.api

@@ -965,6 +965,8 @@ public final class kotlinx/coroutines/flow/FlowKt {
 	public static final fun fold (Lkotlinx/coroutines/flow/Flow;Ljava/lang/Object;Lkotlin/jvm/functions/Function3;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
 	public static final fun forEach (Lkotlinx/coroutines/flow/Flow;Lkotlin/jvm/functions/Function2;)V
 	public static final fun getDEFAULT_CONCURRENCY ()I
+	public static final fun last (Lkotlinx/coroutines/flow/Flow;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
+	public static final fun lastOrNull (Lkotlinx/coroutines/flow/Flow;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
 	public static final fun launchIn (Lkotlinx/coroutines/flow/Flow;Lkotlinx/coroutines/CoroutineScope;)Lkotlinx/coroutines/Job;
 	public static final fun map (Lkotlinx/coroutines/flow/Flow;Lkotlin/jvm/functions/Function2;)Lkotlinx/coroutines/flow/Flow;
 	public static final fun mapLatest (Lkotlinx/coroutines/flow/Flow;Lkotlin/jvm/functions/Function2;)Lkotlinx/coroutines/flow/Flow;

kotlinx-coroutines-core/common/src/flow/terminal/Reduce.kt

@@ -144,3 +144,28 @@ public suspend fun <T> Flow<T>.firstOrNull(predicate: suspend (T) -> Boolean): T
     }
     return result
 }
+
+/**
+ * The terminal operator that returns the last element emitted by the flow.
+ *
+ * Throws [NoSuchElementException] if the flow was empty.
+ */
+public suspend fun <T> Flow<T>.last(): T {
+    var result: Any? = NULL
+    collect {
+        result = it
+    }
+    if (result === NULL) throw NoSuchElementException("Expected at least one element")
+    return result as T
+}
+
+/**
+ * The terminal operator that returns the last element emitted by the flow or `null` if the flow was empty.
+ */
+public suspend fun <T> Flow<T>.lastOrNull(): T? {
+    var result: T? = null
+    collect {
+        result = it
+    }
+    return result
+}

kotlinx-coroutines-core/common/test/flow/terminal/LastTest.kt

@@ -0,0 +1,45 @@
+/*
+ * Copyright 2016-2021 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
+ */
+
+package kotlinx.coroutines.flow
+
+import kotlinx.coroutines.*
+import kotlin.test.*
+
+class LastTest : TestBase() {
+    @Test
+    fun testLast() = runTest {
+        val flow = flowOf(1, 2, 3)
+        assertEquals(3, flow.last())
+        assertEquals(3, flow.lastOrNull())
+    }
+
+    @Test
+    fun testNulls() = runTest {
+        val flow = flowOf(1, null)
+        assertNull(flow.last())
+        assertNull(flow.lastOrNull())
+    }
+
+    @Test
+    fun testNullsLastOrNull() = runTest {
+        val flow = flowOf(null, 1)
+        assertEquals(1, flow.lastOrNull())
+    }
+
+    @Test
+    fun testEmptyFlow() = runTest {
+        assertFailsWith<NoSuchElementException> { emptyFlow<Int>().last() }
+        assertNull(emptyFlow<Int>().lastOrNull())
+    }
+
+    @Test
+    fun testBadClass() = runTest {
+        val instance = BadClass()
+        val flow = flowOf(instance)
+        assertSame(instance, flow.last())
+        assertSame(instance, flow.lastOrNull())
+
+    }
+}