[Fuchsia][EME] Audio decryptor

Implement FuchsiaDecryptor (audio part) with StreamProcessor.

This CL also copies logic from FuchsiaVideoDecoder into
StreamProcessorHelper and SysmemBufferPool.

StreamProcessorHelper is a wrapper for StreamProcessor.

BufferPool is a wrapper of BufferCollection.

Caller is expected to create and own input/output BufferPool in its own
class and interact with StreamProcessorHelper by buffer index.

The new classes will be shared between FuchsiaVideoDecoder and
FuchsiaDecryptor in the future.

component successfully.

Bug: 966191
Test: Shaka player demo "Dig the Uke", log shows connecting to fuchsia
Change-Id: Ia0bd63e9ff39eaf4b6f477ac542fcbefe9c86e46
Reviewed-on: https://chromium-review.googlesource.com/c/chromium/src/+/1764179Reviewed-by: Xiaohan Wang <xhwang@chromium.org>
Reviewed-by: Sergey Ulanov <sergeyu@chromium.org>
Commit-Queue: Yuchen Liu <yucliu@chromium.org>
Cr-Commit-Position: refs/heads/master@{#694024}

media/filters/BUILD.gn

@@ -207,6 +207,14 @@ jumbo_source_set("filters") {
     sources += [
       "fuchsia/fuchsia_video_decoder.cc",
       "fuchsia/fuchsia_video_decoder.h",
+      "fuchsia/stream_processor_helper.cc",
+      "fuchsia/stream_processor_helper.h",
+      "fuchsia/sysmem_buffer_pool.cc",
+      "fuchsia/sysmem_buffer_pool.h",
+      "fuchsia/sysmem_buffer_reader.cc",
+      "fuchsia/sysmem_buffer_reader.h",
+      "fuchsia/sysmem_buffer_writer.cc",
+      "fuchsia/sysmem_buffer_writer.h",
     ]
     deps += [
       "//gpu/command_buffer/client",

media/filters/fuchsia/stream_processor_helper.cc

+// Copyright 2019 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/filters/fuchsia/stream_processor_helper.h"
+
+#include "base/bind.h"
+#include "base/fuchsia/fuchsia_logging.h"
+#include "media/base/bind_to_current_loop.h"
+
+namespace media {
+
+StreamProcessorHelper::IoPacket::IoPacket(size_t index,
+                                          size_t offset,
+                                          size_t size,
+                                          base::TimeDelta timestamp,
+                                          fuchsia::media::FormatDetails format,
+                                          base::OnceClosure destroy_cb)
+    : index_(index),
+      offset_(offset),
+      size_(size),
+      timestamp_(timestamp),
+      format_(std::move(format)),
+      destroy_cb_(std::move(destroy_cb)) {}
+
+StreamProcessorHelper::IoPacket::~IoPacket() = default;
+
+// static
+std::unique_ptr<StreamProcessorHelper::IoPacket>
+StreamProcessorHelper::IoPacket::CreateInput(
+    size_t index,
+    size_t size,
+    base::TimeDelta timestamp,
+    fuchsia::media::FormatDetails format,
+    base::OnceClosure destroy_cb) {
+  return std::make_unique<IoPacket>(index, 0 /* offset */, size, timestamp,
+                                    std::move(format), std::move(destroy_cb));
+}
+
+// static
+std::unique_ptr<StreamProcessorHelper::IoPacket>
+StreamProcessorHelper::IoPacket::CreateOutput(size_t index,
+                                              size_t offset,
+                                              size_t size,
+                                              base::TimeDelta timestamp,
+                                              base::OnceClosure destroy_cb) {
+  return std::make_unique<IoPacket>(index, offset, size, timestamp,
+                                    fuchsia::media::FormatDetails(),
+                                    std::move(destroy_cb));
+}
+
+StreamProcessorHelper::StreamProcessorHelper(
+    fuchsia::media::StreamProcessorPtr processor,
+    Client* client)
+    : processor_(std::move(processor)), client_(client), weak_factory_(this) {
+  DCHECK(processor_);
+  DCHECK(client_);
+  weak_this_ = weak_factory_.GetWeakPtr();
+
+  processor_.set_error_handler(
+      [this](zx_status_t status) {
+        ZX_LOG(ERROR, status)
+            << "The fuchsia.media.StreamProcessor channel was terminated.";
+        OnError();
+      });
+
+  processor_.events().OnStreamFailed =
+      fit::bind_member(this, &StreamProcessorHelper::OnStreamFailed);
+  processor_.events().OnInputConstraints =
+      fit::bind_member(this, &StreamProcessorHelper::OnInputConstraints);
+  processor_.events().OnFreeInputPacket =
+      fit::bind_member(this, &StreamProcessorHelper::OnFreeInputPacket);
+  processor_.events().OnOutputConstraints =
+      fit::bind_member(this, &StreamProcessorHelper::OnOutputConstraints);
+  processor_.events().OnOutputFormat =
+      fit::bind_member(this, &StreamProcessorHelper::OnOutputFormat);
+  processor_.events().OnOutputPacket =
+      fit::bind_member(this, &StreamProcessorHelper::OnOutputPacket);
+  processor_.events().OnOutputEndOfStream =
+      fit::bind_member(this, &StreamProcessorHelper::OnOutputEndOfStream);
+
+  processor_->EnableOnStreamFailed();
+}
+
+StreamProcessorHelper::~StreamProcessorHelper() = default;
+
+void StreamProcessorHelper::Process(std::unique_ptr<IoPacket> input) {
+  DCHECK(input);
+  DCHECK(processor_);
+
+  fuchsia::media::Packet packet;
+  packet.mutable_header()->set_buffer_lifetime_ordinal(
+      input_buffer_lifetime_ordinal_);
+  packet.mutable_header()->set_packet_index(input->index());
+  packet.set_buffer_index(packet.header().packet_index());
+  packet.set_timestamp_ish(input->timestamp().InNanoseconds());
+  packet.set_stream_lifetime_ordinal(stream_lifetime_ordinal_);
+  packet.set_start_offset(input->offset());
+  packet.set_valid_length_bytes(input->size());
+
+  active_stream_ = true;
+
+  if (!input->format().IsEmpty()) {
+    processor_->QueueInputFormatDetails(stream_lifetime_ordinal_,
+                                        fidl::Clone(input->format()));
+  }
+
+  DCHECK(input_packets_.find(input->index()) == input_packets_.end());
+  input_packets_[input->index()] = std::move(input);
+  processor_->QueueInputPacket(std::move(packet));
+}
+
+void StreamProcessorHelper::ProcessEos() {
+  DCHECK(processor_);
+
+  active_stream_ = true;
+  processor_->QueueInputEndOfStream(stream_lifetime_ordinal_);
+  processor_->FlushEndOfStreamAndCloseStream(stream_lifetime_ordinal_);
+}
+
+void StreamProcessorHelper::Reset() {
+  if (active_stream_) {
+    return;
+  }
+
+  processor_->CloseCurrentStream(stream_lifetime_ordinal_,
+                                 /*release_input_buffers=*/false,
+                                 /*release_output_buffers=*/false);
+  stream_lifetime_ordinal_ += 2;
+  active_stream_ = false;
+  input_packets_.clear();
+}
+
+void StreamProcessorHelper::OnStreamFailed(uint64_t stream_lifetime_ordinal,
+                                           fuchsia::media::StreamError error) {
+  if (stream_lifetime_ordinal_ != stream_lifetime_ordinal) {
+    return;
+  }
+
+  if (error == fuchsia::media::StreamError::DECRYPTOR_NO_KEY) {
+    client_->OnNoKey();
+    return;
+  }
+
+  OnError();
+}
+
+void StreamProcessorHelper::OnInputConstraints(
+    fuchsia::media::StreamBufferConstraints constraints) {
+  // Buffer lifetime ordinal is an odd number incremented by 2 for each buffer
+  // generation as required by StreamProcessor.
+  input_buffer_lifetime_ordinal_ += 2;
+
+  // Default settings are used in CompleteInputBuffersAllocation to finish
+  // StreamProcessor input buffers setup.
+  if (!constraints.has_default_settings() ||
+      !constraints.default_settings().has_packet_count_for_server() ||
+      !constraints.default_settings().has_packet_count_for_client()) {
+    DLOG(ERROR)
+        << "Received OnInputConstraints() with missing required fields.";
+    OnError();
+    return;
+  }
+
+  DCHECK(input_packets_.empty());
+  input_buffer_constraints_ = std::move(constraints);
+
+  client_->AllocateInputBuffers(input_buffer_constraints_);
+}
+
+void StreamProcessorHelper::OnFreeInputPacket(
+    fuchsia::media::PacketHeader free_input_packet) {
+  if (!free_input_packet.has_buffer_lifetime_ordinal() ||
+      !free_input_packet.has_packet_index()) {
+    DLOG(ERROR) << "Received OnFreeInputPacket() with missing required fields.";
+    OnError();
+    return;
+  }
+
+  if (free_input_packet.buffer_lifetime_ordinal() !=
+      input_buffer_lifetime_ordinal_) {
+    return;
+  }
+
+  auto it = input_packets_.find(free_input_packet.packet_index());
+  if (it == input_packets_.end()) {
+    DLOG(WARNING) << "Received OnFreeInputPacket() with invalid packet index.";
+    return;
+  }
+
+  input_packets_.erase(it);
+}
+
+void StreamProcessorHelper::OnOutputConstraints(
+    fuchsia::media::StreamOutputConstraints output_constraints) {
+  if (!output_constraints.has_stream_lifetime_ordinal()) {
+    DLOG(ERROR)
+        << "Received OnOutputConstraints() with missing required fields.";
+    OnError();
+    return;
+  }
+
+  if (output_constraints.stream_lifetime_ordinal() !=
+      stream_lifetime_ordinal_) {
+    return;
+  }
+
+  if (!output_constraints.has_buffer_constraints_action_required() ||
+      !output_constraints.buffer_constraints_action_required()) {
+    return;
+  }
+
+  if (!output_constraints.has_buffer_constraints()) {
+    DLOG(ERROR) << "Received OnOutputConstraints() which requires buffer "
+                   "constraints action, but without buffer constraints.";
+    OnError();
+    return;
+  }
+
+  // StreamProcessor API expects odd buffer lifetime ordinal, which is
+  // incremented by 2 for each buffer generation.
+  output_buffer_lifetime_ordinal_ += 2;
+
+  output_buffer_constraints_ =
+      std::move(*output_constraints.mutable_buffer_constraints());
+
+  client_->AllocateOutputBuffers(output_buffer_constraints_);
+}
+
+void StreamProcessorHelper::OnOutputFormat(
+    fuchsia::media::StreamOutputFormat output_format) {
+  if (!output_format.has_stream_lifetime_ordinal() ||
+      !output_format.has_format_details()) {
+    DLOG(ERROR) << "Received OnOutputFormat() with missing required fields.";
+    OnError();
+    return;
+  }
+
+  if (output_format.stream_lifetime_ordinal() != stream_lifetime_ordinal_) {
+    return;
+  }
+
+  client_->OnOutputFormat(std::move(output_format));
+}
+
+void StreamProcessorHelper::OnOutputPacket(fuchsia::media::Packet output_packet,
+                                           bool error_detected_before,
+                                           bool error_detected_during) {
+  if (!output_packet.has_header() ||
+      !output_packet.header().has_buffer_lifetime_ordinal() ||
+      !output_packet.header().has_packet_index() ||
+      !output_packet.has_buffer_index()) {
+    DLOG(ERROR) << "Received OnOutputPacket() with missing required fields.";
+    OnError();
+    return;
+  }
+
+  if (output_packet.header().buffer_lifetime_ordinal() !=
+      output_buffer_lifetime_ordinal_) {
+    return;
+  }
+
+  auto buffer_index = output_packet.buffer_index();
+  auto packet_index = output_packet.header().packet_index();
+  base::TimeDelta timestamp;
+  if (output_packet.has_timestamp_ish()) {
+    timestamp = base::TimeDelta::FromNanoseconds(output_packet.timestamp_ish());
+  }
+
+  client_->OnOutputPacket(IoPacket::CreateOutput(
+      buffer_index, output_packet.start_offset(),
+      output_packet.valid_length_bytes(), timestamp,
+      BindToCurrentLoop(base::BindOnce(
+          &StreamProcessorHelper::OnRecycleOutputBuffer, weak_this_,
+          output_buffer_lifetime_ordinal_, packet_index))));
+}
+
+void StreamProcessorHelper::OnOutputEndOfStream(
+    uint64_t stream_lifetime_ordinal,
+    bool error_detected_before) {
+  if (stream_lifetime_ordinal != stream_lifetime_ordinal_) {
+    return;
+  }
+
+  stream_lifetime_ordinal_ += 2;
+  active_stream_ = false;
+
+  client_->OnProcessEos();
+}
+
+void StreamProcessorHelper::OnError() {
+  processor_.Unbind();
+  client_->OnError();
+}
+
+void StreamProcessorHelper::CompleteInputBuffersAllocation(
+    fuchsia::sysmem::BufferCollectionTokenPtr sysmem_token) {
+  DCHECK(!input_buffer_constraints_.IsEmpty());
+  fuchsia::media::StreamBufferPartialSettings settings;
+  settings.set_buffer_lifetime_ordinal(input_buffer_lifetime_ordinal_);
+  settings.set_buffer_constraints_version_ordinal(
+      input_buffer_constraints_.buffer_constraints_version_ordinal());
+  settings.set_single_buffer_mode(false);
+  settings.set_packet_count_for_server(
+      input_buffer_constraints_.default_settings().packet_count_for_server());
+  settings.set_packet_count_for_client(
+      input_buffer_constraints_.default_settings().packet_count_for_client());
+  settings.set_sysmem_token(std::move(sysmem_token));
+  processor_->SetInputBufferPartialSettings(std::move(settings));
+}
+
+void StreamProcessorHelper::CompleteOutputBuffersAllocation(
+    size_t max_used_output_buffers,
+    fuchsia::sysmem::BufferCollectionTokenPtr collection_token) {
+  DCHECK(!output_buffer_constraints_.IsEmpty());
+  DCHECK_LE(max_used_output_buffers,
+            output_buffer_constraints_.packet_count_for_client_max());
+
+  // Pass new output buffer settings to the stream processor.
+  fuchsia::media::StreamBufferPartialSettings settings;
+  settings.set_buffer_lifetime_ordinal(output_buffer_lifetime_ordinal_);
+  settings.set_buffer_constraints_version_ordinal(
+      output_buffer_constraints_.buffer_constraints_version_ordinal());
+  settings.set_packet_count_for_client(max_used_output_buffers);
+  settings.set_packet_count_for_server(
+      output_buffer_constraints_.packet_count_for_server_recommended());
+  settings.set_sysmem_token(std::move(collection_token));
+  processor_->SetOutputBufferPartialSettings(std::move(settings));
+  processor_->CompleteOutputBufferPartialSettings(
+      output_buffer_lifetime_ordinal_);
+}
+
+void StreamProcessorHelper::OnRecycleOutputBuffer(
+    uint64_t buffer_lifetime_ordinal,
+    uint32_t packet_index) {
+  if (!processor_)
+    return;
+
+  if (buffer_lifetime_ordinal != output_buffer_lifetime_ordinal_)
+    return;
+
+  fuchsia::media::PacketHeader header;
+  header.set_buffer_lifetime_ordinal(buffer_lifetime_ordinal);
+  header.set_packet_index(packet_index);
+  processor_->RecycleOutputPacket(std::move(header));
+}
+
+}  // namespace media

media/filters/fuchsia/stream_processor_helper.h

+// Copyright 2019 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 MEDIA_FILTERS_FUCHSIA_STREAM_PROCESSOR_HELPER_H_
+#define MEDIA_FILTERS_FUCHSIA_STREAM_PROCESSOR_HELPER_H_
+
+#include <fuchsia/media/cpp/fidl.h>
+#include <fuchsia/sysmem/cpp/fidl.h>
+
+#include "base/callback.h"
+#include "base/callback_helpers.h"
+#include "base/containers/flat_map.h"
+#include "base/macros.h"
+#include "base/memory/weak_ptr.h"
+#include "base/optional.h"
+#include "base/time/time.h"
+
+namespace media {
+
+// Helper class of fuchsia::media::StreamProcessor. It's responsible for:
+// 1. Data validation check.
+// 2. Stream/Buffer life time management.
+// 3. Configure StreamProcessor and input/output buffer settings.
+class StreamProcessorHelper {
+ public:
+  class IoPacket {
+   public:
+    static std::unique_ptr<IoPacket> CreateInput(
+        size_t index,
+        size_t size,
+        base::TimeDelta timestamp,
+        fuchsia::media::FormatDetails format,
+        base::OnceClosure destroy_cb);
+
+    static std::unique_ptr<IoPacket> CreateOutput(size_t index,
+                                                  size_t offset,
+                                                  size_t size,
+                                                  base::TimeDelta timestamp,
+                                                  base::OnceClosure destroy_cb);
+
+    IoPacket(size_t index,
+             size_t offset,
+             size_t size,
+             base::TimeDelta timestamp,
+             fuchsia::media::FormatDetails format,
+             base::OnceClosure destroy_cb);
+    ~IoPacket();
+
+    size_t index() const { return index_; }
+
+    size_t offset() const { return offset_; }
+
+    size_t size() const { return size_; }
+
+    base::TimeDelta timestamp() const { return timestamp_; }
+
+    const fuchsia::media::FormatDetails& format() const { return format_; }
+
+   private:
+    size_t index_;
+    size_t offset_;
+    size_t size_;
+    base::TimeDelta timestamp_;
+    fuchsia::media::FormatDetails format_;
+    base::ScopedClosureRunner destroy_cb_;
+  };
+
+  class Client {
+   public:
+    // Allocate input/output buffers with the given constraints. Client should
+    // call ProvideInput/OutputBufferCollectionToken to finish the buffer
+    // allocation flow.
+    virtual void AllocateInputBuffers(
+        const fuchsia::media::StreamBufferConstraints& stream_constraints) = 0;
+    virtual void AllocateOutputBuffers(
+        const fuchsia::media::StreamBufferConstraints& stream_constraints) = 0;
+
+    // Called when all the pushed packets are processed.
+    virtual void OnProcessEos() = 0;
+
+    // Called when output format is available.
+    virtual void OnOutputFormat(fuchsia::media::StreamOutputFormat format) = 0;
+
+    // Called when output packet is available. Deleting |packet| will notify
+    // StreamProcessor the output buffer is available to be re-used.
+    virtual void OnOutputPacket(std::unique_ptr<IoPacket> packet) = 0;
+
+    // Only available for decryption, which indicates currently the
+    // StreamProcessor doesn't have the content key to process.
+    virtual void OnNoKey() = 0;
+
+    // Called when any fatal errors happens.
+    virtual void OnError() = 0;
+
+   protected:
+    virtual ~Client() = default;
+  };
+
+  StreamProcessorHelper(fuchsia::media::StreamProcessorPtr processor,
+                        Client* client);
+  ~StreamProcessorHelper();
+
+  // Process one packet. |packet| is owned by this class until the buffer
+  // represented by |packet| is released.
+  void Process(std::unique_ptr<IoPacket> packet);
+
+  // Push End-Of-Stream to StreamProcessor. No more data should be sent to
+  // StreamProcessor without calling Reset.
+  void ProcessEos();
+
+  // Provide input/output BufferCollectionToken to finish StreamProcessor buffer
+  // setup flow.
+  void CompleteInputBuffersAllocation(
+      fuchsia::sysmem::BufferCollectionTokenPtr token);
+  void CompleteOutputBuffersAllocation(
+      size_t max_used_output_buffers,
+      fuchsia::sysmem::BufferCollectionTokenPtr token);
+
+  void Reset();
+
+ private:
+  // Event handlers for |processor_|.
+  void OnStreamFailed(uint64_t stream_lifetime_ordinal,
+                      fuchsia::media::StreamError error);
+  void OnInputConstraints(
+      fuchsia::media::StreamBufferConstraints input_constraints);
+  void OnFreeInputPacket(fuchsia::media::PacketHeader free_input_packet);
+  void OnOutputConstraints(
+      fuchsia::media::StreamOutputConstraints output_constraints);
+  void OnOutputFormat(fuchsia::media::StreamOutputFormat output_format);
+  void OnOutputPacket(fuchsia::media::Packet output_packet,
+                      bool error_detected_before,
+                      bool error_detected_during);
+  void OnOutputEndOfStream(uint64_t stream_lifetime_ordinal,
+                           bool error_detected_before);
+
+  void OnError();
+
+  void OnRecycleOutputBuffer(uint64_t buffer_lifetime_ordinal,
+                             uint32_t packet_index);
+
+  uint64_t stream_lifetime_ordinal_ = 1;
+
+  // Set to true if we've sent an input packet with the current
+  // stream_lifetime_ordinal_.
+  bool active_stream_ = false;
+
+  // Input buffers.
+  uint64_t input_buffer_lifetime_ordinal_ = 1;
+  fuchsia::media::StreamBufferConstraints input_buffer_constraints_;
+
+  // Map from packet index to corresponding input IoPacket. IoPacket should be
+  // owned by this class until StreamProcessor released the buffer.
+  base::flat_map<size_t, std::unique_ptr<IoPacket>> input_packets_;
+
+  // Output buffers.
+  uint64_t output_buffer_lifetime_ordinal_ = 1;
+  fuchsia::media::StreamBufferConstraints output_buffer_constraints_;
+
+  fuchsia::media::StreamProcessorPtr processor_;
+  Client* const client_;
+
+  base::WeakPtr<StreamProcessorHelper> weak_this_;
+  base::WeakPtrFactory<StreamProcessorHelper> weak_factory_;
+
+  DISALLOW_COPY_AND_ASSIGN(StreamProcessorHelper);
+};
+
+}  // namespace media
+
+#endif  // MEDIA_FILTERS_FUCHSIA_STREAM_PROCESSOR_HELPER_H_

media/filters/fuchsia/sysmem_buffer_pool.cc

+// Copyright 2019 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/filters/fuchsia/sysmem_buffer_pool.h"
+
+#include <zircon/rights.h>
+#include <algorithm>
+
+#include "base/bind.h"
+#include "base/fuchsia/default_context.h"
+#include "base/fuchsia/fuchsia_logging.h"
+#include "media/filters/fuchsia/sysmem_buffer_reader.h"
+#include "media/filters/fuchsia/sysmem_buffer_writer.h"
+
+namespace media {
+namespace {
+
+template <typename T>
+using CreateAccessorCB = base::OnceCallback<void(std::unique_ptr<T>)>;
+
+template <typename T>
+void CreateAccessorInstance(
+    zx_status_t status,
+    fuchsia::sysmem::BufferCollectionInfo_2 buffer_collection_info,
+    CreateAccessorCB<T> create_cb) {
+  if (status != ZX_OK) {
+    ZX_LOG(ERROR, status) << "Fail to enable reader or writer";
+    std::move(create_cb).Run(nullptr);
+    return;
+  }
+
+  std::unique_ptr<T> accessor = T::Create(std::move(buffer_collection_info));
+  if (!accessor) {
+    std::move(create_cb).Run(nullptr);
+    return;
+  }
+
+  std::move(create_cb).Run(std::move(accessor));
+}
+
+template <typename T>
+void CreateAccessor(fuchsia::sysmem::BufferCollection* collection,
+                    CreateAccessorCB<T> create_cb) {
+  collection->WaitForBuffersAllocated(
+      [create_cb = std::move(create_cb)](zx_status_t status,
+                                         fuchsia::sysmem::BufferCollectionInfo_2
+                                             buffer_collection_info) mutable {
+        CreateAccessorInstance<T>(status, std::move(buffer_collection_info),
+                                  std::move(create_cb));
+      });
+}
+
+}  // namespace
+
+SysmemBufferPool::Creator::Creator(
+    fuchsia::sysmem::BufferCollectionPtr collection,
+    std::vector<fuchsia::sysmem::BufferCollectionTokenPtr> shared_tokens)
+    : collection_(std::move(collection)),
+      shared_tokens_(std::move(shared_tokens)) {
+  collection_.set_error_handler(
+      [this](zx_status_t status) {
+        ZX_DLOG(ERROR, status)
+            << "Connection to BufferCollection was disconnected.";
+        collection_.Unbind();
+
+        if (create_cb_)
+          std::move(create_cb_).Run(nullptr);
+      });
+}
+
+SysmemBufferPool::Creator::~Creator() = default;
+
+void SysmemBufferPool::Creator::Create(
+    fuchsia::sysmem::BufferCollectionConstraints constraints,
+    CreateCB create_cb) {
+  DCHECK(!create_cb_);
+  create_cb_ = std::move(create_cb);
+  // BufferCollection needs to be synchronized to ensure that all token
+  // duplicate requests have been processed and sysmem knows about all clients
+  // that will be using this buffer collection.
+  collection_->Sync([this, constraints = std::move(constraints)]() mutable {
+    collection_->SetConstraints(true /* has constraints */,
+                                std::move(constraints));
+
+    DCHECK(create_cb_);
+    std::move(create_cb_)
+        .Run(std::make_unique<SysmemBufferPool>(std::move(collection_),
+                                                std::move(shared_tokens_)));
+  });
+}
+
+SysmemBufferPool::SysmemBufferPool(
+    fuchsia::sysmem::BufferCollectionPtr collection,
+    std::vector<fuchsia::sysmem::BufferCollectionTokenPtr> shared_tokens)
+    : collection_(std::move(collection)),
+      shared_tokens_(std::move(shared_tokens)) {
+  collection_.set_error_handler(
+      [this](zx_status_t status) {
+        ZX_LOG(ERROR, status)
+            << "The fuchsia.sysmem.BufferCollection channel was disconnected.";
+        collection_.Unbind();
+        if (create_reader_cb_)
+          std::move(create_reader_cb_).Run(nullptr);
+        if (create_writer_cb_)
+          std::move(create_writer_cb_).Run(nullptr);
+      });
+}
+
+SysmemBufferPool::~SysmemBufferPool() = default;
+
+fuchsia::sysmem::BufferCollectionTokenPtr SysmemBufferPool::TakeToken() {
+  DCHECK(!shared_tokens_.empty());
+  auto token = std::move(shared_tokens_.back());
+  shared_tokens_.pop_back();
+  return token;
+}
+
+void SysmemBufferPool::CreateReader(CreateReaderCB create_cb) {
+  DCHECK(!create_reader_cb_);
+  create_reader_cb_ = std::move(create_cb);
+
+  CreateAccessor<SysmemBufferReader>(
+      collection_.get(), base::BindOnce(&SysmemBufferPool::OnReaderCreated,
+                                        base::Unretained(this)));
+}
+
+void SysmemBufferPool::OnReaderCreated(
+    std::unique_ptr<SysmemBufferReader> reader) {
+  DCHECK(create_reader_cb_);
+  std::move(create_reader_cb_).Run(std::move(reader));
+}
+
+void SysmemBufferPool::CreateWriter(CreateWriterCB create_cb) {
+  DCHECK(!create_writer_cb_);
+  create_writer_cb_ = std::move(create_cb);
+
+  CreateAccessor<SysmemBufferWriter>(
+      collection_.get(), base::BindOnce(&SysmemBufferPool::OnWriterCreated,
+                                        base::Unretained(this)));
+}
+
+void SysmemBufferPool::OnWriterCreated(
+    std::unique_ptr<SysmemBufferWriter> writer) {
+  DCHECK(create_writer_cb_);
+  std::move(create_writer_cb_).Run(std::move(writer));
+}
+
+BufferAllocator::BufferAllocator() {
+  allocator_ = base::fuchsia::ComponentContextForCurrentProcess()
+                   ->svc()
+                   ->Connect<fuchsia::sysmem::Allocator>();
+
+  allocator_.set_error_handler([](zx_status_t status) {
+    // Just log a warning. We will handle BufferCollection the failure when
+    // trying to create a new BufferCollection.
+    ZX_DLOG(WARNING, status)
+        << "The fuchsia.sysmem.Allocator channel was disconnected.";
+  });
+}
+
+BufferAllocator::~BufferAllocator() = default;
+
+std::unique_ptr<SysmemBufferPool::Creator>
+BufferAllocator::MakeBufferPoolCreator(size_t num_of_tokens) {
+  // Create a new sysmem buffer collection token for the allocated buffers.
+  fuchsia::sysmem::BufferCollectionTokenPtr collection_token;
+  allocator_->AllocateSharedCollection(collection_token.NewRequest());
+
+  // Create collection token for sharing with other components.
+  std::vector<fuchsia::sysmem::BufferCollectionTokenPtr> shared_tokens;
+  for (size_t i = 0; i < num_of_tokens; i++) {
+    fuchsia::sysmem::BufferCollectionTokenPtr token;
+    collection_token->Duplicate(ZX_RIGHT_SAME_RIGHTS, token.NewRequest());
+    shared_tokens.push_back(std::move(token));
+  }
+
+  fuchsia::sysmem::BufferCollectionPtr buffer_collection;
+
+  // Convert the token to a BufferCollection connection.
+  allocator_->BindSharedCollection(std::move(collection_token),
+                                   buffer_collection.NewRequest());
+
+  return std::make_unique<SysmemBufferPool::Creator>(
+      std::move(buffer_collection), std::move(shared_tokens));
+}
+
+}  // namespace media

media/filters/fuchsia/sysmem_buffer_pool.h

+// Copyright 2019 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 MEDIA_FILTERS_FUCHSIA_SYSMEM_BUFFER_POOL_H_
+#define MEDIA_FILTERS_FUCHSIA_SYSMEM_BUFFER_POOL_H_
+
+#include <fuchsia/media/cpp/fidl.h>
+#include <fuchsia/sysmem/cpp/fidl.h>
+#include <lib/sys/cpp/component_context.h>
+
+#include <list>
+#include <vector>
+
+#include "base/callback.h"
+#include "base/containers/span.h"
+#include "base/macros.h"
+
+namespace media {
+
+class SysmemBufferReader;
+class SysmemBufferWriter;
+
+// Pool of buffers allocated by sysmem. It owns BufferCollection. It doesn't
+// provide any function read/write the buffers. Call should use
+// ReadableBufferPool/WritableBufferPool for read/write.
+class SysmemBufferPool {
+ public:
+  using CreateReaderCB =
+      base::OnceCallback<void(std::unique_ptr<SysmemBufferReader>)>;
+  using CreateWriterCB =
+      base::OnceCallback<void(std::unique_ptr<SysmemBufferWriter>)>;
+
+  // Creates SysmemBufferPool asynchronously. It also owns the channel to
+  // fuchsia services.
+  class Creator {
+   public:
+    using CreateCB =
+        base::OnceCallback<void(std::unique_ptr<SysmemBufferPool>)>;
+    Creator(
+        fuchsia::sysmem::BufferCollectionPtr collection,
+        std::vector<fuchsia::sysmem::BufferCollectionTokenPtr> shared_tokens);
+    ~Creator();
+
+    void Create(fuchsia::sysmem::BufferCollectionConstraints constraints,
+                CreateCB build_cb);
+
+   private:
+    fuchsia::sysmem::BufferCollectionPtr collection_;
+    std::vector<fuchsia::sysmem::BufferCollectionTokenPtr> shared_tokens_;
+    CreateCB create_cb_;
+
+    DISALLOW_COPY_AND_ASSIGN(Creator);
+  };
+
+  SysmemBufferPool(
+      fuchsia::sysmem::BufferCollectionPtr collection,
+      std::vector<fuchsia::sysmem::BufferCollectionTokenPtr> shared_tokens);
+  ~SysmemBufferPool();
+
+  fuchsia::sysmem::BufferCollectionTokenPtr TakeToken();
+
+  // Create Reader/Writer to access raw memory. The returned Reader/Writer is
+  // owned by SysmemBufferPool and lives as long as SysmemBufferPool.
+  void CreateReader(CreateReaderCB create_cb);
+  void CreateWriter(CreateWriterCB create_cb);
+
+  // Returns if this object is still usable. Caller must check this before
+  // calling SysmemBufferReader/Writer APIs.
+  bool is_live() const { return collection_.is_bound(); }
+
+ private:
+  friend class BufferAllocator;
+
+  void OnReaderCreated(std::unique_ptr<SysmemBufferReader> reader);
+  void OnWriterCreated(std::unique_ptr<SysmemBufferWriter> reader);
+
+  fuchsia::sysmem::BufferCollectionPtr collection_;
+  std::vector<fuchsia::sysmem::BufferCollectionTokenPtr> shared_tokens_;
+
+  CreateReaderCB create_reader_cb_;
+  CreateWriterCB create_writer_cb_;
+
+  DISALLOW_COPY_AND_ASSIGN(SysmemBufferPool);
+};
+
+// Wrapper of sysmem Allocator.
+class BufferAllocator {
+ public:
+  BufferAllocator();
+  ~BufferAllocator();
+
+  std::unique_ptr<SysmemBufferPool::Creator> MakeBufferPoolCreator(
+      size_t num_shared_token);
+
+ private:
+  fuchsia::sysmem::AllocatorPtr allocator_;
+
+  DISALLOW_COPY_AND_ASSIGN(BufferAllocator);
+};
+
+}  // namespace media
+
+#endif  // MEDIA_FILTERS_FUCHSIA_SYSMEM_BUFFER_POOL_H_

media/filters/fuchsia/sysmem_buffer_reader.cc

+// Copyright 2019 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/filters/fuchsia/sysmem_buffer_reader.h"
+
+#include "base/fuchsia/fuchsia_logging.h"
+
+namespace media {
+
+SysmemBufferReader::SysmemBufferReader(
+    fuchsia::sysmem::BufferCollectionInfo_2 info)
+    : collection_info_(std::move(info)) {}
+
+SysmemBufferReader::~SysmemBufferReader() = default;
+
+bool SysmemBufferReader::Read(size_t index,
+                              size_t offset,
+                              base::span<uint8_t> data) {
+  DCHECK_LT(index, collection_info_.buffer_count);
+  const fuchsia::sysmem::BufferMemorySettings& settings =
+      collection_info_.settings.buffer_settings;
+  fuchsia::sysmem::VmoBuffer& buffer = collection_info_.buffers[index];
+  DCHECK_LE(buffer.vmo_usable_start + offset + data.size(),
+            settings.size_bytes);
+
+  size_t vmo_offset = buffer.vmo_usable_start + offset;
+
+  // Invalidate cache.
+  if (settings.coherency_domain == fuchsia::sysmem::CoherencyDomain::RAM) {
+    zx_status_t status = buffer.vmo.op_range(
+        ZX_VMO_OP_CACHE_CLEAN_INVALIDATE, vmo_offset, data.size(), nullptr, 0);
+    ZX_LOG_IF(ERROR, status != ZX_OK, status) << "Fail to invalidate cache";
+  }
+
+  zx_status_t status = buffer.vmo.read(data.data(), vmo_offset, data.size());
+  ZX_LOG_IF(ERROR, status != ZX_OK, status) << "Fail to read";
+  return status == ZX_OK;
+}
+
+// static
+std::unique_ptr<SysmemBufferReader> SysmemBufferReader::Create(
+    fuchsia::sysmem::BufferCollectionInfo_2 info) {
+  return std::make_unique<SysmemBufferReader>(std::move(info));
+}
+
+// static
+fuchsia::sysmem::BufferCollectionConstraints
+SysmemBufferReader::GetRecommendedConstraints(size_t max_used_output_frames) {
+  fuchsia::sysmem::BufferCollectionConstraints buffer_constraints;
+  buffer_constraints.usage.cpu = fuchsia::sysmem::cpuUsageRead;
+  buffer_constraints.min_buffer_count_for_camping = max_used_output_frames;
+  return buffer_constraints;
+}
+
+}  // namespace media

media/filters/fuchsia/sysmem_buffer_reader.h

+// Copyright 2019 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 MEDIA_FILTERS_FUCHSIA_SYSMEM_BUFFER_READER_H_
+#define MEDIA_FILTERS_FUCHSIA_SYSMEM_BUFFER_READER_H_
+
+#include <fuchsia/media/cpp/fidl.h>
+#include <fuchsia/sysmem/cpp/fidl.h>
+
+#include <memory>
+
+#include "base/containers/span.h"
+
+namespace media {
+
+// Helper class to read content from fuchsia::sysmem::BufferCollection.
+class SysmemBufferReader {
+ public:
+  static fuchsia::sysmem::BufferCollectionConstraints GetRecommendedConstraints(
+      size_t max_used_output_frames);
+
+  static std::unique_ptr<SysmemBufferReader> Create(
+      fuchsia::sysmem::BufferCollectionInfo_2 info);
+
+  explicit SysmemBufferReader(fuchsia::sysmem::BufferCollectionInfo_2 info);
+  ~SysmemBufferReader();
+
+  // Read the buffer content at |index| into |data|, starting from |offset|.
+  bool Read(size_t index, size_t offset, base::span<uint8_t> data);
+
+ private:
+  fuchsia::sysmem::BufferCollectionInfo_2 collection_info_;
+
+  DISALLOW_COPY_AND_ASSIGN(SysmemBufferReader);
+};
+
+}  // namespace media
+
+#endif  // MEDIA_FILTERS_FUCHSIA_SYSMEM_BUFFER_READER_H_

media/filters/fuchsia/sysmem_buffer_writer.cc

+// Copyright 2019 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/filters/fuchsia/sysmem_buffer_writer.h"
+
+#include <zircon/rights.h>
+#include <algorithm>
+
+#include "base/bits.h"
+#include "base/fuchsia/fuchsia_logging.h"
+#include "base/process/process_metrics.h"
+
+namespace media {
+
+class SysmemBufferWriter::Buffer {
+ public:
+  Buffer() = default;
+
+  ~Buffer() {
+    if (!base_address_) {
+      return;
+    }
+
+    size_t mapped_bytes =
+        base::bits::Align(offset_ + size_, base::GetPageSize());
+    zx_status_t status = zx::vmar::root_self()->unmap(
+        reinterpret_cast<uintptr_t>(base_address_), mapped_bytes);
+    ZX_DCHECK(status == ZX_OK, status) << "zx_vmar_unmap";
+  }
+
+  Buffer(Buffer&&) = default;
+  Buffer& operator=(Buffer&&) = default;
+
+  bool Initialize(zx::vmo vmo,
+                  size_t offset,
+                  size_t size,
+                  fuchsia::sysmem::CoherencyDomain coherency_domain) {
+    DCHECK(!base_address_);
+    DCHECK(vmo);
+
+    // zx_vmo_write() doesn't work for sysmem-allocated VMOs (see ZX-4854), so
+    // the VMOs have to be mapped.
+    size_t bytes_to_map = base::bits::Align(offset + size, base::GetPageSize());
+    uintptr_t addr;
+    zx_status_t status = zx::vmar::root_self()->map(
+        /*vmar_offset=*/0, vmo, /*vmo_offset=*/0, bytes_to_map,
+        ZX_VM_PERM_READ | ZX_VM_PERM_WRITE, &addr);
+    if (status != ZX_OK) {
+      ZX_DLOG(ERROR, status) << "zx_vmar_map";
+      return false;
+    }
+
+    base_address_ = reinterpret_cast<uint8_t*>(addr);
+    offset_ = offset;
+    size_ = size;
+    coherency_domain_ = coherency_domain;
+
+    return true;
+  }
+
+  bool is_used() const { return is_used_; }
+  size_t size() const { return size_; }
+
+  // Copies as much data as possible from |data| to this input buffer.
+  size_t Write(base::span<const uint8_t> data) {
+    DCHECK(!is_used_);
+    is_used_ = true;
+
+    size_t bytes_to_fill = std::min(size_, data.size());
+    memcpy(base_address_ + offset_, data.data(), bytes_to_fill);
+
+    // Flush CPU cache if StreamProcessor reads from RAM.
+    if (coherency_domain_ == fuchsia::sysmem::CoherencyDomain::RAM) {
+      zx_status_t status = zx_cache_flush(base_address_ + offset_,
+                                          bytes_to_fill, ZX_CACHE_FLUSH_DATA);
+      ZX_DCHECK(status == ZX_OK, status) << "zx_cache_flush";
+    }
+
+    return bytes_to_fill;
+  }
+
+  void Release() { is_used_ = false; }
+
+ private:
+  uint8_t* base_address_ = nullptr;
+
+  // Buffer settings provided by sysmem.
+  size_t offset_ = 0;
+  size_t size_ = 0;
+  fuchsia::sysmem::CoherencyDomain coherency_domain_;
+
+  // Set to true when this buffer is being used by the codec.
+  bool is_used_ = false;
+};
+
+SysmemBufferWriter::SysmemBufferWriter(std::vector<Buffer> buffers)
+    : buffers_(std::move(buffers)) {}
+
+SysmemBufferWriter::~SysmemBufferWriter() = default;
+
+size_t SysmemBufferWriter::Write(size_t index, base::span<const uint8_t> data) {
+  DCHECK_LT(index, buffers_.size());
+  DCHECK(!buffers_[index].is_used());
+
+  return buffers_[index].Write(data);
+}
+
+base::Optional<size_t> SysmemBufferWriter::Acquire() {
+  auto it = std::find_if(
+      buffers_.begin(), buffers_.end(),
+      [](const SysmemBufferWriter::Buffer& buf) { return !buf.is_used(); });
+
+  if (it == buffers_.end())
+    return base::nullopt;
+
+  return it - buffers_.begin();
+}
+
+void SysmemBufferWriter::Release(size_t index) {
+  DCHECK_LT(index, buffers_.size());
+  buffers_[index].Release();
+}
+
+// static
+std::unique_ptr<SysmemBufferWriter> SysmemBufferWriter::Create(
+    fuchsia::sysmem::BufferCollectionInfo_2 info) {
+  std::vector<SysmemBufferWriter::Buffer> buffers;
+  buffers.resize(info.buffer_count);
+
+  fuchsia::sysmem::BufferMemorySettings& settings =
+      info.settings.buffer_settings;
+  for (size_t i = 0; i < info.buffer_count; ++i) {
+    fuchsia::sysmem::VmoBuffer& buffer = info.buffers[i];
+    if (!buffers[i].Initialize(std::move(buffer.vmo), buffer.vmo_usable_start,
+                               settings.size_bytes,
+                               settings.coherency_domain)) {
+      return nullptr;
+    }
+  }
+
+  return std::make_unique<SysmemBufferWriter>(std::move(buffers));
+}
+
+// static
+base::Optional<fuchsia::sysmem::BufferCollectionConstraints>
+SysmemBufferWriter::GetRecommendedConstraints(
+    const fuchsia::media::StreamBufferConstraints& stream_constraints) {
+  fuchsia::sysmem::BufferCollectionConstraints buffer_constraints;
+
+  if (!stream_constraints.has_default_settings() ||
+      !stream_constraints.default_settings().has_packet_count_for_client()) {
+    DLOG(ERROR)
+        << "Received StreamBufferConstaints with missing required fields.";
+    return base::nullopt;
+  }
+
+  // Currently we have to map buffers VMOs to write to them (see ZX-4854) and
+  // memory cannot be mapped as write-only (see ZX-4872), so request RW access
+  // even though we will never need to read from these buffers.
+  buffer_constraints.usage.cpu =
+      fuchsia::sysmem::cpuUsageRead | fuchsia::sysmem::cpuUsageWrite;
+
+  buffer_constraints.min_buffer_count_for_camping =
+      stream_constraints.default_settings().packet_count_for_client();
+  buffer_constraints.has_buffer_memory_constraints = true;
+  buffer_constraints.buffer_memory_constraints.min_size_bytes =
+      stream_constraints.has_per_packet_buffer_bytes_recommended();
+  buffer_constraints.buffer_memory_constraints.ram_domain_supported = true;
+  buffer_constraints.buffer_memory_constraints.cpu_domain_supported = true;
+
+  return buffer_constraints;
+}
+
+}  // namespace media

media/filters/fuchsia/sysmem_buffer_writer.h

+// Copyright 2019 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 MEDIA_FILTERS_FUCHSIA_SYSMEM_BUFFER_WRITER_H_
+#define MEDIA_FILTERS_FUCHSIA_SYSMEM_BUFFER_WRITER_H_
+
+#include <fuchsia/media/cpp/fidl.h>
+#include <fuchsia/sysmem/cpp/fidl.h>
+
+#include <memory>
+
+#include "base/containers/span.h"
+#include "base/optional.h"
+
+namespace media {
+
+// Helper class to write content into fuchsia::sysmem::BufferCollection.
+class SysmemBufferWriter {
+ public:
+  class Buffer;
+
+  static base::Optional<fuchsia::sysmem::BufferCollectionConstraints>
+  GetRecommendedConstraints(
+      const fuchsia::media::StreamBufferConstraints& stream_constraints);
+
+  static std::unique_ptr<SysmemBufferWriter> Create(
+      fuchsia::sysmem::BufferCollectionInfo_2 info);
+
+  explicit SysmemBufferWriter(std::vector<Buffer> buffers);
+  ~SysmemBufferWriter();
+
+  // Write the content of |data| into buffer at |index|. Return num of bytes
+  // written into the buffer. Write a used buffer will fail. It will mark the
+  // buffer as "used".
+  size_t Write(size_t index, base::span<const uint8_t> data);
+
+  // Acquire unused buffer for write. If |min_size| is provided, the returned
+  // buffer will have available size larger than |min_size|. This will NOT
+  // mark the buffer as "used".
+  base::Optional<size_t> Acquire();
+
+  // Notify the pool buffer at |index| is free to write new data.
+  void Release(size_t index);
+
+ private:
+  std::vector<Buffer> buffers_;
+
+  DISALLOW_COPY_AND_ASSIGN(SysmemBufferWriter);
+};
+
+}  // namespace media
+
+#endif  // MEDIA_FILTERS_FUCHSIA_SYSMEM_BUFFER_WRITER_H_

media/fuchsia/cdm/BUILD.gn

@@ -10,6 +10,10 @@ source_set("cdm") {
     "fuchsia_cdm.h",
     "fuchsia_cdm_factory.cc",
     "fuchsia_cdm_factory.h",
+    "fuchsia_decryptor.cc",
+    "fuchsia_decryptor.h",
+    "stream_processor_decryptor.cc",
+    "stream_processor_decryptor.h",
   ]
 
   deps = [

media/fuchsia/cdm/fuchsia_cdm.cc

@@ -10,6 +10,7 @@
 #include "fuchsia/base/mem_buffer_util.h"
 #include "media/base/callback_registry.h"
 #include "media/base/cdm_promise.h"
+#include "media/fuchsia/cdm/fuchsia_decryptor.h"
 
 namespace media {
 
@@ -218,7 +219,9 @@ FuchsiaCdm::SessionCallbacks& FuchsiaCdm::SessionCallbacks::operator=(
 
 FuchsiaCdm::FuchsiaCdm(fuchsia::media::drm::ContentDecryptionModulePtr cdm,
                        SessionCallbacks callbacks)
-    : cdm_(std::move(cdm)), session_callbacks_(std::move(callbacks)) {
+    : cdm_(std::move(cdm)),
+      session_callbacks_(std::move(callbacks)),
+      decryptor_(new FuchsiaDecryptor(cdm_.get())) {
   DCHECK(cdm_);
   cdm_.set_error_handler([](zx_status_t status) {
     // Error will be handled in CdmSession::OnSessionError.
@@ -393,8 +396,8 @@ std::unique_ptr<CallbackRegistration> FuchsiaCdm::RegisterEventCB(
 }
 
 Decryptor* FuchsiaCdm::GetDecryptor() {
-  NOTIMPLEMENTED();
-  return nullptr;
+  DCHECK(decryptor_);
+  return decryptor_.get();
 }
 
 int FuchsiaCdm::GetCdmId() const {

media/fuchsia/cdm/fuchsia_cdm.h

@@ -15,6 +15,7 @@
 #include "media/base/content_decryption_module.h"
 
 namespace media {
+class FuchsiaDecryptor;
 
 class FuchsiaCdm : public ContentDecryptionModule, public CdmContext {
  public:
@@ -86,6 +87,8 @@ class FuchsiaCdm : public ContentDecryptionModule, public CdmContext {
   fuchsia::media::drm::ContentDecryptionModulePtr cdm_;
   SessionCallbacks session_callbacks_;
 
+  std::unique_ptr<FuchsiaDecryptor> decryptor_;
+
   DISALLOW_COPY_AND_ASSIGN(FuchsiaCdm);
 };
 

media/fuchsia/cdm/fuchsia_decryptor.cc

+// Copyright 2019 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/fuchsia/cdm/fuchsia_decryptor.h"
+
+#include "base/fuchsia/fuchsia_logging.h"
+#include "base/logging.h"
+#include "media/base/decoder_buffer.h"
+#include "media/base/video_frame.h"
+#include "media/fuchsia/cdm/stream_processor_decryptor.h"
+
+namespace media {
+
+FuchsiaDecryptor::FuchsiaDecryptor(
+    fuchsia::media::drm::ContentDecryptionModule* cdm)
+    : cdm_(cdm) {
+  DCHECK(cdm_);
+}
+
+FuchsiaDecryptor::~FuchsiaDecryptor() = default;
+
+void FuchsiaDecryptor::RegisterNewKeyCB(StreamType stream_type,
+                                        const NewKeyCB& key_added_cb) {
+  NOTIMPLEMENTED();
+}
+
+void FuchsiaDecryptor::Decrypt(StreamType stream_type,
+                               scoped_refptr<DecoderBuffer> encrypted,
+                               const DecryptCB& decrypt_cb) {
+  if (stream_type != StreamType::kAudio) {
+    NOTREACHED();
+    decrypt_cb.Run(Status::kError, nullptr);
+    return;
+  }
+
+  if (!audio_decryptor_)
+    audio_decryptor_ = StreamProcessorDecryptor::CreateAudioDecryptor(cdm_);
+
+  audio_decryptor_->Decrypt(std::move(encrypted), decrypt_cb);
+}
+
+void FuchsiaDecryptor::CancelDecrypt(StreamType stream_type) {
+  DCHECK_EQ(stream_type, StreamType::kAudio);
+  audio_decryptor_->CancelDecrypt();
+}
+
+void FuchsiaDecryptor::InitializeAudioDecoder(const AudioDecoderConfig& config,
+                                              const DecoderInitCB& init_cb) {
+  // Only supports decrypt for audio stream.
+  NOTREACHED();
+  init_cb.Run(false);
+}
+
+void FuchsiaDecryptor::InitializeVideoDecoder(const VideoDecoderConfig& config,
+                                              const DecoderInitCB& init_cb) {
+  NOTIMPLEMENTED();
+  init_cb.Run(false);
+}
+
+void FuchsiaDecryptor::DecryptAndDecodeAudio(
+    scoped_refptr<DecoderBuffer> encrypted,
+    const AudioDecodeCB& audio_decode_cb) {
+  // Only supports decrypt for audio stream.
+  NOTREACHED();
+  audio_decode_cb.Run(Status::kError, AudioFrames());
+}
+
+void FuchsiaDecryptor::DecryptAndDecodeVideo(
+    scoped_refptr<DecoderBuffer> encrypted,
+    const VideoDecodeCB& video_decode_cb) {
+  NOTIMPLEMENTED();
+  video_decode_cb.Run(Status::kError, nullptr);
+}
+
+void FuchsiaDecryptor::ResetDecoder(StreamType stream_type) {
+  DCHECK_EQ(stream_type, StreamType::kVideo);
+  NOTIMPLEMENTED();
+}
+
+void FuchsiaDecryptor::DeinitializeDecoder(StreamType stream_type) {
+  DCHECK_EQ(stream_type, StreamType::kVideo);
+  NOTIMPLEMENTED();
+}
+
+bool FuchsiaDecryptor::CanAlwaysDecrypt() {
+  return false;
+}
+
+}  // namespace media

media/fuchsia/cdm/fuchsia_decryptor.h

+// Copyright 2019 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 MEDIA_FUCHSIA_CDM_FUCHSIA_DECRYPTOR_H_
+#define MEDIA_FUCHSIA_CDM_FUCHSIA_DECRYPTOR_H_
+
+#include <memory>
+
+#include "media/base/decryptor.h"
+
+namespace fuchsia {
+namespace media {
+namespace drm {
+class ContentDecryptionModule;
+}  // namespace drm
+}  // namespace media
+}  // namespace fuchsia
+
+namespace media {
+class StreamProcessorDecryptor;
+
+class FuchsiaDecryptor : public Decryptor {
+ public:
+  // Caller should make sure |cdm| lives longer than this class.
+  explicit FuchsiaDecryptor(fuchsia::media::drm::ContentDecryptionModule* cdm);
+  ~FuchsiaDecryptor() override;
+
+  // media::Decryptor implementation:
+  void RegisterNewKeyCB(StreamType stream_type,
+                        const NewKeyCB& key_added_cb) override;
+  void Decrypt(StreamType stream_type,
+               scoped_refptr<DecoderBuffer> encrypted,
+               const DecryptCB& decrypt_cb) override;
+  void CancelDecrypt(StreamType stream_type) override;
+  void InitializeAudioDecoder(const AudioDecoderConfig& config,
+                              const DecoderInitCB& init_cb) override;
+  void InitializeVideoDecoder(const VideoDecoderConfig& config,
+                              const DecoderInitCB& init_cb) override;
+  void DecryptAndDecodeAudio(scoped_refptr<DecoderBuffer> encrypted,
+                             const AudioDecodeCB& audio_decode_cb) override;
+  void DecryptAndDecodeVideo(scoped_refptr<DecoderBuffer> encrypted,
+                             const VideoDecodeCB& video_decode_cb) override;
+  void ResetDecoder(StreamType stream_type) override;
+  void DeinitializeDecoder(StreamType stream_type) override;
+  bool CanAlwaysDecrypt() override;
+
+ private:
+  fuchsia::media::drm::ContentDecryptionModule* const cdm_;
+
+  std::unique_ptr<StreamProcessorDecryptor> audio_decryptor_;
+
+  DISALLOW_COPY_AND_ASSIGN(FuchsiaDecryptor);
+};
+
+}  // namespace media
+
+#endif  // MEDIA_FUCHSIA_CDM_FUCHSIA_DECRYPTOR_H_

media/fuchsia/cdm/stream_processor_decryptor.cc

+// Copyright 2019 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/fuchsia/cdm/stream_processor_decryptor.h"
+
+#include <fuchsia/media/cpp/fidl.h>
+#include <fuchsia/media/drm/cpp/fidl.h>
+
+#include "base/bind.h"
+#include "base/fuchsia/fuchsia_logging.h"
+#include "base/logging.h"
+#include "media/base/decoder_buffer.h"
+#include "media/base/decrypt_config.h"
+#include "media/base/encryption_pattern.h"
+#include "media/base/subsample_entry.h"
+#include "media/filters/fuchsia/sysmem_buffer_reader.h"
+#include "media/filters/fuchsia/sysmem_buffer_writer.h"
+
+namespace media {
+namespace {
+
+// Decryptor will copy decrypted data immediately once it's available. Client
+// just needs one buffer.
+const uint32_t kMaxUsedOutputFrames = 1;
+
+std::string GetEncryptionMode(EncryptionMode mode) {
+  switch (mode) {
+    case EncryptionMode::kCenc:
+      return fuchsia::media::drm::ENCRYPTION_MODE_CENC;
+    case EncryptionMode::kCbcs:
+      return fuchsia::media::drm::ENCRYPTION_MODE_CBCS;
+    default:
+      NOTREACHED() << "unknown encryption mode " << static_cast<int>(mode);
+      return "";
+  }
+}
+
+fuchsia::media::KeyId GetKeyId(const std::string& key_id) {
+  fuchsia::media::KeyId fuchsia_key_id;
+  DCHECK_EQ(key_id.size(), fuchsia_key_id.data.size());
+  std::copy(key_id.begin(), key_id.end(), fuchsia_key_id.data.begin());
+  return fuchsia_key_id;
+}
+
+std::vector<fuchsia::media::SubsampleEntry> GetSubsamples(
+    const std::vector<SubsampleEntry>& subsamples) {
+  std::vector<fuchsia::media::SubsampleEntry> fuchsia_subsamples(
+      subsamples.size());
+
+  for (size_t i = 0; i < subsamples.size(); i++) {
+    fuchsia_subsamples[i].clear_bytes = subsamples[i].clear_bytes;
+    fuchsia_subsamples[i].encrypted_bytes = subsamples[i].cypher_bytes;
+  }
+
+  return fuchsia_subsamples;
+}
+
+fuchsia::media::EncryptionPattern GetEncryptionPattern(
+    EncryptionPattern pattern) {
+  fuchsia::media::EncryptionPattern fuchsia_pattern;
+  fuchsia_pattern.clear_blocks = pattern.skip_byte_block();
+  fuchsia_pattern.encrypted_blocks = pattern.crypt_byte_block();
+  return fuchsia_pattern;
+}
+
+fuchsia::media::FormatDetails GetFormatDetails(const DecryptConfig* config) {
+  DCHECK(config);
+
+  fuchsia::media::EncryptedFormat encrypted_format;
+  encrypted_format.set_mode(GetEncryptionMode(config->encryption_mode()))
+      .set_key_id(GetKeyId(config->key_id()))
+      .set_init_vector(
+          std::vector<uint8_t>(config->iv().begin(), config->iv().end()))
+      .set_subsamples(GetSubsamples(config->subsamples()));
+  if (config->encryption_mode() == EncryptionMode::kCbcs) {
+    DCHECK(config->encryption_pattern().has_value());
+    encrypted_format.set_pattern(
+        GetEncryptionPattern(config->encryption_pattern().value()));
+  }
+
+  fuchsia::media::FormatDetails format;
+  format.set_format_details_version_ordinal(0);
+  format.mutable_domain()->crypto().set_encrypted(std::move(encrypted_format));
+  return format;
+}
+
+}  // namespace
+
+StreamProcessorDecryptor::StreamProcessorDecryptor(
+    fuchsia::media::StreamProcessorPtr processor)
+    : processor_(std::move(processor), this) {}
+
+StreamProcessorDecryptor::~StreamProcessorDecryptor() = default;
+
+void StreamProcessorDecryptor::Decrypt(scoped_refptr<DecoderBuffer> encrypted,
+                                       Decryptor::DecryptCB decrypt_cb) {
+  DCHECK(!encrypted->end_of_stream()) << "EOS frame is always clear.";
+  DCHECK(!decrypt_cb_);
+  DCHECK(!pending_encrypted_buffer_);
+
+  decrypt_cb_ = std::move(decrypt_cb);
+
+  // Input buffer writer is not available. Wait.
+  if (!input_writer_) {
+    pending_encrypted_buffer_ = std::move(encrypted);
+    return;
+  }
+
+  if (!input_pool_->is_live()) {
+    DLOG(ERROR) << "Input buffer pool is dead.";
+    std::move(decrypt_cb_).Run(Decryptor::kError, nullptr);
+    return;
+  }
+
+  // Decryptor can only process one buffer at a time, which means there
+  // should be always enough unused buffers.
+  base::Optional<size_t> buf_index = input_writer_->Acquire();
+
+  // No available input buffer. Just wait for the next available one.
+  if (!buf_index.has_value()) {
+    pending_encrypted_buffer_ = std::move(encrypted);
+    return;
+  }
+
+  size_t index = buf_index.value();
+
+  size_t bytes = input_writer_->Write(
+      index, base::make_span(encrypted->data(), encrypted->data_size()));
+  if (bytes < encrypted->data_size()) {
+    // The encrypted data size is too big. Decryptor should consider
+    // splitting the buffer and update the IV and subsample entries.
+    // TODO(yucliu): Handle large encrypted buffer correctly. For now, just
+    // reject the decryption.
+    DLOG(ERROR) << "Encrypted data size is too big.";
+    std::move(decrypt_cb_).Run(Decryptor::kError, nullptr);
+    return;
+  }
+
+  const DecryptConfig* decrypt_config = encrypted->decrypt_config();
+  DCHECK(decrypt_config);
+
+  auto input_packet = StreamProcessorHelper::IoPacket::CreateInput(
+      index, encrypted->data_size(), encrypted->timestamp(),
+      GetFormatDetails(decrypt_config),
+      base::BindOnce(&StreamProcessorDecryptor::OnInputPacketReleased,
+                     base::Unretained(this), index));
+
+  processor_.Process(std::move(input_packet));
+}
+
+void StreamProcessorDecryptor::CancelDecrypt() {
+  // Close current stream and drop all the cached decoder buffers.
+  // Keep input/output_pool_ to avoid buffer re-allocation.
+  processor_.Reset();
+  pending_encrypted_buffer_ = nullptr;
+
+  // Fire |decrypt_cb_| immediately as required by Decryptor::CancelDecrypt.
+  if (decrypt_cb_)
+    std::move(decrypt_cb_).Run(Decryptor::kSuccess, nullptr);
+}
+
+// StreamProcessorHelper::Client implementation:
+void StreamProcessorDecryptor::AllocateInputBuffers(
+    const fuchsia::media::StreamBufferConstraints& stream_constraints) {
+  base::Optional<fuchsia::sysmem::BufferCollectionConstraints>
+      buffer_constraints =
+          SysmemBufferWriter::GetRecommendedConstraints(stream_constraints);
+
+  if (!buffer_constraints.has_value()) {
+    OnError();
+    return;
+  }
+
+  input_pool_creator_ =
+      allocator_.MakeBufferPoolCreator(1 /* num_shared_token */);
+
+  input_pool_creator_->Create(
+      std::move(buffer_constraints).value(),
+      base::BindOnce(&StreamProcessorDecryptor::OnInputBufferPoolAvailable,
+                     base::Unretained(this)));
+}
+
+void StreamProcessorDecryptor::AllocateOutputBuffers(
+    const fuchsia::media::StreamBufferConstraints& stream_constraints) {
+  if (!stream_constraints.has_packet_count_for_client_max() ||
+      !stream_constraints.has_packet_count_for_client_min()) {
+    DLOG(ERROR) << "StreamBufferConstraints doesn't contain required fields.";
+    OnError();
+    return;
+  }
+
+  size_t max_used_output_buffers = std::min(
+      kMaxUsedOutputFrames, stream_constraints.packet_count_for_client_max());
+  max_used_output_buffers = std::max(
+      max_used_output_buffers,
+      static_cast<size_t>(stream_constraints.packet_count_for_client_min()));
+
+  output_pool_creator_ =
+      allocator_.MakeBufferPoolCreator(1 /* num_shared_token */);
+
+  output_pool_creator_->Create(
+      SysmemBufferReader::GetRecommendedConstraints(max_used_output_buffers),
+      base::BindOnce(&StreamProcessorDecryptor::OnOutputBufferPoolAvailable,
+                     base::Unretained(this), max_used_output_buffers));
+}
+
+void StreamProcessorDecryptor::OnProcessEos() {
+  // Decryptor never pushes EOS frame.
+  NOTREACHED();
+}
+
+void StreamProcessorDecryptor::OnOutputFormat(
+    fuchsia::media::StreamOutputFormat format) {}
+
+void StreamProcessorDecryptor::OnOutputPacket(
+    std::unique_ptr<StreamProcessorHelper::IoPacket> packet) {
+  DCHECK(output_reader_);
+  if (!output_pool_->is_live()) {
+    DLOG(ERROR) << "Output buffer pool is dead.";
+    return;
+  }
+
+  auto clear_buffer = base::MakeRefCounted<DecoderBuffer>(packet->size());
+  clear_buffer->set_timestamp(packet->timestamp());
+
+  bool read_success =
+      output_reader_->Read(packet->index(), packet->offset(),
+                           base::make_span(clear_buffer->writable_data(),
+                                           clear_buffer->data_size()));
+
+  if (!read_success) {
+    DLOG(ERROR) << "Fail to get decrypted result.";
+    std::move(decrypt_cb_).Run(Decryptor::kError, nullptr);
+    return;
+  }
+
+  std::move(decrypt_cb_).Run(Decryptor::kSuccess, std::move(clear_buffer));
+}
+
+void StreamProcessorDecryptor::OnNoKey() {
+  if (decrypt_cb_)
+    std::move(decrypt_cb_).Run(Decryptor::kNoKey, nullptr);
+}
+
+void StreamProcessorDecryptor::OnError() {
+  pending_encrypted_buffer_ = nullptr;
+  if (decrypt_cb_)
+    std::move(decrypt_cb_).Run(Decryptor::kError, nullptr);
+
+  processor_.Reset();
+}
+
+void StreamProcessorDecryptor::OnInputPacketReleased(size_t index) {
+  input_writer_->Release(index);
+
+  if (!pending_encrypted_buffer_)
+    return;
+
+  DCHECK(decrypt_cb_);
+
+  // If there're pending decryption request, handle it now since we have
+  // available input buffers.
+  Decrypt(std::move(pending_encrypted_buffer_), std::move(decrypt_cb_));
+}
+
+void StreamProcessorDecryptor::OnInputBufferPoolAvailable(
+    std::unique_ptr<SysmemBufferPool> pool) {
+  if (!pool) {
+    DLOG(ERROR) << "Fail to allocate input buffer.";
+    OnError();
+    return;
+  }
+
+  input_pool_ = std::move(pool);
+
+  // Provide token before enabling writer. Tokens must be provided to
+  // StreamProcessor before getting the allocated buffers.
+  processor_.CompleteInputBuffersAllocation(input_pool_->TakeToken());
+
+  input_pool_->CreateWriter(
+      base::BindOnce(&StreamProcessorDecryptor::OnInputBufferPoolWriter,
+                     base::Unretained(this)));
+}
+
+void StreamProcessorDecryptor::OnInputBufferPoolWriter(
+    std::unique_ptr<SysmemBufferWriter> writer) {
+  if (!writer) {
+    LOG(ERROR) << "Fail to enable input buffer writer";
+    OnError();
+    return;
+  }
+
+  DCHECK(!input_writer_);
+  input_writer_ = std::move(writer);
+
+  if (pending_encrypted_buffer_) {
+    DCHECK(decrypt_cb_);
+    Decrypt(std::move(pending_encrypted_buffer_), std::move(decrypt_cb_));
+  }
+}
+
+void StreamProcessorDecryptor::OnOutputBufferPoolAvailable(
+    size_t max_used_output_buffers,
+    std::unique_ptr<SysmemBufferPool> pool) {
+  if (!pool) {
+    LOG(ERROR) << "Fail to allocate output buffer.";
+    OnError();
+    return;
+  }
+
+  output_pool_ = std::move(pool);
+
+  // Provide token before enabling reader. Tokens must be provided to
+  // StreamProcessor before getting the allocated buffers.
+  processor_.CompleteOutputBuffersAllocation(max_used_output_buffers,
+                                             output_pool_->TakeToken());
+
+  output_pool_->CreateReader(
+      base::BindOnce(&StreamProcessorDecryptor::OnOutputBufferPoolReader,
+                     base::Unretained(this)));
+}
+
+void StreamProcessorDecryptor::OnOutputBufferPoolReader(
+    std::unique_ptr<SysmemBufferReader> reader) {
+  if (!reader) {
+    LOG(ERROR) << "Fail to enable output buffer reader.";
+    OnError();
+    return;
+  }
+
+  DCHECK(!output_reader_);
+  output_reader_ = std::move(reader);
+}
+
+std::unique_ptr<StreamProcessorDecryptor>
+StreamProcessorDecryptor::CreateAudioDecryptor(
+    fuchsia::media::drm::ContentDecryptionModule* cdm) {
+  DCHECK(cdm);
+
+  fuchsia::media::drm::DecryptorParams params;
+  params.set_require_secure_mode(false);
+  params.mutable_input_details()->set_format_details_version_ordinal(0);
+  fuchsia::media::StreamProcessorPtr stream_processor;
+  cdm->CreateDecryptor(std::move(params), stream_processor.NewRequest());
+
+  return std::make_unique<StreamProcessorDecryptor>(
+      std::move(stream_processor));
+}
+
+}  // namespace media

media/fuchsia/cdm/stream_processor_decryptor.h

+// Copyright 2019 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 MEDIA_FUCHSIA_CDM_STREAM_PROCESSOR_DECRYPTOR_H_
+#define MEDIA_FUCHSIA_CDM_STREAM_PROCESSOR_DECRYPTOR_H_
+
+#include <fuchsia/media/drm/cpp/fidl.h>
+
+#include <memory>
+
+#include "media/base/decryptor.h"
+#include "media/filters/fuchsia/stream_processor_helper.h"
+#include "media/filters/fuchsia/sysmem_buffer_pool.h"
+
+namespace media {
+class SysmemBufferReader;
+class SysmemBufferWriter;
+
+// Class to handle decryption for one stream. All the APIs have the same
+// requirement as Decryptor.
+class StreamProcessorDecryptor : public StreamProcessorHelper::Client {
+ public:
+  static std::unique_ptr<StreamProcessorDecryptor> CreateAudioDecryptor(
+      fuchsia::media::drm::ContentDecryptionModule* cdm);
+
+  explicit StreamProcessorDecryptor(
+      fuchsia::media::StreamProcessorPtr processor);
+  ~StreamProcessorDecryptor() override;
+
+  void Decrypt(scoped_refptr<DecoderBuffer> encrypted,
+               Decryptor::DecryptCB decrypt_cb);
+  void CancelDecrypt();
+
+  // StreamProcessorHelper::Client implementation:
+  void AllocateInputBuffers(const fuchsia::media::StreamBufferConstraints&
+                                stream_constraints) override;
+  void AllocateOutputBuffers(const fuchsia::media::StreamBufferConstraints&
+                                 stream_constraints) override;
+  void OnProcessEos() override;
+  void OnOutputFormat(fuchsia::media::StreamOutputFormat format) override;
+  void OnOutputPacket(
+      std::unique_ptr<StreamProcessorHelper::IoPacket> packet) override;
+  void OnNoKey() override;
+  void OnError() override;
+
+ private:
+  void OnInputPacketReleased(size_t index);
+  void OnInputBufferPoolAvailable(std::unique_ptr<SysmemBufferPool> pool);
+  void OnInputBufferPoolWriter(std::unique_ptr<SysmemBufferWriter> writer);
+  void OnOutputBufferPoolAvailable(size_t max_used_output_buffers,
+                                   std::unique_ptr<SysmemBufferPool> pool);
+  void OnOutputBufferPoolReader(std::unique_ptr<SysmemBufferReader> reader);
+
+  StreamProcessorHelper processor_;
+  BufferAllocator allocator_;
+
+  // Pending buffers due to input buffer pool not available.
+  scoped_refptr<DecoderBuffer> pending_encrypted_buffer_;
+
+  Decryptor::DecryptCB decrypt_cb_;
+
+  std::unique_ptr<SysmemBufferPool::Creator> input_pool_creator_;
+  std::unique_ptr<SysmemBufferPool> input_pool_;
+  std::unique_ptr<SysmemBufferWriter> input_writer_;
+
+  std::unique_ptr<SysmemBufferPool::Creator> output_pool_creator_;
+  std::unique_ptr<SysmemBufferPool> output_pool_;
+  std::unique_ptr<SysmemBufferReader> output_reader_;
+
+  scoped_refptr<DecoderBuffer> clear_buffer_;
+
+  DISALLOW_COPY_AND_ASSIGN(StreamProcessorDecryptor);
+};
+
+}  // namespace media
+
+#endif  // MEDIA_FUCHSIA_CDM_STREAM_PROCESSOR_DECRYPTOR_H_