Merge pull request #39 from srilumpa/xpack_support

Add XPack authentication support

app/org/elastic4play/database/DBConfiguration.scala

@@ -24,6 +24,7 @@ import com.sksamuel.elastic4s.streams.ReactiveElastic.ReactiveElastic
 import com.sksamuel.elastic4s.streams.{ RequestBuilder, ResponseListener }
 import com.sksamuel.elastic4s.update.{ RichUpdateResponse, UpdateDefinition }
 import com.sksamuel.elastic4s.{ ElasticsearchClientUri, TcpClient }
+import com.sksamuel.elastic4s.xpack.security.XPackElasticClient
 import org.elasticsearch.action.admin.cluster.health.ClusterHealthResponse
 import org.elasticsearch.action.admin.indices.create.CreateIndexResponse
 import org.elasticsearch.action.admin.indices.exists.indices.IndicesExistsResponse
@@ -42,6 +43,8 @@ class DBConfiguration(
     searchHost: Seq[String],
     searchCluster: String,
     baseIndexName: String,
+    xpackUsername: String,
+    xpackPassword: String,
     lifecycle: ApplicationLifecycle,
     val version: Int,
     implicit val ec: ExecutionContext,
@@ -57,6 +60,8 @@ class DBConfiguration(
       configuration.get[Seq[String]]("search.host"),
       configuration.get[String]("search.cluster"),
       configuration.get[String]("search.index"),
+      configuration.get[String]("search.username"),
+      configuration.get[String]("search.password"),
       lifecycle,
       version,
       ec,
@@ -65,12 +70,23 @@ class DBConfiguration(
 
   private[DBConfiguration] lazy val logger = Logger(getClass)
 
+  private def connect(): TcpClient = {
+    var uri = ElasticsearchClientUri(s"elasticsearch://${searchHost.mkString(",")}")
+    var settings = Settings.builder()
+    settings.put("cluster.name", searchCluster)
+    if (xpackUsername.isEmpty) {
+      return TcpClient.transport(settings.build(), uri)
+    }
+    else {
+      settings.put("xpack.security.user", s"${xpackUsername}:${xpackPassword}")
+      return XPackElasticClient(settings.build(), uri)
+    }
+  }
+
   /**
     * Underlying ElasticSearch client
     */
-  private[database] val client = TcpClient.transport(
-    Settings.builder().put("cluster.name", searchCluster).build(),
-    ElasticsearchClientUri(s"elasticsearch://${searchHost.mkString(",")}"))
+  private[database] val client = connect()
   // when application close, close also ElasticSearch connection
   lifecycle.addStopHook { () ⇒ Future { client.close() } }
 
@@ -133,5 +149,5 @@ class DBConfiguration(
   /**
     * return a new instance of DBConfiguration that points to the previous version of the index schema
     */
-  def previousVersion: DBConfiguration = new DBConfiguration(searchHost, searchCluster, baseIndexName, lifecycle, version - 1, ec, actorSystem)
-}
+  def previousVersion: DBConfiguration = new DBConfiguration(searchHost, searchCluster, baseIndexName, xpackUsername, xpackPassword, lifecycle, version - 1, ec, actorSystem)
+}

build.sbt

@@ -13,11 +13,14 @@ lazy val elastic4play = (project in file("."))
 
 scalaVersion := "2.12.4"
 
+resolvers += "elasticsearch-releases" at "https://artifacts.elastic.co/maven"
+
 libraryDependencies ++= Seq(
   cacheApi,
   "com.sksamuel.elastic4s" %% "elastic4s-core" % "5.6.0",
   "com.sksamuel.elastic4s" %% "elastic4s-streams" % "5.6.0",
   "com.sksamuel.elastic4s" %% "elastic4s-tcp" % "5.6.0",
+  "com.sksamuel.elastic4s" %% "elastic4s-xpack-security" % "5.6.0",
   "com.typesafe.akka" %% "akka-stream-testkit" % "2.5.6" % Test,
   "org.scalactic" %% "scalactic" % "3.0.4",
   "org.bouncycastle" % "bcprov-jdk15on" % "1.58",
@@ -42,4 +45,4 @@ scalacOptions in ThisBuild ++= Seq(
   "-Ywarn-value-discard", // Warn when non-Unit expression results are unused
   "-Ywarn-inaccessible",
   "-Ywarn-dead-code"
-)
+)