如何在单个事务中组合Doobie的ConnectionIO与异步Effect monad

做Doobie开发的同学应该都知道，官方推荐仓储层返回ConnectionIO类型——这能让咱们把多个数据库操作链式串起来，统一放到单个事务里执行，既清晰又省心。

现在假设咱们要开发一个REST API，流程是这样的：

• 从数据库查询一个对象
• 用cats.effect.IO或者monix.eval.Task对这个对象做异步处理
• 把处理后的对象存回数据库

咱们希望这三步全程在单个事务里完成，但麻烦的是，这时候咱们同时处在ConnectionIO和Task两个monad里——没有transactor.trans()提供的自然转换，直接写for推导式会报错，根本没法组合起来。

比如你可能会写出这样的代码，但它根本跑不起来：

def getObject: ConnectionIO[Request] = ??? 
def saveObject(obj: Request): ConnectionIO[Request] = ??? 
def processObject(obj: Request): monix.eval.Task[Request] = ??? 

val transaction:??? = for { 
  obj <- getObject //ConnectionIO[Request] 
  processed <- processObject(obj) //monix.eval.Task[Request] 
  updated <- saveObject(processed) //ConnectionIO[Request] 
} yield updated 

正确解决方案 #

核心思路是把Effect类型（这里是Task）提升到ConnectionIO的上下文里。因为ConnectionIO实现了Async类型类，咱们可以借助Async[ConnectionIO].liftIO方法完成转换——注意需要先把Task转成IO，因为liftIO接受的是IO参数。调整后的代码如下：

def getObject: ConnectionIO[Request] = ??? 
def saveObject(obj: Request): ConnectionIO[Request] = ??? 
def processObject(obj: Request): monix.eval.Task[Request] = ??? 

val transaction: ConnectionIO[Request] = for { 
  obj <- getObject //ConnectionIO[Request] 
  processed <- Async[ConnectionIO].liftIO(processObject(obj).toIO) //ConnectionIO[Request] 
  updated <- saveObject(processed) //ConnectionIO[Request] 
} yield updated 

val result: Task[Request] = transaction.transact(xa) 

这样一来，整个流程就都处在ConnectionIO的上下文里了。最后调用transact(xa)把整个ConnectionIO转换成Task执行时，所有数据库操作都会被包裹在单个事务中：只要中间任何一步出错，整个事务就会回滚；全部成功的话，就会提交所有变更。

内容的提问来源于stack exchange，提问作者Eugene Zhulkov