Skip to content

HTTPS clone URL

Subversion checkout URL

You can clone with
or
.
Download ZIP
Browse files

fix a bug when body parser finishs before the end of the stream

  • Loading branch information...
commit ef70e641d9114ff8225332bf18b4dd995bd39bcc 1 parent b68f80f
Sadek Drobi authored
View
64 framework/src/play/src/main/scala/play/core/server/netty/RequestBodyHandler.scala
@@ -24,48 +24,56 @@ import scala.collection.JavaConverters._
private[server] trait RequestBodyHandler {
def newRequestBodyHandler[R](firstIteratee: Promise[Iteratee[Array[Byte], Either[Result, R]]], allChannels: DefaultChannelGroup, server: Server): (Promise[Iteratee[Array[Byte], Either[Result, R]]], SimpleChannelUpstreamHandler) = {
- var redeemed = false
+ import scala.concurrent.stm._
+ val redeemed = Ref(false)
var p = Promise[Iteratee[Array[Byte], Either[Result, R]]]()
val MAX_MESSAGE_WATERMARK = 10
val MIN_MESSAGE_WATERMARK = 10
- import scala.concurrent.stm._
val counter = Ref(0)
var iteratee: Ref[Iteratee[Array[Byte], Either[Result, R]]] = Ref(Iteratee.flatten(firstIteratee))
def pushChunk(ctx: ChannelHandlerContext, chunk: Input[Array[Byte]]) {
-
- if (counter.single.transformAndGet { _ + 1 } > MAX_MESSAGE_WATERMARK && ctx.getChannel.isOpen())
+ if (counter.single.transformAndGet { _ + 1 } > MAX_MESSAGE_WATERMARK && ctx.getChannel.isOpen() && !redeemed.single())
ctx.getChannel.setReadable(false)
- if (!redeemed) {
- val itPromise = Promise[Iteratee[Array[Byte], Either[Result, R]]]()
- val current = iteratee.single.swap(Iteratee.flatten(itPromise))
- val next = current.pureFlatFold[Array[Byte], Either[Result, R]] {
- case Step.Done(_, _) => current
- case Step.Cont(k) => k(chunk)
- case Step.Error(e, _) => current
- }
-
- itPromise.redeem(next)
+ val itPromise = Promise[Iteratee[Array[Byte], Either[Result, R]]]()
+ val current = atomic { implicit txn =>
+ if(!redeemed())
+ Some(iteratee.single.swap(Iteratee.flatten(itPromise)))
+ else None
+ }
- next.pureFold {
- case Step.Done(a, e) => if (!redeemed) {
- p.redeem(next);
- iteratee = null; p = null; redeemed = true
- if (ctx.getChannel.isOpen()) ctx.getChannel.setReadable(true)
- }
- case Step.Cont(k) =>
- if (counter.single.transformAndGet { _ - 1 } <= MIN_MESSAGE_WATERMARK && ctx.getChannel.isOpen())
- ctx.getChannel.setReadable(true)
-
- case Step.Error(msg, e) =>
- if (!redeemed) {
- p.redeem(Done(Left(Results.InternalServerError), e))
- iteratee = null; p = null; redeemed = true
+ current.foreach{ i =>
+ i.unflatten.extend1 {
+ case Redeemed(Step.Cont(k)) =>
+ val next = k(chunk)
+ continue(next)
+ case Redeemed(finished) =>
+ finish(finished.it)
+ case Thrown(e) =>
+ if (!redeemed.single.swap(true)) {
+ p.throwing(e)
+ iteratee = null; p = null;
if (ctx.getChannel.isOpen()) ctx.getChannel.setReadable(true)
}
+ itPromise.throwing(e)
+ }
+ }
+
+ def continue(it:Iteratee[Array[Byte], Either[Result, R]]){
+ if (counter.single.transformAndGet { _ - 1 } <= MIN_MESSAGE_WATERMARK && ctx.getChannel.isOpen())
+ ctx.getChannel.setReadable(true)
+ itPromise.redeem(it)
+ }
+
+ def finish(it:Iteratee[Array[Byte], Either[Result, R]]){
+ if (!redeemed.single.swap(true)) {
+ p.redeem(it)
+ iteratee = null; p = null;
+ if (ctx.getChannel.isOpen()) ctx.getChannel.setReadable(true)
}
+ itPromise.redeem(it)
}
}
Please sign in to comment.
Something went wrong with that request. Please try again.