//设置package包名称以及导入依赖的类
package firecrest.guice

import java.net.InetAddress
import javax.inject.Singleton

import com.google.inject.{AbstractModule, Provides}
import firecrest.config.ElasticSearchConfig
import org.elasticsearch.client.support.AbstractClient
import org.elasticsearch.client.transport.TransportClient
import org.elasticsearch.common.transport.InetSocketTransportAddress

class ElasticSearchClientModule extends AbstractModule {

  @Provides
  @Singleton
  def client(esConfig: ElasticSearchConfig): AbstractClient = {
    val esClient: TransportClient = TransportClient.builder().build()
      .addTransportAddress(
        new InetSocketTransportAddress(
          InetAddress.getByName(esConfig.host),
          esConfig.port))
    esClient
  }

  override def configure(): Unit = {}
} 

//设置package包名称以及导入依赖的类
package sinks

import java.net.InetAddress
import java.util

import com.typesafe.config.ConfigFactory
import org.apache.flink.api.common.functions.RuntimeContext
import org.apache.flink.streaming.connectors.elasticsearch.{IndexRequestBuilder, ElasticsearchSink}
import org.elasticsearch.action.index.IndexRequest
import org.elasticsearch.client.Requests
import org.elasticsearch.common.transport.{InetSocketTransportAddress, TransportAddress}
import org.joda.time.DateTime

trait ElasticsearchSinks {

  val appConfiguration = ConfigFactory.load()

  val esConfiguration = new util.HashMap[String, String]
  esConfiguration.put("bulk.flush.max.actions", "1")
  esConfiguration.put("cluster.name", appConfiguration.getString("ElasticSearchCluster"))

  val esTransports = new util.ArrayList[TransportAddress]
  esTransports.add(new InetSocketTransportAddress(InetAddress.getByName(appConfiguration.getString("ElasticSearchHost")), appConfiguration.getInt("ElasticSearchPort")))

  val esPopularLanguagesSink = new ElasticsearchSink(esConfiguration, esTransports, new IndexRequestBuilder[(DateTime, String,Int)] {
    override def createIndexRequest(element: (DateTime,String,Int), ctx: RuntimeContext): IndexRequest = {
      val json = new util.HashMap[String, AnyRef]
      json.put("timestamp", element._1.asInstanceOf[AnyRef])
      json.put("language", element._2.asInstanceOf[AnyRef])
      json.put("cnt", element._3.asInstanceOf[AnyRef])
      Requests.indexRequest.index("twitter").`type`("popular-languages").source(json)
    }
  })

} 

//设置package包名称以及导入依赖的类
package it.agilelab.bigdata.wasp.testApp

import it.agilelab.bigdata.wasp.core.WaspSystem
import it.agilelab.bigdata.wasp.core.elastic.ElasticAdminActor
import it.agilelab.bigdata.wasp.core.logging.WaspLogger
import it.agilelab.bigdata.wasp.core.utils.ElasticConfiguration
import it.agilelab.bigdata.wasp.web.Global
import org.elasticsearch.client.transport.TransportClient
import org.elasticsearch.common.settings.ImmutableSettings
import org.elasticsearch.common.transport.InetSocketTransportAddress
import org.elasticsearch.index.query.FilterBuilders._
import org.elasticsearch.index.query.QueryBuilders
import org.elasticsearch.index.query.QueryBuilders._
import play.api.libs.json.Json


object ElasticApp extends App with WaspSystem with ElasticConfiguration{

    def loggerActorProps = Global.loggerActorProps
    val logger = WaspLogger("ElasticApp")

    val settings = ImmutableSettings.settingsBuilder().put("cluster.name", elasticConfig.cluster_name).build();

    val transportClient = new TransportClient(settings)

    for (connection <- elasticConfig.connections) {
      val address = new InetSocketTransportAddress(connection.host, ElasticAdminActor.port)
      transportClient.addTransportAddress(address)
    }

  println("connected")



