refactor(demos): framing test helpers are split / join, like the pipes

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

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,20 +37,20 @@ 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" >> {
     GrpcFrames.frame(Array[Byte](7, 8, 9)).toList === List[Byte](0, 0, 0, 0, 3, 7, 8, 9)
   }
 
   "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]] =

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))

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 |