Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Return buffer's content after timeout at the latest if synchronous wa… #106

Merged
merged 1 commit into from
Jun 24, 2024
Merged

Return buffer's content after timeout at the latest if synchronous wa… #106

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 4 additions & 6 deletions .scalafix.conf
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,9 +7,7 @@ rules = [
DisableSyntax.noNulls = true
DisableSyntax.noReturns = true

OrganizeImports {
targetDialect = Scala3
groupedImports = Explode
expandRelative = true
removeUnused = false
}
OrganizeImports.targetDialect = Scala3
OrganizeImports.groupedImports = Explode
OrganizeImports.expandRelative = true
OrganizeImports.removeUnused = false
34 changes: 21 additions & 13 deletions molly-core/src/main/scala/molly/core/query/SyncWatchQuery.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,8 @@ package molly.core.query
import cats.effect.kernel.Async
import cats.effect.kernel.Sync
import cats.effect.syntax.spawn.*
import cats.syntax.flatMap.*
import cats.syntax.functor.*
import com.mongodb.client.ChangeStreamIterable
import com.mongodb.client.MongoChangeStreamCursor
import com.mongodb.client.model.changestream.ChangeStreamDocument
Expand All @@ -12,6 +14,8 @@ import fs2.Stream
import molly.core.MollyCodec
import org.bson.BsonDocument

import scala.concurrent.duration.*

final case class SyncWatchQuery[F[_], A] private[core] (private[core] val iterable: ChangeStreamIterable[BsonDocument])(
using
f: Async[F],
Expand All @@ -27,26 +31,30 @@ final case class SyncWatchQuery[F[_], A] private[core] (private[core] val iterab
def fullDocument(fullDocument: FullDocument): SyncWatchQuery[F, A] =
SyncWatchQuery(iterable.fullDocument(fullDocument))

def list(bufferSize: Int = 16): F[List[ChangeStreamDocument[A]]] = stream(bufferSize).compile.toList
def list(bufferSize: Int = 16, timeout: FiniteDuration = 10.seconds): F[List[ChangeStreamDocument[A]]] =
stream(bufferSize, timeout).compile.toList

def stream(bufferSize: Int = 16): Stream[F, ChangeStreamDocument[A]] =
def stream(bufferSize: Int = 16, timeout: FiniteDuration = 10.seconds): Stream[F, ChangeStreamDocument[A]] =
Stream
.bracket(f.delay(iterable.batchSize(bufferSize).cursor()))(cursor => f.delay(cursor.close()))
.flatMap(fromCursor(_, bufferSize))
.flatMap(fromCursor(_, bufferSize, timeout))

private type Cursor = MongoChangeStreamCursor[ChangeStreamDocument[BsonDocument]]

private def fromCursor(cursor: Cursor, bufferSize: Int): Stream[F, ChangeStreamDocument[A]] =
private def fromCursor(cursor: Cursor, bufferSize: Int, timeout: FiniteDuration): Stream[F, ChangeStreamDocument[A]] =
def getNextChunk(cursor: Cursor): F[Option[(Chunk[ChangeStreamDocument[BsonDocument]], Cursor)]] =
f
.suspend(Sync.Type.Blocking):
val bldr = Vector.newBuilder[ChangeStreamDocument[BsonDocument]]
var cnt = 0
while cnt < bufferSize && cursor.hasNext do
bldr += cursor.next()
cnt += 1
if cnt == 0 then None else Some((Chunk.from(bldr.result()), cursor))
.cancelable(f.delay(cursor.close()))
val buffer = Vector.newBuilder[ChangeStreamDocument[BsonDocument]]
f.race(
f
.suspend(Sync.Type.Blocking):
var count = 0
while count < bufferSize && cursor.hasNext do
buffer += cursor.next()
count += 1
if count == 0 then None else Some((Chunk.from(buffer.result()), cursor))
.cancelable(f.delay(cursor.close())),
f.sleep(timeout) >> f.delay(Some((Chunk.from(buffer.result()), cursor)))
).map(_.fold(identity, identity))

Stream
.unfoldChunkEval(cursor)(getNextChunk)
Expand Down
28 changes: 26 additions & 2 deletions molly-core/src/test/scala/molly/core/MollySyncCollectionTest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -42,6 +42,30 @@ object MollySyncCollectionTest extends IOSuite with TestContainerForAll[IO] with
.and(expect(csDocs.exists(_.getDocumentKey() == new BsonDocument("_id", new BsonInt32(3)))))
.and(expect(csDocs.forall(_.getOperationTypeString() == "insert")))

test("watch: return one change per inserted document with buffer size greater than result size"): containers =>
withClient(containers): (client: MollyClient[IO]) =>
withSyncClient(containers): (syncClient: MollySyncClient[IO]) =>
val doc1 = new BsonDocument("_id", new BsonInt32(1)).append("x", new BsonInt32(47))
val doc2 = new BsonDocument("_id", new BsonInt32(2)).append("x", new BsonInt32(20))
val doc3 = new BsonDocument("_id", new BsonInt32(3)).append("x", new BsonInt32(99))

def runChangeStream(coll: MollySyncCollection[IO, BsonDocument]) =
coll.watch().stream(bufferSize = 16, timeout = 3.seconds).take(3).compile.toList

def insert(coll: BsonDocumentCollection[IO]) = IO.sleep(eta) >> coll.insertMany(Seq(doc1, doc2, doc3))

for
db <- client.getDatabase("test")
syncDb <- syncClient.getDatabase("test")
coll <- db.getCollection("watch2")
syncColl <- syncDb.getCollection("watch2")
csDocs <- runChangeStream(syncColl).both(insert(coll)).map(_._1)
yield expect(csDocs.size == 3)
.and(expect(csDocs.exists(_.getDocumentKey() == new BsonDocument("_id", new BsonInt32(1)))))
.and(expect(csDocs.exists(_.getDocumentKey() == new BsonDocument("_id", new BsonInt32(2)))))
.and(expect(csDocs.exists(_.getDocumentKey() == new BsonDocument("_id", new BsonInt32(3)))))
.and(expect(csDocs.forall(_.getOperationTypeString() == "insert")))

test("watch: return different changes"): containers =>
withClient(containers): (client: MollyClient[IO]) =>
withSyncClient(containers): (syncClient: MollySyncClient[IO]) =>
Expand All @@ -61,8 +85,8 @@ object MollySyncCollectionTest extends IOSuite with TestContainerForAll[IO] with
for
db <- client.getDatabase("test")
syncDb <- syncClient.getDatabase("test")
coll <- db.getCollection("watch2")
syncColl <- syncDb.getCollection("watch2")
coll <- db.getCollection("watch3")
syncColl <- syncDb.getCollection("watch3")
csDocs <- runChangeStream(syncColl).both(insertAndUpdate(coll)).map(_._1)
yield expect(csDocs.size == 4)
.and(expect(csDocs.exists(_.getDocumentKey() == new BsonDocument("_id", new BsonInt32(1)))))
Expand Down