  val filter = termFilter(
    "SECURITYID",
    "DE0008404005"
  );

  var res = transportClient.prepareSearch("orderpipeline_index")
                  //.setSearchType(SearchType.)
                  .setQuery(QueryBuilders.boolQuery().must(matchQuery(
                                                            "SECURITYID",
                                                            "DE0008404005")
                  ).must(rangeQuery("LASTPX")))
                  //.setPostFilter(filter)
                  .setExplain(true)
                  .execute().actionGet()
  println(res.toString)
  for(hit <- res.getHits.hits()){
    val json = Json.parse(hit.source)
    //println((json \ "SECURITYID").as[String])
    println(json)
  }
  //println(res)

} 

//设置package包名称以及导入依赖的类
package com.mnubo.flink.streaming.connectors.elasticsearch

import com.mnubo.flink.streaming.connectors._
import org.elasticsearch.action.admin.cluster.health.ClusterHealthStatus
import org.elasticsearch.client.transport.TransportClient
import org.elasticsearch.common.transport.InetSocketTransportAddress

class Elasticsearch17InputFormatSpec extends ElasticsearchInputFormatSpec {
  protected override def elasticSearchVersion = "1.7.5"

  private def createClientInternal(host: String = es.host, port: Int = es.esTransportPort) =
    new TransportClient()
      .addTransportAddress(new InetSocketTransportAddress(host, port))

  protected override def isClusterGreen(host: String, port: Int) =
    using(createClientInternal(host, port))(c => c.admin.cluster.prepareHealth().get.getStatus == ClusterHealthStatus.GREEN)

  protected override def createClient: ESClient = new ESClient {
    private val internalClient = createClientInternal()

    override def createIndex(indexName: String, docType: String, mapping: String) =
      internalClient
        .admin
        .indices
        .prepareCreate(indexName)
        .addMapping(docType, mapping)
        .get

    override def flush(indexName: String) =
      internalClient
        .admin
        .indices
        .prepareFlush(indexName)
        .setForce(true)
        .get


    override def index(indexName: String, docType: String, doc: String) =
      internalClient
        .prepareIndex(indexName, docType)
        .setSource(doc)
        .get

    override def close() =
      internalClient.close()
  }
} 

//设置package包名称以及导入依赖的类
package com.mnubo.flink.streaming.connectors.elasticsearch

import java.net.InetAddress

import com.mnubo.flink.streaming.connectors._
import org.elasticsearch.client.transport.TransportClient
import org.elasticsearch.cluster.health.ClusterHealthStatus
import org.elasticsearch.common.transport.InetSocketTransportAddress

class Elasticsearch23InputFormatSpec extends ElasticsearchInputFormatSpec {
  protected override def elasticSearchVersion = "2.3.3"

  private def createClientInternal(host: String = es.host, port: Int = es.esTransportPort) =
    TransportClient
      .builder()
      .build()
      .addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName(host), port))

  protected override def isClusterGreen(host: String, port: Int) =
    using(createClientInternal(host, port))(c => c.admin.cluster.prepareHealth().get.getStatus == ClusterHealthStatus.GREEN)

  protected override def createClient: ESClient = new ESClient {
    private val internalClient = createClientInternal()

    override def createIndex(indexName: String, docType: String, mapping: String) =
      internalClient
        .admin
        .indices
        .prepareCreate(indexName)
        .addMapping(docType, mapping)
        .get

    override def flush(indexName: String) =
      internalClient
        .admin
        .indices
        .prepareFlush(indexName)
        .setForce(true)
        .get


    override def index(indexName: String, docType: String, doc: String) =
      internalClient
        .prepareIndex(indexName, docType)
        .setSource(doc)
        .get

    override def close() =
      internalClient.close()
  }
} 

//设置package包名称以及导入依赖的类
package houseprices.elasticsearch.config

import org.elasticsearch.client.Client
import org.elasticsearch.client.transport.TransportClient
import org.elasticsearch.common.transport.InetSocketTransportAddress
import org.elasticsearch.node.NodeBuilder
import org.elasticsearch.common.settings.Settings
import java.net.InetAddress
import org.elasticsearch.node.Node

