Media Remoting: UMAs to track session events and measurements.

media/remoting/remote_demuxer_stream_adapter.cc

 // Copyright 2016 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/remoting/remote_demuxer_stream_adapter.h"
 
 #include "base/base64.h"
 #include "base/bind.h"
+#include "base/callback_helpers.h"
 #include "media/base/bind_to_current_loop.h"
 #include "media/base/decoder_buffer.h"
 #include "media/base/timestamp_constants.h"
 #include "media/remoting/rpc/proto_enum_utils.h"
 #include "media/remoting/rpc/proto_utils.h"
 
 // Convenience logging macro used throughout this file.
 #define DEMUXER_VLOG(level) VLOG(level) << __func__ << "[" << name_ << "]: "
 
 namespace media {
 namespace remoting {
 
-namespace {
-constexpr char kErrorLostMojoConnection[] = "Mojo connection error";
-constexpr char kErrorDataPipeWrite[] = "Mojo data pipe write error";
-constexpr char kErrorDuplicateInitialize[] = "Duplicate attempt to initialize";
-constexpr char kErrorMissingMessageFields[] = "Missing required message fields";
-}  // namepsace
-
 // static
 mojo::DataPipe* CreateDataPipe() {
   // Capacity in bytes for Mojo data pipe.
   constexpr int kMojoDataPipeCapacityInBytes = 512 * 1024;
 
   MojoCreateDataPipeOptions options;
   options.struct_size = sizeof(MojoCreateDataPipeOptions);
   options.flags = MOJO_WRITE_DATA_FLAG_NONE;
   options.element_num_bytes = 1;
   options.capacity_num_bytes = kMojoDataPipeCapacityInBytes;
   return new mojo::DataPipe(options);
 }
 
 RemoteDemuxerStreamAdapter::RemoteDemuxerStreamAdapter(
     scoped_refptr<base::SingleThreadTaskRunner> main_task_runner,
     scoped_refptr<base::SingleThreadTaskRunner> media_task_runner,
     const std::string& name,
     ::media::DemuxerStream* demuxer_stream,
     const base::WeakPtr<RpcBroker>& rpc_broker,
     int rpc_handle,
     mojom::RemotingDataStreamSenderPtrInfo stream_sender_info,
-    mojo::ScopedDataPipeProducerHandle producer_handle)
+    mojo::ScopedDataPipeProducerHandle producer_handle,
+    const ErrorCallback& error_callback)
     : main_task_runner_(std::move(main_task_runner)),
       media_task_runner_(std::move(media_task_runner)),
       name_(name),
       rpc_broker_(rpc_broker),
       rpc_handle_(rpc_handle),
       demuxer_stream_(demuxer_stream),
       type_(demuxer_stream ? demuxer_stream->type() : DemuxerStream::UNKNOWN),
+      error_callback_(error_callback),
       remote_callback_handle_(kInvalidHandle),
       read_until_callback_handle_(kInvalidHandle),
       read_until_count_(0),
       last_count_(0),
       pending_flush_(false),
       current_pending_frame_offset_(0),
       pending_frame_is_eos_(false),
       media_status_(::media::DemuxerStream::kOk),
       producer_handle_(std::move(producer_handle)),
+      bytes_written_to_pipe_(0),
       request_buffer_weak_factory_(this),
       weak_factory_(this) {
   DCHECK(main_task_runner_);
   DCHECK(media_task_runner_);
   DCHECK(media_task_runner_->BelongsToCurrentThread());
   DCHECK(demuxer_stream);
+  DCHECK(!error_callback.is_null());
   const RpcBroker::ReceiveMessageCallback receive_callback =
       media::BindToCurrentLoop(
           base::Bind(&RemoteDemuxerStreamAdapter::OnReceivedRpc,
                      weak_factory_.GetWeakPtr()));
   main_task_runner_->PostTask(
       FROM_HERE,
       base::Bind(&remoting::RpcBroker::RegisterMessageReceiverCallback,
                  rpc_broker_, rpc_handle_, receive_callback));
 
   stream_sender_.Bind(std::move(stream_sender_info));
   stream_sender_.set_connection_error_handler(
       base::Bind(&RemoteDemuxerStreamAdapter::OnFatalError,
-                 weak_factory_.GetWeakPtr(), kErrorLostMojoConnection));
+                 weak_factory_.GetWeakPtr(), MOJO_PIPE_ERROR));
 }
 
 RemoteDemuxerStreamAdapter::~RemoteDemuxerStreamAdapter() {
   DCHECK(media_task_runner_->BelongsToCurrentThread());
   main_task_runner_->PostTask(
       FROM_HERE,
       base::Bind(&remoting::RpcBroker::UnregisterMessageReceiverCallback,
                  rpc_broker_, rpc_handle_));
 }
 
