raboof/pekko
1
0
Fork 0
Code Issues Pull requests Projects Releases 6 Packages Wiki Activity Actions 10

Merge pull request #17684 from spray/w/17683-fix-message-parser

Browse source 
=htc #17683 drop already consumed input in `continue` (+ ByteString compaction)
This commit is contained in:
drewhk 2015-06-12 10:46:55 +02:00
commit 446a616064
3 changed files with 9 additions and 7 deletions
Download patch file Download diff file Expand all files Collapse all files

10
akka-http-core/src/main/scala/akka/http/impl/engine/parsing/HttpMessageParser.scala
View file

@ -170,11 +170,11 @@ private[http] abstract class HttpMessageParser[Output >: MessageOutput <: Parser
val remainingInputBytes = input.length - bodyStart
if (remainingInputBytes > 0) {
if (remainingInputBytes < remainingBodyBytes) {
emit(EntityPart(input drop bodyStart))
emit(EntityPart(input.drop(bodyStart).compact))
continue(parseFixedLengthBody(remainingBodyBytes - remainingInputBytes, isLastMessage))
} else {
val offset = bodyStart + remainingBodyBytes.toInt
emit(EntityPart(input.slice(bodyStart, offset)))
emit(EntityPart(input.slice(bodyStart, offset).compact))
emit(MessageEnd)
setCompletionHandling(CompletionOk)
if (isLastMessage) terminate()
@ -211,7 +211,7 @@ private[http] abstract class HttpMessageParser[Output >: MessageOutput <: Parser
if (chunkSize > 0) {
val chunkBodyEnd = cursor + chunkSize
def result(terminatorLen: Int) = {
emit(EntityChunk(HttpEntity.Chunk(input.slice(cursor, chunkBodyEnd), extension)))
emit(EntityChunk(HttpEntity.Chunk(input.slice(cursor, chunkBodyEnd).compact, extension)))
Trampoline(_ ⇒ parseChunk(input, chunkBodyEnd + terminatorLen, isLastMessage))
}
byteChar(input, chunkBodyEnd) match {
@ -252,7 +252,9 @@ private[http] abstract class HttpMessageParser[Output >: MessageOutput <: Parser
def continue(input: ByteString, offset: Int)(next: (ByteString, Int) ⇒ StateResult): StateResult = {
state =
math.signum(offset - input.length) match {
case -1 ⇒ more ⇒ next(input ++ more, offset)
case -1 ⇒
val remaining = input.drop(offset)
(more ⇒ next(remaining ++ more, 0))
case 0 ⇒ next(_, 0)
case 1 ⇒ throw new IllegalStateException
}

2
akka-http-core/src/main/scala/akka/http/impl/engine/parsing/HttpResponseParser.scala
View file

@ -132,7 +132,7 @@ private[http] class HttpResponseParser(_settings: ParserSettings, _headerParser:
// currently we do not check for `settings.maxContentLength` overflow
def parseToCloseBody(input: ByteString, bodyStart: Int): StateResult = {
if (input.length > bodyStart)
emit(EntityPart(input drop bodyStart))
emit(EntityPart(input.drop(bodyStart).compact))
continue(parseToCloseBody)
}
}

4
akka-http-core/src/main/scala/akka/http/impl/engine/ws/FrameEventParser.scala
View file

@ -87,7 +87,7 @@ private[http] class FrameEventParser extends ByteStringParserStage[FrameEvent] {
if (thisFrameData.length == length) ReadFrameHeader
else readData(length - thisFrameData.length)
pushAndBecomeWithRemaining(FrameStart(header, thisFrameData), nextState, remaining, ctx)
pushAndBecomeWithRemaining(FrameStart(header, thisFrameData.compact), nextState, remaining, ctx)
}
}
@ -103,7 +103,7 @@ private[http] class FrameEventParser extends ByteStringParserStage[FrameEvent] {
val frameData = elem.take(remaining.toInt)
val remainingData = elem.drop(remaining.toInt)
pushAndBecomeWithRemaining(FrameData(frameData, lastPart = true), ReadFrameHeader, remainingData, ctx)
pushAndBecomeWithRemaining(FrameData(frameData.compact, lastPart = true), ReadFrameHeader, remainingData, ctx)
}
}