trait EsClientBuilder {
  this: EsConfig =>

  val settings = Settings.settingsBuilder()
    .put("http.enabled", this.httpEnabled)
    .put("path.home", this.pathHome)
    .put("path.data", this.pathData)
    .put("node.local", this.isLocal)
    .put("node.client", this.isClient)
    .put("node.name", "node." + this.clusterName)

  def createNode: Node = {
    NodeBuilder.nodeBuilder()
      .clusterName(clusterName)
      .settings(settings)
      .build()
  }

  def build: Client = {
    createNode.start().client()
  }
}

object EsClientBuilder {
  def buildClient(env: String = "dev"): Client = {
    env match {
      case "qa" => (new EsClientBuilder with QaConfig) build
      case _ => (new EsClientBuilder with DevConfig) build
    }
  }

  def transportClient = {
    val settings = Settings.settingsBuilder().put("cluster.name", "dev.pricepaid")
    TransportClient.builder().settings(settings).build().addTransportAddress(new InetSocketTransportAddress(InetAddress.getLocalHost, 9300))
  }

} 

//设置package包名称以及导入依赖的类
package com.github.jparkie.spark.elasticsearch.transport

import com.github.jparkie.spark.elasticsearch.conf.SparkEsTransportClientConf
import org.apache.spark.Logging
import org.elasticsearch.client.Client
import org.elasticsearch.client.transport.TransportClient
import org.elasticsearch.common.settings.Settings
import org.elasticsearch.common.transport.InetSocketTransportAddress

import scala.collection.mutable

private[elasticsearch] trait SparkEsTransportClientManager extends Serializable with Logging {
  @transient
  private[transport] val internalTransportClients = mutable.HashMap.empty[SparkEsTransportClientConf, TransportClient]

  private[transport] def buildTransportSettings(clientConf: SparkEsTransportClientConf): Settings = {
    val esSettingsBuilder = Settings.builder()

    clientConf.transportSettings foreach { currentSetting =>
      esSettingsBuilder.put(currentSetting._1, currentSetting._2)
    }

    esSettingsBuilder.build()
  }

  private[transport] def buildTransportClient(clientConf: SparkEsTransportClientConf, esSettings: Settings): TransportClient = {
    import SparkEsTransportClientConf._

    val esClient = TransportClient.builder()
      .settings(esSettings)
      .build()

    getTransportAddresses(clientConf.transportAddresses, clientConf.transportPort) foreach { inetSocketAddress =>
      esClient.addTransportAddresses(new InetSocketTransportAddress(inetSocketAddress))
    }

    sys.addShutdownHook {
      logInfo("Closed Elasticsearch Transport Client.")

      esClient.close()
    }

    logInfo(s"Connected to the following Elasticsearch nodes: ${esClient.connectedNodes()}.")

    esClient
  }

  
  def closeTransportClient(clientConf: SparkEsTransportClientConf): Unit = synchronized {
    internalTransportClients.remove(clientConf) match {
      case Some(transportClient) =>
        transportClient.close()
      case None =>
        logError(s"No TransportClient for $clientConf.")
    }
  }
}

object SparkEsTransportClientManager extends SparkEsTransportClientManager 

//设置package包名称以及导入依赖的类
package io.github.chenfh5.java_api

import java.net.InetAddress

import org.elasticsearch.client.transport.TransportClient
import org.elasticsearch.common.settings.Settings
import org.elasticsearch.common.transport.InetSocketTransportAddress
import org.slf4j.LoggerFactory

import io.github.chenfh5.common.OwnConfigReader


object EsClient {
  private val LOG = LoggerFactory.getLogger(getClass.getName)

  private val clusterName = OwnConfigReader.getOwnProperty.clusterName
  private val ips = OwnConfigReader.getOwnProperty.ips
  private val port = OwnConfigReader.getOwnProperty.javaClientPort.toInt

