Changes publish, replay, replay all to clear state in case of sequence termination.

kzaher · kzaher · commit fd947cbe452e · 2017-10-01T13:14:58.000+02:00
diff --git a/RxSwift/Observables/Multicast.swift b/RxSwift/Observables/Multicast.swift
@@ -60,7 +60,7 @@ extension ObservableType {
     - returns: A connectable observable sequence that shares a single subscription to the underlying sequence.
     */
     public func publish() -> ConnectableObservable<E> {
-        return self.multicast(PublishSubject())
+        return self.multicast { PublishSubject() }
     }
 }
 
@@ -78,7 +78,7 @@ extension ObservableType {
      */
     public func replay(_ bufferSize: Int)
         -> ConnectableObservable<E> {
-        return self.multicast(ReplaySubject.create(bufferSize: bufferSize))
+        return self.multicast { ReplaySubject.create(bufferSize: bufferSize) }
     }
 
     /**
@@ -92,7 +92,7 @@ extension ObservableType {
      */
     public func replayAll()
         -> ConnectableObservable<E> {
-        return self.multicast(ReplaySubject.createUnbounded())
+        return self.multicast { ReplaySubject.createUnbounded() }
     }
 }
 
diff --git a/RxTest/ColdObservable.swift b/RxTest/ColdObservable.swift
@@ -26,14 +26,19 @@ final class ColdObservable<Element>
         
         let i = self.subscriptions.count - 1
 
+        var disposed = false
+
         for recordedEvent in recordedEvents {
             _ = testScheduler.scheduleRelativeVirtual((), dueTime: recordedEvent.time, action: { _ in
-                observer.on(recordedEvent.value)
+                if !disposed {
+                    observer.on(recordedEvent.value)
+                }
                 return Disposables.create()
             })
         }
         
         return Disposables.create {
+            disposed = true
             let existing = self.subscriptions[i]
             self.subscriptions[i] = Subscription(existing.subscribe, self.testScheduler.clock)
         }
diff --git a/Tests/RxSwiftTests/Observable+MulticastTests.swift b/Tests/RxSwiftTests/Observable+MulticastTests.swift

Original file line number	Diff line number	Diff line change
`@@ -60,7 +60,7 @@ extension ObservableType {`
`60`	`60`	`- returns: A connectable observable sequence that shares a single subscription to the underlying sequence.`
`61`	`61`	`*/`
`62`	`62`	`public func publish() -> ConnectableObservable<E> {`
`63`		`- return self.multicast(PublishSubject())`
	`63`	`+ return self.multicast { PublishSubject() }`
`64`	`64`	`}`
`65`	`65`	`}`
`66`	`66`
`@@ -78,7 +78,7 @@ extension ObservableType {`
`78`	`78`	`*/`
`79`	`79`	`public func replay(_ bufferSize: Int)`
`80`	`80`	`-> ConnectableObservable<E> {`
`81`		`- return self.multicast(ReplaySubject.create(bufferSize: bufferSize))`
	`81`	`+ return self.multicast { ReplaySubject.create(bufferSize: bufferSize) }`
`82`	`82`	`}`
`83`	`83`
`84`	`84`	`/**`
`@@ -92,7 +92,7 @@ extension ObservableType {`
`92`	`92`	`*/`
`93`	`93`	`public func replayAll()`
`94`	`94`	`-> ConnectableObservable<E> {`
`95`		`- return self.multicast(ReplaySubject.createUnbounded())`
	`95`	`+ return self.multicast { ReplaySubject.createUnbounded() }`
`96`	`96`	`}`
`97`	`97`	`}`
`98`	`98`