Implements RTP header dumping.

chrome/browser/media/webrtc_rtp_dump_writer.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 "chrome/browser/media/webrtc_rtp_dump_writer.h"
+
+#include "base/big_endian.h"
+#include "base/file_util.h"
+#include "base/logging.h"
+#include "content/public/browser/browser_thread.h"
+#include "third_party/zlib/zlib.h"
+
+using content::BrowserThread;
+
+namespace {
+
+// The header of the dump file.
+struct RtpDumpFileHeader {

[tommi (sloooow) - chröme, 2014/05/17 10:18:37]

is there a certain packing that's assumed for these structs?  If so, what is it?
 Is it safe to assume that all build configurations will be configured for that
exact packing or should we be explicit?

Maybe it would be good to have compile time assert for the sizes of these
structs?

[jiayl, 2014/05/19 17:32:59]

Make it independent of the packing by replacing sizeof(...) with a const.

+  static const unsigned char kFirstLine[];
+
+  explicit RtpDumpFileHeader(const base::TimeTicks& start)
+      : start_sec(0), start_usec(0), source(0), port(0), padding(0) {
+    base::TimeDelta interval(start - base::TimeTicks());
+    start_sec = interval.InSeconds();
+    start_usec =
+        interval.InMilliseconds() * base::Time::kMicrosecondsPerMillisecond;
+  }
+
+  void WriteBigEndian(std::vector<uint8>* output) {
+    size_t buffer_start_pos = output->size();
+    output->resize(output->size() + sizeof(RtpDumpFileHeader));
+
+    char* buffer = reinterpret_cast<char*>(output->data() + buffer_start_pos);
+
+    base::WriteBigEndian(buffer, start_sec);
+    buffer += sizeof(start_sec);
+
+    base::WriteBigEndian(buffer, start_usec);
+    buffer += sizeof(start_usec);
+
+    base::WriteBigEndian(buffer, source);
+    buffer += sizeof(source);
+
+    base::WriteBigEndian(buffer, port);
+    buffer += sizeof(port);
+
+    base::WriteBigEndian(buffer, padding);
+  }
+
+  uint32 start_sec;   //  start of recording, the seconds part.

[tommi (sloooow) - chröme, 2014/05/17 10:18:37]

could these members all be const? (sorry, have to ask)

[jiayl, 2014/05/19 17:32:59]

Done.

+  uint32 start_usec;  //  start of recording, the microseconds part.
+  uint32 source;      //  network source (multicast address). Always 0.
+  uint16 port;        //  UDP port. Always 0.
+  uint16 padding;     //  2 bytes padding.
+};
+
+const unsigned char RtpDumpFileHeader::kFirstLine[] =
+    "#!rtpplay1.0 0.0.0.0/0\n";
+
+// The header for each packet dump.
+struct PacketDumpHeader {
+  PacketDumpHeader(const base::TimeTicks& start,
+                   uint16 dump_length,
+                   uint16 packet_length)
+      : packet_dump_length(dump_length),
+        packet_length(packet_length),
+        offset_ms((base::TimeTicks::Now() - start).InMilliseconds()) {}
+
+  void WriteBigEndian(std::vector<uint8>* output) {
+    size_t buffer_start_pos = output->size();
+    output->resize(output->size() + sizeof(PacketDumpHeader));
+
+    char* buffer = reinterpret_cast<char*>(output->data() + buffer_start_pos);
+
+    base::WriteBigEndian(buffer, packet_dump_length);
+    buffer += sizeof(packet_dump_length);
+
+    base::WriteBigEndian(buffer, packet_length);
+    buffer += sizeof(packet_length);
+
+    base::WriteBigEndian(buffer, offset_ms);
+  }
+
+  // Length of the packet dump including this header.
+  uint16 packet_dump_length;

[tommi (sloooow) - chröme, 2014/05/17 10:18:37]

and these ones?

[jiayl, 2014/05/19 17:32:59]

Done.

+
+  // Length of header + payload of the RTP packet.
+  uint16 packet_length;
+
+  // Milliseconds since the start of recording.
+  uint32 offset_ms;
+};
+
+// Append |src_len| bytes from |src| to |dest|.
+bool AppendToBuffer(const uint8* src,
+                    size_t src_len,
+                    std::vector<uint8>* dest) {
+  if (dest->capacity() < dest->size() + src_len)
+    return false;
+
+  for (size_t i = 0; i < src_len; ++i) {

[tommi (sloooow) - chröme, 2014/05/17 10:18:37]

nit: no {}

[jiayl, 2014/05/19 17:32:59]

Done.

+    dest->push_back(src[i]);
+  }
+  return true;
+}
+
+static const size_t kMinimumGzipOutputBufferSize = 256;
+
+}  // namespace
+
+// This class is running on the FILE thread for compressing and writing the
+// dump buffer to disk.
+class WebRtcRtpDumpWriter::FileThreadWorker {
+ public:
+  explicit FileThreadWorker(const base::FilePath& dump_path)
+      : dump_path_(dump_path), stream_initialized_(false) {
+    thread_checker_.DetachFromThread();
+  }
+
+  ~FileThreadWorker() { DCHECK(thread_checker_.CalledOnValidThread()); }
+
+  // Compresses the data in |buffer| write to the dump file. If |end_stream| is
+  // true, the compression stream will be ended and the dump file cannot be
+  // written to any more.
+  void CompressAndWriteToFileOnFileThread(
+      const scoped_ptr<std::vector<uint8> >& buffer,
+      bool end_stream,
+      FlushResult* result,
+      size_t* bytes_written) {
+    DCHECK(thread_checker_.CalledOnValidThread());
+
+    *result = FLUSH_RESULT_SUCCESS;
+    *bytes_written = 0;
+
+    if (buffer->size()) {
+      *bytes_written = CompressAndWriteBufferToFile(buffer.get(), result);
+    } else if (!base::PathExists(dump_path_)) {
+      *result = FLUSH_RESULT_NO_DATA;
+    }
+
+    if (end_stream) {
+      if (!EndDumpFile())
+        *result = FLUSH_RESULT_FAILURE;
+    }
+  }
+
+ private:
+  // Helper for CompressAndWriteToFileOnFileThread to compress and write one
+  // dump.
+  size_t CompressAndWriteBufferToFile(std::vector<uint8>* buffer,
+                                      FlushResult* result) {
+    DCHECK(thread_checker_.CalledOnValidThread());
+    DCHECK(buffer->size());
+
+    *result = FLUSH_RESULT_SUCCESS;
+
+    std::vector<uint8> compressed_buffer;
+    if (!Compress(buffer, &compressed_buffer)) {
+      DVLOG(2) << "Compressing buffer failed.";
+      *result = FLUSH_RESULT_FAILURE;
+      return 0;
+    }
+
+    int bytes_written = -1;
+
+    if (base::PathExists(dump_path_)) {
+      bytes_written = base::AppendToFile(
+          dump_path_,
+          reinterpret_cast<const char*>(compressed_buffer.data()),
+          compressed_buffer.size());
+    } else {
+      bytes_written = base::WriteFile(
+          dump_path_,
+          reinterpret_cast<const char*>(compressed_buffer.data()),
+          compressed_buffer.size());
+    }
+
+    if (bytes_written == -1) {
+      DVLOG(2) << "Writing file failed: " << dump_path_.value();
+      *result = FLUSH_RESULT_FAILURE;
+      return 0;
+    }
+
+    DCHECK_EQ(static_cast<size_t>(bytes_written), compressed_buffer.size());
+
+    return bytes_written;
+  }
+
+  // Compresses |input| into |output|.
+  bool Compress(std::vector<uint8>* input, std::vector<uint8>* output) {

[tommi (sloooow) - chröme, 2014/05/17 10:18:37]

why is |input| writable?

[jiayl, 2014/05/19 17:32:59]

the zlib methods require z_stream::next_in to be writable.

+    DCHECK(thread_checker_.CalledOnValidThread());
+    int result = Z_OK;
+
+    output->resize(std::max(kMinimumGzipOutputBufferSize, input->size()));
+
+    if (!stream_initialized_) {
+      memset(&stream_, 0, sizeof(stream_));
+      result = deflateInit2(&stream_,
+                            Z_DEFAULT_COMPRESSION,
+                            Z_DEFLATED,
+                            // windowBits = 15 is default, 16 is added to
+                            // produce a gzip header + trailer.
+                            15 + 16,
+                            8,  // memLevel = 8 is default.
+                            Z_DEFAULT_STRATEGY);
+      DCHECK_EQ(Z_OK, result);
+      stream_initialized_ = true;
+    }
+
+    stream_.next_in = input->data();
+    stream_.avail_in = input->size();
+    stream_.next_out = output->data();
+    stream_.avail_out = output->size();
+
+    result = deflate(&stream_, Z_SYNC_FLUSH);
+    DCHECK_EQ(Z_OK, result);
+    DCHECK_EQ(0U, stream_.avail_in);
+
+    output->resize(output->size() - stream_.avail_out);
+
+    stream_.next_in = NULL;
+    stream_.next_out = NULL;
+    stream_.avail_out = 0;
+    return true;
+  }
+
+  // Ends the compression stream and completes the dump file.
+  bool EndDumpFile() {
+    DCHECK(thread_checker_.CalledOnValidThread());
+
+    if (!stream_initialized_)
+      return true;
+
+    std::vector<uint8> output_buffer;
+    output_buffer.resize(kMinimumGzipOutputBufferSize);
+
+    stream_.next_in = NULL;
+    stream_.avail_in = 0;
+    stream_.next_out = output_buffer.data();
+    stream_.avail_out = output_buffer.size();
+
+    int result = deflate(&stream_, Z_FINISH);
+    DCHECK_EQ(Z_STREAM_END, result);
+
+    result = deflateEnd(&stream_);
+    DCHECK_EQ(Z_OK, result);
+
+    output_buffer.resize(output_buffer.size() - stream_.avail_out);
+
+    int bytes_written =
+        base::AppendToFile(dump_path_,
+                           reinterpret_cast<const char*>(output_buffer.data()),
+                           output_buffer.size());
+
+    return bytes_written > 0;
+  }
+
+  const base::FilePath dump_path_;
+
+  z_stream stream_;
+  bool stream_initialized_;
+
+  base::ThreadChecker thread_checker_;
+
+  DISALLOW_COPY_AND_ASSIGN(FileThreadWorker);
+};
+
+WebRtcRtpDumpWriter::WebRtcRtpDumpWriter(
+    const base::FilePath& incoming_dump_path,
+    const base::FilePath& outgoing_dump_path,
+    size_t max_dump_size,
+    const base::Closure& max_dump_size_reached_callback)
+    : max_dump_size_(max_dump_size),
+      max_dump_size_reached_callback_(max_dump_size_reached_callback),
+      total_dump_size_on_disk_(0),
+      incoming_file_thread_worker_(new FileThreadWorker(incoming_dump_path)),
+      outgoing_file_thread_worker_(new FileThreadWorker(outgoing_dump_path)),
+      weak_ptr_factory_(this) {
+}
+
+WebRtcRtpDumpWriter::~WebRtcRtpDumpWriter() {
+  DCHECK(thread_checker_.CalledOnValidThread());
+  if (BrowserThread::DeleteSoon(
+          BrowserThread::FILE, FROM_HERE, incoming_file_thread_worker_.get())) {
+    ignore_result(incoming_file_thread_worker_.release());
+  } else {
+    incoming_file_thread_worker_.reset();

[tommi (sloooow) - chröme, 2014/05/17 10:18:37]

if we get here, is that an error or expected case?  Can you document what will
have happened if we hit here?

If this should never happen, I'd prefer a check here and just pass the return
value of relese() to DeleteSoon().  Same below.

[jiayl, 2014/05/19 17:32:59]

Done.

+  }
+
+  if (BrowserThread::DeleteSoon(
+          BrowserThread::FILE, FROM_HERE, outgoing_file_thread_worker_.get())) {
+    ignore_result(outgoing_file_thread_worker_.release());
+  } else {
+    outgoing_file_thread_worker_.reset();
+  }
+}
+
+void WebRtcRtpDumpWriter::WriteRtpPacket(const uint8* packet_header,
+                                         size_t header_length,
+                                         size_t packet_length,
+                                         bool incoming) {
+  DCHECK(thread_checker_.CalledOnValidThread());
+
+  static const size_t kMaxInMemoryBufferSize = 65536;  // 64KB
+
+  std::vector<uint8>* dest_buffer =
+      incoming ? &incoming_buffer_ : &outgoing_buffer_;
+  bool succeeded = true;
+  if (!dest_buffer->capacity()) {
+    dest_buffer->reserve(std::min(kMaxInMemoryBufferSize, max_dump_size_));
+
+    start_time_ = base::TimeTicks::Now();
+
+    // Writes the dump file header.
+    succeeded = AppendToBuffer(RtpDumpFileHeader::kFirstLine,
+                               arraysize(RtpDumpFileHeader::kFirstLine) - 1,
+                               dest_buffer);
+    DCHECK(succeeded);
+
+    RtpDumpFileHeader header(start_time_);
+    header.WriteBigEndian(dest_buffer);
+  }
+
+  size_t packet_dump_length = sizeof(PacketDumpHeader) + header_length;
+
+  // Flushes the buffer to disk if the buffer is full.
+  if (dest_buffer->capacity() < dest_buffer->size() + packet_dump_length)
+    FlushBuffer(incoming, false, FlushDoneCallback());
+
+  // Writes the packet dump header.
+  PacketDumpHeader packet_dump_header(
+      start_time_, packet_dump_length, packet_length);
+  packet_dump_header.WriteBigEndian(dest_buffer);
+
+  // Writes the actual RTP packet header.
+  succeeded = AppendToBuffer(packet_header, header_length, dest_buffer);
+  DCHECK(succeeded);
+}
+
+void WebRtcRtpDumpWriter::EndDump(RtpDumpType type,
+                                  const EndDumpCallback& finished_callback) {
+  DCHECK(thread_checker_.CalledOnValidThread());
+
+  bool incoming = (type == RTP_DUMP_BOTH || type == RTP_DUMP_INCOMING);
+  EndDumpContext context(type, finished_callback);
+
+  FlushBuffer(incoming,
+              true,
+              base::Bind(&WebRtcRtpDumpWriter::OnDumpEnded,
+                         weak_ptr_factory_.GetWeakPtr(),
+                         context,
+                         incoming));
+}
+
+WebRtcRtpDumpWriter::EndDumpContext::EndDumpContext(
+    RtpDumpType type,
+    const EndDumpCallback& callback)
+    : type(type),
+      incoming_succeeded(false),
+      outgoing_succeeded(false),
+      callback(callback) {
+}
+
+WebRtcRtpDumpWriter::EndDumpContext::~EndDumpContext() {
+}
+
+void WebRtcRtpDumpWriter::FlushBuffer(bool incoming,
+                                      bool end_stream,
+                                      const FlushDoneCallback& callback) {
+  DCHECK(thread_checker_.CalledOnValidThread());
+
+  scoped_ptr<std::vector<uint8> > new_buffer(new std::vector<uint8>());
+
+  if (incoming) {
+    new_buffer->reserve(incoming_buffer_.capacity());
+    new_buffer->swap(incoming_buffer_);
+  } else {
+    new_buffer->reserve(outgoing_buffer_.capacity());
+    new_buffer->swap(outgoing_buffer_);
+  }
+
+  scoped_ptr<FlushResult> result(new FlushResult(FLUSH_RESULT_FAILURE));
+
+  scoped_ptr<size_t> bytes_written(new size_t(0));
+
+  FileThreadWorker* worker = incoming ? incoming_file_thread_worker_.get()
+                                      : outgoing_file_thread_worker_.get();
+
+  // Using "Unretained(worker)" because |worker| is owner by this object and it
+  // guaranteed to be deleted on the FILE thread before this object goes away.
+  BrowserThread::PostTaskAndReply(
+      BrowserThread::FILE,
+      FROM_HERE,
+      base::Bind(&FileThreadWorker::CompressAndWriteToFileOnFileThread,
+                 base::Unretained(worker),
+                 Passed(&new_buffer),
+                 end_stream,
+                 result.get(),
+                 bytes_written.get()),
+      // OnFlushDone is necessary to avoid running the callback after this
+      // object is gone.
+      base::Bind(&WebRtcRtpDumpWriter::OnFlushDone,
+                 weak_ptr_factory_.GetWeakPtr(),
+                 callback,
+                 Passed(&result),
+                 Passed(&bytes_written)));
+}
+
+void WebRtcRtpDumpWriter::OnFlushDone(const FlushDoneCallback& callback,
+                                      const scoped_ptr<FlushResult>& result,
+                                      const scoped_ptr<size_t>& bytes_written) {
+  DCHECK(thread_checker_.CalledOnValidThread());
+
+  total_dump_size_on_disk_ += *bytes_written;
+
+  if (total_dump_size_on_disk_ >= max_dump_size_ &&
+      !max_dump_size_reached_callback_.is_null()) {
+    max_dump_size_reached_callback_.Run();
+  }
+
+  // Returns success for FLUSH_RESULT_MAX_SIZE_REACHED since the dump is still
+  // valid.
+  if (!callback.is_null()) {
+    callback.Run(*result != FLUSH_RESULT_FAILURE &&
+                 *result != FLUSH_RESULT_NO_DATA);
+  }
+}
+
+void WebRtcRtpDumpWriter::OnDumpEnded(EndDumpContext context,
+                                      bool incoming,
+                                      bool success) {
+  DCHECK(thread_checker_.CalledOnValidThread());
+
+  DVLOG(2) << "Dump ended, incoming = " << incoming
+           << ", succeeded = " << success;
+
+  if (incoming)
+    context.incoming_succeeded = success;
+  else
+    context.outgoing_succeeded = success;
+
+  // End the outgoing dump if needed.
+  if (incoming && context.type == RTP_DUMP_BOTH) {
+    FlushBuffer(false,
+                true,
+                base::Bind(&WebRtcRtpDumpWriter::OnDumpEnded,
+                           weak_ptr_factory_.GetWeakPtr(),
+                           context,
+                           false));
+    return;
+  }
+
+  context.callback.Run(context.incoming_succeeded, context.outgoing_succeeded);
+}