diff --git a/demos/grpc-filter/src/main/scala/dev/constructive/eo/demo/grpc/GrpcFrames.scala b/demos/grpc-filter/src/main/scala/dev/constructive/eo/demo/grpc/GrpcFrames.scala index 12f10be1..e4744b66 100644 --- a/demos/grpc-filter/src/main/scala/dev/constructive/eo/demo/grpc/GrpcFrames.scala +++ b/demos/grpc-filter/src/main/scala/dev/constructive/eo/demo/grpc/GrpcFrames.scala @@ -19,7 +19,7 @@ object GrpcFrames: /** gRPC's default limit on a received message: 4 MiB. */ val DefaultMaxMessageBytes: Int = 4 * 1024 * 1024 - /** The bytes are not gRPC frames this decoder can read. It fails the stream, like the socket + /** The bytes are not gRPC frames [[splitFrames]] can split. It fails the stream, like the socket * errors around it: a gRPC peer answers a malformed frame by ending the call. */ final case class Malformed(reason: String) extends RuntimeException(reason) with NoStackTrace diff --git a/demos/grpc-filter/src/test/scala/dev/constructive/eo/demo/grpc/GrpcFramesSpec.scala b/demos/grpc-filter/src/test/scala/dev/constructive/eo/demo/grpc/GrpcFramesSpec.scala index 16f9781e..f37d80cf 100644 --- a/demos/grpc-filter/src/test/scala/dev/constructive/eo/demo/grpc/GrpcFramesSpec.scala +++ b/demos/grpc-filter/src/test/scala/dev/constructive/eo/demo/grpc/GrpcFramesSpec.scala @@ -11,9 +11,10 @@ import org.specs2.mutable.Specification /** gRPC framing over bytes that arrive in arbitrary pieces. */ class GrpcFramesSpec extends Specification with ScalaCheck: - import GrpcFramesSpec.pieces + import GrpcFramesSpec.{join, pieces} - def decode( + /** Bytes, in pieces, through [[GrpcFrames.splitFrames]]: the messages, or the failure. */ + def split( chunks: List[Chunk[Byte]], max: Int = GrpcFrames.DefaultMaxMessageBytes, ): Either[Throwable, List[List[Byte]]] = @@ -26,11 +27,8 @@ class GrpcFramesSpec extends Specification with ScalaCheck: .toList .map(_.map(_.toList)) - def framed(messages: List[Array[Byte]]): Array[Byte] = - Stream.emits(messages).through(GrpcFrames.joinFrames).compile.to(Array) - def malformed(bytes: Int*)(reason: String) = - decode(List(Chunk.array(bytes.map(_.toByte).toArray)), max = 10) must beLeft( + split(List(Chunk.array(bytes.map(_.toByte).toArray)), max = 10) must beLeft( GrpcFrames.Malformed(reason) ) @@ -39,7 +37,7 @@ class GrpcFramesSpec extends Specification with ScalaCheck: "messages come back whole however the bytes are cut" >> Prop.forAll(messages, sizes) { (messages, sizes) => - decode(pieces(framed(messages), sizes)) must beRight(messages.map(_.toList)) + split(pieces(join(messages), sizes)) must beRight(messages.map(_.toList)) } "a frame is a zero, the length big-endian, then the message" >> { @@ -47,12 +45,12 @@ class GrpcFramesSpec extends Specification with ScalaCheck: } "no bytes, no messages; an empty message is a message" >> { - decode(Nil) must beRight(Nil) - decode(List(GrpcFrames.frame(Array.emptyByteArray))) must beRight(List(Nil)) + split(Nil) must beRight(Nil) + split(List(GrpcFrames.frame(Array.emptyByteArray))) must beRight(List(Nil)) } "a message at the size limit passes; one byte more fails the stream" >> { - decode(List(GrpcFrames.frame(new Array[Byte](10))), max = 10) must beRight + split(List(GrpcFrames.frame(new Array[Byte](10))), max = 10) must beRight malformed(0, 0, 0, 0, 11)("a 11-byte message: over 10") } @@ -70,6 +68,10 @@ class GrpcFramesSpec extends Specification with ScalaCheck: object GrpcFramesSpec: + /** Messages through [[GrpcFrames.joinFrames]]: their frames, back to back. */ + def join(messages: Seq[Array[Byte]]): Array[Byte] = + Stream.emits(messages).through(GrpcFrames.joinFrames).compile.to(Array) + /** `bytes` in pieces of `sizes` (cycled), as a socket might deliver them. */ def pieces(bytes: Array[Byte], sizes: List[Int]): List[Chunk[Byte]] = @tailrec def go(at: Int, next: LazyList[Int], acc: List[Chunk[Byte]]): List[Chunk[Byte]] = diff --git a/demos/grpc-filter/src/test/scala/dev/constructive/eo/demo/grpc/PipelineSpec.scala b/demos/grpc-filter/src/test/scala/dev/constructive/eo/demo/grpc/PipelineSpec.scala index ed115efe..23133168 100644 --- a/demos/grpc-filter/src/test/scala/dev/constructive/eo/demo/grpc/PipelineSpec.scala +++ b/demos/grpc-filter/src/test/scala/dev/constructive/eo/demo/grpc/PipelineSpec.scala @@ -27,6 +27,7 @@ class PipelineSpec extends Specification with ScalaCheck: .getBytes("UTF-8") private val upstream = messages :+ task + private val upstreamBytes = GrpcFramesSpec.join(upstream) /** What the filter must forward: the messages the decoded issues say to keep, as they are. */ private val expected = upstream.filter(Triage.decoding(policy)(_).keep).map(_.toList) @@ -43,10 +44,9 @@ class PipelineSpec extends Specification with ScalaCheck: "in memory, however the bytes are cut, the filter forwards exactly the kept messages" >> Prop .forAll(Gen.nonEmptyListOf(Gen.choose(1, 20_000))) { sizes => - val bytes = Stream.emits(upstream).through(GrpcFrames.joinFrames).compile.to(Array) val (tally, out) = run(Ref.of[IO, Tally](Tally.empty).flatMap { tally => Stream - .emits(GrpcFramesSpec.pieces(bytes, sizes)) + .emits(GrpcFramesSpec.pieces(upstreamBytes, sizes)) .flatMap(Stream.chunk) .covary[IO] .through(Triage.filter(Triage.optics(policy), tally)) diff --git a/site/docs/cookbook/grpc-filter.md b/site/docs/cookbook/grpc-filter.md index 5c9a3a35..5d585077 100644 --- a/site/docs/cookbook/grpc-filter.md +++ b/site/docs/cookbook/grpc-filter.md @@ -51,9 +51,9 @@ message over gRPC's default 4 MiB limit, or on bytes that end inside a frame. `joinFrames` frames messages again, without copying them. ```scala -socketBytes // Stream[F, Byte], as the socket reads it - .through(GrpcFrames.splitFrames()) // Stream[F, Array[Byte]]: one per message - .through(GrpcFrames.joinFrames) // Stream[F, Byte]: the same frames again +socketBytes // Stream[F, Byte], as the socket reads it + .through(GrpcFrames.splitFrames()) // Stream[F, Array[Byte]]: one per message + .through(GrpcFrames.joinFrames) // Stream[F, Byte]: the same frames again ``` The demo puts these frames straight on TCP rather than inside HTTP/2 streams: @@ -263,7 +263,7 @@ full decode (baseline): 200000 messages, 813.3 MB in 4.16 s: | suite | checks | |---|---| -| `GrpcFramesSpec` | any messages, framed and cut into arbitrary pieces, decode back whole; the frame layout; the size limit (unsigned lengths included); compressed frames and truncated bytes fail the stream | +| `GrpcFramesSpec` | any messages, joined into frames and cut into arbitrary pieces, split back whole; the frame layout; the size limit (unsigned lengths included); compressed frames and truncated bytes fail the stream | | `TriageSpec` | the optics judge every message exactly as decoding does: 3 000 issues the size the stream's are, under the default policy and under 100 random ones, and 500 random issues of every size under random policies, every kind both kept and dropped; the slice `Wire.detail` hands out is the detail's own JSON; pretty-printed JSON is read the same; an unknown kind, a missing `detail` and bytes that aren't JSON are dropped as unclassified | | `TriageSpec` (laws) | `Model`'s optics pass cats-eo-laws' `OptionalTests` | | `PipelineSpec` | the whole filter, in memory with the bytes cut at random, and over real sockets: it forwards exactly the messages the decoded issues say to keep, byte for byte and in order, and its tally adds up; a malformed frame fails it |