 						observer.onError(error)
 					}
 				}
-				if !complete {
-					subscriptions[index] = disposable
-				}
-				else {
+				if let serialDisposable = subscriptions[index] {
+					serialDisposable.disposable = disposable
+				} else {
 					disposable.dispose()
 				}
 			}

			
			for i in 0 ..< flow.count {
				let index = i
				var complete = false
				let disposable = flow[index].subscribe { event in
					
					lock.lock()
					defer { lock.unlock() }
					

		
		return Observable<T>.create { observer in
			var current = 0, initialized = false
			var subscriptions: [SerialDisposable?] = flow.map { _ in SerialDisposable() }

			let lock = NSRecursiveLock()
			lock.lock()
