Cast: Added a new RawEventSubscriber implementation for use in cast extension.

Added corresponding protocol buffer definitions for raw events, but in aggregated format.
Added EncodingEventSubscriber:
- encodes events in protocol buffer format
- (we need to test how much aggregation helps with reducing the size, if at all)
Added unit tests.

TODO in the future CLs: add event filtering, size-based windowing

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

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

media/cast/cast.gyp

@@ -36,6 +36,35 @@
         'logging/simple_event_subscriber.h',
       ], # source
     },
+    {
+      'target_name': 'cast_logging_proto_lib',
+      'type': 'static_library',
+      'sources': [
+        'logging/proto/proto_utils.cc',
+        'logging/proto/raw_events.proto',
+      ],
+      'variables': {
+        'proto_in_dir': 'logging/proto',
+        'proto_out_dir': 'media/cast/logging/proto',
+      },
+      'includes': ['../../build/protoc.gypi'],
+    },
+    {
+      'target_name': 'sender_logging',
+      'type': 'static_library',
+      'include_dirs': [
+        '<(DEPTH)/',
+      ],
+      'dependencies': [
+        'cast_config',
+        'cast_logging_proto_lib',
+        '<(DEPTH)/base/base.gyp:base',
+      ],
+      'sources': [
+        'logging/encoding_event_subscriber.cc',
+        'logging/encoding_event_subscriber.h',
+      ], # source
+    },
   ],  # targets,
   'conditions': [
     ['include_tests==1', {
@@ -45,8 +74,10 @@
           'type': '<(gtest_target_type)',
           'dependencies': [
             'cast_config',
+            'cast_logging_proto_lib',
             'cast_receiver.gyp:cast_receiver',
             'cast_sender.gyp:cast_sender',
+            'sender_logging',
             'test/utility/utility.gyp:cast_test_utility',
             'transport/cast_transport.gyp:cast_transport',
             '<(DEPTH)/base/base.gyp:run_all_unittests',
@@ -69,6 +100,7 @@
             'framer/cast_message_builder_unittest.cc',
             'framer/frame_buffer_unittest.cc',
             'framer/framer_unittest.cc',
+            'logging/encoding_event_subscriber_unittest.cc',
             'logging/logging_impl_unittest.cc',
             'logging/logging_raw_unittest.cc',
             'logging/simple_event_subscriber_unittest.cc',

media/cast/logging/encoding_event_subscriber.cc

+// Copyright 2014 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#include "media/cast/logging/encoding_event_subscriber.h"
+
+#include <utility>
+
+#include "base/logging.h"
+#include "media/cast/logging/proto/proto_utils.h"
+
+using google::protobuf::RepeatedPtrField;
+using media::cast::proto::AggregatedFrameEvent;
+using media::cast::proto::AggregatedGenericEvent;
+using media::cast::proto::AggregatedPacketEvent;
+using media::cast::proto::BasePacketEvent;
+
+namespace media {
+namespace cast {
+
+EncodingEventSubscriber::EncodingEventSubscriber() {}
+
+EncodingEventSubscriber::~EncodingEventSubscriber() {
+  DCHECK(thread_checker_.CalledOnValidThread());
+}
+
+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;
+  }
+
+  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());
+  }
+}
+
+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) {
+      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());
+}
+
+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);
+}
+
+void EncodingEventSubscriber::GetFrameEventsAndReset(
+    FrameEventMap* frame_event_map) {
+  DCHECK(thread_checker_.CalledOnValidThread());
+  frame_event_map->swap(frame_event_map_);
+  frame_event_map_.clear();
+}
+
+void EncodingEventSubscriber::GetPacketEventsAndReset(
+    PacketEventMap* packet_event_map) {
+  DCHECK(thread_checker_.CalledOnValidThread());
+  packet_event_map->swap(packet_event_map_);
+  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();
+}
+
+}  // namespace cast
+}  // namespace media

media/cast/logging/encoding_event_subscriber.h

