I’m trying to log the response entity, however when i unmarshal it within the logResult-directive i get the error: java.lang.NullPointerException: Deflater has been closed.
Example code:
import akka.http.scaladsl.server.directives.LoggingMagnet
import akka.http.scaladsl.server.{Directives, Route, RouteResult}
import akka.stream.Materializer
import org.entur.offers.ComponentRegistry
trait RouteBuilder extends Directives {
val routes: Route =
extractRequestContext { context =>
logResult(LoggingMagnet(_ => {
result: RouteResult => {
result match {
case RouteResult.Complete(response) => {
implicit val mat: Materializer = context.materializer
val a = Unmarshal(response.entity).to[String] // This line causes the error to appear
}
case RouteResult.Rejected(_) =>
}
}
})) {
ComponentRegistry.searchRestApi.routes // These are my api routes
}
}
}
I don’t understand what’s wrong with this code, can someone explain how to do this the correct way?
java.lang.NullPointerException: Deflater has been closed
at java.util.zip.Deflater.ensureOpen(Deflater.java:559)
at java.util.zip.Deflater.deflate(Deflater.java:440)
at java.util.zip.Deflater.deflate(Deflater.java:385)
at akka.http.scaladsl.coding.DeflateCompressor$.drainDeflater(Deflate.scala:82)
at akka.http.scaladsl.coding.DeflateCompressor.compressWithBuffer(Deflate.scala:48)
at akka.http.scaladsl.coding.GzipCompressor.compressWithBuffer(Gzip.scala:40)
at akka.http.scaladsl.coding.DeflateCompressor.compressAndFlush(Deflate.scala:34)
at akka.http.scaladsl.coding.Encoder.encodeChunk$1(Encoder.scala:38)
at akka.http.scaladsl.coding.Encoder.$anonfun$newEncodeTransformer$1(Encoder.scala:41)
at akka.http.impl.util.StreamUtils$$anon$2$$anon$6.onPush(StreamUtils.scala:34)
at akka.stream.impl.fusing.GraphInterpreter.processPush(GraphInterpreter.scala:519)
at akka.stream.impl.fusing.GraphInterpreter.processEvent(GraphInterpreter.scala:508)
at akka.stream.impl.fusing.GraphInterpreter.execute(GraphInterpreter.scala:378)
at akka.stream.impl.fusing.GraphInterpreterShell.runBatch(ActorGraphInterpreter.scala:588)
at akka.stream.impl.fusing.GraphInterpreterShell.init(ActorGraphInterpreter.scala:558)
at akka.stream.impl.fusing.ActorGraphInterpreter.tryInit(ActorGraphInterpreter.scala:679)
at akka.stream.impl.fusing.ActorGraphInterpreter.finishShellRegistration(ActorGraphInterpreter.scala:721)
at akka.stream.impl.fusing.ActorGraphInterpreter.akka$stream$impl$fusing$ActorGraphInterpreter$$shortCircuitBatch(ActorGraphInterpreter.scala:736)
at akka.stream.impl.fusing.ActorGraphInterpreter$$anonfun$receive$1.applyOrElse(ActorGraphInterpreter.scala:761)
at akka.actor.Actor.aroundReceive(Actor.scala:517)
at akka.actor.Actor.aroundReceive$(Actor.scala:515)
at akka.stream.impl.fusing.ActorGraphInterpreter.aroundReceive(ActorGraphInterpreter.scala:670)
at akka.actor.ActorCell.receiveMessage(ActorCell.scala:588)
at akka.actor.ActorCell.invoke(ActorCell.scala:557)
at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:258)
at akka.dispatch.Mailbox.run(Mailbox.scala:225)
at akka.dispatch.Mailbox.exec(Mailbox.scala:235)
at akka.dispatch.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)
at akka.dispatch.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339)
at akka.dispatch.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)
at akka.dispatch.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)
BUMP
I guess it is wrong to unmarshall the request/response-entities twice as I’m doing here, and I could just log the response before it is converted to an HttpEntity, however that means I would have to do logging on each separate endpoint, whereas with this solution it only happens at one place
Yes, akka-http is a streaming HTTP implementation, so you need to hook into the stream without consuming it. That’s the whole point of streaming: not having to keep the whole data in memory while processing. One way to do it is to add something to the end of your processing pipeline and use HttpResponse.transformEntityDataBytes to add a logger stage to the response without already reading the stream. You will then be able to receive the data while it is being streamed to the client. In any case, for streamed entities you will only have received the complete entity stream when it was also sent completely to the client.
Alternatively, a simple way to keep your code is to check whether the response entity is streamed or strict and only log the strict ones which are already in memory anyway. You can e.g. match on response.entity to check whether it’s an HttpEntity.Strict.
@jrudolph
I have tried both suggestions without luck unfortunately and a bit more help would be much appreciated!
HttpResponse.transformEntityDataBytes
I rewrote the code like this:
import akka.http.scaladsl.server.directives.LoggingMagnet
import akka.http.scaladsl.server.{Directives, Route, RouteResult}
import akka.stream.scaladsl.Flow
import akka.util.ByteString
import org.entur.offers.ComponentRegistry
trait RouteBuilder extends Directives {
val routes: Route =
logResult(LoggingMagnet(logger => {
result: RouteResult => {
result match {
case RouteResult.Complete(response) => {
response.transformEntityDataBytes(Flow.fromFunction {
responseEntity: ByteString =>
logger.info(responseEntity.utf8String)
responseEntity
})
}
case RouteResult.Rejected(_) =>
}
}
})) {
ComponentRegistry.searchRestApi.routes // These are my api routes
}
}
The problem now is that the statements within transformEntityDataBytes are never executed (i checked with a debugger) so I get no log message. I also tried to use the mapRouteResult-directive instead of the logResult-directive, but that did not change anything.
HttpEntity.Strict
When i tried matching I could only get a match for HttpEntity.Chunked which is weird because it is an HttpEntity.Strict when the application creates the response. It seems to me that the http-entity is beeing converted somewhere, but I don’t understand why/how that happens.