A basic RxMap

RxMap.kt

/**
 * A reactive map implementation that allows you to observe a key and its value.
 */
class RxMap<K, V>(
        private val map: AbstractMap<K, V> = HashMap(),
        private val rxMap: AbstractMap<K, Relay<V, V>> = HashMap()
) : Map<K, V> {

    /**
     * @return a hot observable that emits the value for the specified key. When you subscribe,
     * the current value in the map for the key will be emitted.
     */
    @Synchronized fun observe(key: K): Observable<V> {
        return rxMap.getOrPut(key, createObservable(key)
        )
    }

    private fun createObservable(key: K): () -> Relay<V, V> {
        return {
            BehaviorRelay.create<V>().toSerialized()
        }
    }

    override fun containsKey(key: K): Boolean {
        return map.containsKey(key)
    }

    override fun containsValue(value: V): Boolean {
        return map.containsValue(value)
    }

    override val entries: Set<Map.Entry<K, V>> = map.entries

    override fun get(key: K): V? {
        return map[key]
    }

    override fun isEmpty(): Boolean {
        return map.isEmpty()
    }

    override val keys: Set<K> = map.keys
    override val size: Int = map.size
    override val values: Collection<V> = map.values

    @Synchronized fun put(key: K, value: V) {
        map.put(key, value)
        rxMap.getOrPut(key, createObservable(key)).call(value)
    }

    @Synchronized fun put(mapEntry: MapEntry<K, V>) {
        put(mapEntry.key, mapEntry.value)
    }

    @Synchronized fun remove(key: K): V? {
        val removed = map.remove(key)
        if (removed != null) {
            rxMap[key]!!.call(null)
        }

        return removed

    }

    /**
     * Preferred over [Map.Entry] because we get equals()/hashCode() for free
     */
    data class MapEntry<out K, out V>(val key: K, val value: V)
}

RxMapTest.kt

/**
 * Test for [RxMap]
 */
class RxMapTest {
    lateinit var map: RxMap<String, Int>
    lateinit var subscriber: TestSubscriber<Int?>
    lateinit var expected: MutableList<Int?>
    @Before
    fun setUp() {
        map = RxMap(HashMap())
        subscriber = TestSubscriber()
        expected = mutableListOf<Int?>()
    }

    @Test
    fun verifyMapEmitsOnInsert() {
        val emission1 = MapEntry(key = "5", value = 10)
        val emission2 = MapEntry(key = emission1.key, value = 42)


        assertEquals(emission1, emission1)
        assertNotEquals(emission1, emission2)

        map.put(emission1)
        expected.add(emission1.value)


        map.observe(emission1.key)
                .subscribe(subscriber)

        assertExpected()

        map.remove(emission1.key)
        expected.add(null)

        assertExpected()

        map.put(emission2)
        expected.add(emission2.value)

        assertExpected()

        assertThat(subscriber)
                .hasNoTerminalEvent()
    }

    private fun assertExpected() {
        assertThat(subscriber)
                .hasReceivedValues(expected)
    }
}