Merge pull request #11 from precog/more-resilient-watch-streams

More resilient watch streams

Author: jsantos17

core/src/main/scala/mongo4cats/collection/queries/WatchQueryBuilder.scala

@@ -32,6 +32,7 @@ import com.mongodb.reactivestreams.client.{
 import fs2.Stream
 import mongo4cats.helpers._
 import mongo4cats.bson.BsonDecoder
+import mongo4cats.bson.BsonDecodeError
 import mongo4cats.bson.syntax._
 import mongo4cats.client.ClientSession
 import mongo4cats.collection.operations.Aggregate
@@ -59,6 +60,7 @@ trait WatchQueryBuilder[F[_]] {
 
   //
   def stream: Stream[F, ChangeStreamDocument[BsonValue]]
+  def updateStreamAttempt[A: BsonDecoder](onError: BsonDecodeError => F[Unit]): Stream[F, A]
   def updateStream[A: BsonDecoder]: Stream[F, A]
 
   def mapK[G[_]](f: F ~> G): WatchQueryBuilder[G]
@@ -120,6 +122,17 @@ object WatchQueryBuilder {
     def stream =
       boundedStreamF(1).translate(transform)
 
+    def updateStreamAttempt[A: BsonDecoder](onError: BsonDecodeError => G[Unit]) =
+      boundedStreamF(1)
+        .map(_.getFullDocument.as[A])
+        .translate(transform)
+        .flatMap {
+          case Left(decodeError) =>
+            Stream.exec(onError(decodeError))
+          case Right(elem) =>
+            Stream(elem)
+        }
+
     def updateStream[A: BsonDecoder] =
       boundedStreamF(1).map(_.getFullDocument).evalMap(_.as[A].liftTo[F]).translate(transform)
 

version.sbt

@@ -1 +1 @@
-version in ThisBuild := "0.6.0"
+version in ThisBuild := "0.7.0"