anchor.diff:    final int port;
   
     int db;
  -  
  +
   
     RedisReply _currentReply;
   
     void handleDone(EventSink<RedisReply> output) {
   
  +    print('Handling done');
       if (_currentReply != null) {
         var error = new UnexpectedRedisClosureError("Some data has already been sent but was not complete.");
         // Apparently some data has already been sent, but the stream is done.
  
positive.code:   }
  
    Function _subscriptionHandler = null;
    
    Future subscribe(List<String> channels, Function onMessage){
      
      Completer subscribeCompleter = new Completer();
      List<String> args = new List <String>()
          ..add("SUBSCRIBE")
          ..addAll(channels);
      
      send(args).receive().then((val){
        _subscriptionHandler = onMessage; 
        subscribeCompleter.complete();
      });
      return subscribeCompleter.future;
    }
    
    Future unsubscribe(List<String> channels){
      Completer unsubscribeCompleter = new Completer();
      List<String> args = new List <String>()
          ..add("UNSUBSCRIBE")
          ..addAll(channels);
      
      _subscriptionHandler = null;
      send(args).receive().then((val){      
        unsubscribeCompleter.complete();
      });
      return unsubscribeCompleter.future;
    }
     
    /// Handles new data received from stream.
    void _onRedisReply(RedisReply redisReply) {
      logger.fine("Received reply: $redisReply");
      
      if(_subscriptionHandler != null){
        Receiver rec = new Receiver()
        ..reply = redisReply;      
        _subscriptionHandler(rec);
        return;
      }
  
negative.code:         .catchError(_onSocketError);
  
    }
  
    Future auth(String _password) {
      this.password = _password;
      return sendCommand(RedisCommand.AUTH, [ _password ]).receive();
  