diff --git a/examples/media_test/media_test_app.cc b/examples/media_test/media_test_app.cc
index 5db2771..c3d91ac 100644
--- a/examples/media_test/media_test_app.cc
+++ b/examples/media_test/media_test_app.cc
@@ -62,8 +62,6 @@
       std::cout << "    genre      <none>" << std::endl;
       std::cout << "    composer   <none>" << std::endl << std::endl;
       std::cout << std::endl << std::endl << kUp << std::flush;
-    } else {
-      std::cout << std::endl;
     }
 
     CreateNewMediaTest();
@@ -136,7 +134,7 @@
       // Do nothing.
     } else if (metadata) {
       metadata_shown_ = true;
-      std::cout << "    duration   " << std::setprecision(1)
+      std::cout << "    duration   " << std::fixed << std::setprecision(1)
                 << double(metadata->duration) / ns_per_second << " seconds"
                 << clear_line() << std::endl;
       std::cout << "    title      "
@@ -158,7 +156,7 @@
                 << (metadata->composer ? metadata->composer : "<none>")
                 << clear_line() << std::endl
                 << std::endl;
-    } else {
+    } else if (paint_) {
       std::cout << "    duration   <none>" << kClearLine << std::endl;
       std::cout << "    title      <none>" << kClearLine << std::endl;
       std::cout << "    artist     <none>" << kClearLine << std::endl;
diff --git a/services/media/framework/BUILD.gn b/services/media/framework/BUILD.gn
index 7747bc0..81c4882 100644
--- a/services/media/framework/BUILD.gn
+++ b/services/media/framework/BUILD.gn
@@ -20,6 +20,7 @@
     "metadata.cc",
     "metadata.h",
     "models/active_multistream_sink.h",
+    "models/active_multistream_source.h",
     "models/active_sink.h",
     "models/active_source.h",
     "models/demand.h",
@@ -46,6 +47,8 @@
     "safe_clone.h",
     "stages/active_multistream_sink_stage.cc",
     "stages/active_multistream_sink_stage.h",
+    "stages/active_multistream_source_stage.cc",
+    "stages/active_multistream_source_stage.h",
     "stages/active_sink_stage.cc",
     "stages/active_sink_stage.h",
     "stages/active_source_stage.cc",
@@ -60,6 +63,8 @@
     "stages/stage.h",
     "stages/transform_stage.cc",
     "stages/transform_stage.h",
+    "stages/util.cc",
+    "stages/util.h",
     "stream_type.cc",
     "stream_type.h",
   ]
diff --git a/services/media/framework/graph.h b/services/media/framework/graph.h
index f0b6482..0d89ebc 100644
--- a/services/media/framework/graph.h
+++ b/services/media/framework/graph.h
@@ -9,6 +9,7 @@
 
 #include "services/media/framework/engine.h"
 #include "services/media/framework/refs.h"
+#include "services/media/framework/stages/active_multistream_source_stage.h"
 #include "services/media/framework/stages/active_sink_stage.h"
 #include "services/media/framework/stages/active_source_stage.h"
 #include "services/media/framework/stages/multistream_source_stage.h"
@@ -40,6 +41,7 @@
 DEFINE_STAGE_CREATOR(Transform, TransformStage);
 DEFINE_STAGE_CREATOR(ActiveSource, ActiveSourceStage);
 DEFINE_STAGE_CREATOR(ActiveSink, ActiveSinkStage);
+DEFINE_STAGE_CREATOR(ActiveMultistreamSource, ActiveMultistreamSourceStage);
 
 #undef DEFINE_STAGE_CREATOR
 
