anchor.diff:          pending = _inFlightRequests[request];
         }
   
  -      // Make cache request (always).
  -      _cache.get(request)
  -        .catchError(active.sendError)
  -        .then((cachedEntity) {
  -          if (cachedEntity != null) {
  -            active.submit(cachedEntity);
  -          }
  -        });
  -
         // RPC replies are handled in one place, but errors for requests are
         // subscribed to individually. This is because only in-flight requests
         // should have error returns, whereas all streams for a request care when
  
positive.code:           return _INTERNAL_ERROR;
          })
          .then((entity) {
            if (entity != _INTERNAL_ERROR) {
              _recordRpcData(entity);
              active.submit(entity);
            }
          })
          .whenComplete(active.close);
      }
      return active.stream;
    }
  
  
negative.code: 
    Multiplexer(this._delegate, {Cache cache: null})
        : this._cache = cache == null ? new AsyncMapCache() : cache;
    
    _newActiveStream(request) {
      // Create a new stream for this request.
      var active = new _ActiveStream(request);
      active.closed.whenComplete(() => _removeActive(active));
  
      return active;
    }
    
    _handleAgeQuery(request, age) {
      var active = _newActiveStream(request);
      
      _cache.get(request)
        .catchError(active.sendError)
        .then((cachedEntity) {
          // If there actually was an entity response, send it to the client.
          if (cachedEntity != null) {
            active.submit(cachedEntity);
          }
          var ts = new DateTime.now().millisecondsSinceEpoch;
          // If we don't need to issue an rpc
          if (age < 0 || (cachedEntity != null && (ts - cachedEntity.streamy.ts) < age)) {
            if (age == AGE_CACHE_LOOKUP_ONCE) {
              
            }
            return;
          }
          
          _sendRpc(request);
        });
        
        return active.stream;
    }
    
    _sendRpc(request, active) {
      // Only cachable requests need to be handled by the multiplexer (right now).
      if (request.isCachable) {
  
  