Akka stream stays idle instead of throwing exception

Question

I am new to Akka/Scala and am trying to debug the code below. When the resultSetParser has an exception, it does not throw it. Instead the service that uses this code just sits idle forever.

How can I make my service throw the exception rather than just waiting forever on the stream? Is there something like a watchException() function in Akka that I could call right after watchTermination() to make it throw the exception it sees when processing a stream?

val chunkSource: Source[ChunkStreamPart, NotUsed] =
  Source
    .fromIterator(() => resultSetParser(resultSet) map ChunkStreamPart.apply)
    .watchTermination()((mat : NotUsed, fut : Future[Done]) => {
      watchTermination(fut)
      mat
    })
val chunkEntity = Chunked(ContentTypes.`application/json`, chunkSource)

user_1111040 · Answer

Have you tried using recover?

For example (not tested):

Source
    .fromIterator(() => resultSetParser(resultSet) map ChunkStreamPart.apply)
    .recover{
        case _: RuntimeException => ??? /* Return ChunkStreamPart here */
    }

Akka stream stays idle instead of throwing exception

1 Answers