Cast: Implemented size limiting and event filtering in EncodingEventSubscriber.

All raw events are categorized into one of three types: video, audio, or other (i.e. not associated with a particular
stream). EncodingEventSubscriber can be configured to only process either audio or video events.

The subscriber can also be configured on having a max limit on number of frames to store in the frame and packet maps.
This helps bound the memory usage of logging and the size of the blob we are uploading at the end.

How they are going to be used: for each CastRtpStream (one audio, one video) we will install a EncodingEventSubscriber,
listening to AUDIO / VIDEO events respectively. The number of events to store is yet to be determined and is
dependent on how well the data can be compressed, or what's the size limit for uploading.

Lastly, EncodingEventSubscriber has been changed to only process frame and packet related events and not generic events. We are not interested in generic events right now, and they can be computed from frame / packet events anyways.

Review URL: https://codereview.chromium.org/165723002

git-svn-id: svn://svn.chromium.org/chrome/trunk/src@251521 0039d316-1c4b-4281-b951-d872f2087c98

5 files changed, 275 insertions, 154 deletions

diff --git a/media/cast/logging/encoding_event_subscriber.cc b/media/cast/logging/encoding_event_subscriber.cc
index 5a7333a..265e39d 100644
--- a/media/cast/logging/encoding_event_subscriber.cc
+++ b/media/cast/logging/encoding_event_subscriber.cc
@@ -4,6 +4,7 @@
 
 #include "media/cast/logging/encoding_event_subscriber.h"
 
+#include <cstring>
 #include <utility>
 
 #include "base/logging.h"
