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

Merge pull request #19982 from wlk/wlk-19500-make-simpleFramingProtocol-decoder-encoder-public

Browse source 
make simpleFramingProtocol decoder and encoder public
This commit is contained in:
Konrad Malawski 2016-03-09 20:59:58 +01:00
commit 87113f9b14
1 changed files with 14 additions and 5 deletions
Download patch file Download diff file Expand all files Collapse all files

19
akka-stream/src/main/scala/akka/stream/scaladsl/Framing.scala
View file

@ -74,8 +74,20 @@ object Framing {
* included in this limit.
*/
def simpleFramingProtocol(maximumMessageLength: Int): BidiFlow[ByteString, ByteString, ByteString, ByteString, NotUsed] = {
val decoder = lengthField(4, 0, maximumMessageLength + 4, ByteOrder.BIG_ENDIAN).map(_.drop(4))
val encoder = Flow[ByteString].transform(() ⇒ new PushStage[ByteString, ByteString] {
BidiFlow.fromFlowsMat(simpleFramingProtocolEncoder(maximumMessageLength), simpleFramingProtocolDecoder(maximumMessageLength))(Keep.left)
}
/**
* Protocol decoder that is used by [[Framing#simpleFramingProtocol]]
*/
def simpleFramingProtocolDecoder(maximumMessageLength: Int): Flow[ByteString, ByteString, NotUsed] =
lengthField(4, 0, maximumMessageLength + 4, ByteOrder.BIG_ENDIAN).map(_.drop(4))
/**
* Protocol encoder that is used by [[Framing#simpleFramingProtocol]]
*/
def simpleFramingProtocolEncoder(maximumMessageLength: Int): Flow[ByteString, ByteString, NotUsed] =
Flow[ByteString].transform(() ⇒ new PushStage[ByteString, ByteString] {
override def onPush(message: ByteString, ctx: Context[ByteString]): SyncDirective = {
val msgSize = message.size
if (msgSize > maximumMessageLength)
@ -87,9 +99,6 @@ object Framing {
}
})
BidiFlow.fromFlowsMat(encoder, decoder)(Keep.left)
}
class FramingException(msg: String) extends RuntimeException(msg)
private final val bigEndianDecoder: (ByteIterator, Int) ⇒ Int = (bs, length) ⇒ {