Pekko/Akka中如何避免请求实体被多次物化？

你遇到的substream source cannot be materialized more than once错误，本质是Pekko Stream的Source只能被物化（启动消费）一次——哈希计算逻辑已经消费了请求实体的Source，而Http().outgoingConnection会尝试再次物化同一个Source，导致报错。不需要重新实现outgoingConnection，可以通过以下两种方案解决：

方案1：将实体转换为Strict类型（适合小请求） #

把请求实体转换成HttpEntity.Strict，实体内容会被缓存到内存中，可多次读取。这种方式简单直接，适合请求体较小的场景：

import org.apache.pekko.actor.ActorSystem
import org.apache.pekko.http.scaladsl.model.{HttpRequest, RawHeader}
import org.apache.pekko.stream.scaladsl.Source
import scala.concurrent.duration._
import scala.concurrent.Future

implicit val system: ActorSystem = ActorSystem()
import system.dispatcher

val originalRequest: HttpRequest = ... // 你的原始请求

// 将实体转换为Strict类型，设置合理的超时时间
val strictRequestFuture: Future[HttpRequest] = originalRequest.entity.toStrict(5.seconds)

// 计算哈希并构建新请求
val hashedRequestFuture: Future[HttpRequest] = strictRequestFuture.map { strictReq =>
  val payloadBytes = strictReq.entity.data
  // 替换成你的computeHashWithPayloadAndPayloadLength实现
  val sha256Hash = computeHashWithPayloadAndPayloadLength(payloadBytes.utf8String, payloadBytes.length)
  val updatedHeaders = strictReq.headers :+ RawHeader("X-SHA256", sha256Hash)
  strictReq.withHeaders(updatedHeaders)
}

// 发送处理后的请求
Source.fromFuture(hashedRequestFuture)
  .via(Http().outgoingConnection("your-target-host"))
  // 后续处理逻辑...

方案2：用广播流共享实体Source（适合大请求） #

如果请求体较大，不想占用过多内存缓存，可以用Pekko Stream的Broadcast操作将实体Source拆分为两个分支：一个分支用于计算哈希，另一个分支保留原流供outgoingConnection使用。这样两个分支共享同一个物化的流，避免重复物化问题：

import org.apache.pekko.actor.ActorSystem
import org.apache.pekko.http.scaladsl.model.{HttpRequest, HttpEntity, RawHeader}
import org.apache.pekko.stream.scaladsl.{Flow, Source, Sink, Broadcast, GraphDSL}
import org.apache.pekko.stream.{FlowShape, Graph}
import org.apache.pekko.util.ByteString
import scala.concurrent.Future

implicit val system: ActorSystem = ActorSystem()
import system.dispatcher

val originalRequest: HttpRequest = ... // 你的原始请求

// 定义哈希计算流：将ByteString流合并为完整 payload，计算哈希
val hashCalculationFlow: Flow[ByteString, String, _] = Flow[ByteString]
  .fold(ByteString.empty)(_ ++ _)
  .map { payload =>
    computeHashWithPayloadAndPayloadLength(payload.utf8String, payload.length)
  }

// 构建共享实体流的图：广播实体流到哈希计算和实体保留两个分支
val sharedEntityGraph: Graph[FlowShape[Source[ByteString, _], (String, Source[ByteString, _])], _] =
  GraphDSL.create() { implicit builder =>
    import GraphDSL.Implicits._

    val broadcast = builder.add(Broadcast[ByteString](2))
    val hashSink = builder.add(Sink.head[String])
    val entityBufferSink = builder.add(Sink.seq[ByteString])

    // 分支1：计算哈希
    broadcast.out(0) ~> hashCalculationFlow ~> hashSink
    // 分支2：缓存实体字节，后续重新生成Source
    broadcast.out(1) ~> entityBufferSink

    FlowShape(
      // 输入：原始实体Source
      broadcast.in,
      // 输出：(哈希值, 可复用的实体Source)
      builder.add(Flow.fromFunction { _ =>
        for {
          hash <- hashSink.materializedValue
          entityBytes <- entityBufferSink.materializedValue
        } yield (hash, Source(entityBytes))
      }).outlet
    )
  }

// 处理请求：共享实体流，生成带哈希头的新请求
val hashedRequestFuture: Future[HttpRequest] = Source.single(originalRequest.entity.dataBytes)
  .via(sharedEntityGraph)
  .runWith(Sink.head)
  .flatMap(_.map { case (hash, newEntitySource) =>
    val updatedHeaders = originalRequest.headers :+ RawHeader("X-SHA256", hash)
    originalRequest.copy(
      headers = updatedHeaders,
      entity = HttpEntity(originalRequest.entity.contentType, originalRequest.entity.contentLengthOption.getOrElse(0), newEntitySource)
    )
  })

// 发送请求
Source.fromFuture(hashedRequestFuture)
  .via(Http().outgoingConnection("your-target-host"))
  // 后续处理逻辑...

为什么.withEntity(out._2)能解决问题 #

out._2是你基于已消费的实体内容重新创建的新Source，这个新Source还未被物化过，所以outgoingConnection可以正常消费它。本质是绕过了原实体Source只能物化一次的限制。

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