text: // Copyright 2018 The Fuchsia 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 "lib/media/test/codec_client.h"
  
  #include <lib/async/cpp/task.h>
  #include <lib/media/test/one_shot_event.h>
  
  #include <algorithm>
  #include <random>
  
  #include "src/lib/syslog/cpp/logger.h"
  
  namespace {
  
  // The client would like there to be at least this many input packets.  Using 1
  // or 2 here can help avoid short stalls while packets are in transit even if
  // the client doesn't actually need to hold any free input packets for any
  // significant duration for any client-specific reason.
  constexpr uint32_t kMinInputPacketsForClient = 1;
  
  // The client would like there to be at least this many output packets.
  //
  // The client _must_ use a non-zero value here if any output packets will be
  // held indefinitely (such as held until the next output packet is available),
  // since otherwise the Codec can become stuck, unable to continue processing due
  // to all output frames used as active reference frames and no free output
  // buffer to decode into.
  constexpr uint32_t kMinOutputPacketsForClient = 1;
  
  // For input, this example doesn't re-configure input buffers, so there's only
  // one buffer_lifetime_ordinal.
  constexpr uint64_t kInputBufferLifetimeOrdinal = 1;
  
  // It's fine to increase this threshold if we add a new usage of CodecClient
  // with new StreamProcessor server that should/must have more buffers.  This is
  // here to check that we're not allocating more output buffers than expected.
  // If the various cases get further apart, it'd probably be worthwhile to plumb
  // per-case from code that's using CodecClient.  For now this is based on what
  // use_h264_decoder_test allocates (max across astro and QEMU).
  //
  // This is basically 16 max DPB for h264, 1 to decode into (assumed separate
  // from DPB for now), and 1 for the client.
  constexpr uint32_t kMaxExpectedBufferCount = 18;
  
  }  // namespace
  
  CodecClient::CodecClient(async::Loop* loop, thrd_t loop_thread,
                           fidl::InterfaceHandle<fuchsia::sysmem::Allocator> sysmem)
      : loop_(loop), dispatcher_(loop_->dispatcher()), loop_thread_(loop_thread) {
    // We haven't created a channel yet, but that's fine, and we want the error
    // handler set up before any error can possibly be generated by the channel so
    // there's no chance of missing an error.  The async::Loop that we'll use is
    // already running separately from the current thread.
    codec_.set_error_handler([this](zx_status_t status) {
      // Obviously a non-example client that continues to have a purpose even if
      // one of its codecs dies would want to handle errors in a more contained
      // way.
      //
      // TODO(dustingreen): get and print epitaph once that's possible.
      if (!in_lax_mode_) {
        FX_PLOGS(FATAL, status) << "codec_ failed - exiting";
      } else {
        FX_PLOGS(WARNING, status) << "codec_ failed - exiting";
        connection_lost_ = true;
        output_pending_condition_.notify_all();
        is_sync_complete_condition_.notify_all();
        input_free_packet_list_not_empty_.notify_all();
      }
    });
  
    // Only one request is ever created, so we create it in the constructor to
    // avoid needing any manual enforcement that we only do this once.
    temp_codec_request_ = codec_.NewRequest(loop_->dispatcher());
  
    // We treat event setup as much as possible like a hidden part of creating the
    // CodecPtr.  If NewBinding() has !is_valid(), we rely on the Codec server to
    // close the Codec channel async.
    codec_.events().OnStreamFailed = fit::bind_member(this, &CodecClient::OnStreamFailed);
    codec_.events().OnInputConstraints = fit::bind_member(this, &CodecClient::OnInputConstraints);
    codec_.events().OnFreeInputPacket = fit::bind_member(this, &CodecClient::OnFreeInputPacket);
    codec_.events().OnOutputConstraints = fit::bind_member(this, &CodecClient::OnOutputConstraints);
    codec_.events().OnOutputFormat = fit::bind_member(this, &CodecClient::OnOutputFormat);
    codec_.events().OnOutputPacket = fit::bind_member(this, &CodecClient::OnOutputPacket);
    codec_.events().OnOutputEndOfStream = fit::bind_member(this, &CodecClient::OnOutputEndOfStream);
  
    // Bind sysmem_ using FIDL thread.  This is ok because all communication with
    // sysmem also happens via FIDL thread so will queue after this posted lambda.
    PostToFidlThread([this, sysmem = std::move(sysmem)]() mutable {
      zx_status_t bind_status = sysmem_.Bind(std::move(sysmem), dispatcher_);
      ZX_ASSERT(bind_status == ZX_OK);
    });
  }
  
  CodecClient::~CodecClient() { Stop(); }
  
  fidl::InterfaceRequest<fuchsia::media::StreamProcessor> CodecClient::GetTheRequestOnce() {
    ZX_DEBUG_ASSERT(!is_start_called_);
    return std::move(temp_codec_request_);
  }
  
  // Can optionally be called before Start(), to set the min buffer size that'll
  // be requested via sysmem.
  void CodecClient::SetMinOutputBufferSize(uint64_t min_output_buffer_size) {
    ZX_DEBUG_ASSERT(!is_start_called_);
    min_output_buffer_size_ = min_output_buffer_size;
  }
  
  void CodecClient::SetMinOutputBufferCount(uint32_t min_output_buffer_count) {
    ZX_DEBUG_ASSERT(!is_start_called_);
    min_output_buffer_count_ = min_output_buffer_count;
  }
  
  void CodecClient::Start() {
    ZX_DEBUG_ASSERT(!is_start_called_);
    is_start_called_ = true;
    // The caller is responsible for calling this method only once, using the main
    // thread.  This method only holds the lock for short periods, and has to
    // release the lock many times during this method, which is reasonable given
    // the nature of this method as an overall state progression sequencer.
  
    // Call Sync() and wait for it's response _only_ to force the Codec server to
    // reach the point of being able response to messages, just for easier
    // debugging if just starting the Codec server fails instead.  Actual clients
    // don't need to use Sync() here.
    CallSyncAndWaitForResponse();
    FX_VLOGS(3) << "Sync() completed, which means the Codec server exists.";
    if (connection_lost_)
      return;
  
    FX_VLOGS(3) << "Waiting for OnInputConstraints() from the Codec server...";
    // The Codec client can rely on an OnInputConstraints() arriving shortly,
    // without any message required from the client first.  The
    // OnInputConstraints() may in future actually be sent by the CodecFactory,
    // but it'll still be sent to the client on the Codec channel in any case.
    {  // scope lock
      std::unique_lock<std::mutex> lock(lock_);
      // In this example we're not paying attention to channel failure here
      // because channel failure calls exit().
      while (!input_constraints_) {
        input_constraints_exist_condition_.wait(lock);
      }
    }  // ~lock
    ZX_ASSERT(input_constraints_);
    FX_VLOGS(3) << "Got OnInputConstraints() from the Codec server.";
  
    // We know input_constraints_ won't change outside the lock because we prevent
    // that in OnInputConstraints() by only accepting input constraints if there
    // aren't already input constraints.
  
    // Now that we have input constraints, we can create all the input buffers and
    // tell the Codec server about them.  We tell the Codec server by using
    // SetInputSettings() followed by one or num_buffers calls to
    // AddInputBuffer().  These are necessary before it becomes permissible to
    // call CreateStream().
    //
    // We're not on the FIDL thread, so we need to async::PostTask() over to the
    // FIDL thread to send any FIDL message.
  
    FX_CHECK(input_constraints_->has_packet_count_for_server_min());
    FX_CHECK(input_constraints_->has_packet_count_for_server_recommended());
    FX_CHECK(input_constraints_->has_packet_count_for_server_max());
    uint32_t packet_count_for_client =
        std::max(kMinInputPacketsForClient, input_constraints_->packet_count_for_client_min());
    // Use min to make sure decode can proceed without getting stuck while using min.
    uint32_t packet_count_for_server = input_constraints_->packet_count_for_server_min();
    if (packet_count_for_client > input_constraints_->packet_count_for_client_max()) {
      FX_LOGS(FATAL) << "server can't accomodate "
                        "kMinExtraInputPacketsForClient - not "
                        "using server - exiting";
    }
  
    uint32_t input_packet_count;
    fuchsia::sysmem::BufferCollectionInfo_2 buffer_collection_info;
    if (!ConfigurePortBufferCollection(false, kInputBufferLifetimeOrdinal,
                                       input_constraints_->buffer_constraints_version_ordinal(),
                                       packet_count_for_server, packet_count_for_client,
                                       &input_packet_count, &input_buffer_collection_,
                                       &buffer_collection_info)) {
      FX_LOGS(FATAL) << "ConfigurePortBufferCollection failed (input)";
    }
  
    ZX_ASSERT(input_free_packet_bits_.empty());
    input_free_packet_bits_.resize(input_packet_count, true);
    all_input_buffers_.reserve(buffer_collection_info.buffer_count);
    for (uint32_t i = 0; i < buffer_collection_info.buffer_count; i++) {
      std::unique_ptr<CodecBuffer> local_buffer = CodecBuffer::CreateFromVmo(
          i, std::move(buffer_collection_info.buffers[i].vmo),
          buffer_collection_info.buffers[i].vmo_usable_start,
          buffer_collection_info.settings.buffer_settings.size_bytes, true,
          buffer_collection_info.settings.buffer_settings.is_physically_contiguous);
      if (!local_buffer) {
        FX_LOGS(FATAL) << "CodecBuffer::CreateFromVmo() failed";
      }
      ZX_ASSERT(all_input_buffers_.size() == i);
      all_input_buffers_.push_back(std::move(local_buffer));
    }
  
    // Now that we've SetInputBufferPartialSettings(), the codec will get the
    // input buffers from sysmem.  The input packets all start as free with the
    // Codec client, per protocol.  Same goes for input buffers - this client
    // happens to track in terms of packets and have buffer_index == packet_index.
    //
    // TODO(dustingreen): Have CodecClient scramble the order of packets vs.
    // buffers to check that CodecImpl is handling that correctly for input
    // packets.
    input_free_packet_list_.reserve(input_packet_count);
    for (uint32_t i = 0; i < input_packet_count; i++) {
      input_free_packet_list_.push_back(i);
    }
    input_packet_index_to_buffer_index_.resize(input_packet_count);
    input_free_buffer_list_.reserve(buffer_collection_info.buffer_count);
    for (uint32_t i = 0; i < buffer_collection_info.buffer_count; ++i) {
      input_free_buffer_list_.push_back(i);
    }
  
    // Shuffle both free lists, so that we'll notice if a StreamProcessor server has inappropriate
    // dependency on ordering of either list or any particular association of packet_index with
    // buffer_index.
    std::random_device random_device;
    std::mt19937 prng(random_device());
    std::shuffle(input_free_packet_list_.begin(), input_free_packet_list_.end(), prng);
    std::shuffle(input_free_buffer_list_.begin(), input_free_buffer_list_.end(), prng);
  }
  
  bool CodecClient::CreateAndSyncBufferCollection(
      fuchsia::sysmem::BufferCollectionSyncPtr* out_buffer_collection,
      fidl::InterfaceHandle<fuchsia::sysmem::BufferCollectionToken>* out_codec_sysmem_token) {
    fuchsia::sysmem::BufferCollectionSyncPtr buffer_collection;
    fidl::InterfaceHandle<fuchsia::sysmem::BufferCollectionToken> codec_sysmem_token;
  
    // Create client_token which will get converted into out_buffer_collection.
    fuchsia::sysmem::BufferCollectionTokenSyncPtr client_token;
    fidl::InterfaceRequest<fuchsia::sysmem::BufferCollectionToken> client_token_request =
        client_token.NewRequest();
  
    // Create codec_sysmem_token that'll get returned via out_codec_sysmem_token.
    client_token->Duplicate(std::numeric_limits<uint32_t>::max(), codec_sysmem_token.NewRequest());
  
    // client_token gets converted into a buffer_collection.
    //
    // Start client_token connection and start converting it into a
    // BufferCollection, so we can Sync() the previous Duplicate().
    PostToFidlThread([this, client_token_request = std::move(client_token_request),
                      client_token = client_token.Unbind(),
                      buffer_collection_request = buffer_collection.NewRequest()]() mutable {
      if (!sysmem_) {
        return;
      }
      sysmem_->AllocateSharedCollection(std::move(client_token_request));
      // codec_sysmem_token will be known to sysmem by the time client_token
      // closure is seen by sysmem, which in turn is before
      // buffer_collection_request will be hooked up, which is why
      // buffer_collection->Sync() completion below is enough to prove that
      // sysmem knows about codec_sysmem_token before codec_sysmem_token is
      // sent to the codec.
      sysmem_->BindSharedCollection(std::move(client_token), std::move(buffer_collection_request));
    });
  
    // After Sync() completes its round trip, we know that sysmem knows about
    // codec_sysmem_token (causally), which is important because we'll shortly
    // send codec_sysmem_token to the codec which will use codec_sysmem_token via
    // a different sysmem channel.
    zx_status_t sync_status = buffer_collection->Sync();
    if (sync_status != ZX_OK) {
      FX_PLOGS(FATAL, sync_status) << "buffer_collection->Sync() failed";
    }
  
    *out_buffer_collection = std::move(buffer_collection);
    *out_codec_sysmem_token = std::move(codec_sysmem_token);
    return true;
  }
  
  bool CodecClient::WaitForSysmemBuffersAllocated(
      bool is_output, fuchsia::sysmem::BufferCollectionSyncPtr* buffer_collection_param,
      fuchsia::sysmem::BufferCollectionInfo_2* out_buffer_collection_info) {
    // The style guide doesn't like non-const &, but the code in this method is
    // easier to read with a non-const &, so treat it that way within this method.
    fuchsia::sysmem::BufferCollectionSyncPtr& buffer_collection = *buffer_collection_param;
    fuchsia::sysmem::BufferCollectionInfo_2 result_buffer_collection_info;
  
    // It's not permitted to send input data until the client knows that sysmem
    // is done allocating.  It's not required that the client know that the
    // codec knows that sysmem is done allocating though - the server will
    // verify that sysmem is done by communicating with sysmem directly as
    // needed.
    zx_status_t allocate_status;
    zx_status_t call_status =
        buffer_collection->WaitForBuffersAllocated(&allocate_status, &result_buffer_collection_info);
    if (call_status != ZX_OK) {
      FX_PLOGS(ERROR, call_status) << "WaitForBuffersAllocated returned failure";
      return false;
    }
    if (allocate_status != ZX_OK) {
      FX_PLOGS(ERROR, allocate_status) << "WaitForBuffersAllocated allocation failed";
      return false;
    }
  
    // This is a little noisy, but it can be useful to see how many buffers are being used.
    FX_LOGS(INFO) << "WaitForSysmemBuffersAllocated() done - is_output: " << is_output
                  << " buffer_count: " << result_buffer_collection_info.buffer_count;
  
    *out_buffer_collection_info = std::move(result_buffer_collection_info);
    return true;
  }
  
  void CodecClient::Stop() {
    ZX_DEBUG_ASSERT(thrd_current() != loop_thread_);
    OneShotEvent unbind_and_loop_lambdas_done;
    PostToFidlThread([this, &unbind_and_loop_lambdas_done] {
      if (codec_.is_bound()) {
        codec_.Unbind();
      }
      if (sysmem_.is_bound()) {
        sysmem_.Unbind();
      }
      if (input_buffer_collection_.is_bound()) {
        input_buffer_collection_.Unbind();
      }
      if (output_buffer_collection_.is_bound()) {
        output_buffer_collection_.Unbind();
      }
      // Any lambdas previously queued (by any handlers for the bindings we're
      // unbinding just above) need to be done also, so fence those by re-posting.
      //
      // This relies on lambdas on fidl thread (other than this one) to not
      // re-post to the fidl thread, which we enforce for all calls to
      // PostToFidlThread() except this one.
      PostToFidlThread([&unbind_and_loop_lambdas_done] { unbind_and_loop_lambdas_done.Signal(); },
                       false);
    });
    unbind_and_loop_lambdas_done.Wait();
  }
  
  void CodecClient::PostToFidlThread(fit::closure to_run, bool enforce_no_re_posting) {
    ZX_DEBUG_ASSERT(thrd_current() != loop_thread_ || !enforce_no_re_posting);
    zx_status_t post_status = async::PostTask(dispatcher_, std::move(to_run));
    ZX_ASSERT(post_status == ZX_OK);
  }
  
  void CodecClient::CallSyncAndWaitForResponse() {
    // |is_sync_complete_condition_| may also be signaled on connection lost, so it
    // needs to be an instance variable.
    bool is_sync_complete = false;
    // Capturing stuff with just "&" is sometimes frowned upon, but in this case
    // there's no chance of any lambda outliving anything, so it's fine.  The
    // outer lambda is because ProxyController isn't thread-safe and the present
    // method is called from the main thread not the FIDL thread, so we have to
    // switch threads to send a FIDL message.  The inner lambda is the completion
    // callback.
    FX_VLOGS(3) << "before calling Sync() (main thread)...";
    async::PostTask(dispatcher_, [&] {
      FX_VLOGS(3) << "before calling Sync() (fidl thread)...";
      codec_->Sync([&]() {
        {  // scope lock
          std::unique_lock<std::mutex> lock(is_sync_complete_lock_);
          is_sync_complete = true;
        }  // ~lock
        is_sync_complete_condition_.notify_all();
      });
    });
    FX_VLOGS(3) << "after calling Sync() - waiting...\n";
    {  // scope lock
      std::unique_lock<std::mutex> lock(is_sync_complete_lock_);
      // We rely on the channel error handler to be doing an exit() for this loop
      // to be reasonable without checking for channel failure here.
      while (!is_sync_complete && !connection_lost_) {
        is_sync_complete_condition_.wait(lock);
      }
    }
    FX_VLOGS(3) << "after calling Sync() - done waiting\n";
    ZX_ASSERT(is_sync_complete || connection_lost_);
  }
  
  void CodecClient::TrackOutputStreamLifetimeOrdinal(uint64_t output_stream_lifetime_ordinal) {
    // must be odd
    ZX_ASSERT(output_stream_lifetime_ordinal % 2 == 1);
    ZX_ASSERT(output_stream_lifetime_ordinal >= output_stream_lifetime_ordinal_);
    if (output_stream_lifetime_ordinal > output_stream_lifetime_ordinal_) {
      // We're allowed to forget format any time there's a stream change, so we
      // do.  This isn't critical for this test code, but it's closer to how a
      // real client will likely track the output format on a per-stream basis.
      ZX_ASSERT(!last_output_format_ ||
                last_output_format_->stream_lifetime_ordinal() == output_stream_lifetime_ordinal_);
      output_stream_lifetime_ordinal_ = output_stream_lifetime_ordinal;
      last_output_format_ = nullptr;
      // We intentionally don't reset is_packet_since_last_format_.
    }
  }
  
  void CodecClient::OnInputConstraints(fuchsia::media::StreamBufferConstraints input_constraints) {
    if (input_constraints_) {
      FX_LOGS(FATAL) << "server sent more than one input constraints";
    }
    {  // scope lock
      std::unique_lock<std::mutex> lock(lock_);
      input_constraints_ =
          std::make_unique<fuchsia::media::StreamBufferConstraints>(std::move(input_constraints));
    }  // ~lock
    input_constraints_exist_condition_.notify_all();
  }
  
  void CodecClient::OnFreeInputPacket(fuchsia::media::PacketHeader free_input_packet) {
    bool free_buffer_list_was_empty;
    bool free_packet_list_was_empty;
    {  // scope lock
      std::unique_lock<std::mutex> lock(lock_);
      if (!free_input_packet.has_packet_index()) {
        FX_LOGS(FATAL) << "OnFreeInputPacket(): Packet has no index.";
      }
      if (input_free_packet_bits_[free_input_packet.packet_index()]) {
        // already free - a normal client wouldn't want to just exit here since
        // this is the server's fault - in this example we just care that we
        // detect it
        FX_LOGS(FATAL) << "OnFreeInputPacket() when already free - server's fault? - "
                          "packet_index: "
                       << free_input_packet.packet_index();
      }
      free_buffer_list_was_empty = input_free_buffer_list_.empty();
      input_free_buffer_list_.push_back(
          input_packet_index_to_buffer_index_[free_input_packet.packet_index()]);
      free_packet_list_was_empty = input_free_packet_list_.empty();
      input_free_packet_list_.push_back(free_input_packet.packet_index());
      input_free_packet_bits_[free_input_packet.packet_index()] = true;
    }  // ~lock
    if (free_buffer_list_was_empty) {
      input_free_buffer_list_not_empty_.notify_all();
    }
    if (free_packet_list_was_empty) {
      input_free_packet_list_not_empty_.notify_all();
    }
  }
  
  std::unique_ptr<fuchsia::media::Packet> CodecClient::BlockingGetFreeInputPacket() {
    uint32_t free_packet_index;
    {  // scope lock
      std::unique_lock<std::mutex> lock(lock_);
      while (input_free_packet_list_.empty()) {
        if (connection_lost_)
          return nullptr;
        input_free_packet_list_not_empty_.wait(lock);
      }
      free_packet_index = input_free_packet_list_.back();
      input_free_packet_list_.pop_back();
      // We intentionally do not modify input_free_packet_bits_ here, as those bits are
      // tracking the protocol level free-ness, so will get updated when the
      // caller queues the input packet.
      ZX_ASSERT(input_free_packet_bits_[free_packet_index]);
    }  // ~lock
    std::unique_ptr<fuchsia::media::Packet> packet = fuchsia::media::Packet::New();
    packet->mutable_header()->set_buffer_lifetime_ordinal(kInputBufferLifetimeOrdinal);
    packet->mutable_header()->set_packet_index(free_packet_index);
    return packet;
  }
  
  const CodecBuffer& CodecClient::BlockingGetFreeInputBufferForPacket(
      fuchsia::media::Packet* packet) {
    uint32_t free_buffer_index;
    {  // scope lock
      std::unique_lock<std::mutex> lock(lock_);
      while (input_free_buffer_list_.empty()) {
        input_free_buffer_list_not_empty_.wait(lock);
      }
      free_buffer_index = input_free_buffer_list_.back();
      input_free_buffer_list_.pop_back();
    }  // ~lock
    input_packet_index_to_buffer_index_[packet->header().packet_index()] = free_buffer_index;
    packet->set_buffer_index(free_buffer_index);
    return *all_input_buffers_[free_buffer_index].get();
  }
  
  const CodecBuffer& CodecClient::GetInputBufferByIndex(uint32_t packet_index) {
    return *all_input_buffers_[packet_index];
  }
  
  const CodecBuffer& CodecClient::GetOutputBufferByIndex(uint32_t packet_index) {
    return *all_output_buffers_[packet_index];
  }
  
  void CodecClient::QueueInputFormatDetails(uint64_t stream_lifetime_ordinal,
                                            fuchsia::media::FormatDetails input_format_details) {
    async::PostTask(dispatcher_, [this, stream_lifetime_ordinal,
                                  input_format_details = std::move(input_format_details)]() mutable {
      codec_->QueueInputFormatDetails(stream_lifetime_ordinal, std::move(input_format_details));
    });
  }
  
  // buffer - the populated input packet buffer, or an empty input buffers with
  //   end_of_stream set on it.
  void CodecClient::QueueInputPacket(std::unique_ptr<fuchsia::media::Packet> packet) {
    ZX_ASSERT(packet->has_header());
    ZX_ASSERT(packet->header().has_buffer_lifetime_ordinal());
    ZX_ASSERT(packet->header().has_packet_index());
    ZX_ASSERT(packet->has_buffer_index());
    ZX_ASSERT(packet->has_stream_lifetime_ordinal());
    ZX_ASSERT(packet->has_start_offset());
    ZX_ASSERT(packet->has_valid_length_bytes());
    // timestamp_ish field is optional start_access_unit field is optional
    // known_end_access_unit is optional
    fuchsia::media::Packet local_packet = fidl::Clone(*packet);
    {  // scope lock
      // This packet is already not on the free list, but is still considered free
      // from a protocol point of view, so update that part.
      std::unique_lock<std::mutex> lock(lock_);
      ZX_ASSERT(input_free_packet_bits_[local_packet.header().packet_index()]);
      input_free_packet_bits_[local_packet.header().packet_index()] = false;
      // From here it's as if this packet is already in flight with the server.
    }  // ~lock
    async::PostTask(dispatcher_, [this, packet = std::move(local_packet)]() mutable {
      codec_->QueueInputPacket(std::move(packet));
    });
  }
  
  void CodecClient::QueueInputEndOfStream(uint64_t stream_lifetime_ordinal) {
    async::PostTask(dispatcher_, [this, stream_lifetime_ordinal] {
      codec_->QueueInputEndOfStream(stream_lifetime_ordinal);
    });
  }
  
  void CodecClient::FlushEndOfStreamAndCloseStream(uint64_t stream_lifetime_ordinal) {
    async::PostTask(dispatcher_, [this, stream_lifetime_ordinal] {
      codec_->FlushEndOfStreamAndCloseStream(stream_lifetime_ordinal);
    });
  }
  
  std::unique_ptr<CodecOutput> CodecClient::BlockingGetEmittedOutput() {
    while (true) {
      // The rule is that a required pending constraints won't be followed by any
      // more output packets until it's no longer pending (in the sense that the
      // output buffers have been suitably re-configured).  We verify the server
      // is following that rule elsewhere, which means we know here that when both
      // packets are pending and constraints is pending, the packets were
      // delivered to the client first. So we drain the packets first.
      std::unique_ptr<CodecOutput> packet;
      std::shared_ptr<const fuchsia::media::StreamOutputConstraints> constraints;
      {  // scope lock
        std::unique_lock<std::mutex> lock(lock_);
        while (!output_pending_) {
          if (connection_lost_)
            return nullptr;
          output_pending_condition_.wait(lock);
        }
        if (!emitted_output_.empty()) {
          packet = std::move(emitted_output_.front());
          emitted_output_.pop_front();
          // This only does anything when the last packet is consumed and there is
          // no config pending.
          if (!ComputeOutputPendingLocked()) {
            output_pending_ = false;
          }
        } else {
          ZX_ASSERT(output_constraints_action_pending_);
          ZX_ASSERT(last_required_output_constraints_);
          constraints = last_required_output_constraints_;
        }
      }  // ~lock
  
      // Now we own a packet or have a required config to deal with, but not both,
      // so it doesn't matter which order we check here, but for clarity we check
      // in the same order as above.
      if (packet) {
        return packet;
      }
  
      // We have a required output config change to deal with here.
  
      // The server implicitly has relinquished ownership of all output packets
      // and all output buffers as a semantic of the required config change.  This
      // shouldn't really be thought of the packets being "emitted" - rather from
      // the server's point of view they're deallocated already.  Now it's the
      // client's turn to deallocate the old buffers.  It is not permitted to
      // re-use a previous buffer as a new buffer, per protocol rules, not even
      // for the same Codec instance, and not even for a mid-stream output config
      // change.
      //
      // The main mechanism used to detect that the server isn't sending output
      // too soon is output_config_action_pending_.  In contrast, the client code
      // in this example permits itself to send RecycleOutputPacket() after the
      // client has already seen OnOutputConstraints() with action required true,
      // even though the client could stop itself from doing so as a potential
      // optimization.  The client is allowed to send RecycleOutputPacket() up
      // until the implied ReleaseOutputBuffers() at the start of
      // SetOutputSettings().  Between then and a given packet_index becoming
      // emitted again (!free), a RecycleOutputPacket() for that packet_index is
      // forbidden.
      //
      // Because of the client allowing itself to send RecycleOutputPacket() for a
      // while longer than fundamentally necessary, we delay upkeep on
      // output_free_packet_bits_ until here.  This upkeep isn't really fundamentally
      // necessary between OnOutputConstraints() with action required true and the
      // last AddOutputBuffer() as part of output re-configuration, but ... this
      // explicit delayed upkeep _may_ help illustrate how it's acceptable for a
      // client to let the completion end of output processing send
      // RecycleOutputPacket() as long as all those will be sent before
      // SetOutputSettings().
      std::shared_ptr<const fuchsia::media::StreamOutputConstraints> snapped_constraints;
      uint64_t new_output_buffer_lifetime_ordinal;
      {  // scope lock
        std::unique_lock<std::mutex> lock(lock_);
  
        // We know this because the previous OnOutputConstraints() set this and
        // because we're only here if it's set.
        ZX_ASSERT(output_constraints_action_pending_);
        // We know this because we reject additional output from the server when
        // output_config_action_pending_ is true, and because we've drained all
        // previous output by this point.
        ZX_ASSERT(emitted_output_.empty());
        // We know this because we're only here if we have pending constraints.
        ZX_ASSERT(constraints);
  
        // Not really critical to do this, as we'll just end up setting these
        // back to true under the same lock hold interval as we set
        // output_config_action_pending_ to false, but see comment above re.
        // how this might help illustrate how late RecycleOutputPacket() can be
        // sent.
        //
        // Think of this assignment as slightly more than a comment in this
        // example, rather than any real need.
        output_free_packet_bits_.resize(0);
  
        // Free the old output buffers, if any.
        all_output_buffers_.clear();
  
        // Here is where we snap which exact constraints version we'll actually
        // use.
        //
        // For a client that's doing output buffer re-config on the FIDL thread
        // during OnOutputConstraints with action required true, this will always
        // just be the constraints being presently received.  But this example
        // shows how to drive the codec in a protocol-valid way without being
        // forced to perform buffer re-configuration on the FIDL thread.
        ZX_ASSERT(output_constraints_action_pending_);
        ZX_ASSERT(last_required_output_constraints_);
        ZX_ASSERT(last_output_constraints_);
        // We'll snap the last_output_constraints_, which is always at least as
        // recent as the last_required_output_constraints_.
        snapped_constraints = last_output_constraints_;
        ZX_ASSERT(snapped_constraints);
        new_output_buffer_lifetime_ordinal = next_output_buffer_lifetime_ordinal_;
        next_output_buffer_lifetime_ordinal_ += 2;
      }  // ~lock
  
      //
      // Tell the server about output settings.
      //
  
      ZX_ASSERT(snapped_constraints->has_buffer_constraints());
      const fuchsia::media::StreamBufferConstraints& buffer_constraints =
          snapped_constraints->buffer_constraints();
      ZX_ASSERT(buffer_constraints.has_packet_count_for_server_min());
      ZX_ASSERT(buffer_constraints.has_packet_count_for_server_recommended());
      // Use min; if decode gets stuck using min, we want to notice that.
      uint32_t packet_count_for_server = buffer_constraints.packet_count_for_server_min();
      uint32_t packet_count_for_client =
          std::max(kMinOutputPacketsForClient, buffer_constraints.packet_count_for_client_min());
      if (packet_count_for_client > buffer_constraints.packet_count_for_client_max()) {
        FX_LOGS(FATAL) << "server can't accomodate "
                          "kMinExtraOutputPacketsForClient - not "
                          "using server - exiting";
      }
  
      uint32_t packet_count;
      fuchsia::sysmem::BufferCollectionInfo_2 buffer_collection_info;
      if (!ConfigurePortBufferCollection(true, new_output_buffer_lifetime_ordinal,
                                         buffer_constraints.buffer_constraints_version_ordinal(),
                                         packet_count_for_server, packet_count_for_client,
                                         &packet_count, &output_buffer_collection_,
                                         &buffer_collection_info)) {
        FX_LOGS(FATAL) << "ConfigurePortBufferCollection failed (output)";
      }
  
      // Configure tracking for output buffers.
      {  // scope lock
        std::lock_guard<std::mutex> lock(lock_);
  
        all_output_buffers_.reserve(packet_count);
        for (uint32_t i = 0; i < packet_count; i++) {
          std::unique_ptr<CodecBuffer> buffer = CodecBuffer::CreateFromVmo(
              i, std::move(buffer_collection_info.buffers[i].vmo),
              buffer_collection_info.buffers[i].vmo_usable_start,
              buffer_collection_info.settings.buffer_settings.size_bytes, true,
              buffer_collection_info.settings.buffer_settings.is_physically_contiguous);
          if (!buffer) {
            FX_LOGS(FATAL) << "CodecBuffer::Allocate() failed (output)";
          }
          ZX_ASSERT(all_output_buffers_.size() == i);
          all_output_buffers_.push_back(std::move(buffer));
  
          if (i == packet_count - 1) {
            output_free_packet_bits_.resize(packet_count, true);
          }
        }
  
        current_output_buffer_lifetime_ordinal_ = new_output_buffer_lifetime_ordinal;
      }  // ~lock
  
      // We're ready to receive output.
      PostToFidlThread([this, output_buffer_lifetime_ordinal = new_output_buffer_lifetime_ordinal] {
        if (!codec_) {
          return;
        }
        if (output_buffer_lifetime_ordinal != current_output_buffer_lifetime_ordinal_) {
          return;
        }
        codec_->CompleteOutputBufferPartialSettings(output_buffer_lifetime_ordinal);
      });
  
      {  // scope lock
        std::lock_guard<std::mutex> lock(lock_);
  
        // So, now that we're done with that output re-config, it's time to see if
        // that re-config was the last one we need to do, or if there's a newer
        // config that's action-required.
  
        if (snapped_constraints->buffer_constraints().buffer_constraints_version_ordinal() >=
            last_required_output_constraints_->buffer_constraints()
                .buffer_constraints_version_ordinal()) {
          // Good.  The client is caught up.  The output_config_action_pending_
          // can become false here, but may very shortly become true again if
          // another OnOutputConstraints() is received after we release the lock
          // (roughly speaking; see code).
          //
          // It's ok that we didn't set output_config_action_pending_ to false
          // before sending the last AddOutputBuffer() above, because
          // OnOutputConstraints() was still able to update
          // last_required_output_config_ as needed, which it's been able to do
          // all along during most of this whole method.  If we had set to false
          // up there, it would probably be less obvious why it works vs. here,
          // but either can work.
          FX_VLOGS(3) << "output_config_action_pending_ = false, because client "
                         "caught up";
          output_constraints_action_pending_ = false;
          // Because this was true for at least pending config reason which we
          // are only just clearing immediately above.
          ZX_ASSERT(output_pending_);
          // There can be output packets by this point so only clear
          // output_pending_ if there are also no packets.
          if (!ComputeOutputPendingLocked()) {
            output_pending_ = false;
          }
        } else {
          // We've received and even more recent constraints that's
          // action-required, so go around again without clearing
          // output_constraints_action_pending_ or output_pending_. Both remain
          // true until we've caught up to a config that's at least as new as the
          // last_required_output_constraints_.
          FX_VLOGS(3) << "output_constraints_action_pending_ remains true because server "
                         "has sent yet another action-required output constraints";
          ZX_ASSERT(output_constraints_action_pending_);
          ZX_ASSERT(output_pending_);
        }
      }  // ~lock
    }
  }
  
  bool CodecClient::ConfigurePortBufferCollection(
      bool is_output, uint64_t new_buffer_lifetime_ordinal,
      uint64_t buffer_constraints_version_ordinal, uint32_t packet_count_for_server,
      uint32_t packet_count_for_client, uint32_t* out_packet_count,
      fuchsia::sysmem::BufferCollectionPtr* out_buffer_collection,
      fuchsia::sysmem::BufferCollectionInfo_2* out_buffer_collection_info) {
    uint32_t packet_count = packet_count_for_server + packet_count_for_client;
  
    fuchsia::media::StreamBufferPartialSettings settings;
    settings.set_buffer_lifetime_ordinal(new_buffer_lifetime_ordinal);
    settings.set_buffer_constraints_version_ordinal(buffer_constraints_version_ordinal);
    settings.set_single_buffer_mode(false);
    settings.set_packet_count_for_server(packet_count_for_server);
    settings.set_packet_count_for_client(packet_count_for_client);
  
    fuchsia::sysmem::BufferCollectionSyncPtr buffer_collection;
    fidl::InterfaceHandle<fuchsia::sysmem::BufferCollectionToken> codec_sysmem_token;
    if (!CreateAndSyncBufferCollection(&buffer_collection, &codec_sysmem_token)) {
      FX_LOGS(FATAL) << "CreateAndSyncBufferCollection failed (output)";
      return false;
    }
  
    settings.set_sysmem_token(std::move(codec_sysmem_token));
  
    fuchsia::sysmem::BufferCollectionConstraints constraints;
    // TODO(SCN-1388): Hardcoded to read/write to allow direct Vulkan import on
    // UMA platforms.
    if (!is_output && is_input_secure_) {
      constraints.usage.video = fuchsia::sysmem::videoUsageDecryptorOutput;
    } else if (is_output && is_output_secure_) {
      // Only used if kVerifySecureOutput is used in test.
      constraints.usage.cpu =
          fuchsia::sysmem::cpuUsageReadOften | fuchsia::sysmem::cpuUsageWriteOften;
    } else {
      constraints.usage.cpu =
          fuchsia::sysmem::cpuUsageReadOften | fuchsia::sysmem::cpuUsageWriteOften;
    }
  
    // TODO(dustingreen): Make this more flexible once we're more flexible on
    // frame_count on output of decoder.
    constraints.min_buffer_count_for_camping = packet_count_for_client;
    ZX_DEBUG_ASSERT(constraints.min_buffer_count_for_dedicated_slack == 0);
    ZX_DEBUG_ASSERT(constraints.min_buffer_count_for_shared_slack == 0);
  
    // 0 is treated as 0xFFFFFFFF.
    ZX_DEBUG_ASSERT(constraints.max_buffer_count == 0);
  
    constraints.has_buffer_memory_constraints = true;
    // Sysmem has a built-in min_size_bytes of 1, so no need to really constrain
    // min_size_bytes here, unless output, in which case setting a min_size_bytes
    // allows for seamless video frame dimension changes.  If the client code says
    // 0 that's fine.
    //
    // Similar for min_buffer_count, but min of 0 means 0 / no constraint in that case.
    ZX_DEBUG_ASSERT(constraints.buffer_memory_constraints.min_size_bytes == 0);
    ZX_DEBUG_ASSERT(constraints.min_buffer_count == 0);
    if (is_output) {
      constraints.buffer_memory_constraints.min_size_bytes = min_output_buffer_size_;
      constraints.min_buffer_count = min_output_buffer_count_;
    }
    constraints.buffer_memory_constraints.max_size_bytes = std::numeric_limits<uint32_t>::max();
    constraints.buffer_memory_constraints.physically_contiguous_required = false;
    constraints.buffer_memory_constraints.secure_required = false;
    if (is_output && is_output_secure_) {
      constraints.buffer_memory_constraints.inaccessible_domain_supported = true;
    } else if (!is_output && is_input_secure_) {
      constraints.buffer_memory_constraints.cpu_domain_supported = false;
      constraints.buffer_memory_constraints.ram_domain_supported = false;
      constraints.buffer_memory_constraints.inaccessible_domain_supported = true;
      constraints.buffer_memory_constraints.secure_required = true;
      constraints.buffer_memory_constraints.heap_permitted_count = 1;
      constraints.buffer_memory_constraints.heap_permitted[0] =
          fuchsia::sysmem::HeapType::AMLOGIC_SECURE_VDEC;
    } else {
      ZX_DEBUG_ASSERT(!constraints.buffer_memory_constraints.inaccessible_domain_supported);
      ZX_DEBUG_ASSERT(constraints.buffer_memory_constraints.cpu_domain_supported);
    }
  
    ZX_DEBUG_ASSERT(!constraints.buffer_memory_constraints.ram_domain_supported);
  
    // Despite being a consumer of output uncompressed video frames (when decoding
    // video and is_output), for now we intentionally don't constrain to the
    // PixelFormatType(s) that we can consume, and instead fail later if we get
    // something unexpected on output. That's just easier than plumbing
    // PixelFormatType(s) to here for now.
    ZX_DEBUG_ASSERT(constraints.image_format_constraints_count == 0);
  
    PostToFidlThread([this, is_output, settings = std::move(settings)]() mutable {
      if (!codec_) {
        return;
      }
      if (is_output) {
        codec_->SetOutputBufferPartialSettings(std::move(settings));
      } else {
        codec_->SetInputBufferPartialSettings(std::move(settings));
      }
    });
  
    buffer_collection->SetConstraints(true, std::move(constraints));
  
    fuchsia::sysmem::BufferCollectionInfo_2 buffer_collection_info;
    // This borrows buffer_collection during
    // the call.
    if (!WaitForSysmemBuffersAllocated(is_output, &buffer_collection, &buffer_collection_info)) {
      FX_LOGS(FATAL) << "WaitForSysmemBuffer"
                        "sAllocated failed";
      return false;
    }
  
    packet_count = std::max(packet_count, buffer_collection_info.buffer_count);
  
    if (!in_lax_mode_) {
      // We don't expect normal cases to go above kMaxExpectedBufferCount, but if
      // min_output_buffer_count_ forces higher buffer counts, that's fine.
      ZX_ASSERT(buffer_collection_info.buffer_count <= kMaxExpectedBufferCount ||
                min_output_buffer_count_ > kMaxExpectedBufferCount);
    }
  
    fuchsia::sysmem::BufferCollectionPtr buffer_collection_ptr;
  
    // For the Bind() we probably don't strictly need to be on FIDL thread, so do
    // the Bind() here.  This does mean we need to set the error handler before
    // the Bind() however.
    buffer_collection_ptr.set_error_handler([is_output](zx_status_t status) {
      FX_PLOGS(FATAL, status) << "BufferCollection failed is_output: " << is_output;
    });
  
    // This implicitly converts buffer_collection from
    // fidl::SynchronousInterfacePtr<> to fidl::InterfaceHandle<>, which is what
    // we want; we're moving handling of the BufferCollection from this thread to
    // the FIDL thread.
    zx_status_t bind_status = buffer_collection_ptr.Bind(std::move(buffer_collection), dispatcher_);
    if (bind_status != ZX_OK) {
      FX_PLOGS(FATAL, bind_status) << "buffer_collection_ptr.Bind() failed is_output: " << is_output;
      return false;
    }
  
    *out_packet_count = packet_count;
    *out_buffer_collection = std::move(buffer_collection_ptr);
    *out_buffer_collection_info = std::move(buffer_collection_info);
    return true;
  }
  
  void CodecClient::RecycleOutputPacket(fuchsia::media::PacketHeader free_packet) {
    ZX_ASSERT(free_packet.has_packet_index());
    {  // scope lock
      std::unique_lock<std::mutex> lock(lock_);
      output_free_packet_bits_[free_packet.packet_index()] = true;
    }  // ~lock
    async::PostTask(dispatcher_, [this, free_packet = std::move(free_packet)]() mutable {
      codec_->RecycleOutputPacket(std::move(free_packet));
    });
  }
  
  void CodecClient::OnOutputConstraints(fuchsia::media::StreamOutputConstraints output_constraints) {
    bool output_pending_notify_needed = false;
    // Not that the std::move() actaully helps here, but that's what we're doing.
    std::shared_ptr<fuchsia::media::StreamOutputConstraints> shared_constraints =
        std::make_shared<fuchsia::media::StreamOutputConstraints>(std::move(output_constraints));
    {  // scope lock
      std::unique_lock<std::mutex> lock(lock_);
  
      if (!shared_constraints->has_stream_lifetime_ordinal()) {
        FX_LOGS(FATAL) << "StreamOutputConstraints missing stream_lifetime_ordinal";
      }
      uint64_t stream_lifetime_ordinal = shared_constraints->stream_lifetime_ordinal();
      TrackOutputStreamLifetimeOrdinal(stream_lifetime_ordinal);
  
      ZX_ASSERT(
          !last_output_constraints_ ||
          (last_output_constraints_->has_buffer_constraints() &&
           last_output_constraints_->buffer_constraints().has_buffer_constraints_version_ordinal()));
      uint64_t previous_buffer_constraints_version_ordinal =
          last_output_constraints_
              ? last_output_constraints_->buffer_constraints().buffer_constraints_version_ordinal()
              : 0;
      ZX_ASSERT(shared_constraints->has_buffer_constraints() &&
                shared_constraints->buffer_constraints().has_buffer_constraints_version_ordinal());
      if (shared_constraints->buffer_constraints().buffer_constraints_version_ordinal() <
          previous_buffer_constraints_version_ordinal) {
        FX_LOGS(FATAL) << "broken server sent badly ordered buffer constraints ordinals";
      }
      if ((shared_constraints->has_buffer_constraints_action_required() &&
           shared_constraints->buffer_constraints_action_required()) &&
          shared_constraints->buffer_constraints().buffer_constraints_version_ordinal() <=
              previous_buffer_constraints_version_ordinal) {
        FX_LOGS(FATAL) << "broken server sent buffer_constraints_action_required without "
                          "increasingbuffer_constraints_version_ordinal";
      }
      last_output_constraints_ = shared_constraints;
      FX_VLOGS(3) << "OnOutputConstraints buffer_constraints_version_ordinal: "
                  << shared_constraints->buffer_constraints().buffer_constraints_version_ordinal()
                  << "buffer_constraints_action_required: "
                  << shared_constraints->buffer_constraints_action_required();
      if (shared_constraints->buffer_constraints_action_required()) {
        last_required_output_constraints_ = shared_constraints;
        // A client is allowed to forget the output format on any action required
        // buffer constraints, so forget here.
        last_output_format_ = nullptr;
        FX_VLOGS(3) << "output_config_action_pending_ = true, because received a "
                       "buffer_constraints_action_required constraints\n";
        output_constraints_action_pending_ = true;
        if (!output_pending_) {
          output_pending_ = true;
          output_pending_notify_needed = true;
        }
      }
    }  // ~lock
    if (output_pending_notify_needed) {
      output_pending_condition_.notify_all();
    }
  }
  
  void CodecClient::OnOutputFormat(fuchsia::media::StreamOutputFormat output_format) {
    // We don't need to notify output_pending_condition_ since in contrast to
    // constraints we don't need to take action, and nobody cares about the format
    // until the next packet arrives.
    std::shared_ptr<fuchsia::media::StreamOutputFormat> shared_format =
        std::make_shared<fuchsia::media::StreamOutputFormat>(std::move(output_format));
    std::unique_lock<std::mutex> lock(lock_);
  
    if (!shared_format->has_stream_lifetime_ordinal()) {
      FX_LOGS(FATAL) << "StreamOutputFormat missing stream_lifetime_ordinal";
    }
    uint64_t stream_lifetime_ordinal = shared_format->stream_lifetime_ordinal();
    TrackOutputStreamLifetimeOrdinal(stream_lifetime_ordinal);
  
    if (is_format_since_last_packet_) {
      // A server can easily elide unnecessary OnOutputFormat by not sending
      // OnOutputFormat until immediately before OnOutputPacket.  The format is
      // logically part of each output packet, with the optimization that we only
      // send output format when it changes, in between packets, to avoid needing
      // to send output format with every packet.
      FX_LOGS(FATAL) << "broken server sent two OnOutputFormat() in a row";
    }
    if (!shared_format->has_stream_lifetime_ordinal()) {
      FX_LOGS(FATAL) << "OnOutputFormat !has_stream_lifetime_ordinal()";
    }
    if (!shared_format->has_format_details()) {
      FX_LOGS(FATAL) << "OnOutputFormat !has_format_details()";
    }
    last_output_format_ = shared_format;
    is_format_since_last_packet_ = true;
  }
  
  void CodecClient::OnOutputPacket(fuchsia::media::Packet output_packet, bool error_detected_before,
                                   bool error_detected_during) {
    FX_CHECK(output_packet.has_header());
    FX_CHECK(output_packet.has_stream_lifetime_ordinal());
    FX_CHECK(output_packet.header().has_packet_index());
    bool output_pending_notify_needed = false;
    std::unique_ptr<const fuchsia::media::Packet> local_packet =
        std::make_unique<fuchsia::media::Packet>(std::move(output_packet));
    uint32_t packet_index = local_packet->header().packet_index();
  
    {  // scope lock
      std::unique_lock<std::mutex> lock(lock_);
  
      if (!local_packet->has_stream_lifetime_ordinal()) {
        FX_LOGS(FATAL) << "Packet missing stream_lifetime_ordinal";
      }
      uint64_t stream_lifetime_ordinal = local_packet->stream_lifetime_ordinal();
      TrackOutputStreamLifetimeOrdinal(stream_lifetime_ordinal);
      if (!last_output_format_ ||
          last_output_format_->stream_lifetime_ordinal() != stream_lifetime_ordinal) {
        FX_LOGS(FATAL) << "OnOutputFormat required before OnOutputPacket, per-stream";
      }
  
      std::unique_ptr<CodecOutput> output =
          std::make_unique<CodecOutput>(stream_lifetime_ordinal, last_output_constraints_,
                                        last_output_format_, std::move(local_packet), false);
      if (output_constraints_action_pending_) {
        // FWIW, we wouldn't be able to detect this if we were using the
        // async::Loop thread to perform output buffer re-configuration.
        FX_LOGS(FATAL) << "server incorrectly sent output packet while required "
                          "constraints change pending";
      }
      if (!output_free_packet_bits_[packet_index]) {
        // The packet was emitted twice by the server without it becoming free in
        // between, which is broken server behavior.
        FX_LOGS(FATAL) << "server incorrectly emitted an output packet without it becoming "
                          "free in between";
      }
      // Emitted by server, so not free until later when we send it back to server
      // with RecycleOutputPacket(), or until we re-configure output buffers in
      // which case all the output packets start free with the server.
      output_free_packet_bits_[packet_index] = false;
      emitted_output_.push_back(std::move(output));
      is_format_since_last_packet_ = false;
      if (!output_pending_) {
        output_pending_ = true;
        output_pending_notify_needed = true;
      }
    }  // ~lock
    if (output_pending_notify_needed) {
      output_pending_condition_.notify_all();
    }
  }
  
  void CodecClient::OnOutputEndOfStream(uint64_t stream_lifetime_ordinal,
                                        bool error_detected_before) {
    bool output_pending_notify_needed = false;
    std::unique_ptr<CodecOutput> output =
        std::make_unique<CodecOutput>(stream_lifetime_ordinal, nullptr, nullptr, nullptr, true);
    {  // scope lock
      std::unique_lock<std::mutex> lock(lock_);
      if (output_constraints_action_pending_) {
        // FWIW, we wouldn't be able to detect this if we were using the
        // async::Loop thread to perform output buffer re-configuration.
        FX_LOGS(FATAL) << "server incorrectly sent OnOutputEndOfStream() while "
                          "required constraints change pending";
      }
      emitted_output_.push_back(std::move(output));
      if (!output_pending_) {
        output_pending_ = true;
        output_pending_notify_needed = true;
      }
    }  // ~lock
    if (output_pending_notify_needed) {
      output_pending_condition_.notify_all();
    }
  }
  
  void CodecClient::OnStreamFailed(uint64_t stream_lifetime_ordinal,
                                   fuchsia::media::StreamError error) {
    FX_LOGS(FATAL) << "OnStreamFailed: stream_lifetime_ordinal: " << stream_lifetime_ordinal
                   << " error: " << std::hex << static_cast<uint32_t>(error);
  }
  