使用Doobie 0.9.0正确使用Monix 3.2.2 Observable

Question

我想将Monix Observable与Doobie (fs2)流一起使用，但似乎无法正常工作。没有流媒体，我的测试应用程序可以很好地退出，但在使用流媒体之后，我的TaskApp似乎挂起了，并且找不到原因。

下面是一个重现问题的最小示例：

package example

import java.util.concurrent.Executors

import doobie.implicits._
import cats.effect.{Blocker, ContextShift, ExitCode, Resource}
import doobie.hikari.HikariTransactor
import monix.eval.{Task, TaskApp}
import com.typesafe.scalalogging.StrictLogging
import fs2.interop.reactivestreams._
import monix.reactive.Observable

import scala.concurrent.ExecutionContext
 
object Hello extends TaskApp with StrictLogging {
    
  private def resources()(implicit contextShift: ContextShift[Task]): Resource[Task, Resources] = {
    for {
      transactor <- Database.transactor("org.postgresql.Driver", "jdbc:postgresql://localhost/fubar", "fubar", "fubar")
    } yield Resources(transactor)
  }

  def run(args: List[String]): Task[ExitCode] = resources().use(task)
    .flatMap(_ => Task { println("All Done!") })
    .flatMap(_ => Task(ExitCode.Success))
  
  def task(resources: Resources): Task[Unit] = {

    val publisher =
      
      sql"""select id from message;"""

        .query[(Long)]
        .stream
        .transact(resources.transactor)
        .toUnicastPublisher()

    Observable.fromReactivePublisher(publisher)
      .foreachL(id => logger.info(id.toString))
    
  }
  
}

case class Resources(transactor: HikariTransactor[Task])

object Database {

  val ecBlocking = ExecutionContext.fromExecutor(Executors.newFixedThreadPool(8))

  def transactor(dbDriver: String, dbUrl: String, dbUser: String, dbPassword: String)(implicit contextShift: ContextShift[Task]): Resource[Task, HikariTransactor[Task]] = {
    HikariTransactor.newHikariTransactor[Task](dbDriver, dbUrl, dbUser, dbPassword, ecBlocking, Blocker.liftExecutionContext(ecBlocking))
  }

}

我已经根据Monix文档将fs2流转换为Monix observable：https://monix.io/docs/current/reactive/observable.html#fs2

我是否需要以某种方式关闭fs2流或Observable以使应用程序干净地退出？感谢任何让它工作的技巧，或者是如何正确调试它的技巧。

Answer 1

问题是ExecutionContext需要关闭。请参阅作者的答案here。

正确的用法可以在in the documentation上看到。