Fix RTT when SSRC rewriting is enabled. (#2237)

* fix: Drop the packet if a preProcessor exists and rejects it.

* fix: Also pre-process RTCP packets.

* fix: Move the notifyRtcpSent call to the sender pipeline.

Notably, this calls the notifier after the RTCP packet has passed
the preProcessor where SSRC rewriting may change the packet.

The callback is also moved from the receiver pipeline thread (calling
into Conference.sendOut -> Endpoint.send) to the sender pipeline thread.

The only code which uses this callback is EndpointConnectionStats,
which saves sent SRs to calculate RTT. This commit fixes the RTT
calculation when SSRC rewriting is used.

jitsi-media-transform/src/main/kotlin/org/jitsi/nlj/RtpSenderImpl.kt

@@ -34,14 +34,15 @@ import org.jitsi.nlj.transform.NodeTeardownVisitor
 import org.jitsi.nlj.transform.node.AudioRedHandler
 import org.jitsi.nlj.transform.node.ConsumerNode
 import org.jitsi.nlj.transform.node.Node
+import org.jitsi.nlj.transform.node.ObserverNode
 import org.jitsi.nlj.transform.node.PacketCacher
 import org.jitsi.nlj.transform.node.PacketLossConfig
 import org.jitsi.nlj.transform.node.PacketLossNode
 import org.jitsi.nlj.transform.node.PacketStreamStatsNode
+import org.jitsi.nlj.transform.node.PluggableTransformerNode
 import org.jitsi.nlj.transform.node.SrtcpEncryptNode
 import org.jitsi.nlj.transform.node.SrtpEncryptNode
 import org.jitsi.nlj.transform.node.ToggleablePcapWriter
-import org.jitsi.nlj.transform.node.TransformerNode
 import org.jitsi.nlj.transform.node.outgoing.AbsSendTime
 import org.jitsi.nlj.transform.node.outgoing.HeaderExtEncoder
 import org.jitsi.nlj.transform.node.outgoing.HeaderExtStripper
@@ -141,12 +142,7 @@ class RtpSenderImpl(
         incomingPacketQueue.setErrorHandler(queueErrorCounter)
 
         outgoingRtpRoot = pipeline {
-            node(object : TransformerNode("Pre-processor") {
-                override fun transform(packetInfo: PacketInfo): PacketInfo? {
-                    return preProcesor?.invoke(packetInfo) ?: packetInfo
-                }
-                override fun trace(f: () -> Unit) {}
-            })
+            node(PluggableTransformerNode("RTP pre-processor") { preProcesor })
             node(AudioRedHandler(streamInformationStore, logger))
             node(HeaderExtStripper(streamInformationStore))
             node(outgoingPacketCache)
@@ -173,6 +169,7 @@ class RtpSenderImpl(
 
         // TODO: are we setting outgoing rtcp sequence numbers correctly? just add a simple node here to rewrite them
         outgoingRtcpRoot = pipeline {
+            node(PluggableTransformerNode("RTCP pre-processor") { preProcesor })
             node(keyframeRequester)
             node(SentRtcpStats())
             // TODO(brian): not sure this is a great idea.  it works as a catch-call but can also be error-prone
@@ -189,6 +186,16 @@ class RtpSenderImpl(
             }
             node(rtcpSrUpdater)
             node(toggleablePcapWriter.newObserverNode(outbound = true))
+            node(object : ObserverNode("RTCP sent notifier") {
+                override fun observe(packetInfo: PacketInfo) {
+                    val packet = packetInfo.packet
+                    if (packet is RtcpPacket) {
+                        rtcpEventNotifier.notifyRtcpSent(packet)
+                    }
+                }
+
+                override fun trace(f: () -> Unit) {}
+            })
             node(srtcpEncryptWrapper)
             node(packetStreamStats.createNewNode())
             node(PacketLossNode(packetLossConfig), condition = { packetLossConfig.enabled })
@@ -216,10 +223,6 @@ class RtpSenderImpl(
      */
     override fun doProcessPacket(packetInfo: PacketInfo) {
         if (running) {
-            val packet = packetInfo.packet
-            if (packet is RtcpPacket) {
-                rtcpEventNotifier.notifyRtcpSent(packet)
-            }
             packetInfo.addEvent(PACKET_QUEUE_ENTRY_EVENT)
             incomingPacketQueue.add(packetInfo)
         } else {

jitsi-media-transform/src/main/kotlin/org/jitsi/nlj/transform/node/Node.kt

@@ -393,6 +393,19 @@ abstract class TransformerNode(name: String) : StatsKeepingNode(name) {
     }
 }
 
+/** A [TransformerNode] which gets its transformation function dynamically. */
+class PluggableTransformerNode(
+    name: String,
+    val transform: () -> ((PacketInfo) -> PacketInfo?)?
+) : TransformerNode(name) {
+    override fun transform(packetInfo: PacketInfo): PacketInfo? {
+        transform()?.let { return it.invoke(packetInfo) }
+        return packetInfo
+    }
+    override fun trace(f: () -> Unit) {}
+    override val aggregationKey = this.name
+}
+
 /**
  * Unlike a [TransformerNode], [ModifierNode] modifies a packet in-place and never
  * outright 'fails', meaning the original [PacketInfo] will *always* be forwarded.