Updates the integration to not use tuples and addresses other PR comments

Author: chedim

couchbase/src/main/scala/akka/stream/alpakka/couchbase/CouchbaseDocument.scala

@@ -0,0 +1,6 @@
+package akka.stream.alpakka.couchbase
+
+class CouchbaseDocument[T](id: String, document: T) {
+  def getId: String = id;
+  def getDocument: T = document;
+}

couchbase/src/main/scala/akka/stream/alpakka/couchbase/impl/CouchbaseCollectionSessionImpl.scala

@@ -1,7 +1,8 @@
 package akka.stream.alpakka.couchbase.impl
 
-import akka.NotUsed
+import akka.{Done, NotUsed}
 import akka.annotation.InternalApi
+import akka.stream.alpakka.couchbase.CouchbaseDocument
 import akka.stream.alpakka.couchbase.scaladsl.{CouchbaseCollectionSession, CouchbaseSession}
 import akka.stream.scaladsl.Source
 import com.couchbase.client.java.codec.{JsonTranscoder, RawBinaryTranscoder, RawStringTranscoder, Transcoder}
@@ -17,8 +18,11 @@ import scala.concurrent.{ExecutionContext, Future}
 import scala.concurrent.duration.FiniteDuration
 import scala.jdk.FutureConverters.CompletionStageOps
 
+/**
+ * INTERNAL API
+ */
 @InternalApi