+// Copyright 2014 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#ifndef MEDIA_CAST_LOGGING_ENCODING_EVENT_SUBSCRIBER_H_
+#define MEDIA_CAST_LOGGING_ENCODING_EVENT_SUBSCRIBER_H_
+
+#include <map>
+
+#include "base/memory/linked_ptr.h"
+#include "base/threading/thread_checker.h"
+#include "media/cast/logging/proto/raw_events.pb.h"
+#include "media/cast/logging/raw_event_subscriber.h"
+
+namespace media {
+namespace cast {
+
+typedef std::map<RtpTimestamp,
+                 linked_ptr<media::cast::proto::AggregatedFrameEvent> >
+    FrameEventMap;
+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();
+
+  virtual ~EncodingEventSubscriber();
+
+  // RawReventSubscriber implementations.
+  virtual void OnReceiveFrameEvent(const FrameEvent& frame_event) OVERRIDE;
+  virtual void OnReceivePacketEvent(const PacketEvent& packet_event) OVERRIDE;
+  virtual void OnReceiveGenericEvent(const GenericEvent& generic_event)
+      OVERRIDE;
+
+  // Assigns frame events received so far to |frame_events| and clears them
+  // from this object.
+  void GetFrameEventsAndReset(FrameEventMap* frame_events);
+
+  // Assigns packet events received so far to |packet_events| and clears them
+  // 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:
+  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_;
+
+  DISALLOW_COPY_AND_ASSIGN(EncodingEventSubscriber);
+};
+
+}  // namespace cast
+}  // namespace media
+
+#endif  // MEDIA_CAST_LOGGING_ENCODING_EVENT_SUBSCRIBER_H_

media/cast/logging/proto/proto_utils.cc

+// Copyright 2014 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#include "media/cast/logging/proto/proto_utils.h"
+
+#include "base/logging.h"
+
+#define TO_PROTO_ENUM(from_enum, to_enum)  \
+  case from_enum:                          \
+    return media::cast::proto::to_enum
+
+namespace media {
+namespace cast {
+
+media::cast::proto::EventType ToProtoEventType(CastLoggingEvent event) {
+  switch (event) {
+    TO_PROTO_ENUM(kUnknown, UNKNOWN);
+    TO_PROTO_ENUM(kRttMs, RTT_MS);
+    TO_PROTO_ENUM(kPacketLoss, PACKET_LOSS);
+    TO_PROTO_ENUM(kJitterMs, JITTER_MS);
+    TO_PROTO_ENUM(kVideoAckReceived, VIDEO_ACK_RECEIVED);
+    TO_PROTO_ENUM(kRembBitrate, REMB_BITRATE);
+    TO_PROTO_ENUM(kAudioAckSent, AUDIO_ACK_SENT);
+    TO_PROTO_ENUM(kVideoAckSent, VIDEO_ACK_SENT);
+    TO_PROTO_ENUM(kAudioFrameReceived, AUDIO_FRAME_RECEIVED);
+    TO_PROTO_ENUM(kAudioFrameCaptured, AUDIO_FRAME_CAPTURED);
+    TO_PROTO_ENUM(kAudioFrameEncoded, AUDIO_FRAME_ENCODED);
+    TO_PROTO_ENUM(kAudioPlayoutDelay, AUDIO_PLAYOUT_DELAY);
+    TO_PROTO_ENUM(kAudioFrameDecoded, AUDIO_FRAME_DECODED);
+    TO_PROTO_ENUM(kVideoFrameCaptured, VIDEO_FRAME_CAPTURED);
+    TO_PROTO_ENUM(kVideoFrameReceived, VIEDO_FRAME_RECEIVED);
+    TO_PROTO_ENUM(kVideoFrameSentToEncoder, VIDEO_FRAME_SENT_TO_ENCODER);
+    TO_PROTO_ENUM(kVideoFrameEncoded, VIDEO_FRAME_ENCODED);
+    TO_PROTO_ENUM(kVideoFrameDecoded, VIDEO_FRAME_DECODED);
+    TO_PROTO_ENUM(kVideoRenderDelay, VIDEO_RENDER_DELAY);
+    TO_PROTO_ENUM(kPacketSentToPacer, PACKET_SENT_TO_PACER);
+    TO_PROTO_ENUM(kPacketSentToNetwork, PACKET_SENT_TO_NETWORK);
+    TO_PROTO_ENUM(kPacketRetransmitted, PACKET_RETRANSMITTED);
+    TO_PROTO_ENUM(kAudioPacketReceived, AUDIO_PACKET_RECEIVED);
+    TO_PROTO_ENUM(kVideoPacketReceived, VIDEO_PACKET_RECEIVED);
+    TO_PROTO_ENUM(kDuplicatePacketReceived, DUPLICATE_PACKET_RECEIVED);
+    case kNumOfLoggingEvents:
+      NOTREACHED();
+      return media::cast::proto::UNKNOWN;
+  }
+  NOTREACHED();
+  return media::cast::proto::UNKNOWN;
+}
+
+}  // namespace cast
+}  // namespace media

media/cast/logging/proto/proto_utils.h

