re-implement debounce processor... and then realise it's difficult to integrate with Kafka Streams DSL so don't use it

modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/misc/DebounceProcessor.kt

@@ -1,29 +1,18 @@
 package dev.adamko.kafkatorio.processor.misc
 
-import dev.adamko.kotka.extensions.component1
+import dev.adamko.kotka.extensions.state.useAll
-import dev.adamko.kotka.extensions.component2
-import dev.adamko.kotka.extensions.state.allAsSequence
 import kotlin.coroutines.CoroutineContext
 import kotlin.time.Duration
-import kotlinx.coroutines.CancellationException
-import kotlinx.coroutines.CoroutineDispatcher
 import kotlinx.coroutines.CoroutineName
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.Job
 import kotlinx.coroutines.SupervisorJob
-import kotlinx.coroutines.cancelChildren
+import kotlinx.coroutines.cancel
-import kotlinx.coroutines.channels.BufferOverflow
 import kotlinx.coroutines.delay
-import kotlinx.coroutines.flow.MutableSharedFlow
-import kotlinx.coroutines.flow.asFlow
-import kotlinx.coroutines.flow.emitAll
-import kotlinx.coroutines.flow.launchIn
-import kotlinx.coroutines.flow.runningFold
 import kotlinx.coroutines.isActive
 import kotlinx.coroutines.launch
 import org.apache.kafka.common.serialization.Serde
-import org.apache.kafka.streams.KeyValue
 import org.apache.kafka.streams.processor.api.Processor
 import org.apache.kafka.streams.processor.api.ProcessorContext
 import org.apache.kafka.streams.processor.api.ProcessorSupplier
