<commit_before>package io.georocket.storage.indexed;

import java.util.ArrayDeque;
import java.util.Queue;

import org.bson.types.ObjectId;

import io.georocket.constants.AddressConstants;
import io.georocket.storage.ChunkMeta;
import io.georocket.storage.IndexMeta;
import io.georocket.storage.Store;
import io.georocket.storage.StoreCursor;
import io.vertx.core.AsyncResult;
import io.vertx.core.Future;
import io.vertx.core.Handler;
import io.vertx.core.Vertx;
import io.vertx.core.eventbus.ReplyException;
import io.vertx.core.json.JsonArray;
import io.vertx.core.json.JsonObject;

/**
 * An abstract base class for chunk stores that are backed by an indexer
 * @author <NAME>
 */
public abstract class IndexedStore implements Store {
  private static final int PAGE_SIZE = 100;
  
  private final Vertx vertx;
  
  /**
   * Constructs the chunk store
   * @param vertx the Vert.x instance
   */
  public IndexedStore(Vertx vertx) {
    this.vertx = vertx;
  }
  
  @Override
  public void add(String chunk, ChunkMeta chunkMeta, String path,
      IndexMeta indexMeta, Handler<AsyncResult<Void>> handler) {
    doAddChunk(chunk, path, ar -> {
      if (ar.failed()) {
        handler.handle(Future.failedFuture(ar.cause()));
      } else {
        // start indexing
        JsonObject indexMsg = new JsonObject()
            .put("path", ar.result())
            .put("meta", chunkMeta.toJsonObject())
            .put("correlationId", indexMeta.getCorrelationId())
            .put("filename", indexMeta.getFromFile())
            .put("importTime", indexMeta.getImportTimeStamp().getTime());

        if (indexMeta != null && indexMeta.getTags() != null) {
          indexMsg.put("tags", new JsonArray(indexMeta.getTags()));
        }
        if (indexMeta != null && indexMeta.getFallbackCRSString() != null) {
          indexMsg.put("fallbackCRSString", indexMeta.getFallbackCRSString());
        }
        vertx.eventBus().publish(AddressConstants.INDEXER_ADD, indexMsg);
        
        // tell sender that writing was successful
        handler.handle(Future.succeededFuture());
      }
    });
  }

  @Override
  public void delete(String search, String path, Handler<AsyncResult<Void>> handler) {
    new IndexedStoreCursor(vertx, PAGE_SIZE, search, path).start(ar -> {
      if (ar.failed()) {
        Throwable cause = ar.cause();
        if (cause instanceof ReplyException) {
          // Cast to get access to the failure code<commit_msg>Only send to one indexer<commit_after>package io.georocket.storage.indexed;

import java.util.ArrayDeque;
import java.util.Queue;

import org.bson.types.ObjectId;

import io.georocket.constants.AddressConstants;
import io.georocket.storage.ChunkMeta;
import io.georocket.storage.IndexMeta;
import io.georocket.storage.Store;
import io.georocket.storage.StoreCursor;
import io.vertx.core.AsyncResult;
import io.vertx.core.Future;
import io.vertx.core.Handler;
import io.vertx.core.Vertx;
import io.vertx.core.eventbus.ReplyException;
import io.vertx.core.json.JsonArray;
import io.vertx.core.json.JsonObject;

/**
 * An abstract base class for chunk stores that are backed by an indexer
 * @author <NAME>
 */
public abstract class IndexedStore implements Store {
  private static final int PAGE_SIZE = 100;
  
  private final Vertx vertx;
  
  /**
   * Constructs the chunk store
   * @param vertx the Vert.x instance
   */
  public IndexedStore(Vertx vertx) {
    this.vertx = vertx;
  }
  
  @Override
  public void add(String chunk, ChunkMeta chunkMeta, String path,
      IndexMeta indexMeta, Handler<AsyncResult<Void>> handler) {
    doAddChunk(chunk, path, ar -> {
      if (ar.failed()) {
        handler.handle(Future.failedFuture(ar.cause()));
      } else {
        // start indexing
        JsonObject indexMsg = new JsonObject()
            .put("path", ar.result())
            .put("meta", chunkMeta.toJsonObject())
            .put("correlationId", indexMeta.getCorrelationId())
            .put("filename", indexMeta.getFromFile())
            .put("importTime", indexMeta.getImportTimeStamp().getTime());

        if (indexMeta != null && indexMeta.getTags() != null) {
          indexMsg.put("tags", new JsonArray(indexMeta.getTags()));
        }
        if (indexMeta != null && indexMeta.getFallbackCRSString() != null) {
          indexMsg.put("fallbackCRSString", indexMeta.getFallbackCRSString());
        }
        vertx.eventBus().send(AddressConstants.INDEXER_ADD, indexMsg);
        
        // tell sender that writing was successful
        handler.handle(Future.succeededFuture());
      }
    });
  }

  @Override
  public void delete(String search, String path, Handler<AsyncResult<Void>> handler) {
    new IndexedStoreCursor(vertx, PAGE_SIZE, search, path).start(ar -> {
      if (ar.failed()) {
        Throwable cause = ar.cause();
        if (cause instanceof ReplyException) {
          // Cast to get access to the failure code
{'fraction_non_alphanumeric': 0.11986997155627793, 'fraction_numerical': 0.001219016659894352, 'mean_word_length': 3.1614539306847, 'pattern_counts': {'":': 0, '<': 13, '<?xml version=': 0, '>': 17, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '536235', 'n_tokens_mistral': 1494, 'n_tokens_neox': 1467, 'n_words': 339}