Caching causes duplication of page impressions and does not cache

[hsssystems]

We observed a lot of impressions in influxdb that have no series assigned. Debugging the issue showed:

1. The caching algorithim is implemented with a concat in class OpencastClient and this effectively injects two responses from Opencast, one from cache and one uncached

  public Flowable<Response<ResponseBody>> getRequest(final String organization, final String episodeId) {
    final Flowable<Response<ResponseBody>> requestUncached = getRequestUncached(organization, episodeId);
    if (this.cache == null)
      return requestUncached;
    final CacheKey cacheKey = new CacheKey(organization, episodeId);
    return Flowable.concat(Util.nullableToFlowable(this.cache.getIfPresent(cacheKey)),
                           requestUncached.doOnNext(response -> addToCache(cacheKey,
                                                                           response,
                                                                           organization,
                                                                           episodeId)));

}

2. Then, the first response from cache (this.cache.getIfPresent(cacheKey))) has a null body, since it can only be read once https://square.github.io/okhttp/3.x/okhttp/

The response body can be consumed only once.

This class may be used to stream very large responses. For example, it is possible to use this class to read a response that is larger than the entire memory allocated to the current process. It can even stream a response larger than the total storage on the current device, which is a common requirement for video streaming applications.

Because this class does not buffer the full response in memory, the application may not re-read the bytes of the response. Use this one shot to read the entire response into memory with bytes() or string(). Or stream the response with either source(), byteStream(), or charStream().

And this happens while the response is first read in OpencastUtils

  private static Flowable<String> checkResponseCode(
          final Logger logger,
          final Response<? extends ResponseBody> x,
          final String organization,
          final String episodeId) {
    final boolean correctResponse = x.code() / 200 == 1;
    if (!correctResponse) {
      logger.error("OCHTTPERROR, episode {}, organization {}: code, {}", x.code(), episodeId, organization);
    } else {
      logger.debug("OCHTTPSUCCESS, episode {}, organization {}", episodeId, organization);
    }
    return correctResponse ?
            Flowable.fromCallable(() -> Objects.requireNonNull(x.body()).string()) :
            Flowable.error(new InvalidOpencastResponse(x.code()));
  }

Finally this results in a duplication of impressions, one with series and one without (null body from cache). And there is nothing cached, each series lookup causes a call to the external API of Opencast

PR with a fix follows.