如何将 akka actors 与 decline 整合

Question

我有一个带有 akka、akka http 和 akka 流的应用程序，所以它不是休息或微服务是一个短暂的应用程序；它是一个来自 rest api 的数据提取器，应用程序写入 postgres 数据库，它工作正常，我怀疑以这种方式集成 decline 和 akka 是否正确。

代码：

object App extends CommandIOApp(name = "app" , header = "extractor", version = "0.1.0") {

 
  val config = Config( token, jdbcDriver = jdbcDriver, jdbcURL = jdbcURL, jdbcUser = jdbcUser, jdbcPassword = jdbcPassword, jdbcSchema = jdbcSchema)

  override def main: Opts[IO[ExitCode]] =
    Opts.subcommand(name="extract-teams", help="Extract Teams from API"){

      val tableName = Opts.option[String]("recursive", short="r", help="Recursive Extraction")
      val outputFile = Opts.option[Path]("output", short="o", help="Output file").withDefault(Paths.get("output.csv"))

      ( tableName, outputFile).mapN{( table, output) =>
        println(table)
        println(output)

        ClickUpExtractions.extractTeams()

        IO(ExitCode.Success)
      }
    }

  object ClickUpExtractions extends ActorGlobalImplicits {

    def extractTeams(): Unit ={

      import com.arkondata.extractors.ClickUpTeamsActions._
      val extractorFetcher = system.actorOf(ClickUpTeamsFetcher.props(config), name = "EngineActor")
      val extractorWriter = system.actorOf(ClickUpTeamsBulk.props(config) , name = "Writer")

      extractorFetcher ! Fetch(extractorWriter)

    }

  }

}

接收码：

override def receive: Receive = {
    case WritePG(teamsData) =>
      println("-------writer")
      println(teamsData)
      val teamsPG: List[Teams] = teamsData.map(data => Teams(data.id, data.name, data.color, data.avatar))
      println(teamsPG)

      println("insert-----------------------")
      //// manejar exception aqui  importante!!!
      val upsertStatement = "insert into Teams (id, name, color, avatar) values ( ? , ? , ? , ?) ON CONFLICT(id) DO UPDATE SET name = EXCLUDED.name"
      BulkWriter(config).insertMany[Teams](teamsPG,upsertStatement).unsafeRunSync()//.unsafeToFuture()//.unsafeRunSync()
      //// manejar exception aqui  importante!!!
      println("insert doobie")
      println("end------------------------")

      self ! "mensaje salida myself!!!!"
    case msg:String =>
      println(msg)
      context.system.terminate()
      System.exit(1) // this is the last step in the akka flow
  }

注：

System.exit(1) // this is the last step in the akka flow

我还使用其他库，如 doobie、decline、cats：https://ben.kirw.in/decline/effect.html。这些应用程序工作只是我需要知道它是否正确或者 akka 和这个之间是否存在更好的集成：IO(ExitCode.Success)

有什么方法可以从 akka 得到一个 tell 响应并验证它，比如 :

val res = extractorFetcher ! Fetch(extractorWriter)
if (validate(res)) 
   IO(ExitCode.Success)
else 

   IO(ExitCode.Error)

Answer 1

一些建议：

您可以将 Akka 与其他基于猫的框架一起使用，但是您的应用程序中会有两种不同的行话：Fibers、ActorSystems、Futures 与 IO effects，...所以，如果您想工作与 Cats 一起使用 Monix 或 FS2，它们直接与 Cats 效果器连接。相反，如果 Akka 对你来说是强制性的，请使用 Akka Streams，它有一个很棒的套件，只处理 Akka 调度程序和 Scala Futures。
使用 Akka Typed。自从 Akka Typed 投入生产以来，Akka 有了很大的改进。您正在接收函数中创建异步计算，这很容易出错。

Answer 2

Akka 生态系统依赖于 Future。 Cats Effect 依赖于一些 F[_] - 内置的是 IO 但也有 Monix 的 Task 和 ZIO.

Translation Future <=> IO 可以像这样完成：

io.unsafeToFuture

IO.fromFuture(IO(start future here)) // Future start is side-effect on its own

同样适用于 Monix：

task.runToFuture // requires Scheduler

Task.deferFuture(future thunk) // or
Task.deferFutureAction(implicit scheduler => future)

如果您使用的是无标签，那么它是：

Async.fromFuture(Sync[F].delay(start future)) // requires implicit Async[F]

val runToFuture: F ~> Future // AFAIR there is no popular type class for it
runToFuture(io)

如果您正在使用流，那么有 streamz 处理 Akka Streams <=> FS2 翻译的库。 Akka Streams、Monix 的 Observable 实现和 FS2 提供了 Reactive Streams 的实现，因此您可以使这些库使用 RS 接口相互通信，因此像 streamz 这样的库只是一个方便的工具。

没有其他集成 AFAIK，也不需要它们。

但是，我建议了解一些基本知识，了解 Akka 和 Future 的工作原理（渴望、记忆、执行上下文等）与猫效应的工作原理（通常是懒惰、没有记忆、除了 Clock 的初始化之外没有 EC或 CE IO 中的 ContextShift，CE IO 和 Monix 之间的细微差别，例如调度程序）。如果你不知道它们之间的区别，你很容易造成一些伤害。

如何将 akka actors 与 decline 整合

how to integrate akka actors with decline

scala

akka

akka-stream

akka-http

scala-cats