diff --git a/services/media/framework/models/active_multistream_source.h b/services/media/framework/models/active_multistream_source.h
new file mode 100644
index 0000000..22b1b11
--- /dev/null
+++ b/services/media/framework/models/active_multistream_source.h
@@ -0,0 +1,38 @@
+// Copyright 2016 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#ifndef MOJO_MEDIA_MODELS_ACTIVE_MULTISTREAM_SOURCE_H_
+#define MOJO_MEDIA_MODELS_ACTIVE_MULTISTREAM_SOURCE_H_
+
+#include "services/media/framework/models/part.h"
+#include "services/media/framework/packet.h"
+
+namespace mojo {
+namespace media {
+
+// Asynchronous source of packets for multiple streams.
+class ActiveMultistreamSource : public Part {
+ public:
+  using SupplyCallback =
+      std::function<void(size_t output_index, PacketPtr packet)>;
+
+  ~ActiveMultistreamSource() override {}
+
+  // TODO(dalesat): Support dynamic output creation.
+
+  // Returns the number of streams the source produces.
+  virtual size_t stream_count() const = 0;
+
+  // Sets the callback that supplies a packet asynchronously.
+  virtual void SetSupplyCallback(const SupplyCallback& supply_callback) = 0;
+
+  // Requests a packet from the source to be supplied asynchronously via
+  // the supply callback.
+  virtual void RequestPacket() = 0;
+};
+
+}  // namespace media
+}  // namespace mojo
+
+#endif  // MOJO_MEDIA_MODELS_ACTIVE_MULTISTREAM_SOURCE_H_
diff --git a/services/media/framework/models/multistream_source.h b/services/media/framework/models/multistream_source.h
index 0782353..9c70808 100644
--- a/services/media/framework/models/multistream_source.h
+++ b/services/media/framework/models/multistream_source.h
@@ -11,8 +11,7 @@
 namespace mojo {
 namespace media {
 
-// Synchronous source of packets for multiple streams. This is currently used
-// by Demux, though it would be better if Demux were asynchronous.
+// Synchronous source of packets for multiple streams.
 class MultistreamSource : public Part {
  public:
   ~MultistreamSource() override {}
diff --git a/services/media/framework/stages/active_multistream_source_stage.cc b/services/media/framework/stages/active_multistream_source_stage.cc
new file mode 100644
index 0000000..a6859bb
--- /dev/null
+++ b/services/media/framework/stages/active_multistream_source_stage.cc
@@ -0,0 +1,117 @@
+// 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/active_multistream_source_stage.h"
+#include "services/media/framework/stages/util.h"
+
+namespace mojo {
+namespace media {
+
+ActiveMultistreamSourceStage::ActiveMultistreamSourceStage(
+    std::shared_ptr<ActiveMultistreamSource> source)
+    : source_(source) {
+  DCHECK(source);
+  outputs_.resize(source->stream_count());
+
+  supply_function_ = [this](size_t output_index, PacketPtr packet) {
+    DCHECK(!cached_packet_) << "source supplied unrequested packet";
+    DCHECK(output_index < outputs_.size());
+    DCHECK(packet);
+
+    cached_packet_output_index_ = output_index;
+    cached_packet_ = std::move(packet);
+
+    if (cached_packet_->end_of_stream()) {
+      ended_streams_++;
+    }
+
+    Output& output = outputs_[cached_packet_output_index_];
+    if (output.demand() != Demand::kNegative) {
+      RequestUpdate();
+    }
+  };
+
+  source_->SetSupplyCallback(supply_function_);
+}
+
+ActiveMultistreamSourceStage::~ActiveMultistreamSourceStage() {}
+
+size_t ActiveMultistreamSourceStage::input_count() const {
+  return 0;
+};
+
+Input& ActiveMultistreamSourceStage::input(size_t index) {
+  CHECK(false) << "input requested from source";
+  abort();
+}
+
+size_t ActiveMultistreamSourceStage::output_count() const {
+  return outputs_.size();
+}
+
+Output& ActiveMultistreamSourceStage::output(size_t index) {
+  DCHECK(index < outputs_.size());
+  return outputs_[index];
+}
+
+PayloadAllocator* ActiveMultistreamSourceStage::PrepareInput(size_t index) {
+  CHECK(false) << "PrepareInput called on source";
+  return nullptr;
+}
+
+void ActiveMultistreamSourceStage::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 ActiveMultistreamSourceStage::UnprepareOutput(
+    size_t index,
+    const UpstreamCallback& callback) {
+  DCHECK(index < outputs_.size());
+  outputs_[index].SetCopyAllocator(nullptr);
+}
+
+void ActiveMultistreamSourceStage::Update(Engine* engine) {
+  DCHECK(engine);
+
+  if (cached_packet_) {
+    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_ && HasPositiveDemand(outputs_)) {
+    // We have no cached packet and positive demand. Request a packet.
+    source_->RequestPacket();
+  }
+}
+
+void ActiveMultistreamSourceStage::FlushInput(
+    size_t index,
+    const DownstreamCallback& callback) {
+  CHECK(false) << "FlushInput called on source";
+}
+
+void ActiveMultistreamSourceStage::FlushOutput(size_t index) {
+  DCHECK(index < outputs_.size());
+  DCHECK(source_);
+  outputs_[index].Flush();
+  cached_packet_.reset(nullptr);
+  cached_packet_output_index_ = 0;
+  ended_streams_ = 0;
+}
+
+}  // namespace media
+}  // namespace mojo
diff --git a/services/media/framework/stages/active_multistream_source_stage.h b/services/media/framework/stages/active_multistream_source_stage.h
new file mode 100644
index 0000000..625c4e3
--- /dev/null
+++ b/services/media/framework/stages/active_multistream_source_stage.h
@@ -0,0 +1,58 @@
+// Copyright 2016 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#ifndef SERVICES_MEDIA_FRAMEWORK_STAGES_ACTIVE_MULTISTREAM_SOURCE_STAGE_H_
+#define SERVICES_MEDIA_FRAMEWORK_STAGES_ACTIVE_MULTISTREAM_SOURCE_STAGE_H_
+
+#include <vector>
+
+#include "services/media/framework/models/active_multistream_source.h"
+#include "services/media/framework/stages/stage.h"
+
+namespace mojo {
+namespace media {
+
+// A stage that hosts an ActiveMultistreamSource.
+class ActiveMultistreamSourceStage : public Stage {
+ public:
+  ActiveMultistreamSourceStage(std::shared_ptr<ActiveMultistreamSource> source);
+
+  ~ActiveMultistreamSourceStage() override;
+
+  // Stage implementation.
+  size_t input_count() const override;
+
+  Input& input(size_t index) override;
+
+  size_t output_count() const override;
+
+  Output& output(size_t index) override;
+
+  PayloadAllocator* PrepareInput(size_t index) override;
+
+  void PrepareOutput(size_t index,
+                     PayloadAllocator* allocator,
+                     const UpstreamCallback& callback) override;
+
+  void UnprepareOutput(size_t index, const UpstreamCallback& callback) override;
+
+  void Update(Engine* engine) override;
+
+  void FlushInput(size_t index, const DownstreamCallback& callback) override;
+
+  void FlushOutput(size_t index) override;
+
+ private:
+  std::vector<Output> outputs_;
+  std::shared_ptr<ActiveMultistreamSource> source_;
+  ActiveMultistreamSource::SupplyCallback supply_function_;
+  PacketPtr cached_packet_;
+  size_t cached_packet_output_index_;
+  size_t ended_streams_ = 0;
+};
+
+}  // namespace media
+}  // namespace mojo
+
+#endif  // SERVICES_MEDIA_FRAMEWORK_STAGES_ACTIVE_MULTISTREAM_SOURCE_STAGE_H_
diff --git a/services/media/framework/stages/active_sink_stage.cc b/services/media/framework/stages/active_sink_stage.cc
index 7c1d90a..1c4f924 100644
--- a/services/media/framework/stages/active_sink_stage.cc
+++ b/services/media/framework/stages/active_sink_stage.cc
@@ -38,7 +38,7 @@
 
 Output& ActiveSinkStage::output(size_t index) {
   CHECK(false) << "output requested from sink";
-  return *(static_cast<Output*>(nullptr));
+  abort();
 }
 
 PayloadAllocator* ActiveSinkStage::PrepareInput(size_t index) {
diff --git a/services/media/framework/stages/active_source_stage.cc b/services/media/framework/stages/active_source_stage.cc
index 1ae91b8..a38200e 100644
--- a/services/media/framework/stages/active_source_stage.cc
+++ b/services/media/framework/stages/active_source_stage.cc
@@ -30,7 +30,7 @@
 
 Input& ActiveSourceStage::input(size_t index) {
   CHECK(false) << "input requested from source";
-  return *(static_cast<Input*>(nullptr));
+  abort();
 }
 
 size_t ActiveSourceStage::output_count() const {
diff --git a/services/media/framework/stages/multistream_source_stage.cc b/services/media/framework/stages/multistream_source_stage.cc
index 65f687f..7f28ba0 100644
--- a/services/media/framework/stages/multistream_source_stage.cc
+++ b/services/media/framework/stages/multistream_source_stage.cc
@@ -3,6 +3,7 @@
 // found in the LICENSE file.
 
 #include "services/media/framework/stages/multistream_source_stage.h"
+#include "services/media/framework/stages/util.h"
 
 namespace mojo {
 namespace media {
@@ -22,7 +23,7 @@
 
 Input& MultistreamSourceStage::input(size_t index) {
   CHECK(false) << "input requested from source";
-  return *(static_cast<Input*>(nullptr));
+  abort();
 }
 
 size_t MultistreamSourceStage::output_count() const {
@@ -61,16 +62,8 @@
 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) {
+    if (cached_packet_ && HasPositiveDemand(outputs_)) {
       DCHECK(cached_packet_output_index_ < outputs_.size());
       Output& output = outputs_[cached_packet_output_index_];
 
diff --git a/services/media/framework/stages/util.cc b/services/media/framework/stages/util.cc
new file mode 100644
index 0000000..0b3393a
--- /dev/null
+++ b/services/media/framework/stages/util.cc
@@ -0,0 +1,21 @@
+// 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/util.h"
+
+namespace mojo {
+namespace media {
+
+bool HasPositiveDemand(const std::vector<Output>& outputs) {
+  for (const Output& output : outputs) {
+    if (output.demand() == Demand::kPositive) {
+      return true;
+    }
+  }
+
+  return false;
+}
+
+}  // namespace media
+}  // namespace mojo
diff --git a/services/media/framework/stages/util.h b/services/media/framework/stages/util.h
new file mode 100644
index 0000000..d04b2df
--- /dev/null
+++ b/services/media/framework/stages/util.h
@@ -0,0 +1,20 @@
+// Copyright 2016 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#ifndef SERVICES_MEDIA_FRAMEWORK_STAGES_UTIL_H_
+#define SERVICES_MEDIA_FRAMEWORK_STAGES_UTIL_H_
+
+#include <vector>
+
+#include "services/media/framework/stages/stage.h"
+
+namespace mojo {
+namespace media {
+
+bool HasPositiveDemand(const std::vector<Output>& outputs);
+
+}  // namespace media
+}  // namespace mojo
+
+#endif  // SERVICES_MEDIA_FRAMEWORK_STAGES_UTIL_H_
diff --git a/services/media/framework_ffmpeg/ffmpeg_decoder_base.cc b/services/media/framework_ffmpeg/ffmpeg_decoder_base.cc
index 2c37acd..ce03f96 100644
--- a/services/media/framework_ffmpeg/ffmpeg_decoder_base.cc
+++ b/services/media/framework_ffmpeg/ffmpeg_decoder_base.cc
@@ -11,7 +11,7 @@
 
 FfmpegDecoderBase::FfmpegDecoderBase(AvCodecContextPtr av_codec_context)
     : av_codec_context_(std::move(av_codec_context)),
-      av_frame_ptr_(av_frame_alloc()) {
+      av_frame_ptr_(ffmpeg::AvFrame::Create()) {
   DCHECK(av_codec_context_);
 }
 
diff --git a/services/media/framework_mojo/mojo_producer.h b/services/media/framework_mojo/mojo_producer.h
index ba37aab..16865d1 100644
--- a/services/media/framework_mojo/mojo_producer.h
+++ b/services/media/framework_mojo/mojo_producer.h
@@ -7,7 +7,6 @@
 
 #include "base/single_thread_task_runner.h"
 #include "base/synchronization/lock.h"
-#include "base/task_runner.h"
 #include "mojo/common/binding_set.h"
 #include "mojo/services/media/common/interfaces/media_state.mojom.h"
 #include "mojo/services/media/common/interfaces/media_transport.mojom.h"