=str Add benchmark for MessageToFrameRenderer change.

http-bench-jmh/src/main/scala/org/apache/pekko/http/impl/engine/MessageToFrameRendererBenchmark.scala

@@ -0,0 +1,94 @@
+package org.apache.pekko.http.impl.engine
+
+import com.typesafe.config.ConfigFactory
+import org.apache.pekko.NotUsed
+import org.apache.pekko.actor.ActorSystem
+import org.apache.pekko.http.impl.engine.ws.FrameEvent
+import org.apache.pekko.http.impl.engine.ws.Protocol.Opcode
+import org.apache.pekko.stream.scaladsl.{ Flow, Keep, Sink, Source }
+import org.apache.pekko.util.ByteString
+import org.openjdk.jmh.annotations.{
+  Benchmark,
+  BenchmarkMode,
+  Mode,
+  OperationsPerInvocation,
+  OutputTimeUnit,
+  Scope,
+  State,
+  TearDown
+}
+
+import java.util.concurrent.TimeUnit
+import scala.concurrent.Await
+import scala.concurrent.duration.{ Duration, DurationInt }
+
+object MessageToFrameRendererBenchmark {
+  final val OperationsPerInvocation = 100000
+}
+
+@State(Scope.Benchmark)
+@OutputTimeUnit(TimeUnit.SECONDS)
+@BenchmarkMode(Array(Mode.Throughput))
+class MessageToFrameRendererBenchmark {
+  import MessageToFrameRendererBenchmark.OperationsPerInvocation
+
+  private val config = ConfigFactory.parseString("""
+    akka.actor.default-dispatcher {
+      executor = "fork-join-executor"
+      fork-join-executor {
+        parallelism-factor = 1
+      }
+    }
+    """)
+
+  private implicit val system: ActorSystem = ActorSystem("MessageToFrameRendererBenchmark", config)
+
+  @TearDown
+  def shutdown(): Unit = {
+    Await.result(system.terminate(), 5.seconds)
+  }
+
+  private val newstreamedFrames = Source.repeat(ByteString.empty)
+    .take(OperationsPerInvocation)
+    .statefulMap(() => true)((isFirst, data) => {
+        val frameOpcode = if (isFirst) Opcode.Pong else Opcode.Continuation
+        (false, FrameEvent.fullFrame(frameOpcode, None, data, fin = false))
+      }, _ => None)
+    .toMat(Sink.ignore)(Keep.right)
+
+  private val oldstreamedFrames =
+    Source
+      .repeat(ByteString.empty)
+      .take(OperationsPerInvocation)
+      .via(statefulMap(() => {
+        var isFirst = true
+
+        { data =>
+          val frameOpcode =
+            if (isFirst) {
+              isFirst = false
+              Opcode.Pong
+            } else Opcode.Continuation
+
+          FrameEvent.fullFrame(frameOpcode, None, data, fin = false)
+        }
+      }))
+      .toMat(Sink.ignore)(Keep.right)
+
+  def statefulMap[T, U](functionConstructor: () => T => U): Flow[T, U, NotUsed] =
+    Flow[T].statefulMapConcat { () =>
+      val f = functionConstructor()
+      i => f(i) :: Nil
+    }
+
+  @Benchmark
+  @OperationsPerInvocation(OperationsPerInvocation)
+  def benchOldStreamedFrames(): Unit =
+    Await.result(oldstreamedFrames.run(), Duration.Inf)
+
+  @Benchmark
+  @OperationsPerInvocation(OperationsPerInvocation)
+  def benchNewStreamedFrames(): Unit =
+    Await.result(newstreamedFrames.run(), Duration.Inf)
+
+}