  private val transportClient = {
    val settings = Settings
        .builder()
        .put("client.transport.sniff", true)
        .put("cluster.name", clusterName)
        .build()

    val transportClient = addIps(TransportClient.builder().settings(settings).build())
    LOG.info("this is the transportClient={}, initialized successfully", transportClient)
    transportClient
  }

  private def addIps(transportClient: TransportClient) = {
    val ipArray = ips.split(",")
    for (oneIP <- ipArray) {
      transportClient
          .addTransportAddresses(new InetSocketTransportAddress(InetAddress.getByName(oneIP.trim), port))
    }
    transportClient
  }

  def getEsClient = this.transportClient

  def getBuckClient = {
    this.transportClient.prepareBulk()
  }

} 

//设置package包名称以及导入依赖的类
package orm.elasticsearch

import java.net.InetAddress
import java.util.concurrent.RejectedExecutionException

import org.elasticsearch.client.Client
import org.elasticsearch.client.transport.TransportClient
import org.elasticsearch.common.settings.Settings
import org.elasticsearch.common.transport.InetSocketTransportAddress
import webby.api.Application
import webby.mvc.AppStub

trait ElasticPluginConf extends webby.api.Plugin {

  private var _client: Client = _

  val clusterName: String
  val httpHost: InetAddress
  val httpPort: Int
  def transportPort: Int = 9300

  def client: Client = {
    if (_client == null) synchronized {
      if (_client == null) _client = startClient()
    }
    _client
  }

  override def onStart() {
    if (AppStub.isRealProd(app)) _client = startClient()
  }

  override def onStop() {
    if (_client != null) {
      try {
        _client.close()
      } catch {
        case e: RejectedExecutionException => webby.api.Logger(getClass).info("ElasticSearch bug on shutdown: " + e.toString)
      }
    }
  }

  protected def app: Application

  protected def startClient(): Client = {
    TransportClient.builder()
      .settings(Settings.builder().put("cluster.name", clusterName))
      .build()
      .addTransportAddress(new InetSocketTransportAddress(httpHost, transportPort))
  }
} 

//设置package包名称以及导入依赖的类
package teleporter.integration.component.elasticsearch

import java.net.InetAddress

import org.elasticsearch.action.index.IndexRequest
import org.elasticsearch.action.update.UpdateRequest
import org.elasticsearch.client.transport.TransportClient
import org.elasticsearch.common.settings.Settings
import org.elasticsearch.common.transport.InetSocketTransportAddress
import org.elasticsearch.index.VersionType
import org.scalatest.FunSuite


class ElasticSearch2$Test extends FunSuite {
  test("elasticsearch") {
    val settings = Settings.settingsBuilder()
      .put("cluster.name", "rube-es").build()
    val client = TransportClient.builder().settings(settings).build()
      .addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName("localhost"), 9300))
    val indexRequest = new IndexRequest()
    val updateRequest = new UpdateRequest()
    client.prepareIndex("test_mapping3", "teststop", "9")
      .setSource("""{"trade_from": "eeeeeeeeeeeggg"}""")
      .setVersionType(VersionType.EXTERNAL)
      .setVersion(System.currentTimeMillis())
      .execute().get()
  }
} 

//设置package包名称以及导入依赖的类
package com.zjlp.face.bigdata.utils

import java.net.InetAddress

import org.elasticsearch.client.transport.TransportClient
import org.elasticsearch.common.settings.Settings
import org.elasticsearch.common.transport.InetSocketTransportAddress

object EsUtils {
  def getEsClient(clusterName: String, hosts: String, port: Int = 9300): TransportClient = {
    val settings: Settings = Settings.settingsBuilder.put("cluster.name", clusterName).put("client.transport.sniff", true).build
    val client: TransportClient = TransportClient.builder.settings(settings).build
    hosts.split(",").foreach {
      host =>
        client.addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName(host), port))

    }
    return client
  }
}