如何在返回Akka Stream的方法中复用ScalikeJDBC调用方事务？

我无法在返回Akka Stream的方法中复用ScalikeJDBC调用方的会话。我正尝试编写一个数据库访问类，返回Source用于Akka流应用。以下代码可正常运行：

def retrieveMembers(): Source[Member, NotUsed] =
    val publisher: DatabasePublisher[Member] =  DB readOnlyStream { Member.streamAll }
    Source.fromPublisher(publisher)

但该方法会创建独立事务，因此我参考AutoSession的建议修改了方法：

def retrieveMembers()(implicit session: DBSession = AutoSession): Source[Member, NotUsed] =
    val publisher: DatabasePublisher[Member] =  session readOnlyStream { Member.streamAll }
    Source.fromPublisher(publisher)

但修改后无法运行，因为readOnlyStream仅在DB类中定义，DBSession无此方法。请问如何修改方法以复用调用方事务？

方式一：通过DB.withSession绑定传入的会话 #

直接利用DB.withSession将传入的会话绑定到流操作中，确保复用调用方的事务上下文：

def retrieveMembers()(implicit session: DBSession = AutoSession): Source[Member, NotUsed] = {
  val publisher: DatabasePublisher[Member] = DB.withSession(session) { implicit s =>
    Member.streamAll()
  }
  Source.fromPublisher(publisher)
}

这里Member.streamAll()直接返回DatabasePublisher[Member]，通过DB.withSession复用传入的会话，流操作会自动加入调用方的事务中。

方式二：给DBSession扩展流操作方法 #

如果希望保持原方法的调用风格，可以给DBSession扩展一个隐式方法，让它支持readOnlyStream：

import scalikejdbc._
import akka.stream.scaladsl.Source
import java.sql.DatabasePublisher

// 扩展DBSession的隐式方法
implicit class DBSessionStreamOps(session: DBSession) {
  def readOnlyStream[A](block: DBSession => DatabasePublisher[A]): DatabasePublisher[A] = {
    DB.withSession(session)(block)
  }
}

之后你的原方法就可以正常运行：

def retrieveMembers()(implicit session: DBSession = AutoSession): Source[Member, NotUsed] = {
  val publisher: DatabasePublisher[Member] = session.readOnlyStream { s => Member.streamAll() }
  Source.fromPublisher(publisher)
}

重要提示 #

• Akka Stream是惰性求值的，必须保证会话在流处理全程保持有效，避免提前关闭会话导致流操作抛出异常。
• 如果调用方的会话处于事务中（比如通过DB.localTx创建的会话），流操作会自动参与该事务，实现事务的复用。

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