attempt-2.kt

//second attempt, now with more code!

fun StyledTextArea<*, *>.asDebouncedTextChangesFlow(): Flow<PropertyChange<String?>>
        = debounceAsTyping(textProperty(), textProperty().asFlow(), focusedProperty())

fun TextInputControl.asDebouncedTextChangesFlow(): Flow<PropertyChange<String?>>
        = debounceAsTyping(textProperty(), textProperty().asFlow(), focusedProperty())

fun <T> Flow<PropertyChange<T>>.debounceAsTyping(source: ObservableValue<T>, focusedProperty: ObservableValue<Boolean> = BooleanConstant.TRUE): Flow<PropertyChange<T>>
        = debounceAsTyping(source, this, focusedProperty)

fun <T> CoroutineScope.launchCollect(flow: Flow<T>, handler: suspend(T) -> Unit){
    var dispatcher: MainCoroutineDispatcher = Dispatchers.JavaFx //overwrite the immediate launcher so we avoid nested event loops + animations
    if(BootstrappingUtilities.isTestingEnvironment) dispatcher = dispatcher.immediate //unless we're testing, in which case run that thing right now.

    launch(dispatcher) {
        flow.flowOn(this.coroutineContext.minusKey(Job)).collect(handler)
    }
}

private object TRANSIENT_PLACEHOLDER

@JvmName("debounceAsTyping_impl") private fun <T> debounceAsTyping(
        source: ObservableValue<T>,
        textFlow: Flow<PropertyChange<T>>,
        focusedProperty: ObservableValue<Boolean> = BooleanConstant.TRUE
): Flow<PropertyChange<T>> {

    var oldValue: Any? = TRANSIENT_PLACEHOLDER

    val focusChange = focusedProperty.asFlow()
            .filter { it.oldValue && ! it.newValue }
            .map { PropertyChange(source, TRANSIENT_PLACEHOLDER, source.value) }

    val textChange = textFlow
            .map {
                require(Platform.isFxApplicationThread())
                if(oldValue == TRANSIENT_PLACEHOLDER) oldValue = it.oldValue
                PropertyChange(it.source, TRANSIENT_PLACEHOLDER, it.newValue)
            }
            .filter {
                focusedProperty.value // must still be focused, else this change is happening while some other element is selected
                && oldValue != TRANSIENT_PLACEHOLDER && oldValue != it.newValue // must have new change to fire...
                        // nope, this isnt good enough. Consider:
                        // 1. user makes change A("1" -> "12")
                        // 2. user clicks away quickly => changePublished, oldValue set null,
                        // 3. user clicks back (A still not published)
                        // 4. user starts typing "123", oldValue set to "123"
                        // 5. 200 ms passes... does the conflated nature mean we're safe? does change A die safely?
                        // Should I encode a "Version" in this state machine, such that focus increments a version?
                        // then we could safely detect A...
            }

    val merged = flowOf(focusChange, textChange).flattenMerge()

    val timeout = fromEnvOrDefault(Long::class.java, "com.empowerops.common.ui.TypingDebounceTime")
            ?: 200L

    val debounced = merged
            .let { if(timeout == 0L || BootstrappingUtilities.isTestingEnvironment) it else it.debounce(timeout) }
            .map { (src, old, new) ->
                check(old is TRANSIENT_PLACEHOLDER)
                val oldActual = oldValue.also { oldValue = TRANSIENT_PLACEHOLDER }
                check(oldActual != TRANSIENT_PLACEHOLDER)
                @Suppress("UNCHECKED_CAST") //only possible way this isn't safe is oldValue, which above line checks for.
                PropertyChange<T>(src as ObservableValue<T>, oldActual as T, new as T)
            }

    return debounced
}

attempt-3.kt

fun StyledTextArea<*, *>.asDebouncedTextChangesFlow(): Flow<PropertyChange<String?>>
        = debounceAsTyping(textProperty().asFlow(), focusedProperty())

fun TextInputControl.asDebouncedTextChangesFlow(): Flow<PropertyChange<String?>>
        = debounceAsTyping(textProperty().asFlow(), focusedProperty())

fun <T> CoroutineScope.launchCollect(flow: Flow<T>, handler: suspend(T) -> Unit){
    var dispatcher: MainCoroutineDispatcher = Dispatchers.JavaFx //overwrite the immediate launcher so we avoid nested event loops + animations
    if(BootstrappingUtilities.isTestingEnvironment) dispatcher = dispatcher.immediate //unless we're testing, in which case run that thing right now.

    launch(dispatcher + CoroutineName("launchCollect($flow)")) {
        flow.flowOn(this.coroutineContext.minusKey(Job)).collect(handler)
    }
}

private object TRANSIENT_PLACEHOLDER

