Addressing code review comments

Signed-off-by: Andriy Redko <andriy.redko@aiven.io>
akka · Apr 6, 2022 · 753c587 · 753c587
1 parent 43dbd3c
commit 753c587
Show file tree

Hide file tree

Showing 24 changed files with 178 additions and 168 deletions.
diff --git a/elasticsearch/src/main/java/akka/stream/alpakka/elasticsearch/ApiVersion.java b/elasticsearch/src/main/java/akka/stream/alpakka/elasticsearch/ApiVersion.java
@@ -4,7 +4,7 @@
 
 package akka.stream.alpakka.elasticsearch;
 
-public enum ApiVersion implements akka.stream.alpakka.common.ApiVersion {
+public enum ApiVersion implements akka.stream.alpakka.elasticsearch.ApiVersionBase {
   V5,
   V7
 }
diff --git a/...kka/stream/alpakka/common/ApiVersion.java → ...alpakka/elasticsearch/ApiVersionBase.java b/...kka/stream/alpakka/common/ApiVersion.java → ...alpakka/elasticsearch/ApiVersionBase.java
@@ -2,11 +2,11 @@
  * Copyright (C) since 2016 Lightbend Inc. <https://www.lightbend.com>
  */
 
-package akka.stream.alpakka.common;
+package akka.stream.alpakka.elasticsearch;
 
 /**
  * Common interface to represent Opensearch / Elasticsearch versions. 
  */
-public interface ApiVersion {
+public interface ApiVersionBase {
 
 }
diff --git a/elasticsearch/src/main/java/akka/stream/alpakka/opensearch/ApiVersion.java b/elasticsearch/src/main/java/akka/stream/alpakka/opensearch/ApiVersion.java
@@ -4,6 +4,6 @@
 
 package akka.stream.alpakka.opensearch;
 
-public enum ApiVersion implements akka.stream.alpakka.common.ApiVersion {
+public enum ApiVersion implements akka.stream.alpakka.elasticsearch.ApiVersionBase {
   V1
 }
diff --git a/elasticsearch/src/main/scala/akka/stream/alpakka/common/WriteSettings.scala b/elasticsearch/src/main/scala/akka/stream/alpakka/common/WriteSettings.scala
diff --git a/...search/src/main/scala/akka/stream/alpakka/elasticsearch/ElasticsearchSourceSettings.scala b/...search/src/main/scala/akka/stream/alpakka/elasticsearch/ElasticsearchSourceSettings.scala
@@ -6,8 +6,6 @@ package akka.stream.alpakka.elasticsearch
 
 import java.util.concurrent.TimeUnit
 
-import akka.stream.alpakka.common.SourceSettings
-
 import scala.concurrent.duration.FiniteDuration
 
 /**
@@ -18,14 +16,18 @@ final class ElasticsearchSourceSettings private (connection: ElasticsearchConnec
                                                  bufferSize: Int,
                                                  includeDocumentVersion: Boolean,
                                                  scrollDuration: FiniteDuration,
-                                                 apiVersion: ApiVersion) 
-                                        extends SourceSettings[ApiVersion, ElasticsearchSourceSettings](connection, bufferSize, includeDocumentVersion, scrollDuration, apiVersion)  {
+                                                 apiVersion: ApiVersion)
+    extends SourceSettingsBase[ApiVersion, ElasticsearchSourceSettings](connection,
+                                                                        bufferSize,
+                                                                        includeDocumentVersion,
+                                                                        scrollDuration,
+                                                                        apiVersion) {
 
   protected override def copy(connection: ElasticsearchConnectionSettings,
-                   bufferSize: Int,
-                   includeDocumentVersion: Boolean,
-                   scrollDuration: FiniteDuration,
-                   apiVersion: ApiVersion): ElasticsearchSourceSettings =
+                              bufferSize: Int,
+                              includeDocumentVersion: Boolean,
+                              scrollDuration: FiniteDuration,
+                              apiVersion: ApiVersion): ElasticsearchSourceSettings =
     new ElasticsearchSourceSettings(connection = connection,
                                     bufferSize = bufferSize,
                                     includeDocumentVersion = includeDocumentVersion,

diff --git a/...csearch/src/main/scala/akka/stream/alpakka/elasticsearch/ElasticsearchWriteSettings.scala b/...csearch/src/main/scala/akka/stream/alpakka/elasticsearch/ElasticsearchWriteSettings.scala
@@ -4,8 +4,6 @@
 
 package akka.stream.alpakka.elasticsearch
 
-import akka.stream.alpakka.common.WriteSettings
-
 import akka.util.JavaDurationConverters._
 
 import scala.concurrent.duration._
@@ -66,15 +64,20 @@ final class ElasticsearchWriteSettings private (connection: ElasticsearchConnect
                                                 retryLogic: RetryLogic,
                                                 versionType: Option[String],
                                                 apiVersion: ApiVersion,
-                                                allowExplicitIndex: Boolean) 
-                                       extends WriteSettings[ApiVersion, ElasticsearchWriteSettings](connection, bufferSize, retryLogic, versionType, apiVersion, allowExplicitIndex) {
+                                                allowExplicitIndex: Boolean)
+    extends WriteSettingsBase[ApiVersion, ElasticsearchWriteSettings](connection,
+                                                                      bufferSize,
+                                                                      retryLogic,
+                                                                      versionType,
+                                                                      apiVersion,
+                                                                      allowExplicitIndex) {
 
   protected override def copy(connection: ElasticsearchConnectionSettings,
-                   bufferSize: Int,
-                   retryLogic: RetryLogic,
-                   versionType: Option[String],
-                   apiVersion: ApiVersion,
-                   allowExplicitIndex: Boolean): ElasticsearchWriteSettings =
+                              bufferSize: Int,
+                              retryLogic: RetryLogic,
+                              versionType: Option[String],
+                              apiVersion: ApiVersion,
+                              allowExplicitIndex: Boolean): ElasticsearchWriteSettings =
     new ElasticsearchWriteSettings(connection, bufferSize, retryLogic, versionType, apiVersion, allowExplicitIndex)
 
   override def toString: String =
@@ -89,6 +92,7 @@ final class ElasticsearchWriteSettings private (connection: ElasticsearchConnect
 }
 
 object ElasticsearchWriteSettings {
+
   /** Scala API */
   def apply(connection: ElasticsearchConnectionSettings): ElasticsearchWriteSettings =
     new ElasticsearchWriteSettings(connection, 10, RetryNever, None, ApiVersion.V7, allowExplicitIndex = true)

diff --git a/...tream/alpakka/common/SourceSettings.scala → ...ka/elasticsearch/SourceSettingsBase.scala b/...tream/alpakka/common/SourceSettings.scala → ...ka/elasticsearch/SourceSettingsBase.scala
@@ -2,7 +2,7 @@
  * Copyright (C) since 2016 Lightbend Inc. <https://www.lightbend.com>
  */
 
-package akka.stream.alpakka.common
+package akka.stream.alpakka.elasticsearch
 
 import akka.util.JavaDurationConverters._
 import java.util.concurrent.TimeUnit
@@ -15,12 +15,13 @@ import scala.concurrent.duration.FiniteDuration
  * Configure Elastiscsearch/Opensearch sources.
  *
  */
-abstract class SourceSettings[Version <: ApiVersion, S <: SourceSettings[Version, S]] private[alpakka] (
-                                                 val connection: ElasticsearchConnectionSettings,
-                                                 val bufferSize: Int,
-                                                 val includeDocumentVersion: Boolean,
-                                                 val scrollDuration: FiniteDuration,
-                                                 val apiVersion: Version) { this: S =>
+abstract class SourceSettingsBase[Version <: ApiVersionBase, S <: SourceSettingsBase[Version, S]] private[alpakka] (
+    val connection: ElasticsearchConnectionSettings,
+    val bufferSize: Int,
+    val includeDocumentVersion: Boolean,
+    val scrollDuration: FiniteDuration,
+    val apiVersion: Version
+) { this: S =>
   def withConnection(value: ElasticsearchConnectionSettings): S = copy(connection = value)
 
   def withBufferSize(value: Int): S = copy(bufferSize = value)
@@ -55,9 +56,9 @@ abstract class SourceSettings[Version <: ApiVersion, S <: SourceSettings[Version
   }
 
   protected def copy(connection: ElasticsearchConnectionSettings = connection,
-                   bufferSize: Int = bufferSize,
-                   includeDocumentVersion: Boolean = includeDocumentVersion,
-                   scrollDuration: FiniteDuration = scrollDuration,
-                   apiVersion: Version = apiVersion): S;
+                     bufferSize: Int = bufferSize,
+                     includeDocumentVersion: Boolean = includeDocumentVersion,
+                     scrollDuration: FiniteDuration = scrollDuration,
+                     apiVersion: Version = apiVersion): S;
 
 }
diff --git a/elasticsearch/src/main/scala/akka/stream/alpakka/elasticsearch/WriteSettingsBase.scala b/elasticsearch/src/main/scala/akka/stream/alpakka/elasticsearch/WriteSettingsBase.scala
@@ -0,0 +1,42 @@
+/*
+ * Copyright (C) since 2016 Lightbend Inc. <https://www.lightbend.com>
+ */
+
+package akka.stream.alpakka.elasticsearch
+
+import akka.stream.alpakka.elasticsearch.ElasticsearchConnectionSettings
+import akka.stream.alpakka.elasticsearch.RetryLogic
+
+/**
+ * Configure Elasticsearch/Opensearch sinks and flows.
+ */
+abstract class WriteSettingsBase[Version <: ApiVersionBase, W <: WriteSettingsBase[Version, W]] private[alpakka] (
+    val connection: ElasticsearchConnectionSettings,
+    val bufferSize: Int,
+    val retryLogic: RetryLogic,
+    val versionType: Option[String],
+    val apiVersion: Version,
+    val allowExplicitIndex: Boolean
+) { this: W =>
+
+  def withConnection(value: ElasticsearchConnectionSettings): W = copy(connection = value)
+
+  def withBufferSize(value: Int): W = copy(bufferSize = value)
+
+  def withRetryLogic(value: RetryLogic): W =
+    copy(retryLogic = value)
+
+  def withVersionType(value: String): W = copy(versionType = Option(value))
+
+  def withApiVersion(value: Version): W =
+    if (apiVersion == value) this else copy(apiVersion = value)
+
+  def withAllowExplicitIndex(value: Boolean): W = copy(allowExplicitIndex = value)
+
+  protected def copy(connection: ElasticsearchConnectionSettings = connection,
+                     bufferSize: Int = bufferSize,
+                     retryLogic: RetryLogic = retryLogic,
+                     versionType: Option[String] = versionType,
+                     apiVersion: Version = apiVersion,
+                     allowExplicitIndex: Boolean = allowExplicitIndex): W;
+}
diff --git a/.../src/main/scala/akka/stream/alpakka/elasticsearch/impl/ElasticsearchSimpleFlowStage.scala b/.../src/main/scala/akka/stream/alpakka/elasticsearch/impl/ElasticsearchSimpleFlowStage.scala
@@ -9,7 +9,6 @@ import akka.http.scaladsl.HttpExt
 import akka.http.scaladsl.model.Uri.Path
 import akka.http.scaladsl.model._
 import akka.http.scaladsl.unmarshalling.Unmarshal
-import akka.stream.alpakka.common.WriteSettings
 import akka.stream.alpakka.elasticsearch._
 import akka.stream.stage._
 import akka.stream._
@@ -25,7 +24,7 @@ import scala.concurrent.{ExecutionContext, Future}
 @InternalApi
 private[elasticsearch] final class ElasticsearchSimpleFlowStage[T, C](
     elasticsearchParams: ElasticsearchParams,
-    settings: WriteSettings[_, _],
+    settings: WriteSettingsBase[_, _],
     writer: MessageWriter[T]
 )(implicit http: HttpExt, mat: Materializer, ec: ExecutionContext)
     extends GraphStage[

diff --git a/...arch/src/main/scala/akka/stream/alpakka/elasticsearch/impl/ElasticsearchSourceStage.scala b/...arch/src/main/scala/akka/stream/alpakka/elasticsearch/impl/ElasticsearchSourceStage.scala
@@ -9,8 +9,7 @@ import akka.http.scaladsl.HttpExt
 import akka.http.scaladsl.model.Uri.Path
 import akka.http.scaladsl.model._
 import akka.http.scaladsl.unmarshalling.Unmarshal
-import akka.stream.alpakka.common.SourceSettings
-import akka.stream.alpakka.elasticsearch.{ApiVersion, ElasticsearchParams, ReadResult}
+import akka.stream.alpakka.elasticsearch.{ApiVersion, ElasticsearchParams, ReadResult, SourceSettingsBase}
 import akka.stream.stage.{GraphStage, GraphStageLogic, OutHandler, StageLogging}
 import akka.stream.{Attributes, Materializer, Outlet, SourceShape}
 import spray.json.DefaultJsonProtocol._
@@ -46,7 +45,7 @@ private[elasticsearch] trait MessageReader[T] {
 private[elasticsearch] final class ElasticsearchSourceStage[T](
     elasticsearchParams: ElasticsearchParams,
     searchParams: Map[String, String],
-    settings: SourceSettings[_, _],
+    settings: SourceSettingsBase[_, _],
     reader: MessageReader[T]
 )(implicit http: HttpExt, mat: Materializer, ec: ExecutionContext)
     extends GraphStage[SourceShape[ReadResult[T]]] {
@@ -72,7 +71,7 @@ object ElasticsearchSourceStage {
 private[elasticsearch] final class ElasticsearchSourceLogic[T](
     elasticsearchParams: ElasticsearchParams,
     searchParams: Map[String, String],
-    settings: SourceSettings[_, _],
+    settings: SourceSettingsBase[_, _],
     out: Outlet[ReadResult[T]],
     shape: SourceShape[ReadResult[T]],
     reader: MessageReader[T]

diff --git a/...icsearch/src/main/scala/akka/stream/alpakka/elasticsearch/javadsl/ElasticsearchFlow.scala b/...icsearch/src/main/scala/akka/stream/alpakka/elasticsearch/javadsl/ElasticsearchFlow.scala
@@ -6,7 +6,6 @@ package akka.stream.alpakka.elasticsearch.javadsl
 
 import akka.NotUsed
 import akka.annotation.ApiMayChange
-import akka.stream.alpakka.common.WriteSettings
 import akka.stream.alpakka.elasticsearch.{scaladsl, _}
 import com.fasterxml.jackson.databind.ObjectMapper
 
@@ -28,7 +27,7 @@ object ElasticsearchFlow {
    */
   def create[T](
       elasticsearchParams: ElasticsearchParams,
-      settings: WriteSettings[_, _],
+      settings: WriteSettingsBase[_, _],
       objectMapper: ObjectMapper
   ): akka.stream.javadsl.Flow[WriteMessage[T, NotUsed], WriteResult[T, NotUsed], NotUsed] =
     create(elasticsearchParams, settings, new JacksonWriter[T](objectMapper))
@@ -44,7 +43,7 @@ object ElasticsearchFlow {
    */
   def create[T](
       elasticsearchParams: ElasticsearchParams,
-      settings: WriteSettings[_, _],
+      settings: WriteSettingsBase[_, _],
       messageWriter: MessageWriter[T]
   ): akka.stream.javadsl.Flow[WriteMessage[T, NotUsed], WriteResult[T, NotUsed], NotUsed] =
     scaladsl.ElasticsearchFlow
@@ -63,7 +62,7 @@ object ElasticsearchFlow {
    */
   def createWithPassThrough[T, C](
       elasticsearchParams: ElasticsearchParams,
-      settings: WriteSettings[_, _],
+      settings: WriteSettingsBase[_, _],
       objectMapper: ObjectMapper
   ): akka.stream.javadsl.Flow[WriteMessage[T, C], WriteResult[T, C], NotUsed] =
     createWithPassThrough(elasticsearchParams, settings, new JacksonWriter[T](objectMapper))
@@ -80,7 +79,7 @@ object ElasticsearchFlow {
    */
   def createWithPassThrough[T, C](
       elasticsearchParams: ElasticsearchParams,
-      settings: WriteSettings[_, _],
+      settings: WriteSettingsBase[_, _],
       messageWriter: MessageWriter[T]
   ): akka.stream.javadsl.Flow[WriteMessage[T, C], WriteResult[T, C], NotUsed] =
     scaladsl.ElasticsearchFlow
@@ -100,7 +99,7 @@ object ElasticsearchFlow {
    */
   def createBulk[T, C](
       elasticsearchParams: ElasticsearchParams,
-      settings: WriteSettings[_, _],
+      settings: WriteSettingsBase[_, _],
       objectMapper: ObjectMapper
   ): akka.stream.javadsl.Flow[java.util.List[WriteMessage[T, C]], java.util.List[WriteResult[T, C]], NotUsed] =
     createBulk(elasticsearchParams, settings, new JacksonWriter[T](objectMapper))
@@ -118,7 +117,7 @@ object ElasticsearchFlow {
    */
   def createBulk[T, C](
       elasticsearchParams: ElasticsearchParams,
-      settings: WriteSettings[_, _],
+      settings: WriteSettingsBase[_, _],
       messageWriter: MessageWriter[T]
   ): akka.stream.javadsl.Flow[java.util.List[WriteMessage[T, C]], java.util.List[WriteResult[T, C]], NotUsed] =
     akka.stream.scaladsl
@@ -143,7 +142,7 @@ object ElasticsearchFlow {
   @ApiMayChange
   def createWithContext[T, C](
       elasticsearchParams: ElasticsearchParams,
-      settings: WriteSettings[_, _],
+      settings: WriteSettingsBase[_, _],
       objectMapper: ObjectMapper
   ): akka.stream.javadsl.FlowWithContext[WriteMessage[T, NotUsed], C, WriteResult[T, C], C, NotUsed] =
     createWithContext(elasticsearchParams, settings, new JacksonWriter[T](objectMapper))
@@ -160,7 +159,7 @@ object ElasticsearchFlow {
   @ApiMayChange
   def createWithContext[T, C](
       elasticsearchParams: ElasticsearchParams,
-      settings: WriteSettings[_, _],
+      settings: WriteSettingsBase[_, _],
       messageWriter: MessageWriter[T]
   ): akka.stream.javadsl.FlowWithContext[WriteMessage[T, NotUsed], C, WriteResult[T, C], C, NotUsed] =
     scaladsl.ElasticsearchFlow

diff --git a/...icsearch/src/main/scala/akka/stream/alpakka/elasticsearch/javadsl/ElasticsearchSink.scala b/...icsearch/src/main/scala/akka/stream/alpakka/elasticsearch/javadsl/ElasticsearchSink.scala
@@ -6,7 +6,6 @@ package akka.stream.alpakka.elasticsearch.javadsl
 
 import java.util.concurrent.CompletionStage
 
-import akka.stream.alpakka.common.WriteSettings
 import akka.stream.alpakka.elasticsearch._
 import akka.stream.javadsl._
 import akka.{Done, NotUsed}
@@ -22,7 +21,7 @@ object ElasticsearchSink {
    */
   def create[T](
       elasticsearchParams: ElasticsearchParams,
-      settings: WriteSettings[_, _],
+      settings: WriteSettingsBase[_, _],
       objectMapper: ObjectMapper
   ): akka.stream.javadsl.Sink[WriteMessage[T, NotUsed], CompletionStage[Done]] =
     ElasticsearchFlow