remoting/host/audio_pump.cc

   1 // Copyright 2015 The Chromium Authors. All rights reserved.
   2 // Use of this source code is governed by a BSD-style license that can be
   3 // found in the LICENSE file.
   4
   5 #include "remoting/host/audio_pump.h"
   6
   7 #include "base/bind.h"
   8 #include "base/location.h"
   9 #include "base/logging.h"
  10 #include "base/single_thread_task_runner.h"
  11 #include "base/thread_task_runner_handle.h"
  12 #include "remoting/codec/audio_encoder.h"
  13 #include "remoting/host/audio_capturer.h"
  14 #include "remoting/proto/audio.pb.h"
  15 #include "remoting/protocol/audio_stub.h"
  16
  17 namespace remoting {
  18
  19 // Limit the data stored in the pending send buffers to 250ms.
  20 const int kMaxBufferedIntervalMs = 250;
  21
  22 class AudioPump::Core {
  23    public:
  24   Core(base::WeakPtr<AudioPump> pump,
  25        scoped_ptr<AudioCapturer> audio_capturer,
  26        scoped_ptr<AudioEncoder> audio_encoder);
  27   ~Core();
  28
  29   void Start();
  30   void Pause(bool pause);
  31
  32   void OnPacketSent(int size);
  33
  34  private:
  35   void EncodeAudioPacket(scoped_ptr<AudioPacket> packet);
  36
  37   base::ThreadChecker thread_checker_;
  38
  39   base::WeakPtr<AudioPump> pump_;
  40
  41   scoped_refptr<base::SingleThreadTaskRunner> pump_task_runner_;
  42
  43   scoped_ptr<AudioCapturer> audio_capturer_;
  44   scoped_ptr<AudioEncoder> audio_encoder_;
  45
  46   bool enabled_;
  47
  48   // Number of bytes in the queue that have been encoded but haven't been sent
  49   // yet.
  50   int bytes_pending_;
  51
  52   DISALLOW_COPY_AND_ASSIGN(Core);
  53 };
  54
  55 AudioPump::Core::Core(base::WeakPtr<AudioPump> pump,
  56                       scoped_ptr<AudioCapturer> audio_capturer,
  57                       scoped_ptr<AudioEncoder> audio_encoder)
  58     : pump_(pump),
  59       pump_task_runner_(base::ThreadTaskRunnerHandle::Get()),
  60       audio_capturer_(audio_capturer.Pass()),
  61       audio_encoder_(audio_encoder.Pass()),
  62       enabled_(true),
  63       bytes_pending_(0) {
  64   thread_checker_.DetachFromThread();
  65 }
  66
  67 AudioPump::Core::~Core() {
  68   DCHECK(thread_checker_.CalledOnValidThread());
  69 }
  70
  71 void AudioPump::Core::Start() {
  72   DCHECK(thread_checker_.CalledOnValidThread());
  73
  74   audio_capturer_->Start(
  75       base::Bind(&Core::EncodeAudioPacket, base::Unretained(this)));
  76 }
  77
  78 void AudioPump::Core::Pause(bool pause) {
  79   DCHECK(thread_checker_.CalledOnValidThread());
  80
  81   enabled_ = !pause;
  82 }
  83
  84 void AudioPump::Core::OnPacketSent(int size) {
  85   DCHECK(thread_checker_.CalledOnValidThread());
  86
  87   bytes_pending_-= size;
  88   DCHECK_GE(bytes_pending_, 0);
  89 }
  90
  91 void AudioPump::Core::EncodeAudioPacket(scoped_ptr<AudioPacket> packet) {
  92   DCHECK(thread_checker_.CalledOnValidThread());
  93   DCHECK(packet);
  94
  95   int max_buffered_bytes =
  96       audio_encoder_->GetBitrate() * kMaxBufferedIntervalMs / 1000 / 8;
  97   if (!enabled_ || bytes_pending_ > max_buffered_bytes)
  98     return;
  99
 100   scoped_ptr<AudioPacket> encoded_packet =
 101       audio_encoder_->Encode(packet.Pass());
 102
 103   // The audio encoder returns a null audio packet if there's no audio to send.
 104   if (!encoded_packet)
 105     return;
 106
 107   int packet_size = encoded_packet->ByteSize();
 108   bytes_pending_ += packet_size;
 109
 110   pump_task_runner_->PostTask(
 111       FROM_HERE, base::Bind(&AudioPump::SendAudioPacket, pump_,
 112                             base::Passed(&encoded_packet), packet_size));
 113 }
 114
 115 AudioPump::AudioPump(
 116     scoped_refptr<base::SingleThreadTaskRunner> audio_task_runner,
 117     scoped_ptr<AudioCapturer> audio_capturer,
 118     scoped_ptr<AudioEncoder> audio_encoder,
 119     protocol::AudioStub* audio_stub)
 120     : audio_task_runner_(audio_task_runner),
 121       audio_stub_(audio_stub),
 122       weak_factory_(this) {
 123   DCHECK(audio_stub_);
 124
 125   core_.reset(new Core(weak_factory_.GetWeakPtr(), audio_capturer.Pass(),
 126                        audio_encoder.Pass()));
 127
 128   audio_task_runner_->PostTask(
 129       FROM_HERE, base::Bind(&Core::Start, base::Unretained(core_.get())));
 130 }
 131
 132 AudioPump::~AudioPump() {
 133   DCHECK(thread_checker_.CalledOnValidThread());
 134
 135   audio_task_runner_->DeleteSoon(FROM_HERE, core_.release());
 136 }
 137
 138 void AudioPump::Pause(bool pause) {
 139   DCHECK(thread_checker_.CalledOnValidThread());
 140
 141   audio_task_runner_->PostTask(
 142       FROM_HERE,
 143       base::Bind(&Core::Pause, base::Unretained(core_.get()), pause));
 144 }
 145
 146 void AudioPump::SendAudioPacket(scoped_ptr<AudioPacket> packet, int size) {
 147   DCHECK(thread_checker_.CalledOnValidThread());
 148   DCHECK(packet);
 149
 150   audio_stub_->ProcessAudioPacket(
 151       packet.Pass(),
 152       base::Bind(&AudioPump::OnPacketSent, weak_factory_.GetWeakPtr(), size));
 153 }
 154
 155 void AudioPump::OnPacketSent(int size) {
 156   audio_task_runner_->PostTask(
 157       FROM_HERE,
 158       base::Bind(&Core::OnPacketSent, base::Unretained(core_.get()), size));
 159 }
 160
 161 }  // namespace remoting