monitor websocket messages sent per endpoint

src/main/scala/Controller.scala

@@ -300,7 +300,8 @@ object Controller:
   final class Endpoint(
       val behavior: ClientEmit => ClientBehavior,
       val rateLimit: RateLimit,
-      val header: RequestHeader
+      val header: RequestHeader,
+      val emitCounter: kamon.metric.Counter
   )
   def endpoint(
       name: String,
@@ -318,7 +319,8 @@ object Controller:
           intervalMillis = interval.toMillis.toInt,
           name = name
         ),
-        header
+        header,
+        Monitor.clientInCounter(name)
       )
 
   type ResponseSync = Either[HttpResponseStatus, Endpoint]

src/main/scala/Monitor.scala

@@ -8,6 +8,7 @@ import kamon.tag.TagSet
 import java.util.concurrent.TimeUnit
 
 import lila.ws.util.Domain
+import kamon.metric.Counter
 
 final class Monitor(
     config: Config,
@@ -75,6 +76,9 @@ object Monitor:
       .withTag("name", name)
       .increment()
 
+  def clientInCounter(endpointName: String): Counter =
+    Kamon.counter("client.in").withTag("name", endpointName)
+
   val historyRoomSize  = Kamon.gauge("history.room.size").withoutTags()
   val historyRoundSize = Kamon.gauge("history.round.size").withoutTags()
 

src/main/scala/netty/ActorChannelConnector.scala

@@ -15,7 +15,11 @@ final private class ActorChannelConnector(clients: ActorRef[Clients.Control])(us
   def apply(endpoint: Endpoint, channel: Channel): Unit =
     val clientPromise = Promise[Client]()
     channel.attr(key.client).set(clientPromise.future)
-    clients ! Clients.Control.Start(endpoint.behavior(emitToChannel(channel)), clientPromise)
+    val channelEmit = emitToChannel(channel)
+    val monitoredEmit: ClientEmit = (msg: ipc.ClientIn) =>
+      endpoint.emitCounter.increment()
+      channelEmit(msg)
+    clients ! Clients.Control.Start(endpoint.behavior(monitoredEmit), clientPromise)
     channel.closeFuture.addListener:
       new GenericFutureListener[NettyFuture[Void]]:
         def operationComplete(f: NettyFuture[Void]): Unit =