             }
         }
         
+        var canceller: Canceller? = nil
+        _bindToUpstream(upstream, nil, reject, configure, canceller)
+        
         upstream.react(&canceller) { (innerStream: Stream<T>) in
+            lock.lock()
+            unfinishedCount++
+            
             pendingInnerStreams += [innerStream]
-            performRecursively()
+            if pendingInnerStreams.count == 1 {
+                performRecursively()
+            }
+            lock.unlock()
+        }.success { _ -> Void in
+            fulfillIfPossible()
         }
         
     }.name("concatInner")

{
    return Stream<T> { progress, fulfill, reject, configure in
        
        // NOTE: when flattening, don't bind upstream's fulfill/reject with flattened-downstream's fulfill/reject
        var canceller: Canceller? = nil
        _bindToUpstream(upstream, nil, nil, configure, canceller)
        
        upstream.react(&canceller) { (innerStream: Stream<T>) in
            
            _bindToUpstream(innerStream, nil, nil, configure, nil)

            innerStream.react { value in
                progress(value)
            }.then { value, errorInfo -> Void in
                if value != nil {
                    //fulfill() // comment-out: don't fulfill for multiple innerStream progresses
                }
                else if let errorInfo = errorInfo {
                    if let error = errorInfo.error {
                        reject(error)
                    }
                    else {
                        let error = _RKError(.CancelledByInternalStream, "One of stream is cancelled in `mergeInner()`.")
                        reject(error)
                    }
                }
            }
        }
        
    }.name("mergeInner")

{
    return Stream<T> { progress, fulfill, reject, configure in
        
        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
            
            lock.lock()
            unfinishedCount++
            lock.unlock()
            
            currentInnerStream?.cancel()
            currentInnerStream = innerStream
            
            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")
