Don't use replay(1) to cache View references

The `replay` operator internally caches N+1 elements instead of just
N, meaning it will keep a strong reference to the previous emitted
element. When this element references an Activity, leaks occur.

See ReactiveX/RxJava#6475.

Author: nhaarman

ext/acorn/acorn-rx/src/main/java/com/nhaarman/acorn/presentation/RxScene.kt

@@ -19,8 +19,6 @@ package com.nhaarman.acorn.presentation
 import androidx.annotation.CallSuper
 import arrow.core.Option
 import arrow.core.toOption
-import com.nhaarman.acorn.presentation.RxScene.ContainerEvent.Attached
-import com.nhaarman.acorn.presentation.RxScene.ContainerEvent.Detached
 import com.nhaarman.acorn.state.SceneState
 import io.reactivex.Observable
 import io.reactivex.disposables.CompositeDisposable
@@ -75,20 +73,13 @@ abstract class RxScene<V : Container>(
 
     private val startedEventsSubject = BehaviorSubject.createDefault(false)
 
-    private val containerEventsSubject = BehaviorSubject.createDefault<ContainerEvent<V>>(Detached)
+    private val viewSubject = BehaviorSubject.createDefault(Option.empty<V>())
 
     /**
      * Publishes a stream of optional [V] instances that are attached to this
      * Scene.
      */
-    protected val view: Observable<Option<V>> = containerEventsSubject
-        .map { event ->
-            when (event) {
-                is Attached<V> -> event.v.toOption()
-                is Detached -> Option.empty()
-            }
-        }
-        .replay(1).autoConnect()
+    protected val view: Observable<Option<V>> = viewSubject.hide()
 
     @CallSuper
     override fun onStart() {
@@ -99,12 +90,12 @@ abstract class RxScene<V : Container>(
     @CallSuper
     override fun attach(v: V) {
         super.attach(v)
-        containerEventsSubject.onNext(Attached(v))
+        viewSubject.onNext(v.toOption())
     }
 
     @CallSuper
     override fun detach(v: V) {
-        containerEventsSubject.onNext(Detached)
+        viewSubject.onNext(Option.empty())
         super.detach(v)
     }
 
@@ -119,12 +110,6 @@ abstract class RxScene<V : Container>(
         sceneDisposables.dispose()
     }
 
-    @Suppress("unused")
-    private sealed class ContainerEvent<out V> {
-        class Attached<V>(val v: V) : ContainerEvent<V>()
-        object Detached : ContainerEvent<Nothing>()
-    }
-
     /**
      * A utility function to automatically subscribe and dispose of source
      * [Observable] instance when this Scene starts and stops.

ext/acorn/acorn-rx/src/test/java/com/nhaarman/acorn/presentation/RxSceneTest.kt

@@ -51,6 +51,18 @@ class RxSceneTest {
         expect(observer.lastValue).toBe(Option.just(testView))
     }
 
+    @Test
+    fun `subscribing after view attach`() {
+        /* Given */
+        scene.attach(testView)
+
+        /* When */
+        val observer = scene.viewObservable.test()
+
+        /* Then */
+        expect(observer.lastValue).toBe(Option.just(testView))
+    }
+
     @Test
     fun `detaching a view notifies observers`() {
         /* Given */