-class CouchbaseCollectionSessionImpl(bucketSession: CouchbaseSession, scopeName: String, collectionName: String) extends CouchbaseCollectionSession{
+private[couchbase] class CouchbaseCollectionSessionImpl(bucketSession: CouchbaseSession, scopeName: String, collectionName: String) extends CouchbaseCollectionSession{
 
   override def bucket: CouchbaseSession = bucketSession
 
@@ -27,32 +31,30 @@ class CouchbaseCollectionSessionImpl(bucketSession: CouchbaseSession, scopeName:
 
   override def asJava = new CouchbaseCollectionSessionJavaAdapter(this)
 
-  override def insert[T] (document: (String, T)): Future[(String, T)] = {
+  override def insert[T] (id: String, document: T): Future[Done] = {
     underlying
-      .insert(document._1, document._2,
+      .insert(id, document,
         InsertOptions.insertOptions()
-          .transcoder(chooseTranscoder(document._2.getClass))
+          .transcoder(chooseTranscoder(document.getClass))
       )
-      .asScala.map(_ => document)(ExecutionContext.parasitic)
+      .asScala
+      .map(_ => Done)(ExecutionContext.parasitic)
   }
 
-  override def insert[T] (document: (String, T), insertOptions: InsertOptions): Future[(String, T)] = {
-    underlying.insert(document._1, document._2, insertOptions)
+  override def insert[T] (id: String, document: T, insertOptions: InsertOptions): Future[Done] = {
+    underlying.insert(id, document, insertOptions)
       .asScala
-      .map(r => document)(ExecutionContext.parasitic)
+      .map(_ => Done)(ExecutionContext.parasitic)
   }
 
-  override def get[T](id: String, target: Class[T]): Future[(String, T)] = {
+  override def get[T](id: String, target: Class[T]): Future[CouchbaseDocument[T]] = {
     underlying.get(id, GetOptions.getOptions.transcoder(chooseTranscoder(target)))
-      .thenApply(gr => {
-        (id, gr.contentAs(target))
-      })
+      .thenApply(gr => new CouchbaseDocument[T](id, gr.contentAs(target)))
       .asScala
-
   }
 
-  override def getDocument(id: String): Future[(String, JsonValue)] =
-    underlying.get(id).thenApply(gr => (id, asJsonValue(gr))).asScala
+  override def getDocument(id: String): Future[CouchbaseDocument[JsonValue]] =
+    underlying.get(id).thenApply(gr => new CouchbaseDocument[JsonValue](id, asJsonValue(gr))).asScala
 
   private def asJsonValue(gr: GetResult) =
     try {
@@ -61,77 +63,78 @@ class CouchbaseCollectionSessionImpl(bucketSession: CouchbaseSession, scopeName:
       case ex: Exception => gr.contentAsArray().asInstanceOf[JsonValue]
     }
 
-  override def getBytes(id: String): Future[(String, Array[Byte])] =
-    underlying.get(id).thenApply(gr => (id, gr.contentAsBytes())).asScala
+  override def getBytes(id: String): Future[CouchbaseDocument[Array[Byte]]] =
+    underlying.get(id).thenApply(gr => new CouchbaseDocument[Array[Byte]](id, gr.contentAsBytes())).asScala
 
-  override def getDocument(id: String, timeout: FiniteDuration): Future[(String, JsonValue)] =
+  override def getDocument(id: String, timeout: FiniteDuration): Future[CouchbaseDocument[JsonValue]] =
     underlying.get(id)
       .orTimeout(timeout.toMillis, TimeUnit.MILLISECONDS)
-      .thenApply(gr => (id, asJsonValue(gr)))
+      .thenApply(gr => new CouchbaseDocument[JsonValue](id, asJsonValue(gr)))
       .asScala
 
-  override def getBytes(id: String, timeout: FiniteDuration): Future[(String, Array[Byte])] =
+  override def getBytes(id: String, timeout: FiniteDuration): Future[CouchbaseDocument[Array[Byte]]] =
     underlying.get(id)
       .orTimeout(timeout.toMillis, TimeUnit.MILLISECONDS)
-      .thenApply(gr => (id, gr.contentAsBytes()))
+      .thenApply(gr => new CouchbaseDocument[Array[Byte]](id, gr.contentAsBytes()))
       .asScala
 
-  override def upsert[T](document: (String, T)): Future[(String, T)] = {
-    underlying.upsert(document._1, document._2,
+  override def upsert[T](id: String, document: T): Future[Done] = {
+    underlying.upsert(id, document,
       UpsertOptions.upsertOptions()
-        .transcoder(chooseTranscoder(document._2.getClass))
+        .transcoder(chooseTranscoder(document.getClass))
     )
-    .asScala.map(_ => document)(ExecutionContext.parasitic)
+    .asScala.map(_ => Done)(ExecutionContext.parasitic)
   }
 
-  override def upsert[T](document: (String, T), upsertOptions: UpsertOptions): Future[(String, T)] =
-    underlying.upsert(document._1, document._2, upsertOptions)
-      .thenApply(_ => document)
+  override def upsert[T](id: String, document: T, upsertOptions: UpsertOptions): Future[Done] =
+    underlying.upsert(id, document, upsertOptions)
+      .thenApply(_ => Done)
       .asScala
 
-  override def upsert[T](document: (String, T), upsertOptions: UpsertOptions, timeout: FiniteDuration): Future[(String, T)] =
-    underlying.upsert(document._1, document._2, upsertOptions)
+  override def upsert[T](id: String, document: T, upsertOptions: UpsertOptions, timeout: FiniteDuration): Future[Done] =
+    underlying.upsert(id, document, upsertOptions)
       .orTimeout(timeout.toMillis, TimeUnit.MILLISECONDS)
-      .thenApply(_ => document)
+      .thenApply(_ => Done)
       .asScala
 
 
-  override def replace[T](document: (String, T)): Future[(String, T)] =
-    underlying.replace(document._1, document._2)
-      .thenApply(_ => document)
+  override def replace[T](id: String, document: T): Future[Done] =
+    underlying.replace(id, document)
+      .thenApply(_ => Done)
       .asScala
 
-  override def replace[T](document: (String, T), replaceOptions: ReplaceOptions): Future[(String, T)] =
-    underlying.replace(document._1, document._2, replaceOptions)
-      .thenApply(_ => document)
+  override def replace[T](id: String, document: T, replaceOptions: ReplaceOptions): Future[Done] =
+    underlying.replace(id, document, replaceOptions)
+      .thenApply(_ => Done)
       .asScala
 
-  override def replace[T](document: (String, T), replaceOptions: ReplaceOptions, timeout: FiniteDuration): Future[(String, T)] =
-    underlying.replace(document._1, document._2, replaceOptions)
+  override def replace[T](id: String, document: T, replaceOptions: ReplaceOptions, timeout: FiniteDuration): Future[Done] =
+    underlying.replace(id, document, replaceOptions)
       .orTimeout(timeout.toMillis, TimeUnit.MILLISECONDS)
-      .thenApply(_ => document)
+      .thenApply(_ => Done)
       .asScala
 
-  override def remove(id: String): Future[String] =
+  override def remove(id: String): Future[Done] =
     underlying.remove(id)
-      .thenApply(_ => id)
+      .thenApply(_ => Done)
       .asScala
 
-  override def remove(id: String, removeOptions: RemoveOptions): Future[String] =
+  override def remove(id: String, removeOptions: RemoveOptions): Future[Done] =
     underlying.remove(id, removeOptions)
-      .thenApply(_ => id)
+      .thenApply(_ => Done)
       .asScala
 
-  override def remove(id: String, removeOptions: RemoveOptions, timeout: FiniteDuration): Future[String] =
+  override def remove(id: String, removeOptions: RemoveOptions, timeout: FiniteDuration): Future[Done] =
     underlying.remove(id, removeOptions)
       .orTimeout(timeout.toMillis, TimeUnit.MILLISECONDS)
-      .thenApply(_ => id)
+      .thenApply(_ => Done)
       .asScala
 
-  override def createIndex(indexName: String, createQueryIndexOptions: CreateQueryIndexOptions, fields: String*): Future[Void] =
+  override def createIndex(indexName: String, createQueryIndexOptions: CreateQueryIndexOptions, fields: String*): Future[Done] =
       underlying
         .queryIndexes()
         .createIndex(indexName, util.Arrays.asList(fields: _*), createQueryIndexOptions)
+        .thenApply(_ => Done)
         .asScala
 
   override def listIndexes(): Source[QueryIndex, NotUsed] =
@@ -142,10 +145,9 @@ class CouchbaseCollectionSessionImpl(bucketSession: CouchbaseSession, scopeName:
       )
     )
 
-  private def chooseTranscoder[T](target: Class[T]): Transcoder =
-    target match {
-      case _: Class[Array[Byte]] => RawBinaryTranscoder.INSTANCE
-      case _: Class[String] => RawStringTranscoder.INSTANCE
-      case _ => bucketSession.cluster().environment().transcoder()
-    }
+  private def chooseTranscoder[T](target: Class[T]): Transcoder = {
+    if (target == classOf[Array[Byte]]) RawBinaryTranscoder.INSTANCE
+    else if (target == classOf[String]) RawStringTranscoder.INSTANCE
+    else bucketSession.cluster().environment().transcoder()
+  }
 }

couchbase/src/main/scala/akka/stream/alpakka/couchbase/impl/CouchbaseCollectionSessionJavaAdapter.scala

@@ -4,10 +4,10 @@
 
 package akka.stream.alpakka.couchbase.impl
 
-import akka.NotUsed
+import akka.{Done, NotUsed}
 import akka.annotation.InternalApi
 import akka.stream.alpakka.couchbase.javadsl.CouchbaseSession
-import akka.stream.alpakka.couchbase.{javadsl, scaladsl}
+import akka.stream.alpakka.couchbase.{CouchbaseDocument, javadsl, scaladsl}
 import akka.stream.javadsl.Source
 import com.couchbase.client.java.json.{JsonArray, JsonObject, JsonValue}
 import com.couchbase.client.java.kv.{InsertOptions, RemoveOptions, ReplaceOptions, UpsertOptions}
@@ -16,6 +16,7 @@ import com.couchbase.client.java.{AsyncCollection, AsyncScope}
 
 import java.time.Duration
 import java.util.concurrent.{CompletionStage, TimeUnit}
+import scala.concurrent.Future
 import scala.concurrent.duration.FiniteDuration
 import scala.jdk.FutureConverters._
 
@@ -40,53 +41,53 @@ private[couchbase] final class CouchbaseCollectionSessionJavaAdapter(delegate: s
    * @param document A tuple where first element is id of the document and second is its value
    * @return A Future that completes with the id of the written document when the write is done
    */
-  override def insert[T](document: (String, T)): CompletionStage[(String, T)] =
-    delegate.insert(document).asJava
+  override def insert[T](id: String, document: T): CompletionStage[Done] =
+    delegate.insert(id, document).asJava
 
-  override def insert[T](document: (String, T), insertOptions: InsertOptions): CompletionStage[(String, T)] =
-    delegate.insert(document, insertOptions).asJava
+  override def insert[T](id: String, document: T, insertOptions: InsertOptions): CompletionStage[Done] =
+    delegate.insert(id, document, insertOptions).asJava
 
-  override def getJsonObject(id: String): CompletionStage[(String, JsonObject)] =
+  override def getJsonObject(id: String): CompletionStage[CouchbaseDocument[JsonObject]] =
     delegate.getJsonObject(id).asJava
 
-  override def getJsonArray(id: String): CompletionStage[(String, JsonArray)] =
+  override def getJsonArray(id: String): CompletionStage[CouchbaseDocument[JsonArray]] =
     delegate.getJsonArray(id).asJava
 
-  override def get[T](id: String, target: Class[T]): CompletionStage[(String, T)] =
+  override def get[T](id: String, target: Class[T]): CompletionStage[CouchbaseDocument[T]] =
     delegate.get(id, target).asJava
   /**
    * @return A document if found or none if there is no document for the id
    */
-  override def getDocument(id: String): CompletionStage[(String, JsonValue)] =
+  override def getDocument(id: String): CompletionStage[CouchbaseDocument[JsonValue]] =
     delegate.getDocument(id).asJava
 
   /**
    * @param id Identifier of the document to fetch
    * @return Raw data for the document or none
    */
-  override def getBytes(id: String): CompletionStage[(String, Array[Byte])] =
+  override def getBytes(id: String): CompletionStage[CouchbaseDocument[Array[Byte]]] =
     delegate.getBytes(id).asJava
 
   /**
    * @param timeout fail the returned future with a TimeoutException if it takes longer than this
    * @return A document if found or none if there is no document for the id
    */
-  override def getDocument(id: String, timeout: Duration): CompletionStage[(String, JsonValue)] =
+  override def getDocument(id: String, timeout: Duration): CompletionStage[CouchbaseDocument[JsonValue]] =
     delegate.getDocument(id, FiniteDuration.apply(timeout.toNanos, TimeUnit.NANOSECONDS)).asJava
 
   /**
    * @return A raw document data if found or none if there is no document for the id
    */
-  override def getBytes(id: String, timeout: Duration): CompletionStage[(String, Array[Byte])] =
+  override def getBytes(id: String, timeout: Duration): CompletionStage[CouchbaseDocument[Array[Byte]]] =
     delegate.getBytes(id, FiniteDuration.apply(timeout.toNanos, TimeUnit.NANOSECONDS)).asJava
 
   /**
    * Upsert using the default write settings.
    *
    * @return a future that completes when the upsert is done
    */
-  override def upsert[T](document: (String, T)): CompletionStage[(String, T)] =
-    delegate.upsert(document).asJava
+  override def upsert[T](id: String, document: T): CompletionStage[Done] =
+    delegate.upsert(id, document).asJava
 
   /**
    * Upsert using the given write settings
@@ -95,19 +96,20 @@ private[couchbase] final class CouchbaseCollectionSessionJavaAdapter(delegate: s
    *
    * @return a future that completes when the upsert is done
    */
-  override def upsert[T](document: (String, T), upsertOptions: UpsertOptions): CompletionStage[(String, T)] =
-    delegate.upsert(document, upsertOptions).asJava
+  override def upsert[T](id: String, document: T, upsertOptions: UpsertOptions): CompletionStage[Done] =
+    delegate.upsert(id, document, upsertOptions).asJava
 
   /**
    * Upsert using given write settings and timeout
    *
-   * @param document      document id and value to upsert
+   * @param id document id
+   * @param document      document value to upsert
    * @param upsertOptions Couchbase UpsertOptions
    * @param timeout       timeout for the operation
-   * @return the document id and value
+   * @return a future that completes after the operation is done
    */
-  override def upsert[T](document: (String, T), upsertOptions: UpsertOptions, timeout: Duration): CompletionStage[(String, T)] =
-    delegate.upsert(document, upsertOptions, FiniteDuration.apply(timeout.toNanos, TimeUnit.NANOSECONDS)).asJava
+  override def upsert[T](id: String, document: T, upsertOptions: UpsertOptions, timeout: Duration): CompletionStage[Done] =
+    delegate.upsert(id, document, upsertOptions, FiniteDuration.apply(timeout.toNanos, TimeUnit.NANOSECONDS)).asJava
 
   /**
    * Replace using the default write settings.
@@ -116,8 +118,8 @@ private[couchbase] final class CouchbaseCollectionSessionJavaAdapter(delegate: s
    *
    * @return a future that completes when the replace is done
    */
-  override def replace[T](document: (String, T)): CompletionStage[(String, T)] =
-    delegate.replace(document).asJava
+  override def replace[T](id: String, document: T): CompletionStage[Done] =
+    delegate.replace(id, document).asJava
 
   /**
    * Replace using the given replace options
@@ -126,27 +128,28 @@ private[couchbase] final class CouchbaseCollectionSessionJavaAdapter(delegate: s
    *
    * @return a future that completes when the replace is done
    */
-  override def replace[T](document: (String, T), replaceOptions: ReplaceOptions): CompletionStage[(String, T)] =
-    delegate.replace(document, replaceOptions).asJava
+  override def replace[T](id: String, document: T, replaceOptions: ReplaceOptions): CompletionStage[Done] =
+    delegate.replace(id, document, replaceOptions).asJava
 
   /**
    * Replace using write settings and timeout
    *
-   * @param document       document id and value to replace
+   * @param id id of the document to replace
+   * @param document       document value to replace
    * @param replaceOptions Couchbase replace options
    * @param timeout        timeout for the operation
-   * @return the document id and value
+   * @return a future that completes after the operation is done
    */
-  override def replace[T](document: (String, T), replaceOptions: ReplaceOptions, timeout: Duration): CompletionStage[(String, T)] =
-    delegate.replace(document, replaceOptions, FiniteDuration.apply(timeout.toNanos, TimeUnit.NANOSECONDS)).asJava
+  override def replace[T](id: String, document: T, replaceOptions: ReplaceOptions, timeout: Duration): CompletionStage[Done] =
+    delegate.replace(id, document, replaceOptions, FiniteDuration.apply(timeout.toNanos, TimeUnit.NANOSECONDS)).asJava
 
   /**
    * Remove a document by id using the default write settings.
    *
    * @return Future that completes when the document has been removed, if there is no such document
    *         the future is failed with a `DocumentDoesNotExistException`
    */
-  override def remove(id: String): CompletionStage[String] =
+  override def remove(id: String): CompletionStage[Done] =
     delegate.remove(id).asJava
 
   /**
@@ -155,7 +158,7 @@ private[couchbase] final class CouchbaseCollectionSessionJavaAdapter(delegate: s
    * @return Future that completes when the document has been removed, if there is no such document
    *         the future is failed with a `DocumentDoesNotExistException`
    */
-  override def remove(id: String, removeOptions: RemoveOptions): CompletionStage[String] =
+  override def remove(id: String, removeOptions: RemoveOptions): CompletionStage[Done] =
     delegate.remove(id, removeOptions).asJava
 
   /**
@@ -179,7 +182,7 @@ private[couchbase] final class CouchbaseCollectionSessionJavaAdapter(delegate: s
    *         if the index existed and `ignoreIfExist` is `true`. Completion of the future does not guarantee the index is online
    *         and ready to be used.
    */
-  override def createIndex(indexName: String, createQueryIndexOptions: CreateQueryIndexOptions, fields: String*): CompletionStage[Void] =
+  override def createIndex(indexName: String, createQueryIndexOptions: CreateQueryIndexOptions, fields: String*): CompletionStage[Done] =
     delegate.createIndex(indexName, createQueryIndexOptions, fields: _*).asJava
 
   /**