हम स्टेटफ्लो की कोशिश करते हैं और समझते हैं

public interface StateFlow<out T> : Flow<T> {
    /**
     * The current value of this state flow.
     */
    public val value: T
}

public interface MutableStateFlow<T> : StateFlow<T> {
    /**
     * The current value of this state flow.
     *
     * Setting a value that is [equal][Any.equals] to the previous one does nothing.
     */
    public override var value: T
}

• एक सरल और कचरा-मुक्त आंतरिक कार्यान्वयन।
• एक डिफ़ॉल्ट आइटम की आवश्यकता है। अशक्त भी संभव है।
• रीड-ओनली और रीड-राइट इंटरफेस में पृथक्करण।
• लिंक की तुलना करने के बजाय समानता के माध्यम से तत्वों की तुलना करना।

class StateFlowRepository<T>(initialValue: T? = null) {
    private val stateFlow = MutableStateFlow(initialValue)

    var value: T?
        get() = stateFlow.value
        set(value) {
            stateFlow.value = value
        }

    val stream: Flow<T?> = stateFlow
}

lifecycleScope.launch {
            simpleRepo.stream.collect {
                addData(it.toString())
            }
        }

private class StateFlowImpl<T>(initialValue: Any) : SynchronizedObject(), MutableStateFlow<T>, FusibleFlow<T> {
    private val _state = atomic(initialValue) // T | NULL
    private var sequence = 0 // serializes updates, value update is in process when sequence is odd
    private var slots = arrayOfNulls<StateFlowSlot?>(INITIAL_SIZE)
    private var nSlots = 0 // number of allocated (!free) slots
    private var nextIndex = 0 // oracle for the next free slot index

. . .
}

private val _state = atomic<Any?>(null)

public override var value: T
        get() = NULL.unbox(_state.value)
        set(value) {
            var curSequence = 0
            var curSlots: Array<StateFlowSlot?> = this.slots // benign race, we will not use it
            val newState = value ?: NULL
            synchronized(this) {
                val oldState = _state.value
                if (oldState == newState) return // Don't do anything if value is not changing
                _state.value = newState
                curSequence = sequence
                if (curSequence and 1 == 0) { // even sequence means quiescent state flow (no ongoing update)
                    curSequence++ // make it odd
                    sequence = curSequence
                } else {
                    // update is already in process, notify it, and return
                    sequence = curSequence + 2 // change sequence to notify, keep it odd
                    return
                }
                curSlots = slots // read current reference to collectors under lock
            }
            /*
               Fire value updates outside of the lock to avoid deadlocks with unconfined coroutines
               Loop until we're done firing all the changes. This is sort of simple flat combining that
               ensures sequential firing of concurrent updates and avoids the storm of collector resumes
               when updates happen concurrently from many threads.
             */
            while (true) {
                // Benign race on element read from array
                for (col in curSlots) {
                    col?.makePending()
                }
                // check if the value was updated again while we were updating the old one
                synchronized(this) {
                    if (sequence == curSequence) { // nothing changed, we are done
                        sequence = curSequence + 1 // make sequence even again
                        return // done
                    }
                    // reread everything for the next loop under the lock
                    curSequence = sequence
                    curSlots = slots
                }
            }
        }

1. नया मान सेट करना।
2. अनुक्रम मार्कर को विषम मान पर सेट करने का मतलब है कि हम पहले से ही अपडेट करने की प्रक्रिया में हैं।
3. makePending () - सभी स्लॉट स्टेट्स (यानी, सभी कनेक्शन) को PENDING में सेट करना - हम जल्द ही एक नया मान भेजेंगे।
4. लूप अनुक्रम == उत्सुकता की जांच करता है कि सभी कार्य पूरे हो चुके हैं और अनुक्रम को एक सम संख्या में सेट करता है।

override suspend fun collect(collector: FlowCollector<T>) {
        val slot = allocateSlot()
        var prevState: Any? = null // previously emitted T!! | NULL (null -- nothing emitted yet)
        try {
            // The loop is arranged so that it starts delivering current value without waiting first
            while (true) {
                // Here the coroutine could have waited for a while to be dispatched,
                // so we use the most recent state here to ensure the best possible conflation of stale values
                val newState = _state.value
                // Conflate value emissions using equality
                if (prevState == null || newState != prevState) {
                    collector.emit(NULL.unbox(newState))
                    prevState = newState
                }
                // Note: if awaitPending is cancelled, then it bails out of this loop and calls freeSlot
                if (!slot.takePending()) { // try fast-path without suspending first
                    slot.awaitPending() // only suspend for new values when needed
                }
            }
        } finally {
            freeSlot(slot)
        }
    }

1. हम एक नए कनेक्शन के लिए एक स्लॉट बनाते हैं या उसका पुन: उपयोग करते हैं।
2. हम अशक्त या राज्य परिवर्तन के लिए राज्य की जाँच करते हैं। एमिटीम एक नया अर्थ है।
3. हम जांचते हैं कि क्या अपडेट करने के लिए स्लॉट तैयार हैं (PENDING राज्य) और यदि नहीं, तो हम नए मूल्यों की प्रत्याशा में स्लॉट को निलंबित कर देते हैं।