class RepositoryDoc implements ApiOperation {

  @Override
  public List<GenericJson> execute(IndexingService service)
      throws IOException, InterruptedException {
    List<ListenableFuture<? extends GenericJson>> futures = new ArrayList<>();
    // Extract original item name here. service.indexItem() call below mutates item.name value by
    // adding data source prefix as well as escaping unsupported chars. Use original item name to
    // create fragments below to ensure consistent encoding is applied to fragments as well.
    String originalItemName = item.getName();
    ListenableFuture<Operation> operation =
        content == null
            ? service.indexItem(item, requestMode)
            : service.indexItemAndContent(item, content, contentHash, contentFormat, requestMode);

    futures.add(
        Futures.catchingAsync(
            operation,
            IOException.class,
            /**
             * Push failed index request into the queue if a backend error is returned, a failed
             * future is always returned so that this operation is not treated as if it succeeded.
             */
            (AsyncFunction<IOException, Item>)
                ex -> {
                  logger.log(Level.WARNING, "Error indexing the item " + item, ex);
                  Optional<RepositoryError> error = getRepositoryErrorForResponseException(ex);
                  if (!error.isPresent()) {
                    return Futures.immediateFailedFuture(ex);
                  }
                  logger.log(Level.INFO, "Pushing this failed item to queue " + item);
                  return Futures.transformAsync(
                      service.push(
                          item.getName(),
                          new PushItem()
                              .setQueue(item.getQueue())
                              .setType("REPOSITORY_ERROR")
                              .setRepositoryError(error.get())
                              .encodePayload(item.decodePayload())),
                      input -> Futures.<Item>immediateFailedFuture(ex),
                      MoreExecutors.directExecutor());
                },
            MoreExecutors.directExecutor()));

    for (Map.Entry<String, PushItem> entry : childIds.entrySet()) {
      futures.add(service.push(entry.getKey(), entry.getValue()));
    }

    if (!fragments.isEmpty()) {
      for (Map.Entry<String, Acl> fragment : fragments.entrySet()) {
        Acl fragmentAcl = fragment.getValue();
        Item fragmentItem = fragmentAcl.createFragmentItemOf(originalItemName, fragment.getKey());
        futures.add(service.indexItem(fragmentItem, requestMode));
      }
    }

    try {
      return Futures.allAsList(futures).get();
    } catch (ExecutionException e) {
      throw new IOException(e.getCause());
    }
  }

  private  RepositoryDoc(Builder builder);

}