[Cast Streaming] Add new StatisticsDispatcher class

This patch resolves a longish standing TODO to move the statistics
generation code out of the Sender class and into its own, unit tested
class. This class is called StatisticsDispatcher.

Bug: 298277160
Change-Id: I755beb6c4222f65a94a9b5503a708c431e52bc1b
Reviewed-on: https://chromium-review.googlesource.com/c/openscreen/+/6398046
Commit-Queue: Jordan Bayles <jophba@chromium.org>
Reviewed-by: Muyao Xu <muyaoxu@google.com>

cast/streaming/BUILD.gn

@@ -100,6 +100,8 @@ openscreen_source_set("common") {
     "impl/statistics_collector.h",
     "impl/statistics_defines.cc",
     "impl/statistics_defines.h",
+    "impl/statistics_dispatcher.cc",
+    "impl/statistics_dispatcher.h",
     "public/answer_messages.cc",
     "public/capture_recommendations.cc",
     "public/encoded_frame.cc",
@@ -284,6 +286,7 @@ openscreen_source_set("unittests") {
     "impl/session_messenger_unittest.cc",
     "impl/statistics_analyzer_unittest.cc",
     "impl/statistics_collector_unittest.cc",
+    "impl/statistics_dispatcher_unittest.cc",
     "impl/statistics_unittest.cc",
     "message_fields_unittest.cc",
     "rtp_time_unittest.cc",

cast/streaming/impl/statistics_dispatcher.cc

@@ -0,0 +1,144 @@
+// Copyright 2025 The Chromium Authors
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#include "cast/streaming/impl/statistics_dispatcher.h"
+
+#include <utility>
+
+#include "cast/streaming/encoded_frame.h"
+#include "cast/streaming/impl/rtcp_common.h"
+#include "cast/streaming/impl/rtp_defines.h"
+#include "cast/streaming/impl/session_config.h"
+#include "cast/streaming/impl/statistics_collector.h"
+#include "cast/streaming/impl/statistics_defines.h"
+#include "cast/streaming/public/environment.h"
+#include "platform/base/trivial_clock_traits.h"
+#include "util/chrono_helpers.h"
+#include "util/osp_logging.h"
+#include "util/std_util.h"
+#include "util/trace_logging.h"
+
+namespace openscreen::cast {
+
+using clock_operators::operator<<;
+
+StatisticsDispatcher::StatisticsDispatcher(Environment& environment)
+    : environment_(environment) {}
+StatisticsDispatcher::~StatisticsDispatcher() = default;
+
+void StatisticsDispatcher::DispatchEnqueueEvents(StreamType stream_type,
+                                                 const EncodedFrame& frame) {
+  if (!environment_.statistics_collector()) {
+    return;
+  }
+  const StatisticsEventMediaType media_type = ToMediaType(stream_type);
+
+  // Submit a capture begin event.
+  FrameEvent capture_begin_event;
+  capture_begin_event.type = StatisticsEventType::kFrameCaptureBegin;
+  capture_begin_event.media_type = media_type;
+  capture_begin_event.rtp_timestamp = frame.rtp_timestamp;
+  capture_begin_event.timestamp =
+      (frame.capture_begin_time > Clock::time_point::min())
+          ? frame.capture_begin_time
+          : environment_.now();
+  environment_.statistics_collector()->CollectFrameEvent(
+      std::move(capture_begin_event));
+
+  // Submit a capture end event.
+  FrameEvent capture_end_event;
+  capture_end_event.type = StatisticsEventType::kFrameCaptureEnd;
+  capture_end_event.media_type = media_type;
+  capture_end_event.rtp_timestamp = frame.rtp_timestamp;
+  capture_end_event.timestamp =
+      (frame.capture_end_time > Clock::time_point::min())
+          ? frame.capture_end_time
+          : environment_.now();
+  environment_.statistics_collector()->CollectFrameEvent(
+      std::move(capture_end_event));
+
+  // Submit an encoded event.
+  FrameEvent encode_event;
+  encode_event.timestamp = environment_.now();
+  encode_event.type = StatisticsEventType::kFrameEncoded;
+  encode_event.media_type = media_type;
+  encode_event.rtp_timestamp = frame.rtp_timestamp;
+  encode_event.frame_id = frame.frame_id;
+  encode_event.size = static_cast<uint32_t>(frame.data.size());
+  encode_event.key_frame =
+      frame.dependency == openscreen::cast::EncodedFrame::Dependency::kKeyFrame;
+
+  environment_.statistics_collector()->CollectFrameEvent(
+      std::move(encode_event));
+}
+
+void StatisticsDispatcher::DispatchAckEvent(StreamType stream_type,
+                                            RtpTimeTicks rtp_timestamp,
+                                            FrameId frame_id) {
+  if (!environment_.statistics_collector()) {
+    return;
+  }
+
+  FrameEvent ack_event;
+  ack_event.timestamp = environment_.now();
+  ack_event.type = StatisticsEventType::kFrameAckReceived;
+  ack_event.media_type = ToMediaType(stream_type);
+  ack_event.rtp_timestamp = rtp_timestamp;
+  ack_event.frame_id = frame_id;
+
+  environment_.statistics_collector()->CollectFrameEvent(std::move(ack_event));
+}
+
+void StatisticsDispatcher::DispatchFrameLogMessages(
+    StreamType stream_type,
+    const std::vector<RtcpReceiverFrameLogMessage>& messages) {
+  if (!environment_.statistics_collector()) {
+    return;
+  }
+
+  const Clock::time_point now = environment_.now();
+  const StatisticsEventMediaType media_type = ToMediaType(stream_type);
+  for (const RtcpReceiverFrameLogMessage& log_message : messages) {
+    for (const RtcpReceiverEventLogMessage& event_message :
+         log_message.messages) {
+      switch (event_message.type) {
+        case StatisticsEventType::kPacketReceived: {
+          PacketEvent event;
+          event.timestamp = event_message.timestamp;
+          event.received_timestamp = now;
+          event.type = event_message.type;
+          event.media_type = media_type;
+          event.rtp_timestamp = log_message.rtp_timestamp;
+          event.packet_id = event_message.packet_id;
+          environment_.statistics_collector()->CollectPacketEvent(
+              std::move(event));
+        } break;
+
+        case StatisticsEventType::kFrameAckSent:
+        case StatisticsEventType::kFrameDecoded:
+        case StatisticsEventType::kFramePlayedOut: {
+          FrameEvent event;
+          event.timestamp = event_message.timestamp;
+          event.received_timestamp = now;
+          event.type = event_message.type;
+          event.media_type = media_type;
+          event.rtp_timestamp = log_message.rtp_timestamp;
+          if (event.type == StatisticsEventType::kFramePlayedOut) {
+            event.delay_delta = event_message.delay;
+          }
+          environment_.statistics_collector()->CollectFrameEvent(
+              std::move(event));
+        } break;
+
+        default:
+          OSP_VLOG << "Received log message via RTCP that we did not expect, "
+                      "StatisticsEventType="
+                   << static_cast<int>(event_message.type);
+          break;
+      }
+    }
+  }
+}
+
+}  // namespace openscreen::cast

cast/streaming/impl/statistics_dispatcher.h

@@ -0,0 +1,51 @@
+// Copyright 2025 The Chromium Authors
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#ifndef CAST_STREAMING_IMPL_STATISTICS_DISPATCHER_H_
+#define CAST_STREAMING_IMPL_STATISTICS_DISPATCHER_H_
+
+#include <vector>
+
+#include "cast/streaming/impl/statistics_defines.h"
+#include "platform/api/time.h"
+#include "platform/base/span.h"
+
+namespace openscreen::cast {
+
+class StatisticsCollector;
+class Environment;
+struct EncodedFrame;
+struct RtcpReceiverFrameLogMessage;
+
+// This class is responsible for dispatching statistics events.
+class StatisticsDispatcher {
+ public:
+  explicit StatisticsDispatcher(Environment& environment);
+
+  StatisticsDispatcher(const StatisticsDispatcher&) = delete;
+  StatisticsDispatcher& operator=(const StatisticsDispatcher&) = delete;
+  StatisticsDispatcher(StatisticsDispatcher&&) noexcept = delete;
+  StatisticsDispatcher& operator=(StatisticsDispatcher&&) = delete;
+  ~StatisticsDispatcher();
+
+  // Dispatches enqueue events for a given frame.
+  void DispatchEnqueueEvents(StreamType stream_type, const EncodedFrame& frame);
+
+  // Dispatches frame log messages.
+  void DispatchFrameLogMessages(
+      StreamType stream_type,
+      const std::vector<RtcpReceiverFrameLogMessage>& messages);
+
+  // Dispatches an ack event.
+  void DispatchAckEvent(StreamType stream_type,
+                        RtpTimeTicks rtp_timestamp,
+                        FrameId frame_id);
+
+ private:
+  Environment& environment_;
+};
+
+}  // namespace openscreen::cast
+
+#endif  // CAST_STREAMING_IMPL_STATISTICS_DISPATCHER_H_

cast/streaming/impl/statistics_dispatcher_unittest.cc

@@ -0,0 +1,212 @@
+// Copyright 2025 The Chromium Authors
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#include "cast/streaming/impl/statistics_dispatcher.h"
+
+#include "cast/streaming/encoded_frame.h"
+#include "cast/streaming/impl/rtcp_common.h"
+#include "cast/streaming/impl/rtp_defines.h"
+#include "cast/streaming/impl/statistics_collector.h"
+#include "cast/streaming/impl/statistics_defines.h"
+#include "cast/streaming/public/frame_id.h"
+#include "cast/streaming/testing/mock_environment.h"
+#include "gmock/gmock.h"
+#include "gtest/gtest.h"
+#include "platform/api/time.h"
+#include "platform/base/error.h"
+#include "platform/test/fake_clock.h"
+#include "platform/test/fake_task_runner.h"
+#include "util/chrono_helpers.h"
+
+namespace openscreen::cast {
+
+namespace {
+
+using ::testing::_;
+using ::testing::ElementsAre;
+using ::testing::IsEmpty;
+using ::testing::Mock;
+using ::testing::SaveArg;
+using ::testing::StrictMock;
+
+class StatisticsDispatcherTest : public ::testing::Test {
+ public:
+  StatisticsDispatcherTest()
+      : environment_(&FakeClock::now, task_runner_),
+        collector_(&clock_.now),
+        dispatcher_(environment_) {
+    environment_.SetStatisticsCollector(&collector_);
+  }
+
+  ~StatisticsDispatcherTest() override {
+    environment_.SetStatisticsCollector(nullptr);
+  }
+
+ protected:
+  FakeClock clock_{Clock::now()};
+  FakeTaskRunner task_runner_{clock_};
+  testing::NiceMock<MockEnvironment> environment_;
+  StatisticsCollector collector_;
+  StatisticsDispatcher dispatcher_;
+};
+
+TEST_F(StatisticsDispatcherTest, DispatchEnqueueEvents) {
+  EncodedFrame frame;
+  frame.rtp_timestamp = RtpTimeTicks(12345);
+  frame.frame_id = FrameId::first();
+  frame.dependency = EncodedFrame::Dependency::kKeyFrame;
+  frame.data = ByteView(reinterpret_cast<const uint8_t*>("test"), 4);
+  frame.capture_begin_time = clock_.now() + milliseconds(10);
+  frame.capture_end_time = clock_.now() + milliseconds(20);
+
+  dispatcher_.DispatchEnqueueEvents(StreamType::kVideo, frame);
+  const std::vector<FrameEvent> events = collector_.TakeRecentFrameEvents();
+  ASSERT_EQ(3u, events.size());
+  EXPECT_EQ(events[0].type, StatisticsEventType::kFrameCaptureBegin);
+  EXPECT_EQ(events[0].media_type, StatisticsEventMediaType::kVideo);
+  EXPECT_EQ(events[0].rtp_timestamp, frame.rtp_timestamp);
+  EXPECT_EQ(events[0].timestamp, frame.capture_begin_time);
+
+  EXPECT_EQ(events[1].type, StatisticsEventType::kFrameCaptureEnd);
+  EXPECT_EQ(events[1].media_type, StatisticsEventMediaType::kVideo);
+  EXPECT_EQ(events[1].rtp_timestamp, frame.rtp_timestamp);
+  EXPECT_EQ(events[1].timestamp, frame.capture_end_time);
+
+  EXPECT_EQ(events[2].type, StatisticsEventType::kFrameEncoded);
+  EXPECT_EQ(events[2].media_type, StatisticsEventMediaType::kVideo);
+  EXPECT_EQ(events[2].rtp_timestamp, frame.rtp_timestamp);
+  EXPECT_EQ(events[2].frame_id, frame.frame_id);
+  EXPECT_EQ(events[2].size, 4u);
+  EXPECT_EQ(events[2].key_frame, true);
+}
+
+TEST_F(StatisticsDispatcherTest, DispatchEnqueueEventsWithDefaultTimes) {
+  EncodedFrame frame;
+  frame.rtp_timestamp = RtpTimeTicks(12345);
+  frame.frame_id = FrameId::first();
+  frame.dependency = EncodedFrame::Dependency::kKeyFrame;
+  frame.data = ByteView(reinterpret_cast<const uint8_t*>("test"), 4);
+
+  dispatcher_.DispatchEnqueueEvents(StreamType::kVideo, frame);
+  const std::vector<FrameEvent> events = collector_.TakeRecentFrameEvents();
+  ASSERT_EQ(3u, events.size());
+
+  EXPECT_EQ(events[0].type, StatisticsEventType::kFrameCaptureBegin);
+  EXPECT_EQ(events[0].media_type, StatisticsEventMediaType::kVideo);
+  EXPECT_EQ(events[0].rtp_timestamp, frame.rtp_timestamp);
+  EXPECT_EQ(events[0].timestamp, clock_.now());
+
+  EXPECT_EQ(events[1].type, StatisticsEventType::kFrameCaptureEnd);
+  EXPECT_EQ(events[1].media_type, StatisticsEventMediaType::kVideo);
+  EXPECT_EQ(events[1].rtp_timestamp, frame.rtp_timestamp);
+  EXPECT_EQ(events[1].timestamp, clock_.now());
+
+  EXPECT_EQ(events[2].type, StatisticsEventType::kFrameEncoded);
+  EXPECT_EQ(events[2].media_type, StatisticsEventMediaType::kVideo);
+  EXPECT_EQ(events[2].rtp_timestamp, frame.rtp_timestamp);
+  EXPECT_EQ(events[2].frame_id, frame.frame_id);
+  EXPECT_EQ(events[2].size, 4u);
+  EXPECT_EQ(events[2].key_frame, true);
+}
+
+TEST_F(StatisticsDispatcherTest, DispatchAckEvent) {
+  const RtpTimeTicks kRtpTimestamp(54321);
+  const FrameId kFrameId = FrameId::first() + 1;
+
+  dispatcher_.DispatchAckEvent(StreamType::kAudio, kRtpTimestamp, kFrameId);
+  const std::vector<FrameEvent> events = collector_.TakeRecentFrameEvents();
+
+  EXPECT_EQ(events[0].type, StatisticsEventType::kFrameAckReceived);
+  EXPECT_EQ(events[0].media_type, StatisticsEventMediaType::kAudio);
+  EXPECT_EQ(events[0].rtp_timestamp, kRtpTimestamp);
+  EXPECT_EQ(events[0].frame_id, kFrameId);
+}
+
+TEST_F(StatisticsDispatcherTest, DispatchFrameLogMessages) {
+  std::vector<RtcpReceiverFrameLogMessage> messages;
+  RtcpReceiverFrameLogMessage log_message;
+  log_message.rtp_timestamp = RtpTimeTicks(98765);
+
+  RtcpReceiverEventLogMessage packet_received_message;
+  packet_received_message.type = StatisticsEventType::kPacketReceived;
+  packet_received_message.timestamp = clock_.now() + milliseconds(5);
+  packet_received_message.packet_id = 10;
+  log_message.messages.push_back(packet_received_message);
+
+  RtcpReceiverEventLogMessage frame_ack_sent_message;
+  frame_ack_sent_message.type = StatisticsEventType::kFrameAckSent;
+  frame_ack_sent_message.timestamp = clock_.now() + milliseconds(10);
+  log_message.messages.push_back(frame_ack_sent_message);
+
+  RtcpReceiverEventLogMessage frame_decoded_message;
+  frame_decoded_message.type = StatisticsEventType::kFrameDecoded;
+  frame_decoded_message.timestamp = clock_.now() + milliseconds(15);
+  log_message.messages.push_back(frame_decoded_message);
+
+  RtcpReceiverEventLogMessage frame_played_out_message;
+  frame_played_out_message.type = StatisticsEventType::kFramePlayedOut;
+  frame_played_out_message.timestamp = clock_.now() + milliseconds(20);
+  frame_played_out_message.delay = milliseconds(10);
+  log_message.messages.push_back(frame_played_out_message);
+  messages.push_back(log_message);
+
+  dispatcher_.DispatchFrameLogMessages(StreamType::kAudio, messages);
+  const std::vector<FrameEvent> frame_events =
+      collector_.TakeRecentFrameEvents();
+  const std::vector<PacketEvent> packet_events =
+      collector_.TakeRecentPacketEvents();
+  ASSERT_EQ(3u, frame_events.size());
+  ASSERT_EQ(1u, packet_events.size());
+
+  EXPECT_EQ(packet_events[0].type, StatisticsEventType::kPacketReceived);
+  EXPECT_EQ(packet_events[0].media_type, StatisticsEventMediaType::kAudio);
+  EXPECT_EQ(packet_events[0].rtp_timestamp, log_message.rtp_timestamp);
+  EXPECT_EQ(packet_events[0].packet_id, packet_received_message.packet_id);
+  EXPECT_EQ(packet_events[0].timestamp, packet_received_message.timestamp);
+  EXPECT_EQ(packet_events[0].received_timestamp, clock_.now());
+
+  EXPECT_EQ(frame_events[0].type, StatisticsEventType::kFrameAckSent);
+  EXPECT_EQ(frame_events[0].media_type, StatisticsEventMediaType::kAudio);
+  EXPECT_EQ(frame_events[0].rtp_timestamp, log_message.rtp_timestamp);
+  EXPECT_EQ(frame_events[0].timestamp, frame_ack_sent_message.timestamp);
+  EXPECT_EQ(frame_events[0].received_timestamp, clock_.now());
+
+  EXPECT_EQ(frame_events[1].type, StatisticsEventType::kFrameDecoded);
+  EXPECT_EQ(frame_events[1].media_type, StatisticsEventMediaType::kAudio);
+  EXPECT_EQ(frame_events[1].rtp_timestamp, log_message.rtp_timestamp);
+  EXPECT_EQ(frame_events[1].timestamp, frame_decoded_message.timestamp);
+  EXPECT_EQ(frame_events[1].received_timestamp, clock_.now());
+
+  EXPECT_EQ(frame_events[2].type, StatisticsEventType::kFramePlayedOut);
+  EXPECT_EQ(frame_events[2].media_type, StatisticsEventMediaType::kAudio);
+  EXPECT_EQ(frame_events[2].rtp_timestamp, log_message.rtp_timestamp);
+  EXPECT_EQ(frame_events[2].timestamp, frame_played_out_message.timestamp);
+  EXPECT_EQ(frame_events[2].received_timestamp, clock_.now());
+  EXPECT_EQ(frame_events[2].delay_delta, frame_played_out_message.delay);
+}
+
+TEST_F(StatisticsDispatcherTest, DispatchFrameLogMessagesWithUnknownEventType) {
+  std::vector<RtcpReceiverFrameLogMessage> messages;
+  RtcpReceiverFrameLogMessage log_message;
+  log_message.rtp_timestamp = RtpTimeTicks(98765);
+
+  RtcpReceiverEventLogMessage unknown_event_message;
+  unknown_event_message.type = StatisticsEventType::kUnknown;
+  unknown_event_message.timestamp = clock_.now() + milliseconds(5);
+  log_message.messages.push_back(unknown_event_message);
+
+  messages.push_back(log_message);
+
+  dispatcher_.DispatchFrameLogMessages(StreamType::kAudio, messages);
+
+  const std::vector<FrameEvent> frame_events =
+      collector_.TakeRecentFrameEvents();
+  const std::vector<PacketEvent> packet_events =
+      collector_.TakeRecentPacketEvents();
+  EXPECT_EQ(0u, frame_events.size());
+  EXPECT_EQ(0u, packet_events.size());
+}
+
+}  // namespace
+}  // namespace openscreen::cast

cast/streaming/public/sender.cc

@@ -22,135 +22,11 @@ namespace openscreen::cast {
 
 using clock_operators::operator<<;
 
-namespace {
-
-void DispatchEnqueueEvents(StreamType stream_type,
-                           const EncodedFrame& frame,
-                           Environment& environment) {
-  if (!environment.statistics_collector()) {
-    return;
-  }
-
-  const StatisticsEventMediaType media_type = ToMediaType(stream_type);
-
-  // Submit a capture begin event.
-  FrameEvent capture_begin_event;
-  capture_begin_event.type = StatisticsEventType::kFrameCaptureBegin;
-  capture_begin_event.media_type = media_type;
-  capture_begin_event.rtp_timestamp = frame.rtp_timestamp;
-  capture_begin_event.timestamp =
-      (frame.capture_begin_time > Clock::time_point::min())
-          ? frame.capture_begin_time
-          : environment.now();
-  environment.statistics_collector()->CollectFrameEvent(
-      std::move(capture_begin_event));
-
-  // Submit a capture end event.
-  FrameEvent capture_end_event;
-  capture_end_event.type = StatisticsEventType::kFrameCaptureEnd;
-  capture_end_event.media_type = media_type;
-  capture_end_event.rtp_timestamp = frame.rtp_timestamp;
-  capture_end_event.timestamp =
-      (frame.capture_end_time > Clock::time_point::min())
-          ? frame.capture_end_time
-          : environment.now();
-  environment.statistics_collector()->CollectFrameEvent(
-      std::move(capture_end_event));
-
-  // Submit an encoded event.
-  FrameEvent encode_event;
-  encode_event.timestamp = environment.now();
-  encode_event.type = StatisticsEventType::kFrameEncoded;
-  encode_event.media_type = media_type;
-  encode_event.rtp_timestamp = frame.rtp_timestamp;
-  encode_event.frame_id = frame.frame_id;
-  encode_event.size = static_cast<uint32_t>(frame.data.size());
-  encode_event.key_frame =
-      frame.dependency == openscreen::cast::EncodedFrame::Dependency::kKeyFrame;
-
-  environment.statistics_collector()->CollectFrameEvent(
-      std::move(encode_event));
-}
-
-void DispatchAckEvent(StreamType stream_type,
-                      RtpTimeTicks rtp_timestamp,
-                      FrameId frame_id,
-                      Environment& environment) {
-  if (!environment.statistics_collector()) {
-    return;
-  }
-
-  FrameEvent ack_event;
-  ack_event.timestamp = environment.now();
-  ack_event.type = StatisticsEventType::kFrameAckReceived;
-  ack_event.media_type = ToMediaType(stream_type);
-  ack_event.rtp_timestamp = rtp_timestamp;
-  ack_event.frame_id = frame_id;
-
-  environment.statistics_collector()->CollectFrameEvent(std::move(ack_event));
-}
-
-// TODO(issuetracker.google.com/298277160): move into a helper file, add tests.
-void DispatchFrameLogMessages(
-    StreamType stream_type,
-    const std::vector<RtcpReceiverFrameLogMessage>& messages,
-    Environment& environment) {
-  if (!environment.statistics_collector()) {
-    return;
-  }
-
-  const Clock::time_point now = environment.now();
-  const StatisticsEventMediaType media_type = ToMediaType(stream_type);
-  for (const RtcpReceiverFrameLogMessage& log_message : messages) {
-    for (const RtcpReceiverEventLogMessage& event_message :
-         log_message.messages) {
-      switch (event_message.type) {
-        case StatisticsEventType::kPacketReceived: {
-          PacketEvent event;
-          event.timestamp = event_message.timestamp;
-          event.received_timestamp = now;
-          event.type = event_message.type;
-          event.media_type = media_type;
-          event.rtp_timestamp = log_message.rtp_timestamp;
-          event.packet_id = event_message.packet_id;
-          environment.statistics_collector()->CollectPacketEvent(
-              std::move(event));
-        } break;
-
-        case StatisticsEventType::kFrameAckSent:
-        case StatisticsEventType::kFrameDecoded:
-        case StatisticsEventType::kFramePlayedOut: {
-          FrameEvent event;
-          event.timestamp = event_message.timestamp;
-          event.received_timestamp = now;
-          event.type = event_message.type;
-          event.media_type = media_type;
-          event.rtp_timestamp = log_message.rtp_timestamp;
-          if (event.type == StatisticsEventType::kFramePlayedOut) {
-            event.delay_delta = event_message.delay;
-          }
-          environment.statistics_collector()->CollectFrameEvent(
-              std::move(event));
-        } break;
-
-        default:
-          OSP_VLOG << "Received log message via RTCP that we did not expect, "
-                      "StatisticsEventType="
-                   << static_cast<int>(event_message.type);
-          break;
-      }
-    }
-  }
-}
-
-}  // namespace
-
 Sender::Sender(Environment& environment,
                SenderPacketRouter& packet_router,
                SessionConfig config,
                RtpPayloadType rtp_payload_type)
-    : environment_(environment),
+    : config_(config),
-      config_(config),
       packet_router_(packet_router),
       rtcp_session_(config.sender_ssrc,
                     config.receiver_ssrc,
@@ -162,6 +38,7 @@ Sender::Sender(Environment& environment,
                       packet_router_.max_packet_size()),
       rtp_timebase_(config.rtp_timebase),
       crypto_(config.aes_secret_key, config.aes_iv_mask),
+      statistics_dispatcher_(environment),
       target_playout_delay_(config.target_playout_delay) {
   OSP_CHECK_NE(rtcp_session_.sender_ssrc(), rtcp_session_.receiver_ssrc());
   OSP_CHECK_GT(rtp_timebase_, 0);
@@ -307,7 +184,7 @@ Sender::EnqueueFrameResult Sender::EnqueueFrame(const EncodedFrame& frame) {
 
   // Re-activate RTP sending if it was suspended.
   packet_router_.RequestRtpSend(rtcp_session_.receiver_ssrc());
-  DispatchEnqueueEvents(config_.stream_type, frame, environment_);
+  statistics_dispatcher_.DispatchEnqueueEvents(config_.stream_type, frame);
 
   return OK;
 }
@@ -470,7 +347,8 @@ void Sender::OnReceiverReport(const RtcpReportBlock& receiver_report) {
 
 void Sender::OnCastReceiverFrameLogMessages(
     std::vector<RtcpReceiverFrameLogMessage> messages) {
-  DispatchFrameLogMessages(config_.stream_type, messages, environment_);
+  statistics_dispatcher_.DispatchFrameLogMessages(config_.stream_type,
+                                                  messages);
 }
 
 void Sender::OnReceiverIndicatesPictureLoss() {
@@ -520,8 +398,8 @@ void Sender::OnReceiverCheckpoint(FrameId frame_id,
     PendingFrameSlot& slot = get_slot_for(checkpoint_frame_id_);
     if (slot.is_active_for_frame(checkpoint_frame_id_)) {
       const RtpTimeTicks rtp_timestamp = slot.frame->rtp_timestamp;
-      DispatchAckEvent(config_.stream_type, rtp_timestamp, checkpoint_frame_id_,
+      statistics_dispatcher_.DispatchAckEvent(
-                       environment_);
+          config_.stream_type, rtp_timestamp, checkpoint_frame_id_);
       CancelPendingFrame(checkpoint_frame_id_, /*was_acked*/ true);
     }
   }
@@ -554,7 +432,8 @@ void Sender::OnReceiverHasFrames(std::vector<FrameId> acks) {
     PendingFrameSlot& slot = get_slot_for(id);
     if (slot.is_active_for_frame(id)) {
       const RtpTimeTicks rtp_timestamp = slot.frame->rtp_timestamp;
-      DispatchAckEvent(config_.stream_type, rtp_timestamp, id, environment_);
+      statistics_dispatcher_.DispatchAckEvent(config_.stream_type,
+                                              rtp_timestamp, id);
     }
     CancelPendingFrame(id, /*was_acked*/ true);
   }

cast/streaming/public/sender.h

@@ -19,6 +19,7 @@
 #include "cast/streaming/impl/rtp_packetizer.h"
 #include "cast/streaming/impl/sender_report_builder.h"
 #include "cast/streaming/impl/session_config.h"
+#include "cast/streaming/impl/statistics_dispatcher.h"
 #include "cast/streaming/public/constants.h"
 #include "cast/streaming/public/frame_id.h"
 #include "cast/streaming/rtp_time.h"
@@ -279,7 +280,6 @@ class Sender final : public SenderPacketRouter::Sender,
                            pending_frames_.size()];
   }
 
-  Environment& environment_;
   const SessionConfig config_;
   SenderPacketRouter& packet_router_;
   RtcpSession rtcp_session_;
@@ -288,6 +288,7 @@ class Sender final : public SenderPacketRouter::Sender,
   RtpPacketizer rtp_packetizer_;
   const int rtp_timebase_;
   FrameCrypto crypto_;
+  StatisticsDispatcher statistics_dispatcher_;
 
   // Ring buffer of PendingFrameSlots. The frame having FrameId x will always
   // be slotted at position x % pending_frames_.size(). Use get_slot_for() to