Drop local implementations of setup operator

akka · Jun 19, 2019 · e925403 · e925403
1 parent 7963773
commit e925403
Show file tree

Hide file tree

Showing 18 changed files with 473 additions and 1,485 deletions.
diff --git a/couchbase/src/main/scala/akka/stream/alpakka/couchbase/impl/SetupStage.scala b/couchbase/src/main/scala/akka/stream/alpakka/couchbase/impl/SetupStage.scala
diff --git a/couchbase/src/main/scala/akka/stream/alpakka/couchbase/scaladsl/CouchbaseFlow.scala b/couchbase/src/main/scala/akka/stream/alpakka/couchbase/scaladsl/CouchbaseFlow.scala
@@ -4,7 +4,6 @@
 
 package akka.stream.alpakka.couchbase.scaladsl
 import akka.NotUsed
-import akka.stream.alpakka.couchbase.impl.Setup
 import akka.stream.alpakka.couchbase._
 import akka.stream.scaladsl.Flow
 import com.couchbase.client.java.document.{Document, JsonDocument}
@@ -18,12 +17,13 @@ object CouchbaseFlow {
    * Create a flow to query Couchbase for by `id` and emit [[com.couchbase.client.java.document.JsonDocument JsonDocument]]s.
    */
   def fromId(sessionSettings: CouchbaseSessionSettings, bucketName: String): Flow[String, JsonDocument, NotUsed] =
-    Setup
-      .flow { materializer => _ =>
-        val session = CouchbaseSessionRegistry(materializer.system).sessionFor(sessionSettings, bucketName)
-        Flow[String]
-          .mapAsync(1)(id => session.flatMap(_.get(id /* timeout? */ ))(materializer.system.dispatcher))
-          .collect { case Some(doc) => doc }
+    Flow
+      .setup {
+        case (materializer, _) =>
+          val session = CouchbaseSessionRegistry(materializer.system).sessionFor(sessionSettings, bucketName)
+          Flow[String]
+            .mapAsync(1)(id => session.flatMap(_.get(id /* timeout? */ ))(materializer.system.dispatcher))
+            .collect { case Some(doc) => doc }
       }
       .mapMaterializedValue(_ => NotUsed)
 
@@ -33,12 +33,13 @@ object CouchbaseFlow {
   def fromId[T <: Document[_]](sessionSettings: CouchbaseSessionSettings,
                                bucketName: String,
                                target: Class[T]): Flow[String, T, NotUsed] =
-    Setup
-      .flow { materializer => _ =>
-        val session = CouchbaseSessionRegistry(materializer.system).sessionFor(sessionSettings, bucketName)
-        Flow[String]
-          .mapAsync(1)(id => session.flatMap(_.get(id /* timeout? */, target))(materializer.system.dispatcher))
-          .collect { case Some(doc) => doc }
+    Flow
+      .setup {
+        case (materializer, _) =>
+          val session = CouchbaseSessionRegistry(materializer.system).sessionFor(sessionSettings, bucketName)
+          Flow[String]
+            .mapAsync(1)(id => session.flatMap(_.get(id /* timeout? */, target))(materializer.system.dispatcher))
+            .collect { case Some(doc) => doc }
       }
       .mapMaterializedValue(_ => NotUsed)
 
@@ -48,13 +49,14 @@ object CouchbaseFlow {
   def upsert(sessionSettings: CouchbaseSessionSettings,
              writeSettings: CouchbaseWriteSettings,
              bucketName: String): Flow[JsonDocument, JsonDocument, NotUsed] =
-    Setup
-      .flow { materializer => _ =>
-        val session = CouchbaseSessionRegistry(materializer.system).sessionFor(sessionSettings, bucketName)
-        Flow[JsonDocument]
-          .mapAsync(writeSettings.parallelism)(
-            doc => session.flatMap(_.upsert(doc, writeSettings))(materializer.system.dispatcher)
-          )
+    Flow
+      .setup {
+        case (materializer, _) =>
+          val session = CouchbaseSessionRegistry(materializer.system).sessionFor(sessionSettings, bucketName)
+          Flow[JsonDocument]
+            .mapAsync(writeSettings.parallelism)(
+              doc => session.flatMap(_.upsert(doc, writeSettings))(materializer.system.dispatcher)
+            )
       }
       .mapMaterializedValue(_ => NotUsed)
 
@@ -64,13 +66,14 @@ object CouchbaseFlow {
   def upsertDoc[T <: Document[_]](sessionSettings: CouchbaseSessionSettings,
                                   writeSettings: CouchbaseWriteSettings,
                                   bucketName: String): Flow[T, T, NotUsed] =
-    Setup
-      .flow { materializer => _ =>
-        val session = CouchbaseSessionRegistry(materializer.system).sessionFor(sessionSettings, bucketName)
-        Flow[T]
-          .mapAsync(writeSettings.parallelism)(
-            doc => session.flatMap(_.upsertDoc(doc, writeSettings))(materializer.system.dispatcher)
-          )
+    Flow
+      .setup {
+        case (materializer, _) =>
+          val session = CouchbaseSessionRegistry(materializer.system).sessionFor(sessionSettings, bucketName)
+          Flow[T]
+            .mapAsync(writeSettings.parallelism)(
+              doc => session.flatMap(_.upsertDoc(doc, writeSettings))(materializer.system.dispatcher)
+            )
       }
       .mapMaterializedValue(_ => NotUsed)
 
@@ -81,21 +84,22 @@ object CouchbaseFlow {
   def upsertDocWithResult[T <: Document[_]](sessionSettings: CouchbaseSessionSettings,
                                             writeSettings: CouchbaseWriteSettings,
                                             bucketName: String): Flow[T, CouchbaseWriteResult[T], NotUsed] =
-    Setup
-      .flow { materializer => _ =>
-        val session = CouchbaseSessionRegistry(materializer.system).sessionFor(sessionSettings, bucketName)
-        Flow[T]
-          .mapAsync(writeSettings.parallelism)(
-            doc => {
-              implicit val executor = materializer.system.dispatcher
-              session
-                .flatMap(_.upsertDoc(doc, writeSettings))
-                .map(_ => CouchbaseWriteSuccess(doc))
-                .recover {
-                  case exception => CouchbaseWriteFailure(doc, exception)
-                }
-            }
-          )
+    Flow
+      .setup {
+        case (materializer, _) =>
+          val session = CouchbaseSessionRegistry(materializer.system).sessionFor(sessionSettings, bucketName)
+          Flow[T]
+            .mapAsync(writeSettings.parallelism)(
+              doc => {
+                implicit val executor = materializer.system.dispatcher
+                session
+                  .flatMap(_.upsertDoc(doc, writeSettings))
+                  .map(_ => CouchbaseWriteSuccess(doc))
+                  .recover {
+                    case exception => CouchbaseWriteFailure(doc, exception)
+                  }
+              }
+            )
       }
       .mapMaterializedValue(_ => NotUsed)
 
@@ -105,18 +109,19 @@ object CouchbaseFlow {
   def delete(sessionSettings: CouchbaseSessionSettings,
              writeSettings: CouchbaseWriteSettings,
              bucketName: String): Flow[String, String, NotUsed] =
-    Setup
-      .flow { materializer => _ =>
-        val session = CouchbaseSessionRegistry(materializer.system).sessionFor(sessionSettings, bucketName)
-        Flow[String]
-          .mapAsync(writeSettings.parallelism)(
-            id => {
-              implicit val executor = materializer.system.dispatcher
-              session
-                .flatMap(_.remove(id, writeSettings))
-                .map(_ => id)
-            }
-          )
+    Flow
+      .setup {
+        case (materializer, _) =>
+          val session = CouchbaseSessionRegistry(materializer.system).sessionFor(sessionSettings, bucketName)
+          Flow[String]
+            .mapAsync(writeSettings.parallelism)(
+              id => {
+                implicit val executor = materializer.system.dispatcher
+                session
+                  .flatMap(_.remove(id, writeSettings))
+                  .map(_ => id)
+              }
+            )
       }
       .mapMaterializedValue(_ => NotUsed)
 
@@ -126,21 +131,22 @@ object CouchbaseFlow {
   def deleteWithResult(sessionSettings: CouchbaseSessionSettings,
                        writeSettings: CouchbaseWriteSettings,
                        bucketName: String): Flow[String, CouchbaseDeleteResult, NotUsed] =
-    Setup
-      .flow { materializer => _ =>
-        val session = CouchbaseSessionRegistry(materializer.system).sessionFor(sessionSettings, bucketName)
-        Flow[String]
-          .mapAsync(writeSettings.parallelism)(
-            id => {
-              implicit val executor = materializer.system.dispatcher
-              session
-                .flatMap(_.remove(id, writeSettings))
-                .map(_ => CouchbaseDeleteSuccess(id))
-                .recover {
-                  case exception => CouchbaseDeleteFailure(id, exception)
-                }
-            }
-          )
+    Flow
+      .setup {
+        case (materializer, _) =>
+          val session = CouchbaseSessionRegistry(materializer.system).sessionFor(sessionSettings, bucketName)
+          Flow[String]
+            .mapAsync(writeSettings.parallelism)(
+              id => {
+                implicit val executor = materializer.system.dispatcher
+                session
+                  .flatMap(_.remove(id, writeSettings))
+                  .map(_ => CouchbaseDeleteSuccess(id))
+                  .recover {
+                    case exception => CouchbaseDeleteFailure(id, exception)
+                  }
+              }
+            )
       }
       .mapMaterializedValue(_ => NotUsed)
 }
diff --git a/couchbase/src/main/scala/akka/stream/alpakka/couchbase/scaladsl/CouchbaseSource.scala b/couchbase/src/main/scala/akka/stream/alpakka/couchbase/scaladsl/CouchbaseSource.scala
@@ -5,7 +5,6 @@
 package akka.stream.alpakka.couchbase.scaladsl
 
 import akka.NotUsed
-import akka.stream.alpakka.couchbase.impl.Setup
 import akka.stream.alpakka.couchbase.{CouchbaseSessionRegistry, CouchbaseSessionSettings}
 import akka.stream.scaladsl.Source
 import com.couchbase.client.java.document.json.JsonObject
@@ -22,12 +21,13 @@ object CouchbaseSource {
   def fromStatement(sessionSettings: CouchbaseSessionSettings,
                     statement: Statement,
                     bucketName: String): Source[JsonObject, NotUsed] =
-    Setup
-      .source { materializer => _ =>
-        val session = CouchbaseSessionRegistry(materializer.system).sessionFor(sessionSettings, bucketName)
-        Source
-          .fromFuture(session.map(_.streamedQuery(statement))(materializer.system.dispatcher))
-          .flatMapConcat(identity)
+    Source
+      .setup {
+        case (materializer, _) =>
+          val session = CouchbaseSessionRegistry(materializer.system).sessionFor(sessionSettings, bucketName)
+          Source
+            .fromFuture(session.map(_.streamedQuery(statement))(materializer.system.dispatcher))
+            .flatMapConcat(identity)
       }
       .mapMaterializedValue(_ => NotUsed)
 
@@ -37,12 +37,13 @@ object CouchbaseSource {
   def fromN1qlQuery(sessionSettings: CouchbaseSessionSettings,
                     query: N1qlQuery,
                     bucketName: String): Source[JsonObject, NotUsed] =
-    Setup
-      .source { materializer => _ =>
-        val session = CouchbaseSessionRegistry(materializer.system).sessionFor(sessionSettings, bucketName)
-        Source
-          .fromFuture(session.map(_.streamedQuery(query))(materializer.system.dispatcher))
-          .flatMapConcat(identity)
+    Source
+      .setup {
+        case (materializer, _) =>
+          val session = CouchbaseSessionRegistry(materializer.system).sessionFor(sessionSettings, bucketName)
+          Source
+            .fromFuture(session.map(_.streamedQuery(query))(materializer.system.dispatcher))
+            .flatMapConcat(identity)
       }
       .mapMaterializedValue(_ => NotUsed)