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

		
		return Observable<T>.create { observer in
			var current = 0, initialized = false
			var subscriptions = [Disposable?](repeating: nil, count: flow.count)

			let lock = NSRecursiveLock()
			lock.lock()

							subscriptions[current] = nil
							current += 1
						}
						if index == current {
							assert(subscriptions[index] != nil)
							observer.onNext(element)
						}
						
					case .completed:
						if index >= current {
							if (initialized) {
								subscriptions[index]?.dispose()
