anchor.diff:    StreamController _controller;
     StreamSubscription _subscription;
     bool _issuedPause = false;
  +  // Only report error if the last message was a user-provided message and not a
  +  // ping or pong message.
  +  bool _reportError = false;
     Completer _closeCompleter = new Completer();
     Completer _completer;
   
  
positive.code:     _completer = new Completer();
      _subscription = stream.listen(
          (data) {
            _controller.add(data);
          },
          onDone: () {
  
negative.code: 
    void add(data) {
      _ensureController();
      _reportError = false;
      _controller.add(data);
    }
  }
  