<commit_before>package com.busymachines.commons.elasticsearch

import org.elasticsearch.Version
import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest
import org.elasticsearch.action.admin.indices.stats.IndexStats
import org.scalastuff.esclient.ActionMagnet
import scala.collection.concurrent.TrieMap
import org.elasticsearch.client.Client
import scala.concurrent.Future
import org.elasticsearch.client.transport.TransportClient
import org.elasticsearch.common.settings.ImmutableSettings
import org.elasticsearch.common.transport.InetSocketTransportAddress
import scala.concurrent.Await
import scala.concurrent.duration._
import org.elasticsearch.action.admin.indices.exists.indices.IndicesExistsRequest
import org.elasticsearch.action.admin.indices.mapping.put.PutMappingRequest
import com.busymachines.commons.Logging
import org.elasticsearch.action.admin.indices.create.CreateIndexRequest
import scala.collection.JavaConversions._

object ESClient extends Logging {
  private val clientsByClusterName = TrieMap[String, ESClient]()

  info("Using ElasticSearch client " + Version.CURRENT)

  def apply(config: ESConfig) = 
    clientsByClusterName.getOrElseUpdate(config.clusterName, {
      new ESClient(
        new TransportClient(ImmutableSettings.settingsBuilder().put("cluster.name", config.clusterName)) {
          addTransportAddresses(config.hostNames.map(new InetSocketTransportAddress(_, config.port)): _*)
        }, config)
    })
}

class ESClient(val javaClient: Client, val config: ESConfig) extends Logging {
  
  def execute[Request, Response](request: Request)(implicit action: ActionMagnet[Request, Response]): Future[Response] =
    action.execute(javaClient, request)

  def executeSync[Request, Response](request: Request)(implicit action: ActionMagnet[Request, Response]): Response =
    Await.result(action.execute(javaClient, request), 30 seconds)

  def indexExists(indexName: String): Boolean =  
    executeSync(new IndicesExistsRequest(indexName)).isExists

  def createIndex(indexName: String) {
    executeSync(new CreateIndexRequest(indexName).settings(
    s"""
       {
        "number_of_shards" : ${config.numberOfShards},
        "number_of_replicas" : ${config.numberOfReplicas},
        "index.mapper.dynamic": false            
      }
      """))
  }

  def dropIndex(indexName: String) =
    if (indexExists(indexName)) {
      executeSync(new DeleteIndexRequest(indexName))
      while (indexExists(indexName)) {
        Thread.sleep(100)
      }
    }

  def listIndexNames: List[String] = {
    javaClient.admin().indices().prepareRecoveries().execute().actionGet.shardResponses.keySet.toList
  }

  def addMapping(indexName: String, typeName: String, mapping: ESMapping[_]) {
    val mappingConfiguration = mapping.mappingDefinition(typeName).toString
    try {
      debug(s"Schema for $indexName/$typeName: $mappingConfiguration")
      executeSync(new PutMappingRequest(indexName).`type`(typeName).source(mappingConfiguration))
    }
    catch {
      case e : Throwable =>
        val msg = s"Invalid schema for $indexName/$typeName: ${e.getMessage} in $mappingConfiguration"
        error(msg, e)
        throw new Exception(msg, e)
    }
  }
}<commit_msg>Attempt to wait for index drop<commit_after>package com.busymachines.commons.elasticsearch

import org.elasticsearch.Version
import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest
import org.elasticsearch.action.admin.indices.stats.IndexStats
import org.scalastuff.esclient.ActionMagnet
import scala.collection.concurrent.TrieMap
import org.elasticsearch.client.Client
import scala.concurrent.Future
import org.elasticsearch.client.transport.TransportClient
import org.elasticsearch.common.settings.ImmutableSettings
import org.elasticsearch.common.transport.InetSocketTransportAddress
import scala.concurrent.Await
import scala.concurrent.duration._
import org.elasticsearch.action.admin.indices.exists.indices.IndicesExistsRequest
import org.elasticsearch.action.admin.indices.mapping.put.PutMappingRequest
import com.busymachines.commons.Logging
import org.elasticsearch.action.admin.indices.create.CreateIndexRequest
import scala.collection.JavaConversions._

object ESClient extends Logging {
  private val clientsByClusterName = TrieMap[String, ESClient]()

  info("Using ElasticSearch client " + Version.CURRENT)

  def apply(config: ESConfig) = 
    clientsByClusterName.getOrElseUpdate(config.clusterName, {
      new ESClient(
        new TransportClient(ImmutableSettings.settingsBuilder().put("cluster.name", config.clusterName)) {
          addTransportAddresses(config.hostNames.map(new InetSocketTransportAddress(_, config.port)): _*)
        }, config)
    })
}

class ESClient(val javaClient: Client, val config: ESConfig) extends Logging {
  
  def execute[Request, Response](request: Request)(implicit action: ActionMagnet[Request, Response]): Future[Response] =
    action.execute(javaClient, request)

  def executeSync[Request, Response](request: Request)(implicit action: ActionMagnet[Request, Response]): Response =
    Await.result(action.execute(javaClient, request), 30 seconds)

  def indexExists(indexName: String): Boolean =  
    executeSync(new IndicesExistsRequest(indexName)).isExists

  def createIndex(indexName: String) {
    executeSync(new CreateIndexRequest(indexName).settings(
    s"""
       {
        "number_of_shards" : ${config.numberOfShards},
        "number_of_replicas" : ${config.numberOfReplicas},
        "index.mapper.dynamic": false            
      }
      """))
  }

  def dropIndex(indexName: String) =
    if (indexExists(indexName)) {
      executeSync(new DeleteIndexRequest(indexName))
      Thread.sleep(2000)
      while (indexExists(indexName)) {
        Thread.sleep(1000)
      }
    }

  def listIndexNames: List[String] = {
    javaClient.admin().indices().prepareRecoveries().execute().actionGet.shardResponses.keySet.toList
  }

  def addMapping(indexName: String, typeName: String, mapping: ESMapping[_]) {
    val mappingConfiguration = mapping.mappingDefinition(typeName).toString
    try {
      debug(s"Schema for $indexName/$typeName: $mappingConfiguration")
      executeSync(new PutMappingRequest(indexName).`type`(typeName).source(mappingConfiguration))
    }
    catch {
      case e : Throwable =>
        val msg = s"Invalid schema for $indexName/$typeName: ${e.getMessage} in $mappingConfiguration"
        error(msg, e)
        throw new Exception(msg, e)
    }
  }
}
{'fraction_non_alphanumeric': 0.10386213263579012, 'fraction_numerical': 0.002308047391906447, 'mean_word_length': 4.681818181818182, 'pattern_counts': {'":': 2, '<': 3, '<?xml version=': 0, '>': 5, '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': '982613', 'n_tokens_mistral': 1813, 'n_tokens_neox': 1798, 'n_words': 328}