Fix race conditions in SharedObservableRequest

[yatsinar]

Replaces synchronisation with ConcurrentHashMap;

Exposes removeRequest(params) to be invoked by SharedSingleRequest in its own Single lifecycle;

[yatsinar]

This test was previously asserting a buggy behaviour where last emit here would be the 4th
Data(cachedDomain, null, true).

This wasn't the idea as can be seen in the ReloadingDataScannerTest:

newEvent(Data(null, loading = true), shouldEmit = true)
newEvent(Data("a", loading = true), shouldEmit = true)
newEvent(Data("a", loading = false), shouldEmit = true)

newEvent(Data("a", loading = true), shouldEmit = true)
newEvent(Data("a", loading = false), shouldEmit = true)

The changes in this PR made for avoiding race conditions, apparently fixed that, so I've rewritten this test for better clarity.

[Karloid]

I guess we need to call removeRequest at dispose as well (and yes, that removal should be more sophisticated and actually check a correspondence of the deleted request to the current one, or enforce somehow At Most Once behaviour for removeRequest)

there is test which checks that especial issue:

    @Test
    fun `GIVEN network not answer before unsubscription of first observe WHEN observe again THEN second observer see network response`() {
        // given
        whenever(fromNetwork.invoke(any())).thenReturn(Single.never())
        storage.remove(params)
        memCache.remove(params)

        val observer1 = dataObservableDelegate.observe(params = params, forceReload = false)
            .skipWhileLoading()
            .test()
        ioScheduler.triggerActions()

        observer1.dispose()
        observer1.assertValues()

        // when
        whenever(fromNetwork.invoke(any())).thenReturn(Single.just(domain))
        val observer2 = dataObservableDelegate.observe(params = params, forceReload = false)
            .skipWhileLoading()
            .test()
        ioScheduler.triggerActions()

        // then
        observer2.assertValues(Data(domain))    // ACTUAL: EMPTY LIST
    }
    ```

[yatsinar]

We do remove request on dispose, it's handled in SharedObservableRequest class via doFinally { removeRequest(params) }. here we only add extra removals on events that are specific to Single lifecycle - i.e. doOnSuccess, which doesn't exist in Observable lifecycle. Actually even doOnError is redundant.

Your test case fails because DoD has internal timeout logic (60s by default ) and it's keeping Single.never() alive during this period. Observer2 basically resubscribes to this request in this case. If you change it slightly to accomodate for timeouts:

        // when
        whenever(fromNetwork.invoke(any())).thenReturn(Single.just(domain))
        ioScheduler.advanceTimeBy(180, TimeUnit.SECONDS)

        val observer2 = dataObservableDelegate.observe(params = params, forceReload = false)
            .skipWhileLoading()
            .test()
        ioScheduler.triggerActions()

then it passes.

It's covered in WHEN unsubscribed from network and data NOT arrives in 60 seconds THEN data not saved so I won't be adding this case

[Karloid]

got it, thanks