Migration to elastic4s 6.3.0

bhermann · bhermann · commit 96652924d7de · 2018-08-25T12:24:47.000+02:00
diff --git a/src/main/scala/de/upb/cs/swt/delphi/crawler/Crawler.scala b/src/main/scala/de/upb/cs/swt/delphi/crawler/Crawler.scala
@@ -18,7 +18,7 @@ package de.upb.cs.swt.delphi.crawler
 
 import akka.actor.ActorSystem
 import akka.stream.ActorMaterializer
-import com.sksamuel.elastic4s.http.{HttpClient}
+import com.sksamuel.elastic4s.http.ElasticClient
 import de.upb.cs.swt.delphi.crawler.control.Server
 import de.upb.cs.swt.delphi.crawler.discovery.maven.MavenCrawlActor
 import de.upb.cs.swt.delphi.crawler.discovery.maven.MavenCrawlActor.Start
@@ -65,7 +65,7 @@ object Crawler extends App with AppLogging {
 
   new Server(configuration.controlServerPort).start()
 
-  val elasticActor = system.actorOf(ElasticActor.props(HttpClient(configuration.elasticsearchClientUri)))
+  val elasticActor = system.actorOf(ElasticActor.props(ElasticClient(configuration.elasticsearchClientUri)))
   val hermesActor = system.actorOf(HermesActor.props(elasticActor))
   val processingDispatchActor = system.actorOf(ProcessingDispatchActor.props(hermesActor))
   val preprocessingDispatchActor = system.actorOf(PreprocessingDispatchActor.props(configuration, processingDispatchActor, elasticActor))
diff --git a/src/main/scala/de/upb/cs/swt/delphi/crawler/Playground.scala b/src/main/scala/de/upb/cs/swt/delphi/crawler/Playground.scala
@@ -18,7 +18,7 @@ package de.upb.cs.swt.delphi.crawler
 
 import akka.actor.{ActorRef, ActorSystem}
 import com.sksamuel.elastic4s.ElasticsearchClientUri
-import com.sksamuel.elastic4s.http.HttpClient
+import com.sksamuel.elastic4s.http.ElasticClient
 import de.upb.cs.swt.delphi.crawler.storage.ElasticActor
 
 /**
@@ -29,7 +29,7 @@ object Playground extends App {
   val system : ActorSystem = ActorSystem("trial")
 
   val clientUri = ElasticsearchClientUri("localhost", 9200)
-  val elastic : ActorRef = system.actorOf(ElasticActor.props(HttpClient(clientUri)), "elastic")
+  val elastic : ActorRef = system.actorOf(ElasticActor.props(ElasticClient(clientUri)), "elastic")
 
   //val maven : ActorRef = system.actorOf(MavenCrawlActor.props(Uri("http://repo1.maven.org/maven2/de/tu-darmstadt/"), elastic), "maven")
 
diff --git a/src/main/scala/de/upb/cs/swt/delphi/crawler/discovery/maven/MavenCrawlActor.scala b/src/main/scala/de/upb/cs/swt/delphi/crawler/discovery/maven/MavenCrawlActor.scala
@@ -18,7 +18,7 @@ package de.upb.cs.swt.delphi.crawler.discovery.maven
 
 import akka.actor.{Actor, ActorLogging, ActorRef, Props}
 import akka.stream.ActorMaterializer
-import com.sksamuel.elastic4s.http.HttpClient
+import com.sksamuel.elastic4s.http.ElasticClient
 import de.upb.cs.swt.delphi.crawler.Configuration
 import de.upb.cs.swt.delphi.crawler.discovery.maven.MavenCrawlActor.Start
 import de.upb.cs.swt.delphi.crawler.storage.ArtifactExistsQuery
@@ -37,7 +37,7 @@ class MavenCrawlActor(configuration: Configuration, nextStep : ActorRef)
       log.info("Starting Maven discovery process...")
 
       implicit val materializer = ActorMaterializer()
-      implicit val client = HttpClient(configuration.elasticsearchClientUri)
+      implicit val client = ElasticClient(configuration.elasticsearchClientUri)
       implicit val logger = log
 
       createSource(configuration.mavenRepoBase)
diff --git a/src/main/scala/de/upb/cs/swt/delphi/crawler/preprocessing/PreprocessingDispatchActor.scala b/src/main/scala/de/upb/cs/swt/delphi/crawler/preprocessing/PreprocessingDispatchActor.scala
@@ -17,22 +17,22 @@
 package de.upb.cs.swt.delphi.crawler.preprocessing
 
 import akka.actor.{Actor, ActorLogging, ActorRef, Props}
-import com.sksamuel.elastic4s.http.HttpClient
-import akka.util.Timeout
 import akka.pattern.ask
+import akka.routing.{BalancingPool, RoundRobinPool}
+import akka.util.Timeout
+import com.sksamuel.elastic4s.http.ElasticClient
 import de.upb.cs.swt.delphi.crawler.Configuration
 import de.upb.cs.swt.delphi.crawler.discovery.maven.MavenIdentifier
-import de.upb.cs.swt.delphi.crawler.storage.ElasticActor
-import akka.routing.{BalancingPool, RoundRobinPool}
 import de.upb.cs.swt.delphi.crawler.processing.CallGraphStream
+import de.upb.cs.swt.delphi.crawler.storage.ElasticActor
 
 import scala.concurrent.duration._
-import scala.util.{Success, Try}
+import scala.util.Success
 
 class PreprocessingDispatchActor(configuration : Configuration, nextStep : ActorRef, elasticActor : ActorRef) extends Actor with ActorLogging {
 
   val elasticPool = context.actorOf(RoundRobinPool(configuration.elasticActorPoolSize)
-    .props(ElasticActor.props(HttpClient(configuration.elasticsearchClientUri))))
+    .props(ElasticActor.props(ElasticClient(configuration.elasticsearchClientUri))))
   val callGraphPool = context.actorOf(BalancingPool(configuration.callGraphStreamPoolSize)
     .props(CallGraphStream.props(configuration)))
 
diff --git a/src/main/scala/de/upb/cs/swt/delphi/crawler/processing/CallGraphStream.scala b/src/main/scala/de/upb/cs/swt/delphi/crawler/processing/CallGraphStream.scala
@@ -10,7 +10,7 @@ import akka.stream.scaladsl.GraphDSL.Implicits._
 import akka.stream.scaladsl.{Flow, GraphDSL, Keep, RunnableGraph, Sink, Source}
 import akka.util.Timeout
 import akka.{Done, NotUsed}
-import com.sksamuel.elastic4s.http.HttpClient
+import com.sksamuel.elastic4s.http.ElasticClient
 import de.upb.cs.swt.delphi.crawler.Configuration
 import de.upb.cs.swt.delphi.crawler.discovery.maven.MavenIdentifier
 import de.upb.cs.swt.delphi.crawler.preprocessing.{JarFile, MavenDownloader, PomFile}
@@ -50,7 +50,7 @@ class CallGraphStream(configuration: Configuration) extends Actor with ActorLogg
   implicit val parallelism = 1
   implicit val ec = ExecutionContext.global
 
-  val esClient = HttpClient(configuration.elasticsearchClientUri)
+  val esClient = ElasticClient(configuration.elasticsearchClientUri)
 
   val mavenDependencyActor = context.actorOf(MavenDependencyActor.props(configuration))
   val opalActor: ActorRef = context.actorOf(OpalActor.props(configuration))
diff --git a/src/main/scala/de/upb/cs/swt/delphi/crawler/processing/ElasticEdgeSearchActor.scala b/src/main/scala/de/upb/cs/swt/delphi/crawler/processing/ElasticEdgeSearchActor.scala
@@ -1,8 +1,9 @@
 package de.upb.cs.swt.delphi.crawler.processing
 
 import akka.actor.{Actor, ActorLogging, Props}
+import com.sksamuel.elastic4s.http.{ElasticClient, Response}
 import com.sksamuel.elastic4s.http.ElasticDsl._
-import com.sksamuel.elastic4s.http.HttpClient
+import com.sksamuel.elastic4s.http.search.MultiSearchResponse
 import de.upb.cs.swt.delphi.crawler.discovery.maven.MavenIdentifier
 import de.upb.cs.swt.delphi.crawler.processing.CallGraphStream.{MappedEdge, unresMCtoStr}
 import org.opalj.ai.analyses.cg.UnresolvedMethodCall
@@ -14,7 +15,7 @@ import org.opalj.ai.analyses.cg.UnresolvedMethodCall
  * method is marked as belonging to that dependency.
  */
 
-class ElasticEdgeSearchActor(client: HttpClient) extends Actor with ActorLogging{
+class ElasticEdgeSearchActor(client: ElasticClient) extends Actor with ActorLogging{
 
   val maxBatchSize = 150
 
@@ -61,13 +62,13 @@ class ElasticEdgeSearchActor(client: HttpClient) extends Actor with ActorLogging
     }
 
     def searchEsDb(calls: List[UnresolvedMethodCall], id: MavenIdentifier): (UnresolvedMethodCall) => Boolean = {
-      val resp = client.execute{
+      val resp: Response[MultiSearchResponse] = client.execute{
         multi(
           for(call <- calls) yield genSearchDef(call, id)
         )
       }.await
 
-      val result = resp.right.get.result.items
+      val result = resp.result.items
       val hitIndices = result.filter(_.response.isRight).filter(_.response.right.get.totalHits > 0).map(_.index)
       val hits = for (i <- hitIndices) yield calls(i)
 
@@ -90,5 +91,5 @@ class ElasticEdgeSearchActor(client: HttpClient) extends Actor with ActorLogging
 }
 
 object ElasticEdgeSearchActor {
-  def props(client: HttpClient) = Props(new ElasticEdgeSearchActor(client))
+  def props(client: ElasticClient) = Props(new ElasticEdgeSearchActor(client))
 }
diff --git a/src/main/scala/de/upb/cs/swt/delphi/crawler/storage/ArtifactExistsQuery.scala b/src/main/scala/de/upb/cs/swt/delphi/crawler/storage/ArtifactExistsQuery.scala
@@ -18,7 +18,7 @@ package de.upb.cs.swt.delphi.crawler.storage
 
 import com.sksamuel.elastic4s.http.ElasticDsl._
 import com.sksamuel.elastic4s.http.search.SearchResponse
-import com.sksamuel.elastic4s.http.{HttpClient, RequestSuccess}
+import com.sksamuel.elastic4s.http.{ElasticClient, RequestSuccess}
 import de.upb.cs.swt.delphi.crawler.discovery.maven.MavenIdentifier
 
 /**
@@ -32,12 +32,12 @@ trait ArtifactExistsQuery {
     * @param client
     * @return
     */
-  def exists(identifier : MavenIdentifier)(implicit client : HttpClient) : Boolean = {
+  def exists(identifier : MavenIdentifier)(implicit client : ElasticClient) : Boolean = {
     client.execute {
       searchWithType(delphiProjectType) query must (
         matchQuery("name", identifier.toUniqueString)
       )
-    }.await.merge match {
+    }.await match {
       case RequestSuccess(_,_,_,SearchResponse(_, false, false, _, _, _, _, hits)) => (hits.total > 0)
       case _ => false
     }
diff --git a/src/main/scala/de/upb/cs/swt/delphi/crawler/storage/ArtifactIdentityQuery.scala b/src/main/scala/de/upb/cs/swt/delphi/crawler/storage/ArtifactIdentityQuery.scala
@@ -16,18 +16,18 @@
 
 package de.upb.cs.swt.delphi.crawler.storage
 
-import com.sksamuel.elastic4s.http.{HttpClient, RequestSuccess}
 import com.sksamuel.elastic4s.http.ElasticDsl._
 import com.sksamuel.elastic4s.http.search.SearchResponse
+import com.sksamuel.elastic4s.http.{ElasticClient, RequestSuccess}
 import de.upb.cs.swt.delphi.crawler.discovery.maven.MavenIdentifier
 
 trait ArtifactIdentityQuery {
-  def elasticId(identifier : MavenIdentifier)(implicit client : HttpClient) : Option[String] = {
+  def elasticId(identifier : MavenIdentifier)(implicit client : ElasticClient) : Option[String] = {
     client.execute {
       searchWithType(delphiProjectType) query must (
         matchQuery("name", identifier.toUniqueString)
       )
-    }.await.merge match {
+    }.await match {
       case RequestSuccess(_,_,_,SearchResponse(_, false, false, _, _, _, _, hits)) => hits.hits.headOption.map { case hit => hit.id }
       case x => None
     }
diff --git a/src/main/scala/de/upb/cs/swt/delphi/crawler/storage/ElasticActor.scala b/src/main/scala/de/upb/cs/swt/delphi/crawler/storage/ElasticActor.scala
@@ -17,9 +17,8 @@
 package de.upb.cs.swt.delphi.crawler.storage
 
 import akka.actor.{Actor, ActorLogging, Props}
-
 import akka.event.LoggingAdapter
-import com.sksamuel.elastic4s.http.HttpClient
+import com.sksamuel.elastic4s.http.ElasticClient
 import de.upb.cs.swt.delphi.crawler.Identifier
 import de.upb.cs.swt.delphi.crawler.discovery.git.GitIdentifier
 import de.upb.cs.swt.delphi.crawler.discovery.maven.MavenIdentifier
@@ -30,8 +29,8 @@ import de.upb.cs.swt.delphi.crawler.processing.HermesResults
   * @param client The currently active elasticsearch client
   * @author Ben Hermann
   */
-class ElasticActor(client: HttpClient) extends Actor with ActorLogging with ArtifactIdentityQuery with ElasticStoreQueries {
-  private implicit val c : HttpClient = client
+class ElasticActor(client: ElasticClient) extends Actor with ActorLogging with ArtifactIdentityQuery with ElasticStoreQueries {
+  private implicit val c : ElasticClient = client
   private implicit val l : LoggingAdapter = log
 
   override def receive: PartialFunction[Any, Unit] = {
@@ -44,7 +43,7 @@ class ElasticActor(client: HttpClient) extends Actor with ActorLogging with Arti
 }
 
 object ElasticActor {
-  def props(client: HttpClient): Props = Props(new ElasticActor(client))
+  def props(client: ElasticClient): Props = Props(new ElasticActor(client))
 
   case class Push(identity: Identifier)
 
diff --git a/src/main/scala/de/upb/cs/swt/delphi/crawler/storage/ElasticCallGraphActor.scala b/src/main/scala/de/upb/cs/swt/delphi/crawler/storage/ElasticCallGraphActor.scala
@@ -1,8 +1,8 @@
 package de.upb.cs.swt.delphi.crawler.storage
 
 import akka.actor.{Actor, ActorLogging, Props}
+import com.sksamuel.elastic4s.http.ElasticClient
 import com.sksamuel.elastic4s.http.ElasticDsl._
-import com.sksamuel.elastic4s.http.HttpClient
 import de.upb.cs.swt.delphi.crawler.discovery.maven.MavenIdentifier
 import de.upb.cs.swt.delphi.crawler.processing.CallGraphStream.MappedEdge
 
@@ -13,7 +13,7 @@ import de.upb.cs.swt.delphi.crawler.processing.CallGraphStream.MappedEdge
  * into a single map (defined in createLibraryMap), then adds these maps to a preexisting document.
  */
 
-class ElasticCallGraphActor(client: HttpClient) extends Actor with ActorLogging {
+class ElasticCallGraphActor(client: ElasticClient) extends Actor with ActorLogging {
   override def receive: Receive = {
     case (i: MavenIdentifier, ex: Set[MappedEdge]) => {
       try {
@@ -48,8 +48,9 @@ class ElasticCallGraphActor(client: HttpClient) extends Actor with ActorLogging
           )
         }.sourceInclude("_id")
       }.await
-      if (resp.isRight) {
-        val hits = resp.right.get.result.hits
+
+      if (resp.isSuccess) {
+        val hits = resp.result.hits
         if (hits.total > 0) {
           hits.hits.head.id
         } else {
@@ -62,7 +63,7 @@ class ElasticCallGraphActor(client: HttpClient) extends Actor with ActorLogging
                 "artifactId" -> id.artifactId,
                 "version" -> id.version))
           }.await
-          resp.right.get.result.id
+          resp.result.id
         }
       } else {
         throw new Exception("Elasticsearch server cannot be reached - call graph for " + id.toString +" lost.")
@@ -91,5 +92,5 @@ class ElasticCallGraphActor(client: HttpClient) extends Actor with ActorLogging
 }
 
 object ElasticCallGraphActor{
-  def props(client: HttpClient) = Props(new ElasticCallGraphActor(client))
+  def props(client: ElasticClient) = Props(new ElasticCallGraphActor(client))
 }
diff --git a/src/main/scala/de/upb/cs/swt/delphi/crawler/storage/ElasticIndexMaintenance.scala b/src/main/scala/de/upb/cs/swt/delphi/crawler/storage/ElasticIndexMaintenance.scala
@@ -18,7 +18,7 @@ package de.upb.cs.swt.delphi.crawler.storage
 
 import akka.actor.ActorSystem
 import com.sksamuel.elastic4s.analyzers.KeywordAnalyzer
-import com.sksamuel.elastic4s.http.HttpClient
+import com.sksamuel.elastic4s.http.ElasticClient
 import com.sksamuel.elastic4s.http.ElasticDsl._
 import de.upb.cs.swt.delphi.crawler.{AppLogging, Configuration}
 
@@ -30,7 +30,7 @@ trait ElasticIndexMaintenance extends AppLogging {
   def createDelphiIndex(configuration: Configuration)(implicit system: ActorSystem): Try[Configuration] = {
     log.warning("Could not find Delphi index. Creating it...")
 
-    val client = HttpClient(configuration.elasticsearchClientUri)
+    val client = ElasticClient(configuration.elasticsearchClientUri)
     val featureList = ElasticFeatureListMapping.getMapAsSeq
 
     val identifierFields = Seq(
diff --git a/src/main/scala/de/upb/cs/swt/delphi/crawler/storage/ElasticIndexPreflightCheck.scala b/src/main/scala/de/upb/cs/swt/delphi/crawler/storage/ElasticIndexPreflightCheck.scala
@@ -19,17 +19,17 @@ package de.upb.cs.swt.delphi.crawler.storage
 import akka.actor.ActorSystem
 import akka.http.scaladsl.model.StatusCodes
 import com.sksamuel.elastic4s.http.ElasticDsl._
-import com.sksamuel.elastic4s.http.{HttpClient, RequestFailure, RequestSuccess}
+import com.sksamuel.elastic4s.http.{ElasticClient, RequestFailure, RequestSuccess}
 import de.upb.cs.swt.delphi.crawler.{Configuration, PreflightCheck}
 
-import scala.concurrent.{Await, ExecutionContext}
 import scala.concurrent.duration.Duration
+import scala.concurrent.{Await, ExecutionContext}
 import scala.util.{Failure, Success, Try}
 
 object ElasticIndexPreflightCheck extends PreflightCheck with ElasticIndexMaintenance {
   override def check(configuration: Configuration)(implicit system: ActorSystem): Try[Configuration] = {
     implicit val ec : ExecutionContext = system.dispatcher
-    lazy val client = HttpClient(configuration.elasticsearchClientUri)
+    lazy val client = ElasticClient(configuration.elasticsearchClientUri)
 
     val f = client.execute {
       indexExists("delphi")
@@ -38,7 +38,7 @@ object ElasticIndexPreflightCheck extends PreflightCheck with ElasticIndexMainte
     }
     val delphiIndexExists = Await.result(f, Duration.Inf)
 
-    delphiIndexExists.merge match {
+    delphiIndexExists match {
       case RequestSuccess(StatusCodes.NotFound.intValue, _, _, _) => createDelphiIndex(configuration)
       case RequestSuccess(_,_,_,_) => {
         isIndexCurrent(configuration) match {
diff --git a/src/main/scala/de/upb/cs/swt/delphi/crawler/storage/ElasticReachablePreflightCheck.scala b/src/main/scala/de/upb/cs/swt/delphi/crawler/storage/ElasticReachablePreflightCheck.scala
@@ -17,8 +17,8 @@
 package de.upb.cs.swt.delphi.crawler.storage
 
 import akka.actor.ActorSystem
+import com.sksamuel.elastic4s.http.ElasticClient
 import com.sksamuel.elastic4s.http.ElasticDsl._
-import com.sksamuel.elastic4s.http.HttpClient
 import de.upb.cs.swt.delphi.crawler.{Configuration, PreflightCheck}
 
 import scala.concurrent.duration._
@@ -40,7 +40,7 @@ object ElasticReachablePreflightCheck extends PreflightCheck {
     */
   override def check(configuration: Configuration)(implicit system: ActorSystem): Try[Configuration] = {
     implicit val ec : ExecutionContext = system.dispatcher
-    lazy val client = HttpClient(configuration.elasticsearchClientUri)
+    lazy val client = ElasticClient(configuration.elasticsearchClientUri)
 
     val f = (client.execute {
       nodeInfo()
diff --git a/src/main/scala/de/upb/cs/swt/delphi/crawler/storage/ElasticStoreQueries.scala b/src/main/scala/de/upb/cs/swt/delphi/crawler/storage/ElasticStoreQueries.scala
@@ -19,7 +19,7 @@ import akka.event.LoggingAdapter
 import com.sksamuel.elastic4s.http.ElasticDsl._
 import com.sksamuel.elastic4s.http.index.IndexResponse
 import com.sksamuel.elastic4s.http.update.UpdateResponse
-import com.sksamuel.elastic4s.http.{HttpClient, Response}
+import com.sksamuel.elastic4s.http.{ElasticClient, Response}
 import de.upb.cs.swt.delphi.crawler.discovery.git.GitIdentifier
 import de.upb.cs.swt.delphi.crawler.discovery.maven.MavenIdentifier
 import de.upb.cs.swt.delphi.crawler.processing.HermesResults
@@ -33,29 +33,29 @@ import de.upb.cs.swt.delphi.crawler.processing.HermesResults
 trait ElasticStoreQueries {
   this: ArtifactIdentityQuery =>
 
-  def store(h: HermesResults)(implicit client: HttpClient, log : LoggingAdapter): Option[Response[UpdateResponse]] = {
+  def store(h: HermesResults)(implicit client: ElasticClient, log : LoggingAdapter): Option[Response[UpdateResponse]] = {
     elasticId(h.identifier) match {
       case Some(id) =>
         log.info(s"Pushing Hermes results for ${h.identifier} under id $id.")
         Some(client.execute {
           update(id).in(delphiProjectType).doc("features" -> h.featureMap)
-        }.await.merge)
+        }.await)
       case None => log.warning(s"Tried to push hermes results for non-existing identifier: ${h.identifier}."); None
     }
   }
 
-  def store(g: GitIdentifier)(implicit client: HttpClient,log : LoggingAdapter): Response[IndexResponse] = {
+  def store(g: GitIdentifier)(implicit client: ElasticClient,log : LoggingAdapter): Response[IndexResponse] = {
     log.info("Pushing new git identifier to elastic: [{}]", g)
     client.execute {
       indexInto(delphiProjectType).fields("name" -> (g.repoUrl + "/" + g.commitId),
         "source" -> "Git",
         "identifier" -> Map(
           "repoUrl" -> g.repoUrl,
           "commitId" -> g.commitId))
-    }.await.merge
+    }.await
   }
 
-  def store(m: MavenIdentifier)(implicit client: HttpClient, log : LoggingAdapter): Response[IndexResponse] = {
+  def store(m: MavenIdentifier)(implicit client: ElasticClient, log : LoggingAdapter): Response[IndexResponse] = {
     log.info("Pushing new maven identifier to elastic: [{}]", m)
     client.execute {
       indexInto(delphiProjectType).fields("name" -> m.toUniqueString,
@@ -64,6 +64,6 @@ trait ElasticStoreQueries {
           "groupId" -> m.groupId,
           "artifactId" -> m.artifactId,
           "version" -> m.version))
-    }.await.merge
+    }.await
   }
 }
