danielwegener
commented
8 years ago After some first experiments I'd like to share my observations and ideas.
An akka-stream is potentially infinite, demand-driven sequence of elements that can be terminated (completed/cancelled) from the upstream or downstream side either successfully or with a failure at any time.
Akka uses ByteString as representation for a bunch of bytes while scodec uses BitVector/ByteVector. ByteString and ByteVector
are isomorphic while a BitVector can only be converted into a ByteVector by possibly introducing some padding.
Akka-streams have a single type T on each port such that each element emitted or consumed must be of the same type.
I see the following use-cases for StreamCodec[T]s:
Decoding a continious potentially huge (or infinite) blob Source[BitVector] into a Source[T]
- A decoder should preferably emit elements as early as possible and keep the internal buffer as small as possible
- If a single
BitVectorfrom the upstream is not sufficient to decode aT, theStreamCodecshould not fail but wait for the nextBitVector(or multiple, just enough to fulfill theInsufficientBitserr) - If the upstream completes while a
StreamCodecis still in the middle of a decoding, the stage should fail - If the downstream cancels at any time, the stage just complete normally (cancelling upstream and discarding any buffers)
- Examples: Video file that decodes single frames, durable TCP connections that decode a sequence of protocol messages.
Encoding a potentially infinite Source[T] into a Source[BitVector]
- An encoder should preferable emit elements as early as possible and keep the internal buffer as small as possible
- clarify do we need combinators that require the
StreamCodecto buffer the bitVector over multiple upstream elements? creating checksums could be a valid case.
Decoding a framed, possibly infinite, Source[BitVector]
- each
BitVectorshould be treated as a single decodable unit - Each single decoding attempt of an element can fail independently (while it should still be possible to fail the stage if a decoding error happens)
- It can possibly fail if there are remaining bits left after decoding
- Failed decoding attempts should be handled in-band (without failing the stage, maybe configurable)
- Examples: UDP datagrams
It turns out that this use case does not need a StreamCodec but can simply be done by mapping a Decoder over a Source[BitVector] to get a Attempt[DecodingResult] that can act propertly on remaning bits and decoding failures.
Encoding a framed Source[T] where each T encodes into zero or more BitVectors
- Each encoding attempt of a
Tcan fail independently (while it should be possible to fail the stage if an encoding error happens)
Turns out that this use case also does not need a StreamCodec since a Source[T] can be brought into a shape Source[T1] such that each T1 encodes in to zero or one Attempt[BitVector].
Further ideas
- Every stage we build, should be fusable
- We should not deal with throughput control, that can be handled by up-/downstream with a throttle stage
- Would be great if we could reuse the concept of
withToStringandwithContextfromCodec. Eg. if we have aStreamCodecuses aCodecand the inner codec fails, theStreamCodecshould emit an Err with the context of theCodecand theStreamCodec.
Closing the gap between BitVector/ByteVector and ByteString
The actual conversion mechanism is already solved by scodec-akka. We could additionally provide the following stages:
Flow[ByteVector,ByteString]andFlow[ByteString,ByteVector](simple)Flow[ByteString,BitVector](simple)pad:Flow[BitVector,ByteString]that pads every incomplete byte with zeroesjoin:Flow[BitVector,ByteString]that emits every complete byte and buffers the remainder and prepends it to the nextBitVector. It should fail if there are incomplete bytes left after upstream completes.
API
The following API proposal is inspired by scodecs own api and the scodec-stream API. The general idea is to describe a StreamCodec by combining different strict Codecs into a sequence of StreamCodecs that describe a lazily runnable chain of decoding computations that have safe exit points (StreamCodec becomes Complete) where upstream is allowed to complete without producing an Err.
I have yet looked merely at the decoding side of the problem since it seemed more complicated. I doubt that all of these combinators have useful/possible counterparts in the encoding domain. I tend to follow the scodec-streams example and leave the construction of StreamEncoders and StreamDecoders seperate beside the most simple combinators (see https://github.com/scodec/scodec-stream/blob/series/1.0.x/src/main/scala/scodec/stream/codec/StreamCodec.scala).
trait StreamDecoder[+A] extends StreamDecoderOps[A] { self =>
/**
* This is probably a very clunky attempt. I am happy to hear better ideas!
* @returns stream decoding state after running this decoder with the remainding BitVector.
* Could be: Emitting, NextDecoder, Fail, Complete
* - Emitting signals that this decoder emitted an element ready to be pushed downstream.
* - NextDecoder signals that this decoder is complete and has a StreamDecoder that will decode the remainding BitVector
* - Complete signals that this decoder does not expect any futher elements.
* - Fail signals that this decoder (remainder=bits)
* Turns out that a StreamDecoder can signal Emitting and Complete/Fail/NextDecoder at the same time.
* An implementing stage can control when to run the next step of this continuation like structure.
*/
def decode(bits:BitVector): ???
}
trait StreamDecoderOps[+A] {
/** Decode `this` and then decodes `sd` after `this` completes. Completes when `sd` completes.
* (should support A>:B too) */
def ++:[B>:A](sd:StreamDecoder[B]):StreamDecoder[B]
/** alias for [[StreamDecoder$.or]] */
def |[B>:A](d:StreamDecoder[B]):StreamDecoder[B]
}
object StreamDecoder {
/** Completes instantly.
* This is a neutral element for `++:` and `or` */
val complete: StreamCodec[Nothing]
/** Instantly fails with the given Err. */
def fail(err: =>Err): StreamDecoder[Nothing]
/** Decodes a single `A` using `d` and then completes. */
def once[A](d: Decoder[A]): StreamDecoder[A]
/** Like [[once]], but halts normally and leaves the
* input unconsumed in the event of a decoding error. */
def tryOnce[A](dec: Decoder[A]): StreamDecoder[A]
/** Decodes an `Iterable[A]` using `d`, emits each of its elements and then completes. */
def all[A](d:Decoder[Iterable[A]]):StreamDecoder[A]
/** Repeatedly runs `d` until upstream completes. Fails if upstream completes while the current
* decoding of `d` is incomplete. Or if the */
def many[A](d: StreamDecoder[A]): StreamDecoder[A]
/** Decodes `countCodec` to `n` and then decodes `valueCodec` `n` times.
* Completes after `n` values have been decoded successfully. */
def streamOfN[A](countCodec: Decoder[Int], valueCodec: StreamDecoder[A]):StreamDecoder[A]
/** Wait until `numberOfBits` bits are available, then runs `dec`.
* Runs `dec` if upstream completes even if the buffer has not yet reached `numberOfBits`.
* Should have an `chunkedBytes` equivalent. */
def chunked[A](dec:StreamDecoder[A])(numberOfBits: Long):StreamDecoder[A]
/** Runs `d` but discards all its emitted values */
def drop(d:StreamDecoder[_]): StreamDecoder[Nothing]
/** Instantly emits `a` without consuming any bits */
def provide[A](a:A):StreamDecoder[A]
/** Consumes `numberOfBits` and completes. Completes or fails if upstream terminates and the
* given `numberOfBits` are not yet consumed. Should have an ignoreBytes equivalent. */
def ignore(numberOfBits: Long, failIfUpstreamTerminates:Boolean = false):StreamDecoder[Nothing]
/** Run `d`` using only the first `numberOfBits` bits of
* the current stream, then advance the cursor by that many bits on completion. */
def isolate[A](numberOfBits: Long)(d: StreamDecoder[A]): StreamDecoder[A]
/** Runs `s1`, then runs `s2` if `s1` emits no elements.
* Fails if `s1` or `s2` fails.
* Completes if either `s1` completes after emitting an element or `d` completes. */
def or[A](s1: StreamDecoder[A], s2: StreamDecoder[A]):StreamDecoder[A]
}StreamDecoder API should look similar but is TBD
Stage Implementation considerations
The stage has to deal with possible states:
- Upstream completed:
- A
StreamDecoderStageshould try to decode the remaining bits from the buffer and emit elements if possible and then complete too (or fail if the usedStreamDecoderfails). - A
StreamEncoderStageshould try to encode the remaining elements from the buffer and then complete too (or fail if the usedStreamEncoderfails).
- A
- Downstream canceled
- For both stages: Simply cancel upstream and discard the buffer
- Upstream failed: TBD
- Downstream failed: TBD
I came up with an experimental (and not very elegant) prototype for decoding (ongoing): https://github.com/danielwegener/scodec-akka/blob/wip-stream-codecs/src/test/scala/scodec/interop/akka/stream/StreamDecoderTest.scala
WDYT? /cc @mpilquist
mpilquist
aloiscochard
eirirlar
unoexperto
Given scodec-streams, It would be great to have a similar mechanism for akka-streams that allows one to lift a
StreamEncoder[T]into aFlow[T,ByteString,_]StreamDecoder[T]into aFlow[ByteString,T,_]StreamCodec[T]into aBidiFlow[ByteString,T,T,ByteString,_]A dependency on scalaz-streams should not be required.