Fix: Factory closure lifetime differs between .create and .deferred, …

…subscriptions to the latter keep its factory closure alive #2533 (#2534)

RxSwift/Observables/Deferred.swift

@@ -22,18 +22,16 @@ extension ObservableType {
 }
 
 final private class DeferredSink<Source: ObservableType, Observer: ObserverType>: Sink<Observer>, ObserverType where Source.Element == Observer.Element {
-    typealias Element = Observer.Element 
-
-    private let observableFactory: () throws -> Source
-
-    init(observableFactory: @escaping () throws -> Source, observer: Observer, cancel: Cancelable) {
-        self.observableFactory = observableFactory
+    typealias Element = Observer.Element
+    typealias Parent = Deferred<Source>
+
+    override init(observer: Observer, cancel: Cancelable) {
         super.init(observer: observer, cancel: cancel)
     }
 
-    func run() -> Disposable {
+    func run(_ parent: Parent) -> Disposable {
         do {
-            let result = try self.observableFactory()
+            let result = try parent.observableFactory()
             return result.subscribe(self)
         }
         catch let e {
@@ -60,16 +58,16 @@ final private class DeferredSink<Source: ObservableType, Observer: ObserverType>
 final private class Deferred<Source: ObservableType>: Producer<Source.Element> {
     typealias Factory = () throws -> Source
 
-    private let observableFactory : Factory
+    let observableFactory : Factory
 
     init(observableFactory: @escaping Factory) {
         self.observableFactory = observableFactory
     }
 
     override func run<Observer: ObserverType>(_ observer: Observer, cancel: Cancelable) -> (sink: Disposable, subscription: Disposable)
              where Observer.Element == Source.Element {
-        let sink = DeferredSink(observableFactory: self.observableFactory, observer: observer, cancel: cancel)
-        let subscription = sink.run()
+        let sink = DeferredSink<Source, Observer>(observer: observer, cancel: cancel)
+        let subscription = sink.run(self)
         return (sink: sink, subscription: subscription)
     }
 }

Tests/RxSwiftTests/Observable+Tests.swift

@@ -228,6 +228,78 @@ extension ObservableTest {
     }
 }
 
+// MARK: - Deferred
+extension ObservableTest {
+    func testDeferredFactoryClosureLifetime() {
+        class Foo {
+            let expectation: XCTestExpectation
+
+            init(expectation: XCTestExpectation) {
+                self.expectation = expectation
+            }
+
+            func bar() -> Observable<Void> {
+                Observable<Void>
+                    .deferred {
+                        self.expectation.fulfill()
+                        return .never()
+                    }
+            }
+        }
+
+        let factoryClosureInvoked = expectation(description: "Factory closure has been invoked")
+        var foo: Foo? = Foo(expectation: factoryClosureInvoked)
+        weak var initialFoo = foo
+
+        let disposable = foo?.bar().subscribe()
+
+        wait(for: [factoryClosureInvoked])
+
+        // reset foo to let the initial instance deallocate
+        foo = nil
+
+        // we know that the factory closure has already been executed,
+        // and the foo reference has been nilled, so there should be nothing
+        // keeping the object alive
+        XCTAssertNil(initialFoo)
+
+        disposable?.dispose()
+    }
+
+    func testObservableFactoryClosureLifetime() {
+        class Foo {
+            let expectation: XCTestExpectation
+
+            init(expectation: XCTestExpectation) {
+                self.expectation = expectation
+            }
+
+            func bar() -> Observable<Void> {
+                Observable<Void>
+                    .create { _ in
+                        self.expectation.fulfill()
+                        return Disposables.create()
+                    }
+            }
+        }
+
+        let factoryClosureInvoked = expectation(description: "Factory closure has been invoked")
+        var foo: Foo? = Foo(expectation: factoryClosureInvoked)
+        weak var initialFoo = foo
+
+        let disposable = foo?.bar().subscribe()
+
+        wait(for: [factoryClosureInvoked])
+
+        // reset foo to let the initial instance deallocate
+        foo = nil
+
+        XCTAssertNil(initialFoo)
+
+        disposable?.dispose()
+    }
+}
+
 private class TestObject: NSObject {
     var id = UUID()
 }