从 akka-stream 到 fs2 的旅程 - 如何使用 http4s 在 fs2 中定义类似阶段的 akka-stream http 流

Question

我正在加深我在 fs2 方面的知识，并想尝试 fs2-kafka 来替换 akka 流的用例。这个想法很简单，从 kafka 读取数据并通过 http 请求将数据发布到接收器，然后在成功时提交回 kafka。到目前为止，我无法真正弄清楚 http 部分。在 akka 流/akka http 中，你有一个开箱即用的流 https://doc.akka.io/docs/akka-http/current/client-side/host-level.html#using-a-host-connection-pool

Flow[(HttpRequest, T), (Try[HttpResponse], T), HostConnectionPool]

与 akka 流完美集成。

我试图看看我是否可以用 http4s 和 fs2 做类似的事情。

有没有人有任何参考资料、代码示例、博客以及其他说明如何进行这种集成的东西。到目前为止，我唯一能想到的是将流包装到客户端资源的使用方法中，即

BlazeClientBuilder[IO](IORuntime.global.compute).resource.use { ..... run stream here ..... }

即使这样我也不确定整个事情

Answer 1

typelevel 生态系统的特点是，一切都只是一个库，您不需要说明它们之间有多少交互的示例，您只需要了解每个库的工作原理和基本的组成规则。

def createClient(/** whatever arguments you need */): Resource[IO, Client[IO]] = {
  // Fill this based on the documentation of the client of your choice:
  // I would recommend the ember client from http4s:
  // https://http4s.org/v0.23/api/org/http4s/ember/client/emberclientbuilder 
}


def sendHttpRequest(client: Client[IO])(data: Data): IO[Result] = {
  // Fill this based on the documentation of your client:
  // https://http4s.org/v0.23/client/
  // https://http4s.org/v0.23/api/org/http4s/client/client
}

def getStreamOfRecords(/** whatever arguments you need */): Stream[IO, CommittableConsumerRecord[IO, Key, Data]] = {
  // Fill this based on the documentation of fs2-kafka:
  // https://fd4s.github.io/fs2-kafka/docs/consumers
}

def program(/** whatever arguments you need */): Stream[IO, Unit] = {
  // Based on the documentation of fs2 and fs2-kafka I would guess something like this:
  Stream.fromResource(createClient(...)).flatMap { client =>
    getStreamOfRecords(...).evalMapFilter { committable =>
      sendHttpRequest(client)(data = committable.record).map { result =>
        if (result.isSuccess) Some(committable.offset)
        else None
      }
    }.through(commitBatchWithin(...))
  }
}

object Main extends IOApp.Simple {
  override final val run: IO[Unit] =
    program(...).compile.drain
}

请注意，我是在头脑中编写了所有这些内容，并且只是快速浏览了文档，您需要更改许多内容（尤其是类型，例如 Data 和 Result） .以及调整错误处理和何时提交回 Kafka 等内容。
不过，我希望这有助于您了解如何构建代码。