A simple utility for staging updates on a stream of records without holding merged record state

DataUpdatesStaging.kt

private typealias Update<T> = T.() -> Unit

class DataUpdatesStaging<T> {
    private val updateBuffer: MutableList<Update<T>> = ArrayList()
    private val updateStream: BehaviorSubject<List<Update<T>>> = BehaviorSubject.create(updateBuffer)

    fun stageUpdate(update: Update<T>) {
        synchronized(updateBuffer) {
            updateBuffer.add(update)
            updateStream.onNext(updateBuffer)
        }
    }

    fun applyUpdates(recordStream: Observable<T>): Observable<T> = Observable
            .combineLatest(recordStream, updateStream, { rec, updates ->
                updates.forEach { it(rec) }
                rec
            })

    fun discardUpdates() {
        synchronized(updateBuffer) {
            updateBuffer.clear()
        }
    }

    fun finalizeUpdates(record: T?): T? {
        synchronized(updateBuffer) {
            if (record != null) {
                updateBuffer.forEach { it(record) }
                updateBuffer.clear()
            }
        }
        return record
    }
}