added some nicer utilities for transactions + some tests

Author: Roman Janusz

commons-mongo/jvm/src/main/scala/com/avsystem/commons/mongo/typed/TypedClientSession.scala

@@ -1,6 +1,7 @@
 package com.avsystem.commons
 package mongo.typed
 
+import cats.effect.ExitCase
 import com.mongodb.reactivestreams.client.ClientSession
 import com.mongodb.session.ServerSession
 import com.mongodb.{ClientSessionOptions, ServerAddress, TransactionOptions}
@@ -29,6 +30,18 @@ class TypedClientSession(val nativeSession: ClientSession)
   def abortTransaction: Task[Unit] =
     single(nativeSession.abortTransaction()).void
 
+  def inTransaction[T](
+    transactionOptions: TransactionOptions = TransactionOptions.builder().build()
+  )(
+    task: Task[T]
+  ): Task[T] = Task.defer {
+    startTransaction(transactionOptions)
+    task.guaranteeCase {
+      case ExitCase.Completed => commitTransaction
+      case _ => abortTransaction
+    }
+  }
+
   def pinnedServerAddress: Option[ServerAddress] =
     Option(nativeSession.getPinnedServerAddress)
 

commons-mongo/jvm/src/main/scala/com/avsystem/commons/mongo/typed/TypedMongoClient.scala

@@ -3,9 +3,9 @@ package mongo.typed
 
 import com.avsystem.commons.mongo.BsonValueInput
 import com.avsystem.commons.serialization.GenCodec
+import com.mongodb._
 import com.mongodb.connection.ClusterDescription
 import com.mongodb.reactivestreams.client.{MongoClient, MongoClients}
-import com.mongodb.{ClientSessionOptions, ConnectionString, MongoClientSettings, MongoDriverInformation}
 import monix.eval.Task
 import monix.reactive.Observable
 import org.bson.Document
@@ -70,6 +70,21 @@ class TypedMongoClient(
   ): Task[TypedClientSession] =
     single(nativeClient.startSession(options)).map(new TypedClientSession(_))
 
+  def inSession[T](
+    options: ClientSessionOptions = ClientSessionOptions.builder().build()
+  )(
+    task: TypedClientSession => Task[T]
+  ): Task[T] =
+    startSession(options).bracket(task)(s => Task(s.close()))
+
+  def inTransaction[T](
+    sessionOptions: ClientSessionOptions = ClientSessionOptions.builder().build(),
+    transactionOptions: TransactionOptions = TransactionOptions.builder().build(),
+  )(
+    task: TypedClientSession => Task[T]
+  ): Task[T] =
+    inSession(sessionOptions)(s => s.inTransaction(transactionOptions)(task(s)))
+
   def clusterDescription(): ClusterDescription =
     nativeClient.getClusterDescription
 

commons-mongo/jvm/src/test/scala/com/avsystem/commons/mongo/typed/TypedMongoCollectionTest.scala

@@ -225,4 +225,46 @@ class TypedMongoCollectionTest extends AnyFunSuite with ScalaFutures with Before
     val filter = Rtaie.ref(_.str).is("foo")
     assert(rtaieColl.findOne(filter, Rtaie.IdRef zip Rtaie.SelfRef).value.contains((id, entity)))
   }
+
+  test("successful transaction") {
+    val query = Rte.IdRef.is(entities(0).id)
+    val prop = Rte.ref(_.int)
+
+    val transaction =
+      client.inTransaction() { session =>
+        val coll = rteColl.withSession(session)
+        for {
+          _ <- coll.updateOne(query, prop.inc(1))
+          fromOutside <- rteColl.findOne(query, prop)
+          fromInside <- coll.findOne(query, prop)
+        } yield (fromOutside, fromInside)
+      }
+
+    val (fromOutside, fromInside) = transaction.value
+    val committed = rteColl.findOne(query, prop).value
+
+    // no dirty reads, old value should be visible before commit
+    assert(fromOutside.contains(entities(0).int))
+    // we should see the new value from within the transaction
+    assert(fromInside.contains(entities(0).int + 1))
+    // we should see the new value after commit
+    assert(committed.contains(entities(0).int + 1))
+  }
+
+  test("failed transaction") {
+    val query = Rte.IdRef.is(entities(0).id)
+    val prop = Rte.ref(_.int)
+
+    val transaction =
+      client.inTransaction() { session =>
+        val coll = rteColl.withSession(session)
+        coll.updateOne(query, prop.inc(1)) *> Task.raiseError(new Exception)
+      }
+
+    transaction.materialize.value
+
+    val committed = rteColl.findOne(query, prop).value
+    // we should see the old value because the transaction should be aborted
+    assert(committed.contains(entities(0).int))
+  }
 }