Add UnicastSubject and UnicastWorkSubject

Author: akarnokd

README.md

@@ -23,6 +23,8 @@ Table of contents
   - [PublishSubject](#publishsubject)
   - [ReplaySubject](#replaysubject)
   - [BehaviorSubject](#behaviorsubject)
+  - [UnicastSubject](#unicastsubject)
+  - [UnicastWorkSubject](#unicastworksubject)
 - Sources
   - `range`
   - `timer`
@@ -209,3 +211,52 @@ can be used to hold back the upstream until the expected number of collectors ha
 In the example, it is known `merge` will establish 2 collectors, thus the `publish` can be instructed to await those 2.
 Without the argument, `range` would rush through its items as `merge` doesn't start collecting in time, causing an
 empty result list.
+
+## UnicastSubject
+
+Buffers items until a single collector starts collecting items. Use `collectorCancelled` to
+detect when the collector no longer wants to collect items.
+
+Note that the subject uses an unbounded inner buffer and does not suspend its input side if
+the collector never arrives or can't keep up.
+
+```kotlin
+val us = UnicastSubject()
+
+launchIn(Dispatchers.IO) {
+    for (i in 1..200) {
+        println("Emitting $i")
+        us.emit(i)
+        delay(1)
+    }
+    emit.complete()
+}
+
+// collector arrives late for some reason
+delay(100)
+
+us.collect { println("Collecting $it") }
+```
+
+## UnicastWorkSubject
+
+Buffers items until and inbetween a single collector is able to collect items. If the current
+collector cancels, the next collector will receive the subsequent items.
+
+Note that the subject uses an unbounded inner buffer and does not suspend its input side if
+the collector never arrives or can't keep up.
+
+```kotlin
+val uws = UnicastWorkSubject()
+
+generateInts(uws, 1, 15)
+
+// prints lines 1..5
+uws.take(5).collect { println(it) }
+
+// prints lines 6..10
+uws.take(5).collect { println(it) }
+
+// prints lines 11..15
+uws.take(5).collect { println(it) }
+```

gradle.properties

@@ -1,5 +1,5 @@
 GROUP=com.github.akarnokd
-VERSION_NAME=0.0.7
+VERSION_NAME=0.0.8
 
 POM_ARTIFACT_ID=kotlin-flow-extensions
 POM_NAME=Kotlin Flow Extensions

src/main/kotlin/hu/akarnokd/kotlin/flow/UnicastSubject.kt

@@ -0,0 +1,115 @@
+/*
+ * Copyright 2019-2020 David Karnok
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package hu.akarnokd.kotlin.flow
+
+import kotlinx.coroutines.flow.AbstractFlow
+import kotlinx.coroutines.flow.FlowCollector
+import java.util.concurrent.ConcurrentLinkedQueue
+import java.util.concurrent.atomic.AtomicReference
+
+/**
+ * Buffers items until a single collector collects them.
+ *
+ * @param <T> the item type
+ */
+class UnicastSubject<T> : AbstractFlow<T>(), SubjectAPI<T> {
+
+    companion object {
+        val terminatedCollector = object : FlowCollector<Any> {
+            override suspend fun emit(value: Any) {
+            }
+        }
+        val terminated = Throwable("No more elements")
+    }
+
+    val queue = ConcurrentLinkedQueue<T>()
+
+    @Volatile
+    var terminal : Throwable? = null
+
+    val resumable = Resumable()
+
+    val current = AtomicReference<FlowCollector<T>>()
+
+    override suspend fun collectSafely(collector: FlowCollector<T>) {
+        while (true) {
+            val curr = current.get()
+            if (curr != null) {
+                throw IllegalStateException("Only one collector allowed")
+            }
+            if (current.compareAndSet(curr, collector)) {
+                break
+            }
+        }
+
+        while (true) {
+            val t = terminal
+            val v = queue.poll()
+
+            if (t != null && v == null) {
+                current.getAndSet(terminatedCollector as FlowCollector<T>)
+                if (t != terminated) {
+                    throw t
+                }
+                return
+            }
+            if (v != null) {
+                try {
+                    collector.emit(v)
+                } catch (e: Throwable) {
+                    current.getAndSet(terminatedCollector as FlowCollector<T>)
+                    queue.clear()
+                    throw e
+                }
+            } else {
+                resumable.await()
+            }
+        }
+    }
+
+    override suspend fun emitError(ex: Throwable) {
+        terminal = ex
+        resumable.resume()
+    }
+
+    override suspend fun complete() {
+        terminal = terminated
+        resumable.resume()
+    }
+
+    override fun hasCollectors(): Boolean {
+        val curr = current.get()
+        return curr != null && curr != terminatedCollector
+    }
+
+    override fun collectorCount(): Int {
+        return if (hasCollectors()) 1 else 0
+    }
+
+    override suspend fun emit(value: T) {
+        if (current.get() != terminatedCollector) {
+            queue.offer(value)
+            resumable.resume()
+        } else {
+            queue.clear()
+        }
+    }
+
+    fun collectorCancelled() : Boolean {
+        return current.get() == terminatedCollector
+    }
+}

src/main/kotlin/hu/akarnokd/kotlin/flow/UnicastWorkSubject.kt

@@ -0,0 +1,102 @@
+/*
+ * Copyright 2019-2020 David Karnok
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package hu.akarnokd.kotlin.flow
+
+import kotlinx.coroutines.flow.AbstractFlow
+import kotlinx.coroutines.flow.FlowCollector
+import java.util.concurrent.ConcurrentLinkedQueue
+import java.util.concurrent.atomic.AtomicReference
+
+/**
+ * Buffers items until a single collector collects them.
+ *
+ * @param <T> the item type
+ */
+class UnicastWorkSubject<T> : AbstractFlow<T>(), SubjectAPI<T> {
+
+    companion object {
+        val terminated = Throwable("No more elements")
+    }
+
+    val queue = ConcurrentLinkedQueue<T>()
+
+    @Volatile
+    var terminal : Throwable? = null
+
+    val resumable = Resumable()
+
+    val current = AtomicReference<FlowCollector<T>>()
+
+    override suspend fun collectSafely(collector: FlowCollector<T>) {
+        while (true) {
+            val curr = current.get()
+            if (curr != null) {
+                throw IllegalStateException("Only one collector allowed")
+            }
+            if (current.compareAndSet(curr, collector)) {
+                break
+            }
+        }
+
+        while (true) {
+            val t = terminal
+            val v = queue.poll()
+
+            if (t != null && v == null) {
+                current.getAndSet(null)
+                if (t != terminated) {
+                    throw t
+                }
+                return
+            }
+            if (v != null) {
+                try {
+                    collector.emit(v)
+                } catch (e: Throwable) {
+                    current.getAndSet(null)
+                    throw e
+                }
+            } else {
+                resumable.await()
+            }
+        }
+    }
+
+    override suspend fun emitError(ex: Throwable) {
+        terminal = ex
+        resumable.resume()
+    }
+
+    override suspend fun complete() {
+        terminal = terminated
+        resumable.resume()
+    }
+
+    override fun hasCollectors(): Boolean {
+        val curr = current.get()
+        return curr != null
+    }
+
+    override fun collectorCount(): Int {
+        return if (hasCollectors()) 1 else 0
+    }
+
+    override suspend fun emit(value: T) {
+        queue.offer(value)
+        resumable.resume()
+    }
+}