anchor.diff:          })
           .then((entity) {
             if (entity != _INTERNAL_ERROR) {
  -            _maybeRecordRpcData(entity);
  +            _recordRpcData(entity);
               active.submit(entity);
             }
           })
  
positive.code:         pending = _delegate.handle(request).single;
          pending
            .catchError((_) {}) // Ignore errors here, they are caught separately below.
            .then((entity) => _handleRpcReply(request, entity));
          _inFlightRequests[request] = pending;
        } else {
          pending = _inFlightRequests[request];
  
negative.code:   }
  
    _handleRpcReply(Request request, Entity entity) {
      if (entity == _INTERNAL_ERROR) {
        // An error occurred, no need to handle it here.
        return;
      }
  
      _recordRpcData(entity);
  
      // Publish this new entity on every channel.
      _activeIndex[request].forEach((act) => runAsync(() => act.submit(entity)));
  