 {
     return Stream<T> { progress, fulfill, reject, configure in
         
-        var canceller: Canceller? = nil
-        _bindToUpstream(upstream, nil, nil, configure, canceller)
-        
         var pendingInnerStreams = [Stream<T>]()
+        
+        var unfinishedCount = 1
+        let lock = NSRecursiveLock()
+        
+        let fulfillIfPossible: Void -> Void = {
+            lock.lock()
+            unfinishedCount--
+            if unfinishedCount == 0 {
+                fulfill()
+            }
+            lock.unlock()
+        }
         
         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 {
+                    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 {
+                            lock.lock()
                             pendingInnerStreams.removeAtIndex(0)
                             recurse()
-                        }.failure { errorInfo -> Void in
+                            lock.unlock()
+                            
+                            fulfillIfPossible()
+                        }
+                        else if let errorInfo = errorInfo {
                             if let error = errorInfo.error {
                                 reject(error)
                             }
                             else {
-                                let error = _RKError(.CancelledByInternalStream, "One of stream is cancelled in `concatInner()`.")
+                                let error = _RKError(.CancelledByInternalStream, "One of the inner stream in `concatInner()` is cancelled.")
                                 reject(error)
                             }
                         }

{
    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")


    // NOTE: downstream should capture upstream
    if let downstreamConfigure = downstreamConfigure {
        let oldPause = downstreamConfigure.pause
        downstreamConfigure.pause = {
            oldPause?()
            upstream.pause()
        }
        
        let oldResume = downstreamConfigure.resume
        downstreamConfigure.resume = {
            oldResume?()
            upstream.resume()
        }
        
        // NOTE: `configure.cancel()` is always called on downstream-finish
        let oldCancel = downstreamConfigure.cancel
        downstreamConfigure.cancel = {
            oldCancel?()
            
            let canceller = reactCanceller()
            canceller?.cancel()
            upstream.cancel()
