Motown: Remove LPCM optimizations, fix prepare, add flush, add ActiveMultistreamSink model/stage

services/media/framework/stages/multistream_source_stage.cc

Index: services/media/framework/stages/multistream_source_stage.cc
diff --git a/services/media/framework/stages/multistream_source_stage.cc b/services/media/framework/stages/multistream_source_stage.cc
new file mode 100644
index 0000000000000000000000000000000000000000..4de36759f854958a077c075ed0560e7af2dd2f56
--- /dev/null
+++ b/services/media/framework/stages/multistream_source_stage.cc
@@ -0,0 +1,124 @@
+// Copyright 2016 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#include "services/media/framework/stages/multistream_source_stage.h"
+
+namespace mojo {
+namespace media {
+
+MultistreamSourceStage::MultistreamSourceStage(
+    std::shared_ptr<MultistreamSource> source) :
+    source_(source),
+    ended_streams_(0) {
+  DCHECK(source);
+  outputs_.resize(source->stream_count());
+}
+
+MultistreamSourceStage::~MultistreamSourceStage() {}
+
+size_t MultistreamSourceStage::input_count() const {
+  return 0;
+};
+
+Input& MultistreamSourceStage::input(size_t index) {
+  LOG(ERROR) << "input requested from source";
+  return *(static_cast<Input*>(nullptr));

[johngro, 2016/02/09 21:00:36]

This needs to CHECK(false).  The only reason to ever do this trick is to fool
the compiler into thinking that you would have returned something valid when, in
fact, the code will never ever execute.  Also, note CHECK and not DCHECK. 
DCHECK only runs in debug builds while CHECK runs in all builds.  IIRC,
NOTREACHED is a DCHECK, so is probably not appropriate here.

This needs to be done wherever else you have followed this pattern.

[jamesr, 2016/02/09 21:15:56]

+100 to this. This line as written is undefined behavior in C++ which means the
compiler is free to do literally anything within this function and do other fun
things like optimize callers assuming that they can never call this function
(since if they could, the program would be undefined) which can really ruin your
day.

[dalesat, 2016/02/10 21:37:24]

Done.

+}
+
+size_t MultistreamSourceStage::output_count() const {
+  return outputs_.size();
+}
+
+Output& MultistreamSourceStage::output(size_t index) {
+  DCHECK(index < outputs_.size());
+  return outputs_[index];
+}
+
+PayloadAllocator* MultistreamSourceStage::PrepareInput(size_t index) {
+  LOG(ERROR) << "PrepareInput called on source";
+  return nullptr;
+}
+
+void MultistreamSourceStage::PrepareOutput(
+    size_t index,
+    PayloadAllocator* allocator,
+    const UpstreamCallback& callback) {
+  DCHECK(index < outputs_.size());
+
+  if (allocator != nullptr) {
+    // Currently, we don't support a source that uses provided allocators. If
+    // we're provided an allocator, the output must have it so supplied packets
+    // can be copied.
+    outputs_[index].SetCopyAllocator(allocator);
+  }
+}
+
+void MultistreamSourceStage::UnprepareOutput(
+    size_t index,
+    const UpstreamCallback& callback) {
+  DCHECK(index < outputs_.size());
+  outputs_[index].SetCopyAllocator(nullptr);
+}
+
+void MultistreamSourceStage::Update(Engine* engine) {
+  DCHECK(engine);
+
+  bool has_positive_demand = false;
+  for (Output& output : outputs_) {
+    if (output.demand() == Demand::kPositive) {
+      has_positive_demand = true;
+      break;
+    }
+  }
+
+  while (true) {
+    if (cached_packet_ && has_positive_demand) {
+      DCHECK(cached_packet_output_index_ < outputs_.size());
+      Output& output = outputs_[cached_packet_output_index_];
+
+      if (output.demand() != Demand::kNegative) {
+        // cached_packet_ is intended for an output which will accept packets.
+        output.SupplyPacket(std::move(cached_packet_), engine);
+      }
+    }
+
+    if (cached_packet_) {
+      // There's still a cached packet. We're done for now.
+      return;
+    }
+
+    if (ended_streams_ == outputs_.size()) {
+      // We've seen end-of-stream for all streams. All done.
+      return;
+    }
+
+    // Pull a packet from the source.
+    cached_packet_ = source_->PullPacket(&cached_packet_output_index_);
+    DCHECK(cached_packet_);
+    DCHECK(cached_packet_output_index_ < outputs_.size());
+
+    if (cached_packet_->end_of_stream()) {
+      ended_streams_++;
+    }
+  }
+}
+
+void MultistreamSourceStage::FlushInput(
+    size_t index,
+    const DownstreamCallback& callback) {
+  LOG(ERROR) << "FlushInput called on source";
+}
+
+void MultistreamSourceStage::FlushOutput(size_t index) {
+  DCHECK(index < outputs_.size());
+  DCHECK(source_);
+  outputs_[index].Flush();
+  source_->Flush();
+  cached_packet_.reset(nullptr);
+  cached_packet_output_index_ = 0;
+  ended_streams_ = 0;
+}
+
+}  // namespace media
+}  // namespace mojo