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 {
+  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.
+  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;
+
+  // 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) {
+    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:
+  FileThreadWorker(const base::FilePath& incoming_dump_path,
+                   const base::FilePath& outgoing_dump_path,
+                   size_t max_dump_size)
+      : incoming_dump_path_(incoming_dump_path),
+        outgoing_dump_path_(outgoing_dump_path),
+        max_dump_size_(max_dump_size),
+        total_dump_size_(0),
+        incoming_stream_initialized_(false),
+        outgoing_stream_initialized_(false) {
+    thread_checker_.DetachFromThread();
+  }
+
+  ~FileThreadWorker() { DCHECK(thread_checker_.CalledOnValidThread()); }
+
+  // Compresses the data in |incoming_buffer| and |outgoing_buffer| and write
+  // to the dump file. If |end_stream| is true, both the incoming and outgoing
+  // compression streams will be ended and the dump file cannot be written to
+  // any more.
+  void CompressAndWriteToFileOnFileThread(
+      scoped_ptr<std::vector<uint8> > incoming_buffer,
+      scoped_ptr<std::vector<uint8> > outgoing_buffer,
+      bool end_stream,
+      FlushResult* result) {
+    DCHECK(thread_checker_.CalledOnValidThread());
+
+    *result = FLUSH_RESULT_SUCCESS;
+
+    if (incoming_buffer->size())
+      CompressAndWriteBufferToFile(incoming_buffer.Pass(), true, result);
+
+    if (*result != FLUSH_RESULT_FAILURE && outgoing_buffer->size())
+      CompressAndWriteBufferToFile(outgoing_buffer.Pass(), false, result);
+
+    if (end_stream) {
+      bool succeeded = false;
+
+      succeeded = EndDumpFile(
+          incoming_dump_path_, incoming_stream_initialized_, &incoming_stream_);
+
+      succeeded = succeeded && EndDumpFile(outgoing_dump_path_,
+                                           outgoing_stream_initialized_,
+                                           &outgoing_stream_);
+
+      if (!succeeded)
+        *result = FLUSH_RESULT_FAILURE;
+    }
+  }
+
+ private:
+  // Helper for CompressAndWriteToFileOnFileThread to compress and write one
+  // dump.
+  void CompressAndWriteBufferToFile(scoped_ptr<std::vector<uint8> > buffer,
+                                    bool incoming,
+                                    FlushResult* result) {
+    DCHECK(thread_checker_.CalledOnValidThread());
+    DCHECK(buffer->size());
+
+    *result = FLUSH_RESULT_SUCCESS;
+
+    base::FilePath* file_path =
+        incoming ? &incoming_dump_path_ : &outgoing_dump_path_;
+
+    std::vector<uint8> compressed_buffer;
+    if (!Compress(buffer.get(), incoming, &compressed_buffer)) {
+      DVLOG(2) << "Compressing buffer failed.";
+      *result = FLUSH_RESULT_FAILURE;
+      return;
+    }
+
+    int bytes_written = -1;
+
+    if (base::PathExists(*file_path)) {
+      bytes_written = base::AppendToFile(
+          *file_path,
+          reinterpret_cast<const char*>(compressed_buffer.data()),
+          compressed_buffer.size());
+    } else {
+      bytes_written = base::WriteFile(
+          *file_path,
+          reinterpret_cast<const char*>(compressed_buffer.data()),
+          compressed_buffer.size());
+    }
+
+    if (bytes_written == -1) {
+      DVLOG(2) << "Writing file failed: " << file_path->value();
+      *result = FLUSH_RESULT_FAILURE;
+      return;
+    }
+
+    DCHECK_EQ(static_cast<size_t>(bytes_written), compressed_buffer.size());
+
+    total_dump_size_ += bytes_written;
+
+    if (total_dump_size_ > max_dump_size_)
+      *result = FLUSH_RESULT_MAX_SIZE_REACHED;
+  }
+
+  // Compresses |input| into |output| using the incoming or outgoing compression
+  // stream depending on |incoming|.
+  bool Compress(std::vector<uint8>* input,
+                bool incoming,
+                std::vector<uint8>* output) {
+    DCHECK(thread_checker_.CalledOnValidThread());
+    int result = Z_OK;
+
+    output->resize(std::max(kMinimumGzipOutputBufferSize, input->size()));
+
+    z_stream* stream = incoming ? &incoming_stream_ : &outgoing_stream_;
+
+    bool* stream_initialized = incoming ? &incoming_stream_initialized_
+                                        : &outgoing_stream_initialized_;
+
+    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 specified by
+  // |dump_path|.
+  bool EndDumpFile(const base::FilePath& dump_path,
+                   bool stream_initialized,
+                   z_stream* stream) {
+    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;
+  }
+
+  base::FilePath incoming_dump_path_;
+  base::FilePath outgoing_dump_path_;
+
+  size_t max_dump_size_;
+  size_t total_dump_size_;
+
+  z_stream incoming_stream_;
+  z_stream outgoing_stream_;
+  bool incoming_stream_initialized_;
+  bool outgoing_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),
+      file_thread_worker_(new FileThreadWorker(incoming_dump_path,
+                                               outgoing_dump_path,
+                                               max_dump_size)),
+      weak_ptr_factory_(this) {
+}
+
+WebRtcRtpDumpWriter::~WebRtcRtpDumpWriter() {
+  DCHECK(thread_checker_.CalledOnValidThread());
+  if (BrowserThread::DeleteSoon(
+          BrowserThread::FILE, FROM_HERE, file_thread_worker_.get())) {
+    ignore_result(file_thread_worker_.release());
+  } else {
+    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)
+    FlushBuffers(incoming, !incoming, false, base::Callback<void(bool)>());
+
+  // 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(
+    const base::Callback<void(bool)>& finished_callback) {
+  DCHECK(thread_checker_.CalledOnValidThread());
+
+  FlushBuffers(true, true, true, finished_callback);
+}
+
+void WebRtcRtpDumpWriter::FlushBuffers(
+    bool incoming,
+    bool outgoing,
+    bool end_stream,
+    const base::Callback<void(bool)>& callback) {
+  DCHECK(thread_checker_.CalledOnValidThread());
+
+  scoped_ptr<std::vector<uint8> > new_incoming_buffer(new std::vector<uint8>());
+  scoped_ptr<std::vector<uint8> > new_outgoing_buffer(new std::vector<uint8>());
+
+  if (incoming) {
+    new_incoming_buffer->reserve(incoming_buffer_.capacity());
+    new_incoming_buffer->swap(incoming_buffer_);
+  }
+
+  if (outgoing) {
+    new_outgoing_buffer->reserve(outgoing_buffer_.capacity());
+    new_outgoing_buffer->swap(outgoing_buffer_);
+  }
+
+  scoped_ptr<FlushResult> result(new FlushResult(FLUSH_RESULT_FAILURE));
+
+  // OnFlushDone is necessary to avoid running the callback after this object is
+  // gone.
+  BrowserThread::PostTaskAndReply(
+      BrowserThread::FILE,
+      FROM_HERE,
+      base::Bind(&FileThreadWorker::CompressAndWriteToFileOnFileThread,
+                 base::Unretained(file_thread_worker_.get()),
+                 Passed(&new_incoming_buffer),
+                 Passed(&new_outgoing_buffer),
+                 end_stream,
+                 result.get()),
+      base::Bind(&WebRtcRtpDumpWriter::OnFlushDone,
+                 weak_ptr_factory_.GetWeakPtr(),
+                 callback,
+                 Passed(&result)));
+}
+
+void WebRtcRtpDumpWriter::OnFlushDone(
+    const base::Callback<void(bool)>& callback,
+    const scoped_ptr<FlushResult>& result) {
+  DCHECK(thread_checker_.CalledOnValidThread());
+
+  if (*result == FLUSH_RESULT_MAX_SIZE_REACHED &&
+      !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);
+}