Mongo 集合的 Monix 任务:错误处理

Question

我正在尝试将MonixTask与mongo-scala-driver一起使用。我很难理解Error Handling

val mongoClient: Resource[Task, MongoConnection[Task, DomainModel]] =
MongoTypedConnection.create[Task, DomainModel](
"mongodb:...&authMechanism=SCRAM-SHA-1"
)
mongoClient.use { client =>
val changeStream: Task[ChangeStreamObservable[DomainModel]] =
for {
collection <- client.getMongoCollection("myDatabase", "myCollection")
changes    <- client.watchCollection(collection)
} yield changes
...
...
...
.as(ExitCode.Success)
}

当没有错误时，这非常有效。我想为此添加错误处理(例如处理不正确的database和collection名称(。我基于文档的初步尝试是尝试：

val changeObs: io.Serializable =
Await.result(changeStream
.onErrorHandleWith {
case _: TimeoutException =>
// Oh, we know about timeouts, recover it
Task.now("Recovered!")
case other =>
// We have no idea what happened, raise error!
Task.raiseError(other)
}.runToFuture, 5.seconds)

但这给了我一个io.Serializable.如何在保留ChangeStreamObservable[DomainModel]的同时进行某种整洁的错误处理？欣赏我可以研究的任何模式的指针。

BR

Answer 1

事实证明我看错了。

Task[ChangeStreamObservable[DomainModel]]已经有了MonadError.对于像我这样的菜鸟来说，这本质上意味着它不会丢失错误。因此，这可以在代码库的最后完成：

changeStream //Or any other Task/Observable which (is composed)composes (from)this Task
.onErrorHandle {
case timeout: MongoTimeoutException =>
logger.error(timeout.getMessage)
case illegal: java.lang.IllegalArgumentException =>
logger.error(illegal.getMessage)
case unauthorized: com.mongodb.MongoCommandException =>
logger.error(unauthorized.getMessage)

我试图运行任务只是为了处理代码库中间的错误，认为如果我编写多个任务/可观察，我会丢失初始错误。