akka issue Substream Source(EntitySource) cannot be materialized more than once

user_8817285 0
Viewed 492

I have this error when I try to do an unmarshal twice to the response, this is because I need the response as case class and I also need it as the json-string because with the object I use it in a first-flow and the json-string in another second-flow so that's why I need both.

error log:

21:08:16.174 [Extractor-akka.actor.default-dispatcher-14] DEBUG akka.actor.RepointableActorRef - Aborting tcp connection to api.clickup.com:443 because of upstream failure: java.lang.IllegalStateException: Substream Source(EntitySource) cannot be materialized more than once
akka.stream.impl.fusing.SubSource$$anon$11.createMaterializedTwiceException(StreamOfStreams.scala:833)
akka.stream.impl.fusing.SubSource$$anon$11.<init>(StreamOfStreams.scala:804)

code:

Http().singleRequest(HttpRequest( method = GET ,uri = s"https://api.xxxxxx.com/task?archived=false&page=${pag}&subtasks=true&include_closed=true" )
        .withHeaders(RawHeader( CLICKUP_AUTH, config.token )))
      .flatMap {
        res =>
          val bodyString:Future[String] = Unmarshal(res).to[String]
          val tasks: Future[TasksItem] = Unmarshal(res).to[TasksItem]

          (tasks zip( bodyString)).map{
            case (a,b) =>
              TaskUnmarshall( a, b)
          }
      }

How can i unmarshal twice a single response from akka streams and get both ?

akka-stream akka-http akka scala
1 Answers

The problem is that the body is the actual stream of bytes across the wire, so that's why you cannot consume them twice, they only arrive once.

If it is fine for your use case to pull the entire body into memory, you can first make the response body strict (load it into memory) and then unmarshall it any number of times:

Http().singleRequest(HttpRequest(uri = "https://somewhere.com/"))
  .flatMap { response =>
    response.entity.toStrict(5.seconds).map { allBodyInMemory =>
      response.withEntity(allBodyInMemory)
    }
  }.flatMap { inMemoryResponse =>
    val json = Unmarshal(inMemoryResponse).to[String]
    val other = Unmarshal(inMemoryResponse).to[TasksItem]
    json.zip(other)
  }

Not sure if it would be possible to consume the stream with two downstreams using the Akka HTTP Unmarshalling API, but in general with Akka Streams that is possible with something like someSource.alsoTo(downstream1).runWith(downstream2)

user_12191250
Related