anchor.diff:      return _pipeFuture.then((_) => this);
     }
   
  +  void _completeWriteStreamCompleter([error]) {
  +    if (_writeStreamCompleter == null) return;
  +    var tmp = _writeStreamCompleter;
  +    _writeStreamCompleter = null;
  +    if (error == null) {
  +      _bindSubscription = null;
  +      tmp.complete();
  +    } else {
  +      tmp.completeError(error);
  +    }
  +  }
  +
     StreamController<List<int>> get _controller {
       if (_controllerInstance == null) {
         _controllerInstance = new StreamController<List<int>>(
             onPauseStateChange: _onPauseStateChange,
             onSubscriptionStateChange: _onSubscriptionStateChange);
  -      _pipeFuture = _controller.stream.pipe(_target).then((_) => this);
  +      var future = _controller.stream.pipe(_target);
  +      future.then((_) => _completeWriteStreamCompleter(),
  +                  onError: (error) => _completeWriteStreamCompleter(error));
  +      _pipeFuture = future.then((_) => this);
       }
       return _controllerInstance;
     }
  
positive.code: class _IOSinkImpl<T> implements IOSink<T> {
    final StreamConsumer<List<int>, T> _target;
  
    StreamController<List<int>> _controllerInstance;
    Future<T> _pipeFuture;
    StreamSubscription<List<int>> _bindSubscription;
  
negative.code:         onError: _controller.signalError);
      if (_paused) _pause();
      if (unbind) {
        return _writeStreamCompleter.future;
      } else {
        return _pipeFuture.then((_) => this);
      }
  