@@ -37,117 +26,227 @@ import org.apache.kafka.streams.state.ValueAndTimestamp
 class DebounceProcessor<K, V>(
   private val inactivityDuration: Duration,
   private val recordsTimestampedStoreName: String,
-  coroutineDispatcher: CoroutineDispatcher = Dispatchers.Default,
+) : Processor<K, V, K, V> {
-) :
-  Processor<K, V, K, V>,
-  CoroutineScope {
 
-  override var coroutineContext: CoroutineContext =
+  private var state: State<K, V> = Uninitialised()
-    SupervisorJob() +
-        coroutineDispatcher +
-        CoroutineName("DebounceProcessor")
 
-  private lateinit var kafkaContext: ProcessorContext<K, V>
+  override fun init(kafkaContext: ProcessorContext<K, V>) {
-  private lateinit var recordsTimestampedStore: TimestampedKeyValueStore<K, V?>
+    state = Active(kafkaContext)
-
-  private val recordsFlow: MutableSharedFlow<DebounceTaskInitializer<K>> =
-    MutableSharedFlow(
-      replay = 0,
-      extraBufferCapacity = Int.MAX_VALUE,
-      onBufferOverflow = BufferOverflow.DROP_OLDEST,
-    )
-
-  override fun init(context: ProcessorContext<K, V>) {
-    this.kafkaContext = context
-
-    this.coroutineContext += CoroutineName("DebounceProcessor-${context.applicationId()}-${context.taskId()}")
-
-    recordsTimestampedStore = kafkaContext.getStateStore(recordsTimestampedStoreName)
-
-    // start collecting records - create a debounce-timer for each record
-    recordsFlow
-      .runningFold(mapOf<K, Job>()) { acc, (k, createTask) ->
-        require(coroutineContext.isActive)
-
-        // cancel any previously existing debounce task
-        acc[k]?.cancel()
-
-        // add a new debounce task
-        when (createTask) {
-          true -> acc.plus(k to createDebounceTask(k))
-          false -> acc.minus(k)
-        }
-
-      }.launchIn(this)
-
-    // re-initialise the flow from previously-stored records
-    launch {
-      recordsFlow.emitAll(
-        recordsTimestampedStore
-          .allAsSequence()
-          .map { keyValue ->
-            require(coroutineContext.isActive)
-            DebounceTaskInitializer<K>(keyValue)
-          }
-          .asFlow()
-      )
-    }
   }
 
-  private fun createDebounceTask(key: K): Job {
+  override fun process(record: Record<K, V>) {
-    return launch {
+    when (val state = state) {
-      delay(inactivityDuration)
+      is Active -> state.addRecord(record)
-      val (value, timestamp) = recordsTimestampedStore.get(key)
+      else      -> error("state is not Active (actual: $state)")
-      if (coroutineContext.isActive) {
-        kafkaContext.forward(Record(key, value, timestamp))
-      }
-    }
-  }
-
-  override fun process(record: Record<K, V?>) {
-    launch {
-      recordsTimestampedStore.put(
-        record.key(),
-        ValueAndTimestamp.make(record.value(), record.timestamp())
-      )
-      recordsFlow.emit(DebounceTaskInitializer(record))
     }
   }
 
   override fun close() {
-    coroutineContext.cancelChildren(CancellationException("processor go sleepy bye bye"))
+    (state as? Active)?.close()
   }
 
-}
+  private sealed interface State<K, V>
 
-internal data class DebounceTaskInitializer<K>(
+  private class Uninitialised<K, V> : State<K, V>
-  val key: K,
+
-  val createTask: Boolean,
+  private class Closed<K, V> : State<K, V>
-) {
+
-  constructor(keyValue: KeyValue<K, *>) : this(keyValue.key, keyValue.value != null)
+  private inner class Active(
-  constructor(record: Record<K, *>) : this(record.key(), record.value() != null)
+    val kafkaContext: ProcessorContext<K, V>,
-}
+  ) : State<K, V>, CoroutineScope {
+    override val coroutineContext: CoroutineContext =
+      SupervisorJob() +
+          Dispatchers.Unconfined +
+          CoroutineName("DebounceProcessor")
+
+    val recordsTimestampedStore: TimestampedKeyValueStore<K, V> =
+      kafkaContext.getStateStore(recordsTimestampedStoreName)
+
+    private val recordJobs = mutableMapOf<K, TimestampedJob>()
+
+    init {
+      recordsTimestampedStore.useAll {
+        forEach { r ->
+          val record = Record(r.key, r.value.value(), r.value.timestamp())
+          addRecord(record)
+        }
+      }
+    }
+
+    fun addRecord(record: Record<K, V>) {
+      recordJobs.compute(record.key()) { _, existing ->
+        if (existing != null && existing.timestamp < record.timestamp()) {
+          existing.job.cancel("replaced by newer record ${existing.timestamp} < ${record.timestamp()}")
+          createTimestampedJob(record)
+        } else {
+          null
+        }
+      }
+    }
+
+    private fun createTimestampedJob(record: Record<K, V>): TimestampedJob {
+      val job = launch {
+        recordsTimestampedStore.put(
+          record.key(),
+          ValueAndTimestamp.make(record.value(), record.timestamp())
+        )
+        delay(inactivityDuration)
+        if (isActive) {
+          kafkaContext.forward(record)
+          recordsTimestampedStore.delete(record.key())
+        }
+      }
+      return TimestampedJob(job, record.timestamp())
+    }
+
+    fun close(): Closed<K, V> {
+      coroutineContext.cancel()
+      return Closed()
+    }
+  }
+
+  private data class TimestampedJob(
+    val job: Job,
+    val timestamp: Long,
+  )
 
 
-fun <K, V> debounceProcessor(
+  companion object {
-  inactivityDuration: Duration,
+    class Supplier<K, V>(
-  storeName: String,
+      private val inactivityDuration: Duration,
-  keySerde: Serde<K>,
+      private val storeName: String,
-  valueSerde: Serde<V>,
+      private val keySerde: Serde<K>,
-): ProcessorSupplier<K, V, K, V> =
+      private val valueSerde: Serde<V>,
-  object : ProcessorSupplier<K, V, K, V> {
+    ) : ProcessorSupplier<K, V, K, V> {
 
-    override fun get(): Processor<K, V, K, V> = DebounceProcessor(inactivityDuration, storeName)
+      override fun get(): Processor<K, V, K, V> = DebounceProcessor(inactivityDuration, storeName)
 
-    override fun stores(): Set<StoreBuilder<*>> = setOf(
+      override fun stores(): Set<StoreBuilder<*>> = setOf(
-      Stores.timestampedKeyValueStoreBuilder(
+        Stores.timestampedKeyValueStoreBuilder(
-        Stores.persistentTimestampedKeyValueStore(storeName),
+          Stores.persistentTimestampedKeyValueStore(storeName),
-        keySerde,
+          keySerde,
-        valueSerde,
+          valueSerde,
+        )
       )
-    )
+    }
-
   }
+}
+
+
+//class DebounceProcessor<K, V>(
+//  private val inactivityDuration: Duration,
+//  private val recordsTimestampedStoreName: String,
+//  coroutineDispatcher: CoroutineDispatcher = Dispatchers.Default
+//) :
+//  Processor<K, V, K, V>,
+//  CoroutineScope {
+//
+//  override var coroutineContext: CoroutineContext =
+//    SupervisorJob() +
+//        coroutineDispatcher +
+//        CoroutineName("DebounceProcessor")
+//
+//  private lateinit var kafkaContext: ProcessorContext<K, V>
+//  private lateinit var recordsTimestampedStore: TimestampedKeyValueStore<K, V?>
+//
+//  private val recordsFlow: MutableSharedFlow<DebounceTaskInitializer<K>> =
+//    MutableSharedFlow(
+//      replay = 0,
+//      extraBufferCapacity = Int.MAX_VALUE,
+//      onBufferOverflow = BufferOverflow.DROP_OLDEST,
+//    )
+//
+//  override fun init(context: ProcessorContext<K, V>) {
+//    this.kafkaContext = context
+//
+//    this.coroutineContext += CoroutineName("DebounceProcessor-${context.applicationId()}-${context.taskId()}")
+//
+//    recordsTimestampedStore = kafkaContext.getStateStore(recordsTimestampedStoreName)
+//
+//    // start collecting records - create a debounce-timer for each record
+//    recordsFlow
+//      .runningFold(mapOf<K, Job>()) { acc, (k, createTask) ->
+//        require(coroutineContext.isActive)
+//
+//        // cancel any previously existing debounce task
+//        acc[k]?.cancel()
+//
+//        // add a new debounce task
+//        when (createTask) {
+//          true -> acc.plus(k to createDebounceTask(k))
+//          false -> acc.minus(k)
+//        }
+//
+//      }.launchIn(this)
+//
+//    // re-initialise the flow from previously-stored records
+//    launch {
+//      recordsFlow.emitAll(
+//        recordsTimestampedStore
+//          .allAsSequence()
+//          .map { keyValue ->
+//            require(coroutineContext.isActive)
+//            DebounceTaskInitializer<K>(keyValue)
+//          }
+//          .asFlow()
+//      )
+//    }
+//  }
+//
+//  private fun createDebounceTask(key: K): Job {
+//    return launch {
+//      delay(inactivityDuration)
+//      val (value, timestamp) = recordsTimestampedStore.get(key)
+//      if (coroutineContext.isActive) {
+//        kafkaContext.forward(Record(key, value, timestamp))
+//      }
+//    }
+//  }
+//
+//  override fun process(record: Record<K, V?>) {
+//    launch {
+//      recordsTimestampedStore.put(
+//        record.key(),
+//        ValueAndTimestamp.make(record.value(), record.timestamp())
+//      )
+//      recordsFlow.emit(DebounceTaskInitializer(record))
+//    }
+//  }
+//
+//  override fun close() {
+//    coroutineContext.cancelChildren(CancellationException("processor go sleepy bye bye"))
+//  }
+//
+//}
+//
+//internal data class DebounceTaskInitializer<K>(
+//  val key: K,
+//  val createTask: Boolean,
+//) {
+//  constructor(keyValue: KeyValue<K, *>) : this(keyValue.key, keyValue.value != null)
+//  constructor(record: Record<K, *>) : this(record.key(), record.value() != null)
+//}
+//
+//
+//fun <K, V> debounceProcessor(
+//  inactivityDuration: Duration,
+//  storeName: String,
+//  keySerde: Serde<K>,
+//  valueSerde: Serde<V>,
+//): ProcessorSupplier<K, V, K, V> =
+//  object : ProcessorSupplier<K, V, K, V> {
+//
+//    override fun get(): Processor<K, V, K, V> = DebounceProcessor(inactivityDuration, storeName)
+//
+//    override fun stores(): Set<StoreBuilder<*>> = setOf(
+//      Stores.timestampedKeyValueStoreBuilder(
+//        Stores.persistentTimestampedKeyValueStore(storeName),
+//        keySerde,
+//        valueSerde,
+//      )
+//    )
+//
+//  }
 
 
 //class SuppressProcessor<K, V>(

modules/events-server/src/test/kotlin/dev/adamko/kafkatorio/processor/misc/DebounceProcessorTest.kt

@@ -19,17 +19,17 @@ class DebounceProcessorTest : FunSpec({
   fun blah() {
 
 
-    val topology = Topology()
+//    val topology = Topology()
-
+//
-    topology.addProcessor(
+//    topology.addProcessor(
-      "blah",
+//      "blah",
-      debounceProcessor(
+//      debounceProcessor(
-        1.minutes,
+//        1.minutes,
-        "asdasda",
+//        "asdasda",
-        Serdes.String(),
+//        Serdes.String(),
-        Serdes.String()
+//        Serdes.String()
-      )
+//      )
-    )
+//    )
 
 
 //    val streamsBuilder = StreamsBuilder()