RxJava Disposable that can hold onto multiple other Disposables and only keeps the last Disposable added with the same key.

HashCompositeDisposable.kt

package com.example

import io.reactivex.rxjava3.disposables.Disposable
import io.reactivex.rxjava3.exceptions.CompositeException
import io.reactivex.rxjava3.exceptions.Exceptions
import io.reactivex.rxjava3.internal.util.ExceptionHelper

/**
 * A disposable that can hold onto multiple other Disposables
 * and only keeps the last Disposable added with the same key.
 *
 * Note: Existing Disposables will be disposed if a new one is added with the same key.
 */
class HashCompositeDisposable : Disposable {

    private var resources: HashMap<Long, Disposable>? = null

    @Volatile
    private var disposed = false

    override fun dispose() {
        if (disposed) {
            return
        }
        var map: HashMap<Long, Disposable>?
        synchronized(this) {
            if (disposed) {
                return
            }
            disposed = true
            map = resources
            resources = null
        }
        dispose(map)
    }

    override fun isDisposed(): Boolean =
        disposed

    fun containsKey(key: Long): Boolean =
        resources?.containsKey(key) ?: false

    fun size(): Int =
        resources?.size ?: 0

    fun add(key: Long, disposable: Disposable): Boolean {
        if (!disposed) {
            synchronized(this) {
                if (!disposed) {
                    var map = resources
                    if (map == null) {
                        map = hashMapOf()
                        resources = map
                    } else if (map.containsKey(key)) {
                        map[key]?.dispose()
                        map.remove(key)
                    }
                    map[key] = disposable
                    return true
                }
            }
        }
        disposable.dispose()
        return false
    }

    fun remove(key: Long): Boolean {
        if (disposed) {
            return false
        }
        synchronized(this) {
            if (disposed) {
                return false
            }
            val map = resources
            if (map?.containsKey(key) != null) {
                map[key]?.dispose()
                return map.remove(key) != null
            }
        }
        return false
    }

    fun remove(disposable: Disposable): Boolean {
        if (disposed) {
            return false
        }
        synchronized(this) {
            if (disposed) {
                return false
            }
            val map = resources
            map?.entries?.find { it.value == disposable }?.let { entry ->
                entry.value.dispose()
                return map.remove(entry.key) != null
            }
        }
        return false
    }

    fun clear() {
        if (disposed) {
            return
        }
        var map: HashMap<Long, Disposable>?
        synchronized(this) {
            if (disposed) {
                return
            }
            map = resources
            resources = null
        }
        dispose(map)
    }

    private fun dispose(map: HashMap<Long, Disposable>?) {
        val disposables = map?.values ?: return
        val errors = mutableListOf<Throwable>()
        disposables.forEach { disposable ->
            try {
                disposable.dispose()
            } catch (ex: Throwable) {
                Exceptions.throwIfFatal(ex)
                errors.add(ex)
            }
        }
        if (errors.isNotEmpty()) {
            if (errors.size == 1) {
                throw ExceptionHelper.wrapOrThrow(errors[0])
            }
            throw CompositeException(errors)
        }
    }
}