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

services/media/framework_mojo/mojo_consumer.h

+// 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.
+
+#ifndef SERVICES_MEDIA_FRAMEWORK_MOJO_MOJO_CONSUMER_H_
+#define SERVICES_MEDIA_FRAMEWORK_MOJO_MOJO_CONSUMER_H_
+
+
+#include "base/single_thread_task_runner.h"
+#include "base/task_runner.h"
+#include "mojo/common/binding_set.h"
+#include "mojo/services/media/common/cpp/mapped_shared_buffer.h"
+#include "mojo/services/media/common/interfaces/media_transport.mojom.h"
+#include "services/media/framework/models/active_source.h"
+
+namespace mojo {
+namespace media {
+
+// Implements MediaConsumer::Flush on behalf of MediaConsumer to avoid name
+// conflict with Part::Flush.
+class MojoConsumerMediaConsumer : public MediaConsumer {
+  // MediaConsumer implementation.
+  void Flush(const FlushCallback& callback) override;
+
+  // Implements MediaConsumer::Flush.
+  virtual void MediaConsumerFlush(const FlushCallback& callback) = 0;
+};
+
+// Implements MediaConsumer to receive a stream from across mojo.
+class MojoConsumer : public MojoConsumerMediaConsumer, public ActiveSource {
+ public:
+  using FlushRequestedCallback = std::function<void(const FlushCallback&)>;
+
+  static std::shared_ptr<MojoConsumer> Create() {
+    return std::shared_ptr<MojoConsumer>(new MojoConsumer());
+  }
+
+  ~MojoConsumer() override;
+
+  // Adds a binding.
+  void AddBinding(InterfaceRequest<MediaConsumer> consumer);
+
+  // Sets a callback signalling that a flush has been requested from the
+  // MediaConsumer client.
+  void SetFlushRequestedCallback(const FlushRequestedCallback& callback);
+
+  // MediaConsumer implementation.
+  void SetBuffer(
+      ScopedSharedBufferHandle buffer,
+      uint64_t size,
+      const SetBufferCallback& callback) override;
+
+  void PushPacket(MediaPacketPtr packet, const PushPacketCallback& callback)
+      override;
+
+  void MediaConsumerFlush(const FlushCallback& callback) override;
+
+  // ActiveSource implementation.
+  bool can_accept_allocator() const override;
+
+  void set_allocator(PayloadAllocator* allocator) override;
+
+  void SetSupplyCallback(const SupplyCallback& supply_callback) override;
+
+  void SetDownstreamDemand(Demand demand) override;
+
+ private:
+  MojoConsumer();
+
+  // Specialized packet implementation.
+  class PacketImpl : public Packet {
+   public:
+    static PacketPtr Create(
+        MediaPacketPtr media_packet,
+        const PushPacketCallback& callback,
+        scoped_refptr<base::SingleThreadTaskRunner> task_runner,
+        const MappedSharedBuffer& buffer) {
+      return PacketPtr(new PacketImpl(
+          media_packet.Pass(),
+          callback,
+          task_runner,
+          buffer));
+    }
+
+    // Packet implementation.
+    int64_t presentation_time() const override {
+      return media_packet_->pts;
+    }
+
+    uint64_t duration() const override {

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

overall, when you get the chance, I'd like to discuss the roll of duration in
the system.  My primary concern centers on expressing duration as a flat uint64
expressed in presentation time units.  It is possible that this is not a
accurate representation if the media units are coprime with the presentation
time units.

For example, if you have 44100 audio in a container which uses nanoseconds as
the presentation timeline, then you cannot express the duration of an integer
number of frames of audio in nSec units without the the chance of error.  If the
duration ends up being used to interpolate presentation times later on, this
error could accumulate and eventually cause trouble.

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

I think we can probably remove this. We no longer support non-interleaved LPCM,
so this value can be inferred from the payload size.

+      return media_packet_->duration;
+    }
+
+    bool end_of_stream() const override {
+      return media_packet_->end_of_stream;
+    }
+
+    size_t size() const override {
+      return media_packet_->payload->length;
+    }
+
+    void* payload() const override { return payload_; }
+
+   protected:
+    void Release() override;
+
+   private:
+    PacketImpl(
+        MediaPacketPtr media_packet,
+        const PushPacketCallback& callback,
+        scoped_refptr<base::SingleThreadTaskRunner> task_runner,
+        const MappedSharedBuffer& buffer);
+
+    ~PacketImpl() override;
+
+    static void RunCallback(const PushPacketCallback& callback);
+
+    MediaPacketPtr media_packet_;
+    const PushPacketCallback callback_;
+    scoped_refptr<base::SingleThreadTaskRunner> task_runner_;
+    void* payload_;
+  };
+
+  BindingSet<MediaConsumer> bindings_;
+  FlushRequestedCallback flush_requested_callback_;
+  scoped_refptr<base::SingleThreadTaskRunner> task_runner_;
+  MappedSharedBuffer buffer_;
+  SupplyCallback supply_callback_;
+};
+
+}  // namespace media
+}  // namespace mojo
+
+#endif  // SERVICES_MEDIA_FRAMEWORK_MOJO_MOJO_CONSUMER_H_