@@ -18,7 +19,9 @@ using media::cast::proto::BasePacketEvent;
 namespace media {
 namespace cast {
 
-EncodingEventSubscriber::EncodingEventSubscriber() {}
+EncodingEventSubscriber::EncodingEventSubscriber(
+    EventMediaType event_media_type, size_t max_frames)
+    : event_media_type_(event_media_type), max_frames_(max_frames) {}
 
 EncodingEventSubscriber::~EncodingEventSubscriber() {
   DCHECK(thread_checker_.CalledOnValidThread());
@@ -28,91 +31,93 @@ void EncodingEventSubscriber::OnReceiveFrameEvent(
     const FrameEvent& frame_event) {
   DCHECK(thread_checker_.CalledOnValidThread());
 
-  FrameEventMap::iterator it = frame_event_map_.find(frame_event.rtp_timestamp);
-  linked_ptr<AggregatedFrameEvent> event_proto;
-
-  // Look up existing entry. If not found, create a new entry and add to map.
-  if (it == frame_event_map_.end()) {
-    event_proto.reset(new AggregatedFrameEvent);
-    event_proto->set_rtp_timestamp(frame_event.rtp_timestamp);
-    frame_event_map_.insert(
-        std::make_pair(frame_event.rtp_timestamp, event_proto));
-  } else {
-    event_proto = it->second;
-  }
+  if (ShouldProcessEvent(frame_event.type)) {
+    FrameEventMap::iterator it =
+        frame_event_map_.find(frame_event.rtp_timestamp);
+    linked_ptr<AggregatedFrameEvent> event_proto;
+
+    // Look up existing entry. If not found, create a new entry and add to map.
+    if (it == frame_event_map_.end()) {
+      event_proto.reset(new AggregatedFrameEvent);
+      event_proto->set_rtp_timestamp(frame_event.rtp_timestamp);
+      frame_event_map_.insert(
+          std::make_pair(frame_event.rtp_timestamp, event_proto));
+    } else {
+      event_proto = it->second;
+    }
 
-  event_proto->add_event_type(ToProtoEventType(frame_event.type));
-  event_proto->add_event_timestamp_micros(
-      frame_event.timestamp.ToInternalValue());
+    event_proto->add_event_type(ToProtoEventType(frame_event.type));
+    event_proto->add_event_timestamp_micros(
+        frame_event.timestamp.ToInternalValue());
 
-  if (frame_event.type == kAudioFrameEncoded ||
-      frame_event.type == kVideoFrameEncoded) {
-    event_proto->set_encoded_frame_size(frame_event.size);
-  } else if (frame_event.type == kAudioPlayoutDelay ||
-             frame_event.type == kVideoRenderDelay) {
-    event_proto->set_delay_millis(frame_event.delay_delta.InMilliseconds());
+    if (frame_event.type == kAudioFrameEncoded ||
+        frame_event.type == kVideoFrameEncoded) {
+      event_proto->set_encoded_frame_size(frame_event.size);
+    } else if (frame_event.type == kAudioPlayoutDelay ||
+               frame_event.type == kVideoRenderDelay) {
+      event_proto->set_delay_millis(frame_event.delay_delta.InMilliseconds());
+    }
+
+    TruncateFrameEventMapIfNeeded();
   }
+
+  DCHECK(frame_event_map_.size() <= max_frames_);
 }
 
 void EncodingEventSubscriber::OnReceivePacketEvent(
     const PacketEvent& packet_event) {
   DCHECK(thread_checker_.CalledOnValidThread());
 
-  PacketEventMap::iterator it =
-      packet_event_map_.find(packet_event.rtp_timestamp);
-  linked_ptr<AggregatedPacketEvent> event_proto;
-  BasePacketEvent* base_packet_event_proto = NULL;
-
-  // Look up existing entry. If not found, create a new entry and add to map.
-  if (it == packet_event_map_.end()) {
-    event_proto.reset(new AggregatedPacketEvent);
-    event_proto->set_rtp_timestamp(packet_event.rtp_timestamp);
-    packet_event_map_.insert(
-        std::make_pair(packet_event.rtp_timestamp, event_proto));
-    base_packet_event_proto = event_proto->add_base_packet_event();
-    base_packet_event_proto->set_packet_id(packet_event.packet_id);
-  } else {
-    // Found existing entry, now look up existing BasePacketEvent using packet
-    // ID. If not found, create a new entry and add to proto.
-    event_proto = it->second;
-    RepeatedPtrField<BasePacketEvent>* field =
-        event_proto->mutable_base_packet_event();
-    for (RepeatedPtrField<BasePacketEvent>::pointer_iterator it =
-             field->pointer_begin();
-         it != field->pointer_end(); ++it) {
-      if ((*it)->packet_id() == packet_event.packet_id) {
-        base_packet_event_proto = *it;
-        break;
-      }
-    }
-    if (!base_packet_event_proto) {
+  if (ShouldProcessEvent(packet_event.type)) {
+    PacketEventMap::iterator it =
+        packet_event_map_.find(packet_event.rtp_timestamp);
+    linked_ptr<AggregatedPacketEvent> event_proto;
+    BasePacketEvent* base_packet_event_proto = NULL;
+
+    // Look up existing entry. If not found, create a new entry and add to map.
+    if (it == packet_event_map_.end()) {
+      event_proto.reset(new AggregatedPacketEvent);
+      event_proto->set_rtp_timestamp(packet_event.rtp_timestamp);
+      packet_event_map_.insert(
+          std::make_pair(packet_event.rtp_timestamp, event_proto));
       base_packet_event_proto = event_proto->add_base_packet_event();
       base_packet_event_proto->set_packet_id(packet_event.packet_id);
+    } else {
+      // Found existing entry, now look up existing BasePacketEvent using packet
+      // ID. If not found, create a new entry and add to proto.
+      event_proto = it->second;
+      RepeatedPtrField<BasePacketEvent>* field =
+          event_proto->mutable_base_packet_event();
+      for (RepeatedPtrField<BasePacketEvent>::pointer_iterator it =
+               field->pointer_begin();
+           it != field->pointer_end();
+           ++it) {
+        if ((*it)->packet_id() == packet_event.packet_id) {
+          base_packet_event_proto = *it;
+          break;
+        }
+      }
+      if (!base_packet_event_proto) {
+        base_packet_event_proto = event_proto->add_base_packet_event();
+        base_packet_event_proto->set_packet_id(packet_event.packet_id);
+      }
     }
+
+    base_packet_event_proto->add_event_type(
+        ToProtoEventType(packet_event.type));
+    base_packet_event_proto->add_event_timestamp_micros(
+        packet_event.timestamp.ToInternalValue());
+
+    TruncatePacketEventMapIfNeeded();
   }
 
-  base_packet_event_proto->add_event_type(ToProtoEventType(packet_event.type));
-  base_packet_event_proto->add_event_timestamp_micros(
-      packet_event.timestamp.ToInternalValue());
+  DCHECK(packet_event_map_.size() <= max_frames_);
 }
 
 void EncodingEventSubscriber::OnReceiveGenericEvent(
     const GenericEvent& generic_event) {
   DCHECK(thread_checker_.CalledOnValidThread());
-
-  GenericEventMap::iterator it = generic_event_map_.find(generic_event.type);
-  linked_ptr<AggregatedGenericEvent> event_proto;
-  if (it == generic_event_map_.end()) {
-    event_proto.reset(new AggregatedGenericEvent);
-    event_proto->set_event_type(ToProtoEventType(generic_event.type));
-    generic_event_map_.insert(std::make_pair(generic_event.type, event_proto));
-  } else {
-    event_proto = it->second;
-  }
-
-  event_proto->add_event_timestamp_micros(
-      generic_event.timestamp.ToInternalValue());
-  event_proto->add_value(generic_event.value);
+  // Do nothing, there are no generic events we are interested in.
 }
 
 void EncodingEventSubscriber::GetFrameEventsAndReset(
@@ -129,11 +134,23 @@ void EncodingEventSubscriber::GetPacketEventsAndReset(
   packet_event_map_.clear();
 }
 
-void EncodingEventSubscriber::GetGenericEventsAndReset(
-    GenericEventMap* generic_event_map) {
-  DCHECK(thread_checker_.CalledOnValidThread());
-  generic_event_map->swap(generic_event_map_);
-  generic_event_map_.clear();
+
+bool EncodingEventSubscriber::ShouldProcessEvent(CastLoggingEvent event) {
+  return GetEventMediaType(event) == event_media_type_;
+}
+
+void EncodingEventSubscriber::TruncateFrameEventMapIfNeeded() {
+  // This works because this is called everytime an event is inserted and
+  // we only insert events one at a time.
+  if (frame_event_map_.size() > max_frames_)
+    frame_event_map_.erase(frame_event_map_.begin());
+}
+
+void EncodingEventSubscriber::TruncatePacketEventMapIfNeeded() {
+  // This works because this is called everytime an event is inserted and
+  // we only insert events one at a time.
+  if (packet_event_map_.size() > max_frames_)
+    packet_event_map_.erase(packet_event_map_.begin());
 }
 
 }  // namespace cast
diff --git a/media/cast/logging/encoding_event_subscriber.h b/media/cast/logging/encoding_event_subscriber.h
index e80061d..acd1d14 100644
--- a/media/cast/logging/encoding_event_subscriber.h
+++ b/media/cast/logging/encoding_event_subscriber.h
@@ -9,6 +9,7 @@
 
 #include "base/memory/linked_ptr.h"
 #include "base/threading/thread_checker.h"
+#include "media/cast/logging/logging_defines.h"
 #include "media/cast/logging/proto/raw_events.pb.h"
 #include "media/cast/logging/raw_event_subscriber.h"
 
@@ -21,17 +22,21 @@ typedef std::map<RtpTimestamp,
 typedef std::map<RtpTimestamp,
                  linked_ptr<media::cast::proto::AggregatedPacketEvent> >
     PacketEventMap;
-typedef std::map<CastLoggingEvent,
-                 linked_ptr<media::cast::proto::AggregatedGenericEvent> >
-    GenericEventMap;
 
 // A RawEventSubscriber implementation that subscribes to events,
 // encodes them in protocol buffer format, and aggregates them into a more
 // compact structure.
-// TODO(imcheng): Implement event filtering and windowing based on size.
 class EncodingEventSubscriber : public RawEventSubscriber {
  public:
-  EncodingEventSubscriber();
+  // |event_media_type|: The subscriber will only process events that
+  // corresponds to this type.
+  // |max_frames|: How many events to keep in the frame / packet map.
+  // This helps keep memory usage bounded.
+  // Every time one of |OnReceive[Frame,Packet]Event()| is
+  // called, it will check if the respective map size has exceeded |max_frames|.
+  // If so, it will remove the oldest aggregated entry (ordered by RTP
+  // timestamp).
+  EncodingEventSubscriber(EventMediaType event_media_type, size_t max_frames);
 
   virtual ~EncodingEventSubscriber();
 
@@ -49,14 +54,20 @@ class EncodingEventSubscriber : public RawEventSubscriber {
   // from this object.
   void GetPacketEventsAndReset(PacketEventMap* packet_events);
 
-  // Assigns generic events received so far to |generic_events| and clears them
-  // from this object.
-  void GetGenericEventsAndReset(GenericEventMap* generic_events);
-
  private:
+  bool ShouldProcessEvent(CastLoggingEvent event);
+
+  // Removes oldest entry from |frame_event_map_| (ordered by RTP timestamp).
+  void TruncateFrameEventMapIfNeeded();
+
+  // Removes oldest entry from |packet_event_map_| (ordered by RTP timestamp).
+  void TruncatePacketEventMapIfNeeded();
+
+  const EventMediaType event_media_type_;
+  const size_t max_frames_;
+
   FrameEventMap frame_event_map_;
   PacketEventMap packet_event_map_;
-  GenericEventMap generic_event_map_;
 
   // All functions must be called on the main thread.
   base::ThreadChecker thread_checker_;
diff --git a/media/cast/logging/encoding_event_subscriber_unittest.cc b/media/cast/logging/encoding_event_subscriber_unittest.cc
index 831bea7..1194ed9 100644
--- a/media/cast/logging/encoding_event_subscriber_unittest.cc
+++ b/media/cast/logging/encoding_event_subscriber_unittest.cc
@@ -13,7 +13,6 @@
 #include "testing/gtest/include/gtest/gtest.h"
 
 using media::cast::proto::AggregatedFrameEvent;
-using media::cast::proto::AggregatedGenericEvent;
 using media::cast::proto::AggregatedPacketEvent;
 using media::cast::proto::BasePacketEvent;
 
@@ -26,23 +25,119 @@ class EncodingEventSubscriberTest : public ::testing::Test {
       : testing_clock_(new base::SimpleTestTickClock()),
         task_runner_(new test::FakeSingleThreadTaskRunner(testing_clock_)),
         cast_environment_(new CastEnvironment(
-            scoped_ptr<base::TickClock>(testing_clock_).Pass(), task_runner_,
-            task_runner_, task_runner_, task_runner_, task_runner_,
-            task_runner_, GetLoggingConfigWithRawEventsAndStatsEnabled())) {
-    cast_environment_->Logging()->AddRawEventSubscriber(&event_subscriber_);
+            scoped_ptr<base::TickClock>(testing_clock_).Pass(),
+            task_runner_,
+            task_runner_,
+            task_runner_,
+            task_runner_,
+            task_runner_,
+            task_runner_,
+            GetLoggingConfigWithRawEventsAndStatsEnabled())) {}
+
+  void Init(EventMediaType event_media_type) {
+    DCHECK(!event_subscriber_);
+    event_subscriber_.reset(new EncodingEventSubscriber(event_media_type, 10));
+    cast_environment_->Logging()->AddRawEventSubscriber(
+        event_subscriber_.get());
   }
 
   virtual ~EncodingEventSubscriberTest() {
-    cast_environment_->Logging()->RemoveRawEventSubscriber(&event_subscriber_);
+    if (event_subscriber_) {
+      cast_environment_->Logging()->RemoveRawEventSubscriber(
+          event_subscriber_.get());
+    }
   }
 
   base::SimpleTestTickClock* testing_clock_;  // Owned by CastEnvironment.
   scoped_refptr<test::FakeSingleThreadTaskRunner> task_runner_;
   scoped_refptr<CastEnvironment> cast_environment_;
-  EncodingEventSubscriber event_subscriber_;
+  scoped_ptr<EncodingEventSubscriber> event_subscriber_;
 };
 
+TEST_F(EncodingEventSubscriberTest, FrameEventTruncating) {
+  Init(VIDEO_EVENT);
+
+  base::TimeTicks now(testing_clock_->NowTicks());
+
+  // Entry with RTP timestamp 0 should get dropped.
+  for (int i = 0; i < 11; i++) {
+    cast_environment_->Logging()->InsertFrameEvent(now,
+                                                   kVideoFrameCaptured,
+                                                   i * 100,
+                                                   /*frame_id*/ 0);
+    cast_environment_->Logging()->InsertFrameEvent(now,
+                                                   kVideoFrameDecoded,
+                                                   i * 100,
+                                                   /*frame_id*/ 0);
+  }
+
+  FrameEventMap frame_events;
+  event_subscriber_->GetFrameEventsAndReset(&frame_events);
+
+  ASSERT_EQ(10u, frame_events.size());
+  EXPECT_EQ(100u, frame_events.begin()->first);
+  EXPECT_EQ(1000u, frame_events.rbegin()->first);
+}
+
+TEST_F(EncodingEventSubscriberTest, PacketEventTruncating) {
+  Init(AUDIO_EVENT);
+
+  base::TimeTicks now(testing_clock_->NowTicks());
+
+  // Entry with RTP timestamp 0 should get dropped.
+  for (int i = 0; i < 11; i++) {
+    cast_environment_->Logging()->InsertPacketEvent(now,
+                                                    kAudioPacketReceived,
+                                                    /*rtp_timestamp*/ i * 100,
+                                                    /*frame_id*/ 0,
+                                                    /*packet_id*/ i,
+                                                    /*max_packet_id*/ 10,
+                                                    /*size*/ 123);
+  }
+
+  PacketEventMap packet_events;
+  event_subscriber_->GetPacketEventsAndReset(&packet_events);
+
+  ASSERT_EQ(10u, packet_events.size());
+  EXPECT_EQ(100u, packet_events.begin()->first);
+  EXPECT_EQ(1000u, packet_events.rbegin()->first);
+}
+
+TEST_F(EncodingEventSubscriberTest, EventFiltering) {
+  Init(VIDEO_EVENT);
+
+  base::TimeTicks now(testing_clock_->NowTicks());
+  RtpTimestamp rtp_timestamp = 100;
+  cast_environment_->Logging()->InsertFrameEvent(now,
+                                                 kVideoFrameDecoded,
+                                                 rtp_timestamp,
+                                                 /*frame_id*/ 0);
+
+  // This is an AUDIO_EVENT and shouldn't be processed by the subscriber.
+  cast_environment_->Logging()->InsertFrameEvent(now,
+                                                 kAudioFrameDecoded,
+                                                 rtp_timestamp,
+                                                 /*frame_id*/ 0);
+
+  FrameEventMap frame_events;
+  event_subscriber_->GetFrameEventsAndReset(&frame_events);
+
+  FrameEventMap::iterator frame_it = frame_events.find(rtp_timestamp);
+  ASSERT_TRUE(frame_it != frame_events.end());
+
+  linked_ptr<AggregatedFrameEvent> frame_event = frame_it->second;
+
+  ASSERT_EQ(1, frame_event->event_type_size());
+  EXPECT_EQ(media::cast::proto::VIDEO_FRAME_DECODED,
+            frame_event->event_type(0));
+
+  PacketEventMap packet_events;
+  event_subscriber_->GetPacketEventsAndReset(&packet_events);
+  EXPECT_TRUE(packet_events.empty());
+}
+
 TEST_F(EncodingEventSubscriberTest, FrameEvent) {
+  Init(VIDEO_EVENT);
   base::TimeTicks now(testing_clock_->NowTicks());
   RtpTimestamp rtp_timestamp = 100;
   cast_environment_->Logging()->InsertFrameEvent(now, kVideoFrameDecoded,
@@ -50,7 +145,7 @@ TEST_F(EncodingEventSubscriberTest, FrameEvent) {
                                                  /*frame_id*/ 0);
 
   FrameEventMap frame_events;
-  event_subscriber_.GetFrameEventsAndReset(&frame_events);
+  event_subscriber_->GetFrameEventsAndReset(&frame_events);
 
   ASSERT_EQ(1u, frame_events.size());
 
@@ -69,11 +164,12 @@ TEST_F(EncodingEventSubscriberTest, FrameEvent) {
   EXPECT_EQ(0, event->encoded_frame_size());
   EXPECT_EQ(0, event->delay_millis());
 
-  event_subscriber_.GetFrameEventsAndReset(&frame_events);
+  event_subscriber_->GetFrameEventsAndReset(&frame_events);
   EXPECT_TRUE(frame_events.empty());
 }
 
 TEST_F(EncodingEventSubscriberTest, FrameEventDelay) {
+  Init(AUDIO_EVENT);
   base::TimeTicks now(testing_clock_->NowTicks());
   RtpTimestamp rtp_timestamp = 100;
   int delay_ms = 100;
@@ -82,7 +178,7 @@ TEST_F(EncodingEventSubscriberTest, FrameEventDelay) {
       /*frame_id*/ 0, base::TimeDelta::FromMilliseconds(delay_ms));
 
   FrameEventMap frame_events;
-  event_subscriber_.GetFrameEventsAndReset(&frame_events);
+  event_subscriber_->GetFrameEventsAndReset(&frame_events);
 
   ASSERT_EQ(1u, frame_events.size());
 
@@ -103,6 +199,7 @@ TEST_F(EncodingEventSubscriberTest, FrameEventDelay) {
 }
 
 TEST_F(EncodingEventSubscriberTest, FrameEventSize) {
+  Init(VIDEO_EVENT);
   base::TimeTicks now(testing_clock_->NowTicks());
   RtpTimestamp rtp_timestamp = 100;
   int size = 123;
@@ -111,7 +208,7 @@ TEST_F(EncodingEventSubscriberTest, FrameEventSize) {
       /*frame_id*/ 0, size);
 
   FrameEventMap frame_events;
-  event_subscriber_.GetFrameEventsAndReset(&frame_events);
+  event_subscriber_->GetFrameEventsAndReset(&frame_events);
 
   ASSERT_EQ(1u, frame_events.size());
 
@@ -132,6 +229,7 @@ TEST_F(EncodingEventSubscriberTest, FrameEventSize) {
 }
 
 TEST_F(EncodingEventSubscriberTest, MultipleFrameEvents) {
+  Init(AUDIO_EVENT);
   RtpTimestamp rtp_timestamp1 = 100;
   RtpTimestamp rtp_timestamp2 = 200;
   base::TimeTicks now1(testing_clock_->NowTicks());
@@ -151,7 +249,7 @@ TEST_F(EncodingEventSubscriberTest, MultipleFrameEvents) {
       now3, kAudioFrameDecoded, rtp_timestamp1, /*frame_id*/ 0);
 
   FrameEventMap frame_events;
-  event_subscriber_.GetFrameEventsAndReset(&frame_events);
+  event_subscriber_->GetFrameEventsAndReset(&frame_events);
 
   ASSERT_EQ(2u, frame_events.size());
 
@@ -185,6 +283,7 @@ TEST_F(EncodingEventSubscriberTest, MultipleFrameEvents) {
 }
 
 TEST_F(EncodingEventSubscriberTest, PacketEvent) {
+  Init(AUDIO_EVENT);
   base::TimeTicks now(testing_clock_->NowTicks());
   RtpTimestamp rtp_timestamp = 100;
   int packet_id = 2;
@@ -194,7 +293,7 @@ TEST_F(EncodingEventSubscriberTest, PacketEvent) {
       /*max_packet_id*/ 10, size);
 
   PacketEventMap packet_events;
-  event_subscriber_.GetPacketEventsAndReset(&packet_events);
+  event_subscriber_->GetPacketEventsAndReset(&packet_events);
 
   ASSERT_EQ(1u, packet_events.size());
 
@@ -214,11 +313,12 @@ TEST_F(EncodingEventSubscriberTest, PacketEvent) {
   ASSERT_EQ(1, base_event.event_timestamp_micros_size());
   EXPECT_EQ(now.ToInternalValue(), base_event.event_timestamp_micros(0));
 
-  event_subscriber_.GetPacketEventsAndReset(&packet_events);
+  event_subscriber_->GetPacketEventsAndReset(&packet_events);
   EXPECT_TRUE(packet_events.empty());
 }
 
 TEST_F(EncodingEventSubscriberTest, MultiplePacketEventsForPacket) {
+  Init(OTHER_EVENT);
   base::TimeTicks now1(testing_clock_->NowTicks());
   RtpTimestamp rtp_timestamp = 100;
   int packet_id = 2;
@@ -234,7 +334,7 @@ TEST_F(EncodingEventSubscriberTest, MultiplePacketEventsForPacket) {
       /*max_packet_id*/ 10, size);
 
   PacketEventMap packet_events;
-  event_subscriber_.GetPacketEventsAndReset(&packet_events);
+  event_subscriber_->GetPacketEventsAndReset(&packet_events);
 
   ASSERT_EQ(1u, packet_events.size());
 
@@ -258,6 +358,7 @@ TEST_F(EncodingEventSubscriberTest, MultiplePacketEventsForPacket) {
 }
 
 TEST_F(EncodingEventSubscriberTest, MultiplePacketEventsForFrame) {
+  Init(OTHER_EVENT);
   base::TimeTicks now1(testing_clock_->NowTicks());
   RtpTimestamp rtp_timestamp = 100;
   int packet_id_1 = 2;
@@ -274,7 +375,7 @@ TEST_F(EncodingEventSubscriberTest, MultiplePacketEventsForFrame) {
       /*max_packet_id*/ 10, size);
 
   PacketEventMap packet_events;
-  event_subscriber_.GetPacketEventsAndReset(&packet_events);
+  event_subscriber_->GetPacketEventsAndReset(&packet_events);
 
   ASSERT_EQ(1u, packet_events.size());
 
@@ -303,6 +404,7 @@ TEST_F(EncodingEventSubscriberTest, MultiplePacketEventsForFrame) {
 }
 
 TEST_F(EncodingEventSubscriberTest, MultiplePacketEvents) {
+  Init(OTHER_EVENT);
   base::TimeTicks now1(testing_clock_->NowTicks());
   RtpTimestamp rtp_timestamp_1 = 100;
   RtpTimestamp rtp_timestamp_2 = 200;
@@ -320,7 +422,7 @@ TEST_F(EncodingEventSubscriberTest, MultiplePacketEvents) {
       /*max_packet_id*/ 10, size);
 
   PacketEventMap packet_events;
-  event_subscriber_.GetPacketEventsAndReset(&packet_events);
+  event_subscriber_->GetPacketEventsAndReset(&packet_events);
 
   ASSERT_EQ(2u, packet_events.size());
 
@@ -356,59 +458,5 @@ TEST_F(EncodingEventSubscriberTest, MultiplePacketEvents) {
   EXPECT_EQ(now2.ToInternalValue(), base_event_2.event_timestamp_micros(0));
 }
 
-TEST_F(EncodingEventSubscriberTest, GenericEvent) {
-  base::TimeTicks now(testing_clock_->NowTicks());
-  int value = 123;
-  cast_environment_->Logging()->InsertGenericEvent(now, kJitterMs, value);
-
-  GenericEventMap generic_events;
-  event_subscriber_.GetGenericEventsAndReset(&generic_events);
-
-  ASSERT_EQ(1u, generic_events.size());
-
-  GenericEventMap::iterator it = generic_events.find(kJitterMs);
-  ASSERT_TRUE(it != generic_events.end());
-
-  linked_ptr<AggregatedGenericEvent> event = it->second;
-
-  ASSERT_EQ(media::cast::proto::JITTER_MS, event->event_type());
-  ASSERT_EQ(1, event->event_timestamp_micros_size());
-  EXPECT_EQ(now.ToInternalValue(), event->event_timestamp_micros(0));
-  ASSERT_EQ(1, event->value_size());
-  EXPECT_EQ(value, event->value(0));
-
-  event_subscriber_.GetGenericEventsAndReset(&generic_events);
-  EXPECT_TRUE(generic_events.empty());
-}
-
-TEST_F(EncodingEventSubscriberTest, MultipleGenericEventsForType) {
-  base::TimeTicks now1(testing_clock_->NowTicks());
-  int value1 = 123;
-  cast_environment_->Logging()->InsertGenericEvent(now1, kJitterMs, value1);
-
-  testing_clock_->Advance(base::TimeDelta::FromMilliseconds(20));
-  base::TimeTicks now2(testing_clock_->NowTicks());
-  int value2 = 456;
-  cast_environment_->Logging()->InsertGenericEvent(now2, kJitterMs, value2);
-
-  GenericEventMap generic_events;
-  event_subscriber_.GetGenericEventsAndReset(&generic_events);
-
-  ASSERT_EQ(1u, generic_events.size());
-
-  GenericEventMap::iterator it = generic_events.find(kJitterMs);
-  ASSERT_TRUE(it != generic_events.end());
-
-  linked_ptr<AggregatedGenericEvent> event = it->second;
-
-  ASSERT_EQ(media::cast::proto::JITTER_MS, event->event_type());
-  ASSERT_EQ(2, event->event_timestamp_micros_size());
-  EXPECT_EQ(now1.ToInternalValue(), event->event_timestamp_micros(0));
-  EXPECT_EQ(now2.ToInternalValue(), event->event_timestamp_micros(1));
-  ASSERT_EQ(2, event->value_size());
-  EXPECT_EQ(value1, event->value(0));
-  EXPECT_EQ(value2, event->value(1));
-}
-
 }  // namespace cast
 }  // namespace media
diff --git a/media/cast/logging/logging_defines.cc b/media/cast/logging/logging_defines.cc
index 59c9a27..5adb574 100644
--- a/media/cast/logging/logging_defines.cc
+++ b/media/cast/logging/logging_defines.cc
@@ -73,6 +73,46 @@ std::string CastLoggingToString(CastLoggingEvent event) {
   return "";
 }
 
+EventMediaType GetEventMediaType(CastLoggingEvent event) {
+  switch (event) {
+    case kUnknown:
+    case kRttMs:
+    case kPacketLoss:
+    case kJitterMs:
+    case kRembBitrate:
+    // TODO(imcheng): These need to be split into video/audio events.
+    case kPacketSentToPacer:
+    case kPacketSentToNetwork:
+    case kPacketRetransmitted:
+      return OTHER_EVENT;
+    case kAudioAckSent:
+    case kAudioFrameReceived:
+    case kAudioFrameCaptured:
+    case kAudioFrameEncoded:
+    case kAudioPlayoutDelay:
+    case kAudioFrameDecoded:
+    case kAudioPacketReceived:
+    case kDuplicateAudioPacketReceived:
+      return AUDIO_EVENT;
+    case kVideoAckReceived:
+    case kVideoAckSent:
+    case kVideoFrameCaptured:
+    case kVideoFrameReceived:
+    case kVideoFrameSentToEncoder:
+    case kVideoFrameEncoded:
+    case kVideoFrameDecoded:
+    case kVideoRenderDelay:
+    case kVideoPacketReceived:
+    case kDuplicateVideoPacketReceived:
+      return VIDEO_EVENT;
+    case kNumOfLoggingEvents:
+      NOTREACHED();
+      return OTHER_EVENT;
+  }
+  NOTREACHED();
+  return OTHER_EVENT;
+}
+
 FrameEvent::FrameEvent()
     : rtp_timestamp(0u), frame_id(kFrameIdUnknown), size(0u), type(kUnknown) {}
 FrameEvent::~FrameEvent() {}
diff --git a/media/cast/logging/logging_defines.h b/media/cast/logging/logging_defines.h
index b517f1a..10529a2 100644
--- a/media/cast/logging/logging_defines.h
+++ b/media/cast/logging/logging_defines.h
@@ -16,6 +16,8 @@ namespace cast {
 
 static const uint32 kFrameIdUnknown = 0xFFFFFFFF;
 
+typedef uint32 RtpTimestamp;
+
 struct CastLoggingConfig {
   // Constructs default config - all logging is disabled.
   CastLoggingConfig();
@@ -74,7 +76,10 @@ enum CastLoggingEvent {
 
 std::string CastLoggingToString(CastLoggingEvent event);
 
-typedef uint32 RtpTimestamp;
+// CastLoggingEvent are classified into one of three following types.
+enum EventMediaType { AUDIO_EVENT, VIDEO_EVENT, OTHER_EVENT };
+
+EventMediaType GetEventMediaType(CastLoggingEvent event);
 
 struct FrameEvent {
   FrameEvent();