Motown: Framework parts for mojo transport (producer/consumer/mediapipe) and control (audiotrack).

services/media/framework_mojo/push_producer_base.cc

Index: services/media/framework_mojo/push_producer_base.cc
diff --git a/services/media/framework_mojo/push_producer_base.cc b/services/media/framework_mojo/push_producer_base.cc
new file mode 100644
index 0000000000000000000000000000000000000000..5cc60536f36679e37442ebbc1863fac2f5060e80
--- /dev/null
+++ b/services/media/framework_mojo/push_producer_base.cc
@@ -0,0 +1,151 @@
+// 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 "base/bind.h"
+#include "base/bind_helpers.h"
+#include "base/logging.h"
+#include "base/message_loop/message_loop.h"
+#include "services/media/framework_mojo/push_producer_base.h"
+
+namespace mojo {
+namespace media {
+
+PushProducerBase::PushProducerBase() :
+    state_(MediaState::UNPREPARED),
+    end_of_stream_(false),
+    max_pushes_outstanding_(0),
+    current_pushes_outstanding_(0) {
+  task_runner_ = base::MessageLoop::current()->task_runner();
+  DCHECK(task_runner_);
+}
+
+PushProducerBase::~PushProducerBase() {
+  base::AutoLock lock(lock_);
+}
+
+void PushProducerBase::SetStatusCallback(
+    const StatusCallback& callback) {
+  status_callback_ = callback;
+}
+
+PayloadAllocator* PushProducerBase::allocator() {
+  return &mojo_allocator_;
+}
+
+void PushProducerBase::SetDemandCallback(
+    const DemandCallback& demand_callback) {
+  demand_callback_ = demand_callback;
+}
+
+void PushProducerBase::Prime() {
+  {
+    base::AutoLock lock(lock_);
+    max_pushes_outstanding_ = 10; // TODO(dalesat): Made up!
+  }
+
+  DCHECK(demand_callback_);
+  demand_callback_(Demand::kPositive);
+  SetState(MediaState::PAUSED);
+}
+
+Demand PushProducerBase::SupplyPacket(PacketPtr packet) {
+  DCHECK(packet);
+
+  // If we're no longer connected, throw the packet away.
+  if (!IsConnected()) {
+    SetState(MediaState::UNPREPARED);
+    // TODO(dalesat): More shutdown?
+    return Demand::kNegative;
+  }
+
+  Demand demand;
+
+  {
+    base::AutoLock lock(lock_);
+    DCHECK(current_pushes_outstanding_ < max_pushes_outstanding_);
+    DCHECK(!end_of_stream_) << "packet pushed after end-of-stream";
+
+    ++current_pushes_outstanding_;
+
+    if (packet->end_of_stream()) {
+      end_of_stream_ = true;
+      demand = Demand::kNegative;
+      max_pushes_outstanding_ = 0;
+    } else {
+      demand = current_pushes_outstanding_ < max_pushes_outstanding_ ?
+          Demand::kPositive :
+          Demand::kNegative;
+    }
+  }
+
+  MediaPacketPtr media_packet = CreateMediaPacket(packet);
+  task_runner_->PostTask(FROM_HERE, base::Bind(
+      &PushProducerBase::PushPacket,
+      base::Unretained(this),
+      packet.release(),
+      base::Passed(media_packet.Pass())));
+
+  return demand;
+}
+
+void PushProducerBase::PushPacket(
+    Packet* packet_raw_ptr,
+    MediaPacketPtr media_packet) {

[johngro, 2016/02/23 00:49:42]

why do you need this level of indirection?  IOW - why isn't PushPacketInternal
just PushPacket which you bind to the task you send to your task runner?

[dalesat, 2016/02/23 20:34:34]

There was probably some work going on here previously. This is going away
shortly in any case. There will be only one producer.

+  PushPacketInternal(packet_raw_ptr, media_packet.Pass());
+}
+
+void PushProducerBase::PushCompleted(const PacketPtr& packet) {
+  DCHECK(packet);
+
+  Demand demand;
+
+  {
+    base::AutoLock lock(lock_);

[johngro, 2016/02/23 00:49:42]

DCHECK(current_pushes_outstanding_)

[dalesat, 2016/02/23 20:34:34]

Done.

+    demand = --current_pushes_outstanding_ < max_pushes_outstanding_ ?
+        Demand::kPositive :
+        Demand::kNegative;
+  }
+
+  DCHECK(demand_callback_);
+  demand_callback_(demand);
+
+  if (end_of_stream_ && packet->end_of_stream()) {
+    SetState(MediaState::ENDED);
+  }
+}
+
+void PushProducerBase::OnConnectionLost() {
+  DCHECK(demand_callback_);
+  demand_callback_(Demand::kNegative);
+  SetState(MediaState::UNPREPARED);
+}
+
+void PushProducerBase::SetState(MediaState state) {
+  if (state_ != state) {
+    state_ = state;
+    if (status_callback_) {
+      status_callback_(state_);
+    }
+  }
+}
+
+MediaPacketPtr PushProducerBase::CreateMediaPacket(
+    const PacketPtr& packet) {
+  DCHECK(packet);
+
+  MediaPacketRegionPtr region = MediaPacketRegion::New();
+  region->offset = mojo_allocator_.OffsetFromPtr(packet->payload());
+  region->length = packet->size();
+
+  MediaPacketPtr media_packet = MediaPacket::New();
+  media_packet->pts = packet->presentation_time();
+  media_packet->duration = packet->duration();
+  media_packet->end_of_stream = packet->end_of_stream();
+  media_packet->payload = region.Pass();
+
+  return media_packet.Pass();
+}
+
+} // namespace media
+} // namespace mojo