@JvmName("debounceAsTyping_impl") private fun <T> debounceAsTyping(
        textFlow: Flow<PropertyChange<T>>,
        focusedProperty: ObservableValue<Boolean> = BooleanConstant.TRUE
): Flow<PropertyChange<T>> {

    // firstly, im defining this code as out-of-scope for all unit fixtures,
    // which likely call `setText` as a way of typing
    if(BootstrappingUtilities.isTestingEnvironment) {
        return object: Flow<PropertyChange<T>> by textFlow {
            override fun toString() = "$textFlow.[not]debounceAsTyping()"
        }
    }

    // note: we're going to merge in a synthetic event on focus lost.
    // this is in support of this typical use case:
    //
    // 1. the controller is publishing changes to the model via
    //    `thisFlowSource.onDeboucedTextChange { model.stuff = newText; eventBus.post(ModelInvalidated()) }}`
    // 2. the controller is reactive with
    //    `@Subscribe fun rebindViewToModelOn(event: ModelInvalidated) = rebindViewToModel()`
    // 3. rebindViewToModel contains the code
    //    `if( ! thisFlowSource.isFocused && thisFlowSource.text != newValue) thisFlowSource.text = model.stuff`
    //
    // a key point: the `isFocused` guard on rebind view to model prevents updates from
    // _changing the text while the user is editing it_
    //
    // but in order for us to be certain that we're in a consistent state
    // we need to make sure an event goes off when the text field isn't focused,
    // that way the update can simply be ignored by the gaurd, but a later event will ensure what the user is typing
    // is written to the model

    var processionSpan: PropertyChange<T>? = null

    // THIS IS HEAP POLLUTION PLAIN AND SIMPLE:
    val transientPlaceholder: T = TRANSIENT_PLACEHOLDER as T
    // but It makes runtime debugging easier and gives me a shim to insert some assertions about correctness.
    // apologies if your up-stack looking at class-cast exceptions.

    val focusLostFlow = focusedProperty.asFlow()
            .filter { it.oldValue && ! it.newValue } //used to trigger commits => only interested in focus lost events
            .map { processionSpan?.copy(oldValue = transientPlaceholder) }
            //.map { it.also { println("FOCUS-CHANGE: $it") } }
            .filterNotNull() //if its null it means focus toggled but no changes were ever made

    val textChange: Flow<PropertyChange<T>> = textFlow
            .map {
                require(Platform.isFxApplicationThread())
                val oldSpan = processionSpan
                processionSpan = when {
                    oldSpan == null -> it //this is the first change since a commit
                    ! focusedProperty.value -> it //this is a programmatic change => write-through, effectively filtering it out
                    else -> oldSpan.copy(newValue = it.newValue) //else just update the new value for commit
                }
                //println("UPDATE: saw $it\nSTATE: precession now $processionSpan")
                PropertyChange(it.source, transientPlaceholder, it.newValue)
            }
            .filter {
                focusedProperty.value // must still be focused, else this change is happening while some other element is selected
                && processionSpan?.oldValue != it.newValue //filter by new-values only to avoid noise
            }

    val timeout = fromEnvOrDefault(Long::class.java, "com.empowerops.common.ui.TypingDebounceTime")
            ?: 200L

    val debouncedTextChanges: Flow<PropertyChange<T?>> = textChange.debounce(timeout)

    val merged: Flow<PropertyChange<T?>> = flowOf(debouncedTextChanges, focusLostFlow).flattenMerge()

    val interrupted = merged
            .map { (src, old, new) ->
                check(old == transientPlaceholder) { "got intermediate with oldValue=$old" }
                processionSpan.also {
                    check(it?.oldValue != transientPlaceholder) { "publishing a transient value $it" }
                    processionSpan = null
                }
            }
            .filterNotNull() //implies change went off after a commit --possible because of focus lost

//    return interrupted
            //.map { it.also { println("COMMIT: $it, span=$processionSpan") } }

    return object: Flow<PropertyChange<T>> by interrupted {
        override fun toString() = "$interrupted.debounceAsTyping()"
    }
}

first-attempt.kt

private fun CoroutineScope.onDebouncedTextChange(
        textProperty: ObservableValue<String>,
        focusedProperty: ObservableValue<Boolean>,
        handler: suspend (PropertyChange<String?>) -> Unit)
{
    val focusChange = focusedProperty.asFlow()
            .filter { it.oldValue && ! it.newValue }
            .map { PropertyChange(textProperty, textProperty.value, textProperty.value) }

    val textChange = textProperty.asFlow()
            .filter { focusedProperty.value }

    val merged = flowOf(focusChange, textChange).flattenMerge()

    launchCollect(merged.debounce(200), handler)
}