 {
     return Stream<T> { progress, fulfill, reject, configure in
         
-        var canceller: Canceller? = nil
-        _bindToUpstream(upstream, nil, nil, configure, canceller)
-        
         var currentInnerStream: Stream<T>?
         
+        var unfinishedCount = 1
+        let lock = NSRecursiveLock()
+        
+        let finishIfPossible: (Void -> Void) -> Void -> Void = { finish in
+            return {
+                lock.lock()
+                unfinishedCount--
+                if unfinishedCount == 0 {
+                    finish()
+                }
+                lock.unlock()
+            }
+        }
+        let fulfillIfPossible = finishIfPossible { fulfill() }
+        let rejectIfPossible = finishIfPossible {
+            let error = _RKError(.CancelledByInternalStream, "Last inner stream in `switchLatestInner()` is cancelled.")
+            reject(error)
+        }
+        
+        var canceller: Canceller? = nil
+        _bindToUpstream(upstream, nil, reject, configure, canceller)
+        
         upstream.react(&canceller) { (innerStream: Stream<T>) in
             
-            _bindToUpstream(innerStream, nil, nil, configure, nil)
+            lock.lock()
+            unfinishedCount++
+            lock.unlock()
             
             currentInnerStream?.cancel()
             currentInnerStream = innerStream
             
-            innerStream.react { value in
+            var innerCanceller: Canceller? = nil
+            _bindToUpstream(innerStream, nil, nil, configure, innerCanceller)
+            
+            innerStream.react(&innerCanceller) { value in
                 progress(value)
-            }
+            }.then { value, errorInfo -> Void in
+                if value != nil {
+                    fulfillIfPossible()
+                }
+                else if let errorInfo = errorInfo {
+                    if let error = errorInfo.error {
+                        reject(error)
+                    }
+                    else {
+                        rejectIfPossible()  // will reject if upstream finished & last innerStream is cancelled
+                    }
+                }
+            }
+            
+        }.success { _ -> Void in
+            fulfillIfPossible()
         }
         
     }.name("switchLatestInner")

{
    return Stream<T> { progress, fulfill, reject, configure in
        
        var canceller: Canceller? = nil
        _bindToUpstream(upstream, nil, nil, configure, canceller)
        
        var pendingInnerStreams = [Stream<T>]()
        
        let performRecursively: Void -> Void = _fix { recurse in
            return {
                if let innerStream = pendingInnerStreams.first {
                    if pendingInnerStreams.count == 1 {
                        _bindToUpstream(innerStream, nil, nil, configure, nil)
                        
                        innerStream.react { value in
                            progress(value)
                        }.success {
                            pendingInnerStreams.removeAtIndex(0)
                            recurse()
                        }.failure { errorInfo -> Void in
                            if let error = errorInfo.error {
                                reject(error)
                            }
                            else {
                                let error = _RKError(.CancelledByInternalStream, "One of stream is cancelled in `concatInner()`.")
                                reject(error)
                            }
                        }

{
    return Stream<T> { progress, fulfill, reject, configure in
        
        var unfinishedCount = 1
        let lock = NSRecursiveLock()
        
        let fulfillIfPossible: Void -> Void = {
            lock.lock()
            unfinishedCount--
            if unfinishedCount == 0 {
                fulfill()
            }
            lock.unlock()
        }
        
        var canceller: Canceller? = nil
        _bindToUpstream(upstream, nil, reject, configure, canceller)
        
        upstream.react(&canceller) { (innerStream: Stream<T>) in
            
            lock.lock()
            unfinishedCount++
            lock.unlock()
            
            var innerCanceller: Canceller? = nil
            _bindToUpstream(innerStream, nil, nil, configure, innerCanceller)

            innerStream.react(&innerCanceller) { value in
                progress(value)
            }.then { value, errorInfo -> Void in
                if value != nil {
                    fulfillIfPossible()
                }
                else if let errorInfo = errorInfo {
                    if let error = errorInfo.error {
                        reject(error)
                    }
                    else {
                        let error = _RKError(.CancelledByInternalStream, "One of the inner stream in `mergeInner()` is cancelled.")
                        reject(error)
                    }
                }
            }
        }.success {
            fulfillIfPossible()
        }
        
    }.name("mergeInner")