+// Copyright 2014 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#ifndef MEDIA_CAST_LOGGING_PROTO_PROTO_UTILS_H_
+#define MEDIA_CAST_LOGGING_PROTO_PROTO_UTILS_H_
+
+#include "media/cast/logging/logging_defines.h"
+#include "media/cast/logging/proto/raw_events.pb.h"
+
+// Utility functions for cast logging protos.
+namespace media {
+namespace cast {
+
+// Converts |event| to a corresponding value in |media::cast::proto::EventType|.
+media::cast::proto::EventType ToProtoEventType(CastLoggingEvent event);
+
+}  // namespace cast
+}  // namespace media
+
+#endif  // MEDIA_CAST_LOGGING_PROTO_PROTO_UTILS_H_

media/cast/logging/proto/raw_events.proto

+// Copyright 2014 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+// Protocol for audio messages.
+
+syntax = "proto2";
+
+option optimize_for = LITE_RUNTIME;
+
+package media.cast.proto;
+
+// Keep in sync with media/cast/logging/logging_defines.h.
+// For compatibility reasons, existing values in this enum must not be changed.
+enum EventType {
+  // Generic events.
+  UNKNOWN = 0;
+  RTT_MS = 1;
+  PACKET_LOSS = 2;
+  JITTER_MS = 3;
+  VIDEO_ACK_RECEIVED = 4;
+  REMB_BITRATE = 5;
+  AUDIO_ACK_SENT = 6;
+  VIDEO_ACK_SENT = 7;
+  // Audio sender.
+  AUDIO_FRAME_RECEIVED = 8;
+  AUDIO_FRAME_CAPTURED = 9;
+  AUDIO_FRAME_ENCODED = 10;
+  // Audio receiver.
+  AUDIO_PLAYOUT_DELAY = 11;
+  AUDIO_FRAME_DECODED = 12;
+  // Video sender.
+  VIDEO_FRAME_CAPTURED = 13;
+  VIEDO_FRAME_RECEIVED = 14;
+  VIDEO_FRAME_SENT_TO_ENCODER = 15;
+  VIDEO_FRAME_ENCODED = 16;
+  // Video receiver.
+  VIDEO_FRAME_DECODED = 17;
+  VIDEO_RENDER_DELAY = 18;
+  // Send-side packet events.
+  PACKET_SENT_TO_PACER = 19;
+  PACKET_SENT_TO_NETWORK = 20;
+  PACKET_RETRANSMITTED = 21;
+  // Receiver-side packet events.
+  AUDIO_PACKET_RECEIVED = 22;
+  VIDEO_PACKET_RECEIVED = 23;
+  DUPLICATE_PACKET_RECEIVED = 24;
+}
+
+message AggregatedFrameEvent {
+  optional uint32 rtp_timestamp = 1;
+
+  repeated EventType event_type = 2 [packed = true];
+  repeated int64 event_timestamp_micros = 3 [packed = true];
+
+  // Size is set only for kAudioFrameEncoded and kVideoFrameEncoded.
+  optional int32 encoded_frame_size = 4;
+
+  // Delay is only set for kAudioPlayoutDelay and kVideoRenderDelay.
+  optional int32 delay_millis = 5;
+};
+
+message BasePacketEvent {
+  optional int32 packet_id = 1;
+  repeated EventType event_type = 2 [packed = true];
+  repeated int64 event_timestamp_micros = 3 [packed = true];
+}
+
+message AggregatedPacketEvent {
+  optional uint32 rtp_timestamp = 1;
+  repeated BasePacketEvent base_packet_event = 2;
+};
+
+message AggregatedGenericEvent {
+  optional EventType event_type = 1;
+  repeated int64 event_timestamp_micros = 2 [packed = true];
+  repeated int32 value = 3 [packed = true];
+};

media/cast/logging/simple_event_subscriber_unittest.cc

@@ -39,9 +39,9 @@ class SimpleEventSubscriberTest : public ::testing::Test {
 
 TEST_F(SimpleEventSubscriberTest, GetAndResetEvents) {
   // Log some frame events.
-  cast_environment_->Logging()->InsertFrameEventWithDelay(
-      testing_clock_->NowTicks(), kAudioFrameDecoded, /*rtp_timestamp*/ 100u,
-      /*frame_id*/ 0u, /*delay*/ base::TimeDelta::FromMilliseconds(100));
+  cast_environment_->Logging()->InsertFrameEventWithSize(
+      testing_clock_->NowTicks(), kAudioFrameEncoded, /*rtp_timestamp*/ 100u,
+      /*frame_id*/ 0u, /*frame_size*/ 123);
   cast_environment_->Logging()->InsertFrameEventWithDelay(
       testing_clock_->NowTicks(), kAudioPlayoutDelay, /*rtp_timestamp*/ 100u,
       /*frame_id*/ 0u, /*delay*/ base::TimeDelta::FromMilliseconds(100));