+int64_t RemoteDemuxerStreamAdapter::GetBytesWrittenAndReset() {
+  DCHECK(media_task_runner_->BelongsToCurrentThread());
+  const int64_t current_count = bytes_written_to_pipe_;
+  bytes_written_to_pipe_ = 0;
+  return current_count;
+}
+
 base::Optional<uint32_t> RemoteDemuxerStreamAdapter::SignalFlush(
     bool flushing) {
   DCHECK(media_task_runner_->BelongsToCurrentThread());
   DEMUXER_VLOG(2) << "flushing=" << flushing;
 
   // Ignores if |pending_flush_| states is same.
   if (pending_flush_ == flushing)
     return base::nullopt;
 
   // Cleans up pending frame data.
@@ skipping 44 matching lines @@
 
   // Checks if initialization had been called or not.
   if (remote_callback_handle_ != kInvalidHandle) {
     DEMUXER_VLOG(1) << "Duplicated initialization. Have: "
                     << remote_callback_handle_
                     << ", Given: " << remote_callback_handle;
     // Shuts down data pipe if available if providing different remote callback
     // handle for initialization. Otherwise, just silently ignore the duplicated
     // request.
     if (remote_callback_handle_ != remote_callback_handle) {
-      OnFatalError(kErrorDuplicateInitialize);
+      OnFatalError(PEERS_OUT_OF_SYNC);
     }
     return;
   }
   remote_callback_handle_ = remote_callback_handle;
 
   // Issues RPC_DS_INITIALIZE_CALLBACK RPC message.
   std::unique_ptr<remoting::pb::RpcMessage> rpc(new remoting::pb::RpcMessage());
   rpc->set_handle(remote_callback_handle_);
   rpc->set_proc(remoting::pb::RpcMessage::RPC_DS_INITIALIZE_CALLBACK);
   auto* init_cb_message = rpc->mutable_demuxerstream_initializecb_rpc();
@@ skipping 35 matching lines @@
                                     weak_factory_.GetWeakPtr()));
   }
 }
 
 void RemoteDemuxerStreamAdapter::ReadUntil(
     std::unique_ptr<remoting::pb::RpcMessage> message) {
   DCHECK(media_task_runner_->BelongsToCurrentThread());
   DCHECK(message);
   if (!message->has_demuxerstream_readuntil_rpc()) {
     DEMUXER_VLOG(1) << "Missing required DemuxerStreamReadUntil struct in RPC";
-    OnFatalError(kErrorMissingMessageFields);
+    OnFatalError(RPC_INVALID);
     return;
   }
 
   const pb::DemuxerStreamReadUntil& rpc_message =
       message->demuxerstream_readuntil_rpc();
   DEMUXER_VLOG(2) << "Received RPC_DS_READUNTIL with callback_handle="
                   << rpc_message.callback_handle()
                   << ", count=" << rpc_message.count();
 
   if (pending_flush_) {
@@ skipping 72 matching lines @@
     } break;
   }
 }
 
 void RemoteDemuxerStreamAdapter::TryWriteData(MojoResult result) {
   DCHECK(media_task_runner_->BelongsToCurrentThread());
   // The Mojo watcher will also call TryWriteData() sometimes as a notification
   // that data pipe is ready. But that does not necessarily mean the data for a
   // read request is ready to be written into the pipe.
   if (!IsProcessingReadRequest() || pending_flush_) {
-    DEMUXER_VLOG(2) << "Skip actions since it's not in the reading state";
+    DEMUXER_VLOG(3) << "Skip actions since it's not in the reading state";
     return;
   }
 
   if (pending_frame_.empty()) {
     DEMUXER_VLOG(3) << "No data available, waiting for demuxer";
     return;
   }
 
   if (!stream_sender_ || !producer_handle_.is_valid()) {
     DEMUXER_VLOG(1) << "Ignore since data pipe stream sender is invalid";
     return;
   }
 
   uint32_t num_bytes = pending_frame_.size() - current_pending_frame_offset_;
   MojoResult mojo_result =
       WriteDataRaw(producer_handle_.get(),
                    pending_frame_.data() + current_pending_frame_offset_,
                    &num_bytes, MOJO_WRITE_DATA_FLAG_NONE);
   if (mojo_result != MOJO_RESULT_OK) {
     if (mojo_result != MOJO_RESULT_SHOULD_WAIT) {
       DEMUXER_VLOG(1) << "Pipe was closed unexpectedly (or a bug). result:"
                       << mojo_result;
-      OnFatalError(kErrorDataPipeWrite);
+      OnFatalError(MOJO_PIPE_ERROR);
     }
     return;
   }
 
   stream_sender_->ConsumeDataChunk(current_pending_frame_offset_, num_bytes,
                                    pending_frame_.size());
   current_pending_frame_offset_ += num_bytes;
+  bytes_written_to_pipe_ += num_bytes;
 
   // Checks if all buffer was written to browser process.
   if (current_pending_frame_offset_ != pending_frame_.size()) {
     // Returns and wait for mojo watcher to notify to write more data.
     return;
   }
 
   // Signal mojo remoting service that all frame buffer is written to data pipe.
   stream_sender_->SendFrame();
 
@@ skipping 65 matching lines @@
     video_config_ = ::media::VideoDecoderConfig();
 }
 
 void RemoteDemuxerStreamAdapter::ResetPendingFrame() {
   DCHECK(media_task_runner_->BelongsToCurrentThread());
   current_pending_frame_offset_ = 0;
   pending_frame_.clear();
   pending_frame_is_eos_ = false;
 }
 
-void RemoteDemuxerStreamAdapter::OnFatalError(const char* reason) {
+void RemoteDemuxerStreamAdapter::OnFatalError(StopTrigger stop_trigger) {
   DCHECK(media_task_runner_->BelongsToCurrentThread());
 
-  DEMUXER_VLOG(1) << reason;
+  DEMUXER_VLOG(1) << __func__ << " with StopTrigger " << stop_trigger;
 
-  // Resets mojo data pipe producer handle.
-  producer_handle_.reset();
-
-  // Resetting |stream_sender_| will close Mojo message pipe, which will cause
-  // the entire remoting session to be shut down.
-  if (stream_sender_) {
-    DEMUXER_VLOG(2) << "Reset data stream sender";
-    stream_sender_.reset();
-  }
+  if (error_callback_.is_null())
+    return;
 
   if (write_watcher_.IsWatching()) {
     DEMUXER_VLOG(2) << "Cancel mojo data pipe watcher";
     write_watcher_.Cancel();
   }
+
+  base::ResetAndReturn(&error_callback_).Run(stop_trigger);
 }
 
 }  // namespace remoting
 }  // namespace media