从akka stream到fs2的旅程——如何使用http4s在fs2中定义akka stream http flow-like阶段

Question

我正在努力加深对fs2的了解，并想尝试fs2Kafka的一个用例，在这个用例中我将取代akka stream。想法很简单，从Kafka读取数据，通过http请求将数据发布到接收器，然后在成功后提交回Kafka。到目前为止，我还不能真正理解http部分。在akka stream/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 stream完美融合。

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

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

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

即使这样，我也不确定整件事

Answer 1

典型级生态系统的特点是，一切都只是一个库，你不需要例子来说明它们中有多少相互作用，你只需要了解每个库是如何工作的以及组成的基本规则。

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