diff --git a/garnet/packages/tests/BUILD.gn b/garnet/packages/tests/BUILD.gn
index 447fcd5..04178e7 100644
--- a/garnet/packages/tests/BUILD.gn
+++ b/garnet/packages/tests/BUILD.gn
@@ -784,9 +784,8 @@
     ":virtual_audio",
     "//garnet/bin/media/codecs:tests",
     "//garnet/public/lib/media/timeline:media_lib_timeline_tests",
-    "//src/media/audio:tests",
+    "//src/media:tests",
     "//src/media/playback/mediaplayer:tests",
-    "//src/media/sessions:tests",
   ]
 }
 
diff --git a/src/media/BUILD.gn b/src/media/BUILD.gn
index e80129f..c22e97b 100644
--- a/src/media/BUILD.gn
+++ b/src/media/BUILD.gn
@@ -7,6 +7,7 @@
   deps = [
     ":tests",
     "audio",
+    "codecs",
     "lib",
     "playback",
     "sessions",
@@ -19,5 +20,6 @@
   data_deps = [
     "audio:tests",
     "sessions:tests",
+    "codecs:tests",
   ]
 }
diff --git a/src/media/codecs/BUILD.gn b/src/media/codecs/BUILD.gn
new file mode 100644
index 0000000..5f6b40d
--- /dev/null
+++ b/src/media/codecs/BUILD.gn
@@ -0,0 +1,14 @@
+# Copyright 2019 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.
+
+group("codecs") {
+  deps = []
+}
+
+group("tests") {
+  testonly = true
+  deps = [
+    "stress",
+  ]
+}
diff --git a/src/media/codecs/stress/BUILD.gn b/src/media/codecs/stress/BUILD.gn
new file mode 100644
index 0000000..bb3f4e4
--- /dev/null
+++ b/src/media/codecs/stress/BUILD.gn
@@ -0,0 +1,67 @@
+# Copyright 2019 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.
+
+import("//build/config.gni")
+import("//build/rust/rustc_library.gni")
+import("//build/rust/rustc_macro.gni")
+import("//build/test/test_package.gni")
+import("//build/testing/environments.gni")
+
+rustc_library("lib") {
+  name = "codec_stress"
+  edition = "2018"
+  with_unit_tests = true
+
+  deps = [
+    "//garnet/public/lib/fidl/rust/fidl",
+    "//garnet/public/rust/fuchsia-async",
+    "//garnet/public/rust/fuchsia-component",
+    "//garnet/public/rust/fuchsia-syslog",
+    "//garnet/public/rust/fuchsia-zircon",
+    "//sdk/fidl/fuchsia.media:fuchsia.media-rustc",
+    "//sdk/fidl/fuchsia.mediacodec:fuchsia.mediacodec-rustc",
+    "//src/lib/fidl_table_validation",
+    "//third_party/rust_crates:failure",
+    "//third_party/rust_crates:futures-preview",
+    "//third_party/rust_crates:lazy_static",
+    "//third_party/rust_crates:log",
+    "//third_party/rust_crates:parking_lot",
+    "//zircon/public/fidl/fuchsia-sysmem:fuchsia-sysmem-rustc",
+  ]
+}
+
+test_package("codec_stress_tests") {
+  deps = [
+    ":lib",
+  ]
+
+  meta = [
+    {
+      path = rebase_path("meta/codec_stress_tests.cmx")
+      dest = "codec_stress_tests.cmx"
+    },
+  ]
+
+  tests = [
+    {
+      name = "codec_stress_lib_test"
+      environments = basic_envs
+    },
+  ]
+
+  resources = [
+    {
+      path = rebase_path(
+              "../../../../garnet/test_data/media/third_party/chromium_media_test_data/bear.h264")
+      dest = "bear.h264"
+    },
+  ]
+}
+
+group("stress") {
+  testonly = true
+  deps = [
+    ":codec_stress_tests",
+  ]
+}
diff --git a/src/media/codecs/stress/meta/codec_stress_lib_test.cmx b/src/media/codecs/stress/meta/codec_stress_lib_test.cmx
new file mode 100644
index 0000000..f97e33b
--- /dev/null
+++ b/src/media/codecs/stress/meta/codec_stress_lib_test.cmx
@@ -0,0 +1,26 @@
+{
+    "facets": {
+        "fuchsia.test": {
+            "injected-services": {
+                "fuchsia.mediacodec.CodecFactory": "fuchsia-pkg://fuchsia.com/codec_factory#meta/codec_factory.cmx",
+                "fuchsia.sysmem.Allocator": "fuchsia-pkg://fuchsia.com/sysmem_connector#meta/sysmem_connector.cmx"
+            }
+        }
+    },
+    "program": {
+        "binary": "test/codec_stress_lib_test"
+    },
+    "sandbox": {
+        "dev": [
+            "class/media-codec",
+            "class/sysmem"
+        ],
+        "features": [],
+        "services": [
+            "fuchsia.mediacodec.CodecFactory",
+            "fuchsia.tracelink.Registry",
+            "fuchsia.sysmem.Allocator",
+            "fuchsia.logger.LogSink"
+        ]
+    }
+}
diff --git a/src/media/codecs/stress/meta/codec_stress_tests.cmx b/src/media/codecs/stress/meta/codec_stress_tests.cmx
new file mode 100644
index 0000000..79b6d89
--- /dev/null
+++ b/src/media/codecs/stress/meta/codec_stress_tests.cmx
@@ -0,0 +1,9 @@
+{
+    "program": {
+        "binary": "bin/app"
+    },
+    "sandbox": {
+        "features": [],
+        "services": []
+    }
+}
diff --git a/src/media/codecs/stress/src/buffer_collection_constraints.rs b/src/media/codecs/stress/src/buffer_collection_constraints.rs
new file mode 100644
index 0000000..e5b2f4f
--- /dev/null
+++ b/src/media/codecs/stress/src/buffer_collection_constraints.rs
@@ -0,0 +1,57 @@
+// Copyright 2019 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.
+
+use fidl_fuchsia_sysmem::*;
+
+pub const IMAGE_FORMAT_CONSTRAINTS_DEFAULT: ImageFormatConstraints = ImageFormatConstraints {
+    pixel_format: PixelFormat {
+        type_: PixelFormatType::Nv12,
+        has_format_modifier: false,
+        format_modifier: FormatModifier { value: 0 },
+    },
+    color_spaces_count: 0,
+    color_space: [ColorSpace { type_: ColorSpaceType::Invalid }; 32],
+    min_coded_width: 0,
+    max_coded_width: 0,
+    min_coded_height: 0,
+    max_coded_height: 0,
+    min_bytes_per_row: 0,
+    max_bytes_per_row: 0,
+    max_coded_width_times_coded_height: 0,
+    layers: 0,
+    coded_width_divisor: 0,
+    coded_height_divisor: 0,
+    bytes_per_row_divisor: 0,
+    start_offset_divisor: 0,
+    display_width_divisor: 0,
+    display_height_divisor: 0,
+    required_min_coded_width: 0,
+    required_max_coded_width: 0,
+    required_min_coded_height: 0,
+    required_max_coded_height: 0,
+    required_min_bytes_per_row: 0,
+    required_max_bytes_per_row: 0,
+};
+
+pub const BUFFER_COLLECTION_CONSTRAINTS_DEFAULT: BufferCollectionConstraints =
+    BufferCollectionConstraints {
+        usage: BufferUsage { cpu: 1, vulkan: 0, display: 0, video: 1 },
+        min_buffer_count_for_camping: 0,
+        min_buffer_count_for_dedicated_slack: 0,
+        min_buffer_count_for_shared_slack: 0,
+        min_buffer_count: 0,
+        max_buffer_count: 0,
+        has_buffer_memory_constraints: false,
+        buffer_memory_constraints: BufferMemoryConstraints {
+            min_size_bytes: 0,
+            max_size_bytes: std::u32::MAX,
+            physically_contiguous_required: false,
+            secure_required: false,
+            secure_permitted: false,
+            ram_domain_supported: false,
+            cpu_domain_supported: true,
+        },
+        image_format_constraints_count: 0,
+        image_format_constraints: [IMAGE_FORMAT_CONSTRAINTS_DEFAULT; 32],
+    };
diff --git a/src/media/codecs/stress/src/buffer_set.rs b/src/media/codecs/stress/src/buffer_set.rs
new file mode 100644
index 0000000..5864348
--- /dev/null
+++ b/src/media/codecs/stress/src/buffer_set.rs
@@ -0,0 +1,349 @@
+// Copyright 2019 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.
+
+//! Handles negotiating buffer sets with the codec server and sysmem.
+
+use crate::{buffer_collection_constraints::*, Result};
+use failure::{self, Fail, ResultExt};
+use fidl::endpoints::{create_endpoints, ClientEnd};
+use fidl_fuchsia_media::*;
+use fidl_fuchsia_sysmem::*;
+use fidl_table_validation::{ValidFidlTable, Validate};
+use fuchsia_component::client;
+use fuchsia_zircon as zx;
+use std::{
+    convert::TryFrom,
+    fmt,
+    iter::{IntoIterator, StepBy},
+    ops::RangeFrom,
+};
+
+#[derive(Debug)]
+pub enum Error {
+    ReclaimClientTokenChannel,
+    ServerOmittedBufferVmo,
+    PacketReferencesInvalidBuffer,
+    VmoReadFail(zx::Status),
+}
+
+impl fmt::Display for Error {
+    fn fmt(&self, w: &mut fmt::Formatter) -> fmt::Result {
+        fmt::Debug::fmt(&self, w)
+    }
+}
+
+impl Fail for Error {}
+
+#[allow(unused)]
+#[derive(ValidFidlTable, Copy, Clone, Debug, PartialEq)]
+#[fidl_table_src(StreamBufferSettings)]
+pub struct ValidStreamBufferSettings {
+    buffer_lifetime_ordinal: u64,
+    buffer_constraints_version_ordinal: u64,
+    packet_count_for_server: u32,
+    packet_count_for_client: u32,
+    per_packet_buffer_bytes: u32,
+    #[fidl_field_type(default = false)]
+    single_buffer_mode: bool,
+}
+
+#[allow(unused)]
+#[derive(ValidFidlTable)]
+#[fidl_table_src(StreamBufferConstraints)]
+#[fidl_table_validator(StreamBufferConstraintsValidator)]
+pub struct ValidStreamBufferConstraints {
+    buffer_constraints_version_ordinal: u64,
+    default_settings: ValidStreamBufferSettings,
+    per_packet_buffer_bytes_min: u32,
+    per_packet_buffer_bytes_recommended: u32,
+    per_packet_buffer_bytes_max: u32,
+    packet_count_for_server_min: u32,
+    packet_count_for_server_recommended: u32,
+    packet_count_for_server_recommended_max: u32,
+    packet_count_for_server_max: u32,
+    packet_count_for_client_min: u32,
+    packet_count_for_client_max: u32,
+    single_buffer_mode_allowed: bool,
+    #[fidl_field_type(default = false)]
+    is_physically_contiguous_required: bool,
+    #[fidl_field_type(optional)]
+    very_temp_kludge_bti_handle: Option<zx::Handle>,
+}
+
+#[derive(ValidFidlTable)]
+#[fidl_table_src(StreamOutputConstraints)]
+pub struct ValidStreamOutputConstraints {
+    pub stream_lifetime_ordinal: u64,
+    pub buffer_constraints_action_required: bool,
+    pub buffer_constraints: ValidStreamBufferConstraints,
+}
+
+pub struct StreamBufferConstraintsValidator;
+
+#[derive(Debug)]
+pub enum StreamBufferConstraintsError {
+    VersionOrdinalZero,
+    SingleBufferMode,
+    ConstraintsNoBtiHandleForPhysicalBuffers,
+}
+
+impl Validate<ValidStreamBufferConstraints> for StreamBufferConstraintsValidator {
+    type Error = StreamBufferConstraintsError;
+    fn validate(candidate: &ValidStreamBufferConstraints) -> std::result::Result<(), Self::Error> {
+        if candidate.buffer_constraints_version_ordinal == 0 {
+            // An ordinal of 0 in StreamBufferConstraints is not allowed.
+            return Err(StreamBufferConstraintsError::VersionOrdinalZero);
+        }
+
+        if candidate.default_settings.single_buffer_mode {
+            // StreamBufferConstraints should never suggest single buffer mode.
+            return Err(StreamBufferConstraintsError::SingleBufferMode);
+        }
+
+        if candidate.is_physically_contiguous_required
+            && candidate
+                .very_temp_kludge_bti_handle
+                .as_ref()
+                .map(|h| h.is_invalid())
+                .unwrap_or(true)
+        {
+            // The bti handle must be provided if the buffers need to be physically contiguous.
+            return Err(StreamBufferConstraintsError::ConstraintsNoBtiHandleForPhysicalBuffers);
+        }
+
+        Ok(())
+    }
+}
+
+/// The pattern to use when advancing ordinals.
+#[derive(Debug, Clone, Copy)]
+pub enum OrdinalPattern {
+    /// Odd ordinal pattern starts at 1 and moves in increments of 2: [1,3,5..]
+    Odd,
+    /// All ordinal pattern starts at 1 and moves in increments of 1: [1,2,3..]
+    All,
+}
+
+impl IntoIterator for OrdinalPattern {
+    type Item = u64;
+    type IntoIter = StepBy<RangeFrom<Self::Item>>;
+    fn into_iter(self) -> Self::IntoIter {
+        let (start, step) = match self {
+            OrdinalPattern::Odd => (1, 2),
+            OrdinalPattern::All => (1, 1),
+        };
+        (start..).step_by(step)
+    }
+}
+
+pub fn get_ordinal(pattern: &mut <OrdinalPattern as IntoIterator>::IntoIter) -> u64 {
+    pattern.next().expect("Getting next item in infinite pattern")
+}
+
+pub enum BufferSetType {
+    Input,
+    Output,
+}
+
+pub struct BufferSetFactory;
+
+impl BufferSetFactory {
+    pub async fn buffer_set(
+        buffer_lifetime_ordinal: u64,
+        constraints: ValidStreamBufferConstraints,
+        codec: &mut StreamProcessorProxy,
+        buffer_set_type: BufferSetType,
+        buffer_collection_constraints: Option<BufferCollectionConstraints>,
+    ) -> Result<BufferSet> {
+        let (collection_client, settings) = await!(Self::settings(
+            buffer_lifetime_ordinal,
+            constraints,
+            buffer_collection_constraints
+        ))?;
+
+        vlog!(2, "Got settings; waiting for buffers.");
+
+        match buffer_set_type {
+            BufferSetType::Input => codec
+                .set_input_buffer_partial_settings(settings)
+                .context("Sending input partial settings to codec")?,
+            BufferSetType::Output => codec
+                .set_output_buffer_partial_settings(settings)
+                .context("Sending output partial settings to codec")?,
+        };
+
+        let (status, collection_info) = await!(collection_client.wait_for_buffers_allocated())
+            .context("Waiting for buffers")?;
+        vlog!(2, "Sysmem responded: {:?}", status);
+        let collection_info = zx::Status::ok(status).map(|_| collection_info)?;
+
+        if let BufferSetType::Output = buffer_set_type {
+            vlog!(2, "Completing settings for output.");
+            codec.complete_output_buffer_partial_settings(buffer_lifetime_ordinal)?;
+        }
+
+        //collection_client.close()?;
+
+        vlog!(
+            2,
+            "Got {} buffers of size {:?}",
+            collection_info.buffer_count,
+            collection_info.settings.buffer_settings.size_bytes
+        );
+        vlog!(3, "Buffer collection is: {:#?}", collection_info.settings);
+        for (i, buffer) in collection_info.buffers.iter().enumerate() {
+            // We enumerate beyond collection_info.buffer_count just for debugging
+            // purposes at this log level.
+            vlog!(3, "Buffer {} is : {:#?}", i, buffer);
+        }
+
+        Ok(BufferSet::try_from(BufferSetSpec {
+            proxy: collection_client,
+            buffer_lifetime_ordinal,
+            collection_info,
+        })?)
+    }
+
+    async fn settings(
+        buffer_lifetime_ordinal: u64,
+        constraints: ValidStreamBufferConstraints,
+        buffer_collection_constraints: Option<BufferCollectionConstraints>,
+    ) -> Result<(BufferCollectionProxy, StreamBufferPartialSettings)> {
+        let (client_token, client_token_request) =
+            create_endpoints::<BufferCollectionTokenMarker>()?;
+        let (codec_token, codec_token_request) = create_endpoints::<BufferCollectionTokenMarker>()?;
+        let client_token = client_token.into_proxy()?;
+
+        let sysmem_client =
+            client::connect_to_service::<AllocatorMarker>().context("Connecting to sysmem")?;
+
+        sysmem_client
+            .allocate_shared_collection(client_token_request)
+            .context("Allocating shared collection")?;
+        client_token.duplicate(std::u32::MAX, codec_token_request)?;
+
+        let (collection_client, collection_request) = create_endpoints::<BufferCollectionMarker>()?;
+        sysmem_client.bind_shared_collection(
+            ClientEnd::new(
+                client_token
+                    .into_channel()
+                    .map_err(|_| Error::ReclaimClientTokenChannel)?
+                    .into_zx_channel(),
+            ),
+            collection_request,
+        )?;
+        let collection_client = collection_client.into_proxy()?;
+        await!(collection_client.sync()).context("Syncing codec_token_request with sysmem")?;
+
+        let mut collection_constraints =
+            buffer_collection_constraints.unwrap_or(BUFFER_COLLECTION_CONSTRAINTS_DEFAULT);
+        assert_eq!(
+            collection_constraints.min_buffer_count_for_camping, 0,
+            "Codecs assert that buffer_count == packet count, so we can't change this yet."
+        );
+        collection_constraints.min_buffer_count_for_camping =
+            constraints.default_settings.packet_count_for_client;
+
+        vlog!(3, "Our buffer collection constraints are: {:#?}", collection_constraints);
+
+        // By design we must say true even if all our fields are left at
+        // default, or sysmem will not give us buffer handles.
+        let has_constraints = true;
+        collection_client
+            .set_constraints(has_constraints, &mut collection_constraints)
+            .context("Sending buffer constraints to sysmem")?;
+
+        Ok((
+            collection_client,
+            StreamBufferPartialSettings {
+                buffer_lifetime_ordinal: Some(buffer_lifetime_ordinal),
+                buffer_constraints_version_ordinal: Some(
+                    constraints.buffer_constraints_version_ordinal,
+                ),
+                single_buffer_mode: Some(constraints.default_settings.single_buffer_mode),
+                packet_count_for_server: Some(constraints.default_settings.packet_count_for_server),
+                packet_count_for_client: Some(constraints.default_settings.packet_count_for_client),
+                sysmem_token: Some(codec_token),
+            },
+        ))
+    }
+}
+
+#[derive(ValidFidlTable, Clone, Copy, Debug, PartialEq)]
+#[fidl_table_src(PacketHeader)]
+pub struct ValidPacketHeader {
+    pub buffer_lifetime_ordinal: u64,
+    pub packet_index: u32,
+}
+
+#[derive(ValidFidlTable, Clone, Copy, Debug, PartialEq)]
+#[fidl_table_src(Packet)]
+pub struct ValidPacket {
+    pub header: ValidPacketHeader,
+    pub buffer_index: u32,
+    pub stream_lifetime_ordinal: u64,
+    pub start_offset: u32,
+    pub valid_length_bytes: u32,
+    #[fidl_field_type(optional)]
+    pub timestamp_ish: Option<u64>,
+    #[fidl_field_type(default = false)]
+    pub start_access_unit: bool,
+    #[fidl_field_type(default = false)]
+    pub known_end_access_unit: bool,
+}
+
+struct BufferSetSpec {
+    proxy: BufferCollectionProxy,
+    buffer_lifetime_ordinal: u64,
+    collection_info: BufferCollectionInfo2,
+}
+
+#[derive(Debug, PartialEq)]
+pub struct Buffer {
+    pub data: zx::Vmo,
+    pub start: u64,
+    pub size: u64,
+}
+
+#[derive(Debug)]
+pub struct BufferSet {
+    pub proxy: BufferCollectionProxy,
+    pub buffers: Vec<Buffer>,
+    pub buffer_lifetime_ordinal: u64,
+}
+
+impl TryFrom<BufferSetSpec> for BufferSet {
+    type Error = failure::Error;
+    fn try_from(mut src: BufferSetSpec) -> std::result::Result<Self, Self::Error> {
+        let mut buffers = vec![];
+        for (i, buffer) in src.collection_info.buffers
+            [0..(src.collection_info.buffer_count as usize)]
+            .iter_mut()
+            .enumerate()
+        {
+            buffers.push(Buffer {
+                data: buffer.vmo.take().ok_or(Error::ServerOmittedBufferVmo).context(format!(
+                    "Trying to ingest {}th buffer of {}: {:#?}",
+                    i, src.collection_info.buffer_count, buffer
+                ))?,
+                start: buffer.vmo_usable_start,
+                size: src.collection_info.settings.buffer_settings.size_bytes as u64,
+            });
+        }
+
+        Ok(Self { proxy: src.proxy, buffers, buffer_lifetime_ordinal: src.buffer_lifetime_ordinal })
+    }
+}
+
+impl BufferSet {
+    pub fn read_packet(&self, packet: &ValidPacket) -> Result<Vec<u8>> {
+        let buffer = self
+            .buffers
+            .get(packet.buffer_index as usize)
+            .ok_or(Error::PacketReferencesInvalidBuffer)?;
+        let mut dest = vec![0; packet.valid_length_bytes as usize];
+        buffer.data.read(&mut dest, packet.start_offset as u64).map_err(Error::VmoReadFail)?;
+        Ok(dest)
+    }
+}
diff --git a/src/media/codecs/stress/src/codecs.rs b/src/media/codecs/stress/src/codecs.rs
new file mode 100644
index 0000000..0e7170d
--- /dev/null
+++ b/src/media/codecs/stress/src/codecs.rs
@@ -0,0 +1,5 @@
+// Copyright 2019 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.
+
+pub mod h264;
diff --git a/src/media/codecs/stress/src/codecs/h264.rs b/src/media/codecs/stress/src/codecs/h264.rs
new file mode 100644
index 0000000..69a5caf
--- /dev/null
+++ b/src/media/codecs/stress/src/codecs/h264.rs
@@ -0,0 +1,134 @@
+// Copyright 2019 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.
+
+use crate::{elementary_stream::*, Result};
+use fidl_fuchsia_media::FormatDetails;
+use std::{fs, path::Path};
+
+pub const BEAR_TEST_FILE: &str = "/pkg/data/bear.h264";
+
+/// Represents an H264 elementary stream.
+pub struct H264Stream {
+    data: Vec<u8>,
+}
+
+impl H264Stream {
+    /// Constructs an H264 elementary stream from a file with raw elementary stream data.
+    pub fn from_file(filename: impl AsRef<Path>) -> Result<Self> {
+        Ok(Self { data: fs::read(filename)? })
+    }
+
+    /// Returns an iterator over H264 NALs that does not copy.
+    fn nal_iter(&self) -> impl Iterator<Item = H264Nal> {
+        H264NalIter { data: &self.data, pos: 0 }
+    }
+}
+
+impl ElementaryStream for H264Stream {
+    fn format_details(&self, version_ordinal: u64) -> FormatDetails {
+        FormatDetails {
+            format_details_version_ordinal: Some(version_ordinal),
+            mime_type: Some(String::from("video/h264")),
+            oob_bytes: None,
+            domain: None,
+            pass_through_parameters: None,
+        }
+    }
+
+    fn is_access_units(&self) -> bool {
+        true
+    }
+
+    fn stream<'a>(&'a self) -> Box<dyn Iterator<Item = ElementaryStreamChunk> + 'a> {
+        Box::new(self.nal_iter().map(|nal| ElementaryStreamChunk {
+            start_access_unit: true,
+            known_end_access_unit: true,
+            data: nal.data,
+            significance: match nal.kind {
+                H264NalKind::Picture => Significance::Video(VideoSignificance::Picture),
+                H264NalKind::NotPicture => Significance::Video(VideoSignificance::NotPicture),
+            },
+            timestamp: None,
+        }))
+    }
+}
+
+pub struct H264Nal<'a> {
+    pub kind: H264NalKind,
+    pub data: &'a [u8],
+}
+
+pub enum H264NalKind {
+    Picture,
+    NotPicture,
+}
+
+impl H264NalKind {
+    const NON_IDR_PICTURE_CODE: u8 = 1;
+    const IDR_PICTURE_CODE: u8 = 5;
+
+    fn from_header(header: u8) -> Self {
+        let kind = header & 0xf;
+        if kind == Self::NON_IDR_PICTURE_CODE || kind == Self::IDR_PICTURE_CODE {
+            H264NalKind::Picture
+        } else {
+            H264NalKind::NotPicture
+        }
+    }
+}
+
+struct H264NalStart<'a> {
+    /// Position in the h264 stream of the start.
+    pos: usize,
+    /// All the data from the start of the NAL onward.
+    data: &'a [u8],
+    kind: H264NalKind,
+}
+
+/// An iterator over NALs in an H264 stream.
+struct H264NalIter<'a> {
+    data: &'a [u8],
+    pos: usize,
+}
+
+impl<'a> H264NalIter<'a> {
+    fn next_nal(&self, pos: usize) -> Option<H264Nal<'a>> {
+        // This won't need to search if pos already at a start code.
+        let nal_start = self.next_nal_start(pos)?;
+        // We search 3 bytes after the found nal's start, because that will
+        // ensure we don't just find the same start code again.
+        match self.next_nal_start(nal_start.pos + 3) {
+            Some(next_start) => Some(H264Nal {
+                kind: nal_start.kind,
+                data: &nal_start.data[0..(next_start.pos - nal_start.pos)],
+            }),
+            None => Some(H264Nal { kind: nal_start.kind, data: nal_start.data }),
+        }
+    }
+
+    fn next_nal_start(&self, pos: usize) -> Option<H264NalStart<'a>> {
+        // This search size will find 3 and 4 byte start codes, and the
+        // header value.
+        const NAL_SEARCH_SIZE: usize = 5;
+
+        let data = self.data.get(pos..)?;
+        data.windows(NAL_SEARCH_SIZE).enumerate().find_map(|(i, candidate)| match candidate {
+            [0, 0, 0, 1, h] | [0, 0, 1, h, _] => Some(H264NalStart {
+                pos: i + pos,
+                data: data.get(i..).expect("Getting slice starting where we just matched"),
+                kind: H264NalKind::from_header(*h),
+            }),
+            _ => None,
+        })
+    }
+}
+
+impl<'a> Iterator for H264NalIter<'a> {
+    type Item = H264Nal<'a>;
+    fn next(&mut self) -> Option<Self::Item> {
+        let nal = self.next_nal(self.pos);
+        self.pos += nal.as_ref().map(|n| n.data.len()).unwrap_or(0);
+        nal
+    }
+}
diff --git a/src/media/codecs/stress/src/elementary_stream.rs b/src/media/codecs/stress/src/elementary_stream.rs
new file mode 100644
index 0000000..906b130
--- /dev/null
+++ b/src/media/codecs/stress/src/elementary_stream.rs
@@ -0,0 +1,78 @@
+// Copyright 2019 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.
+
+use fidl_fuchsia_media::FormatDetails;
+
+pub trait ElementaryStream {
+    fn format_details(&self, version_ordinal: u64) -> FormatDetails;
+
+    /// Whether _all_ chunks in the elementary stream will be on access unit boundaries. These are
+    /// units for decoder input (e.g. in H264, NALs). When input is not in access units, the server
+    /// must parse and/or buffer the bitstream.
+    fn is_access_units(&self) -> bool;
+
+    fn stream<'a>(&'a self) -> Box<dyn Iterator<Item = ElementaryStreamChunk> + 'a>;
+
+    fn video_frame_count(&self) -> usize {
+        self.stream()
+            .filter(|chunk| match chunk.significance {
+                Significance::Video(VideoSignificance::Picture) => true,
+                _ => false,
+            })
+            .count()
+    }
+}
+
+#[derive(Copy, Clone, Debug)]
+pub struct ElementaryStreamChunk<'a> {
+    pub start_access_unit: bool,
+    pub known_end_access_unit: bool,
+    pub data: &'a [u8],
+    pub significance: Significance,
+    pub timestamp: Option<u64>,
+}
+
+#[derive(Copy, Clone, Debug)]
+pub enum Significance {
+    Video(VideoSignificance),
+}
+
+#[derive(Copy, Clone, Debug)]
+pub enum VideoSignificance {
+    Picture,
+    NotPicture,
+}
+
+/// Wraps an elementary stream and adds sequential dummy timestamps to its chunks.
+pub struct TimestampedStream<S, I> {
+    pub source: S,
+    pub timestamps: I,
+}
+
+impl<S, I> ElementaryStream for TimestampedStream<S, I>
+where
+    S: ElementaryStream,
+    I: Iterator<Item = u64> + Clone,
+{
+    fn format_details(&self, version_ordinal: u64) -> FormatDetails {
+        self.source.format_details(version_ordinal)
+    }
+
+    fn is_access_units(&self) -> bool {
+        self.source.is_access_units()
+    }
+
+    fn stream<'a>(&'a self) -> Box<dyn Iterator<Item = ElementaryStreamChunk> + 'a> {
+        let mut timestamps = self.timestamps.clone();
+        Box::new(self.source.stream().map(move |mut chunk| {
+            match chunk.significance {
+                Significance::Video(VideoSignificance::Picture) => {
+                    chunk.timestamp = timestamps.next();
+                }
+                _ => {}
+            };
+            chunk
+        }))
+    }
+}
diff --git a/src/media/codecs/stress/src/input_packet_stream.rs b/src/media/codecs/stress/src/input_packet_stream.rs
new file mode 100644
index 0000000..bd7bbfe
--- /dev/null
+++ b/src/media/codecs/stress/src/input_packet_stream.rs
@@ -0,0 +1,132 @@
+use crate::{buffer_set::*, elementary_stream::*};
+// Copyright 2019 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.
+
+use failure::Fail;
+use fidl_fuchsia_media::*;
+use fuchsia_zircon as zx;
+use std::{collections::HashMap, fmt};
+
+type PacketIdx = u32;
+type BufferIdx = u32;
+
+/// A stream converting elementary stream chunks into input packets for a stream processor.
+pub struct InputPacketStream<I> {
+    packet_and_buffer_pairs: HashMap<PacketIdx, (BufferIdx, UsageStatus)>,
+    buffer_set: BufferSet,
+    stream_lifetime_ordinal: u64,
+    stream: I,
+    sent_eos: bool,
+}
+
+#[derive(Copy, Clone, PartialEq, Debug)]
+enum UsageStatus {
+    Free,
+    InUse,
+}
+
+#[derive(Debug)]
+pub enum Error {
+    PacketRefersToInvalidBuffer,
+    BufferTooSmall { buffer_size: usize, stream_chunk_size: usize },
+    VmoWriteFail(zx::Status),
+}
+
+impl fmt::Display for Error {
+    fn fmt(&self, w: &mut fmt::Formatter) -> fmt::Result {
+        fmt::Debug::fmt(&self, w)
+    }
+}
+
+impl Fail for Error {}
+
+pub enum PacketPoll {
+    Ready(Packet),
+    Eos,
+    NotReady,
+}
+
+impl<'a, I: Iterator<Item = ElementaryStreamChunk<'a>>> InputPacketStream<I> {
+    pub fn new(buffer_set: BufferSet, stream: I, stream_lifetime_ordinal: u64) -> Self {
+        let packets = 0..(buffer_set.buffers.len() as u32);
+        let buffers = packets.clone().rev().map(|idx| (idx, UsageStatus::Free));
+        Self {
+            packet_and_buffer_pairs: packets.zip(buffers).collect(),
+            buffer_set,
+            stream_lifetime_ordinal,
+            stream,
+            sent_eos: false,
+        }
+    }
+
+    pub fn add_free_packet(&mut self, packet: ValidPacketHeader) -> Result<(), Error> {
+        let (_, ref mut status) = *self
+            .packet_and_buffer_pairs
+            .get_mut(&packet.packet_index)
+            .ok_or(Error::PacketRefersToInvalidBuffer)?;
+        *status = UsageStatus::Free;
+        Ok(())
+    }
+
+    fn free_packet_and_buffer(&mut self) -> Option<(u32, u32)> {
+        // This is a linear search. This may not be appropriate in prod code.
+        self.packet_and_buffer_pairs.iter_mut().find_map(|(packet, (buffer, usage))| match usage {
+            UsageStatus::Free => {
+                *usage = UsageStatus::InUse;
+                Some((*packet, *buffer))
+            }
+            UsageStatus::InUse => None,
+        })
+    }
+
+    pub fn next_packet(&mut self) -> Result<PacketPoll, Error> {
+        let (packet_idx, buffer_idx) = if let Some(idxs) = self.free_packet_and_buffer() {
+            idxs
+        } else {
+            return Ok(PacketPoll::NotReady);
+        };
+
+        let chunk = if let Some(chunk) = self.stream.next() {
+            chunk
+        } else if !self.sent_eos {
+            self.sent_eos = true;
+            return Ok(PacketPoll::Eos);
+        } else {
+            return Ok(PacketPoll::NotReady);
+        };
+
+        let buffer = self
+            .buffer_set
+            .buffers
+            .get(buffer_idx as usize)
+            .ok_or(Error::PacketRefersToInvalidBuffer)?;
+
+        if (buffer.size as usize) < chunk.data.len() {
+            return Err(Error::BufferTooSmall {
+                buffer_size: buffer.size as usize,
+                stream_chunk_size: chunk.data.len(),
+            });
+        }
+
+        buffer.data.write(chunk.data, 0).map_err(Error::VmoWriteFail)?;
+
+        Ok(PacketPoll::Ready(Packet {
+            header: Some(PacketHeader {
+                packet_index: Some(packet_idx),
+                buffer_lifetime_ordinal: Some(self.buffer_set.buffer_lifetime_ordinal),
+            }),
+            buffer_index: Some(buffer_idx),
+            stream_lifetime_ordinal: Some(self.stream_lifetime_ordinal),
+            start_offset: Some(0),
+            valid_length_bytes: Some(chunk.data.len() as u32),
+            timestamp_ish: chunk.timestamp,
+            start_access_unit: Some(chunk.start_access_unit),
+            known_end_access_unit: Some(chunk.known_end_access_unit),
+        }))
+    }
+
+    pub fn take_buffer_set(self) -> BufferSet {
+        self.buffer_set
+    }
+}
diff --git a/src/media/codecs/stress/src/lib.rs b/src/media/codecs/stress/src/lib.rs
new file mode 100644
index 0000000..a222c54
--- /dev/null
+++ b/src/media/codecs/stress/src/lib.rs
@@ -0,0 +1,161 @@
+// Copyright 2019 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.
+
+#![feature(async_await, await_macro)]
+#![cfg(test)]
+
+#[macro_use]
+mod log_macros;
+mod buffer_collection_constraints;
+mod buffer_set;
+mod codecs;
+mod elementary_stream;
+mod input_packet_stream;
+mod output_validator;
+mod stream;
+mod stream_runner;
+mod test_spec;
+
+use crate::codecs::h264::*;
+use crate::elementary_stream::*;
+use crate::output_validator::*;
+use crate::stream::*;
+use crate::test_spec::*;
+use failure::{Error, Fail};
+use fuchsia_async as fasync;
+use lazy_static::lazy_static;
+use parking_lot::Mutex;
+use std::{fmt, rc::Rc, sync::Arc};
+
+type Result<T> = std::result::Result<T, Error>;
+
+lazy_static! {
+    // We use a test lock to prevent any tests from running in parallel.
+    //
+    // Running in parallel is something we want to control for specific tests cases especially
+    // when testing hardware stream processors.
+    //
+    // This can be removed in the future if we get a static way to specify environment variables
+    // for a component, so we can set `RUST_TEST_THREADS=1` for this component.
+    static ref TEST_LOCK: Arc<Mutex<()>> = Arc::new(Mutex::new(()));
+    static ref LOGGER: () = ::fuchsia_syslog::init().expect("Initializing syslog");
+}
+
+#[derive(Debug)]
+pub struct FatalError(String);
+
+impl fmt::Display for FatalError {
+    fn fmt(&self, w: &mut fmt::Formatter) -> fmt::Result {
+        write!(w, "FatalError: {}", self.0)
+    }
+}
+
+impl Fail for FatalError {}
+
+// TODO(turnage): Add test spec for buffers released between streams.
+// TODO(turnage): Add hash validator for NV12 and YV12.
+
+#[fasync::run_singlethreaded]
+#[test]
+async fn serial_bear_new_codec_for_each() -> Result<()> {
+    let _lock = TEST_LOCK.lock();
+    *LOGGER;
+
+    eprintln!("reading stream file...");
+
+    let stream = Rc::new(TimestampedStream {
+        source: H264Stream::from_file(BEAR_TEST_FILE)?,
+        timestamps: 0..,
+    });
+
+    eprintln!("got file");
+
+    let frame_count_validator = Rc::new(OutputPacketCountValidator {
+        expected_output_packet_count: stream.video_frame_count(),
+    });
+
+    let spec1 = TestSpec {
+        cases: vec![TestCase {
+            name: "Simple bear test run 1 on new channel",
+            stream: stream.clone(),
+            validators: vec![
+                Rc::new(TerminatesWithValidator {
+                    expected_terminal_output: Output::Eos { stream_lifetime_ordinal: 1 },
+                }),
+                frame_count_validator.clone(),
+            ],
+            stream_options: None,
+        }],
+        relation: CaseRelation::Serial,
+    };
+
+    let spec2 = TestSpec {
+        cases: vec![TestCase {
+            name: "Simple bear test run 2 on new channel",
+            stream,
+            validators: vec![
+                Rc::new(TerminatesWithValidator {
+                    expected_terminal_output: Output::Eos { stream_lifetime_ordinal: 1 },
+                }),
+                frame_count_validator,
+            ],
+            stream_options: None,
+        }],
+        relation: CaseRelation::Serial,
+    };
+
+    await!(spec1.run())?;
+    await!(spec2.run())?;
+
+    Ok(())
+}
+
+#[fasync::run_singlethreaded]
+#[test]
+async fn serial_bear_no_release_buffers() -> Result<()> {
+    let _lock = TEST_LOCK.lock();
+    *LOGGER;
+
+    let stream = Rc::new(TimestampedStream {
+        source: H264Stream::from_file(BEAR_TEST_FILE)?,
+        timestamps: 0..,
+    });
+
+    let frame_count_validator = Rc::new(OutputPacketCountValidator {
+        expected_output_packet_count: stream.video_frame_count(),
+    });
+
+    let spec = TestSpec {
+        cases: vec![
+            TestCase {
+                name: "Simple bear test run 1 on same channel",
+                stream: stream.clone(),
+                validators: vec![
+                    Rc::new(TerminatesWithValidator {
+                        expected_terminal_output: Output::Eos { stream_lifetime_ordinal: 1 },
+                    }),
+                    frame_count_validator.clone(),
+                ],
+                stream_options: None,
+            },
+            TestCase {
+                name: "Simple bear test run 2 on same channel",
+                stream,
+                validators: vec![
+                    Rc::new(TerminatesWithValidator {
+                        expected_terminal_output: Output::Eos { stream_lifetime_ordinal: 3 },
+                    }),
+                    frame_count_validator,
+                ],
+                stream_options: Some(StreamOptions {
+                    queue_format_details: false,
+                    ..StreamOptions::default()
+                }),
+            },
+        ],
+        relation: CaseRelation::Serial,
+    };
+
+    await!(spec.run())
+}
diff --git a/src/media/codecs/stress/src/log_macros.rs b/src/media/codecs/stress/src/log_macros.rs
new file mode 100644
index 0000000..b35a55d
--- /dev/null
+++ b/src/media/codecs/stress/src/log_macros.rs
@@ -0,0 +1,8 @@
+// Copyright 2019 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.
+
+#[macro_export]
+macro_rules! vlog {
+    ($v:expr, $($arg:tt)+) => (::fuchsia_syslog::fx_vlog!(tag: "codec_stress_tests", $v, $($arg)+))
+}
diff --git a/src/media/codecs/stress/src/output_validator.rs b/src/media/codecs/stress/src/output_validator.rs
new file mode 100644
index 0000000..56f3c7d
--- /dev/null
+++ b/src/media/codecs/stress/src/output_validator.rs
@@ -0,0 +1,96 @@
+// Copyright 2019 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.
+
+use crate::{buffer_set::*, FatalError};
+use failure::Error;
+use fidl_fuchsia_media::{FormatDetails, StreamOutputFormat};
+use fidl_table_validation::*;
+use std::rc::Rc;
+
+#[derive(ValidFidlTable, Debug, PartialEq)]
+#[fidl_table_src(StreamOutputFormat)]
+pub struct ValidStreamOutputFormat {
+    pub stream_lifetime_ordinal: u64,
+    pub format_details: FormatDetails,
+}
+
+/// An output packet from the stream.
+#[derive(Debug, PartialEq)]
+pub struct OutputPacket {
+    pub data: Vec<u8>,
+    pub format: Rc<ValidStreamOutputFormat>,
+    pub packet: ValidPacket,
+}
+
+/// Output represents any output from a stream we might want to validate programmatically.
+///
+/// This may extend to contain not just explicit events but certain stream control behaviors or
+/// even errors.
+#[derive(Debug, PartialEq)]
+pub enum Output {
+    Packet(OutputPacket),
+    Eos { stream_lifetime_ordinal: u64 },
+    CodecChannelClose,
+}
+
+/// Checks all output packets, which are provided to the validator in the order in which they
+/// were received from the stream processor.
+///
+/// Failure should be indicated by returning an error, not by panic, so that the full context of
+/// the error will be available in the failure output.
+pub trait OutputValidator {
+    fn validate(&self, output: &[Output]) -> Result<(), Error>;
+}
+
+/// Validates that the output contains the expected number of packets.
+pub struct OutputPacketCountValidator {
+    pub expected_output_packet_count: usize,
+}
+
+impl OutputValidator for OutputPacketCountValidator {
+    fn validate(&self, output: &[Output]) -> Result<(), Error> {
+        let actual_output_packet_count: usize = output
+            .iter()
+            .filter(|output| match output {
+                Output::Packet(_) => true,
+                _ => false,
+            })
+            .count();
+
+        if actual_output_packet_count != self.expected_output_packet_count {
+            return Err(FatalError(format!(
+                "actual output packet count: {}; expected output packet count: {}",
+                actual_output_packet_count, self.expected_output_packet_count
+            ))
+            .into());
+        }
+
+        Ok(())
+    }
+}
+
+/// Validates that a stream terminates with Eos.
+pub struct TerminatesWithValidator {
+    pub expected_terminal_output: Output,
+}
+
+impl OutputValidator for TerminatesWithValidator {
+    fn validate(&self, output: &[Output]) -> Result<(), Error> {
+        let actual_terminal_output = output.last().ok_or(FatalError(format!(
+            "In terminal output: expected {:?}; found: None",
+            Some(&self.expected_terminal_output)
+        )))?;
+
+        if *actual_terminal_output == self.expected_terminal_output {
+            Ok(())
+        } else {
+            Err(FatalError(format!(
+                "In terminal output: expected {:?}; found: {:?}",
+                Some(&self.expected_terminal_output),
+                actual_terminal_output
+            ))
+            .into())
+        }
+    }
+}
diff --git a/src/media/codecs/stress/src/stream.rs b/src/media/codecs/stress/src/stream.rs
new file mode 100644
index 0000000..d86dd09
--- /dev/null
+++ b/src/media/codecs/stress/src/stream.rs
@@ -0,0 +1,226 @@
+// Copyright 2019 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.
+
+use crate::{
+    buffer_set::*, elementary_stream::*, input_packet_stream::*, output_validator::*, FatalError,
+    Result,
+};
+use fidl_fuchsia_media::*;
+use fidl_fuchsia_sysmem::BufferCollectionConstraints;
+use std::{convert::TryFrom, rc::Rc};
+
+pub type OrdinalSequence = <OrdinalPattern as IntoIterator>::IntoIter;
+
+#[derive(Debug, Copy, Clone)]
+pub struct StreamOptions {
+    /// When true, the stream runner will queue format details for each stream. Otherwise it will
+    /// inherit format details from the codec factory.
+    pub queue_format_details: bool,
+    pub release_input_buffers_at_end: bool,
+    pub release_output_buffers_at_end: bool,
+    pub input_buffer_collection_constraints: Option<BufferCollectionConstraints>,
+    pub output_buffer_collection_constraints: Option<BufferCollectionConstraints>,
+}
+
+impl Default for StreamOptions {
+    fn default() -> Self {
+        Self {
+            queue_format_details: true,
+            release_input_buffers_at_end: false,
+            release_output_buffers_at_end: false,
+            input_buffer_collection_constraints: None,
+            output_buffer_collection_constraints: None,
+        }
+    }
+}
+
+pub struct Stream<'a> {
+    pub format_details_version_ordinal: u64,
+    pub stream_lifetime_ordinal: u64,
+    pub input_buffer_ordinals: &'a mut OrdinalSequence,
+    pub input_packet_stream:
+        Option<InputPacketStream<Box<dyn Iterator<Item = ElementaryStreamChunk<'a>> + 'a>>>,
+    pub output_buffer_ordinals: &'a mut OrdinalSequence,
+    pub output_buffer_set: Option<BufferSet>,
+    pub current_output_format: Option<Rc<ValidStreamOutputFormat>>,
+    pub codec: &'a mut StreamProcessorProxy,
+    pub stream: &'a ElementaryStream,
+    pub options: StreamOptions,
+    pub output: Vec<Output>,
+}
+
+pub enum StreamControlFlow {
+    Continue,
+    Stop,
+}
+
+impl<'a: 'b, 'b> Stream<'a> {
+    pub async fn start(&'b mut self) -> Result<()> {
+        if self.options.queue_format_details && self.input_packet_stream.is_some() {
+            self.codec.queue_input_format_details(
+                self.stream_lifetime_ordinal,
+                self.stream.format_details(self.format_details_version_ordinal),
+            )?;
+        }
+
+        self.send_available_input()?;
+
+        Ok(())
+    }
+
+    pub async fn handle_event(
+        &'b mut self,
+        event: StreamProcessorEvent,
+    ) -> Result<StreamControlFlow> {
+        match event {
+            StreamProcessorEvent::OnInputConstraints { input_constraints } => {
+                vlog!(2, "Received input constraints.");
+                vlog!(3, "Input constraints are: {:#?}", input_constraints);
+
+                let buffer_set = await!(BufferSetFactory::buffer_set(
+                    get_ordinal(self.input_buffer_ordinals),
+                    ValidStreamBufferConstraints::try_from(input_constraints)?,
+                    self.codec,
+                    BufferSetType::Input,
+                    self.options.input_buffer_collection_constraints,
+                ))?;
+
+                self.codec.queue_input_format_details(
+                    self.stream_lifetime_ordinal,
+                    self.stream.format_details(self.format_details_version_ordinal),
+                )?;
+
+                self.input_packet_stream = Some(InputPacketStream::new(
+                    buffer_set,
+                    self.stream.stream(),
+                    self.stream_lifetime_ordinal,
+                ));
+                self.send_available_input()?;
+            }
+            StreamProcessorEvent::OnOutputConstraints { output_config } => {
+                vlog!(2, "Received output constraints.");
+                vlog!(3, "Output constraints are: {:#?}", output_config);
+
+                let constraints = ValidStreamOutputConstraints::try_from(output_config)?;
+                if constraints.buffer_constraints_action_required {
+                    self.output_buffer_set = Some(await!(BufferSetFactory::buffer_set(
+                        get_ordinal(self.output_buffer_ordinals),
+                        constraints.buffer_constraints,
+                        self.codec,
+                        BufferSetType::Output,
+                        self.options.output_buffer_collection_constraints,
+                    ))?);
+                }
+            }
+            StreamProcessorEvent::OnFreeInputPacket { free_input_packet } => {
+                vlog!(2, "Received freed input packet.");
+                vlog!(2, "Freed input packet is: {:#?}", free_input_packet);
+
+                let free_input_packet = ValidPacketHeader::try_from(free_input_packet)?;
+                let input_packet_stream = self.input_packet_stream.as_mut().expect(concat!(
+                    "Unwrapping packet stream; ",
+                    "it should be set before we ",
+                    "get free input packets back."
+                ));
+                input_packet_stream.add_free_packet(free_input_packet)?;
+                self.send_available_input()?;
+            }
+            StreamProcessorEvent::OnOutputFormat { output_format } => {
+                vlog!(2, "Received output format.");
+                vlog!(3, "Output format is: {:#?}", output_format);
+
+                let output_format = ValidStreamOutputFormat::try_from(output_format)?;
+                assert_eq!(output_format.stream_lifetime_ordinal, self.stream_lifetime_ordinal);
+                self.current_output_format = Some(Rc::new(output_format));
+            }
+            StreamProcessorEvent::OnOutputPacket {
+                output_packet,
+                error_detected_before,
+                error_detected_during,
+            } => {
+                assert!(!error_detected_before);
+                assert!(!error_detected_during);
+                vlog!(2, "Received output packet.");
+                vlog!(3, "Output packet is: {:#?}", output_packet);
+
+                let output_packet = ValidPacket::try_from(output_packet)?;
+                self.output.push(Output::Packet(OutputPacket {
+                    data: self
+                        .output_buffer_set
+                        .as_ref()
+                        .ok_or(FatalError(String::from(concat!(
+                            "There should be an output buffer set ",
+                            "if we are receiving output packets"
+                        ))))?
+                        .read_packet(&output_packet)?,
+                    format: self.current_output_format.clone().ok_or(FatalError(String::from(
+                        concat!(
+                            "There should be an output format set ",
+                            "if we are receiving output packets"
+                        ),
+                    )))?,
+                    packet: output_packet,
+                }));
+
+                self.codec.recycle_output_packet(PacketHeader {
+                    buffer_lifetime_ordinal: Some(output_packet.header.buffer_lifetime_ordinal),
+                    packet_index: Some(output_packet.header.packet_index),
+                })?;
+            }
+            StreamProcessorEvent::OnOutputEndOfStream {
+                stream_lifetime_ordinal,
+                error_detected_before,
+            } => {
+                assert!(!error_detected_before);
+                vlog!(2, "Received output end of stream.");
+                vlog!(3, "End of stream is for stream {}", stream_lifetime_ordinal);
+
+                // TODO(turnage): Enable the flush method of ending stream in options.
+                self.output.push(Output::Eos { stream_lifetime_ordinal });
+                self.codec.close_current_stream(
+                    self.stream_lifetime_ordinal,
+                    self.options.release_input_buffers_at_end,
+                    self.options.release_output_buffers_at_end,
+                )?;
+                await!(self.codec.sync())?;
+
+                // TODO(turnage): Some codecs return all input packets explicitly, not
+                //                implicitly. All codecs should return explicitly. For now
+                //                we forgive it but soon we want to check that all input
+                //                packets will come back.
+                return Ok(StreamControlFlow::Stop);
+            }
+            e => {
+                vlog!(2, "Got other event: {:#?}", e);
+            }
+        }
+
+        Ok(StreamControlFlow::Continue)
+    }
+
+    fn send_available_input(&'b mut self) -> Result<()> {
+        let input_packet_stream =
+            if let Some(input_packet_stream) = self.input_packet_stream.as_mut() {
+                input_packet_stream
+            } else {
+                return Ok(());
+            };
+
+        loop {
+            match input_packet_stream.next_packet()? {
+                PacketPoll::Ready(input_packet) => {
+                    vlog!(2, "Sending input packet.");
+                    break Ok(self.codec.queue_input_packet(input_packet)?);
+                }
+                PacketPoll::Eos => {
+                    vlog!(2, "Sending stream close");
+                    break Ok(self
+                        .codec
+                        .queue_input_end_of_stream(self.stream_lifetime_ordinal)?);
+                }
+                PacketPoll::NotReady => break Ok(()),
+            }
+        }
+    }
+}
diff --git a/src/media/codecs/stress/src/stream_runner.rs b/src/media/codecs/stress/src/stream_runner.rs
new file mode 100644
index 0000000..ea6df9f
--- /dev/null
+++ b/src/media/codecs/stress/src/stream_runner.rs
@@ -0,0 +1,145 @@
+// Copyright 2019 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.
+
+use crate::{
+    buffer_set::*, elementary_stream::*, input_packet_stream::*, output_validator::*, stream::*,
+    Result,
+};
+use fidl::endpoints::*;
+use fidl_fuchsia_media::*;
+use fidl_fuchsia_mediacodec::*;
+use fuchsia_component::client;
+use futures::TryStreamExt;
+use std::rc::Rc;
+
+/// Runs elementary streams through a stream processor.
+pub struct StreamRunner {
+    input_buffer_ordinals: OrdinalSequence,
+    output_buffer_ordinals: OrdinalSequence,
+    stream_lifetime_ordinals: OrdinalSequence,
+    format_details_ordinals: OrdinalSequence,
+    output_buffer_set: Option<BufferSet>,
+    input_buffer_set: Option<BufferSet>,
+    current_codec: Option<StreamProcessorProxy>,
+}
+
+impl StreamRunner {
+    pub fn new() -> Self {
+        Self {
+            input_buffer_ordinals: OrdinalPattern::Odd.into_iter(),
+            output_buffer_ordinals: OrdinalPattern::Odd.into_iter(),
+            stream_lifetime_ordinals: OrdinalPattern::Odd.into_iter(),
+            format_details_ordinals: OrdinalPattern::All.into_iter(),
+            input_buffer_set: None,
+            output_buffer_set: None,
+            current_codec: None,
+        }
+    }
+
+    pub async fn run_stream(
+        &mut self,
+        stream: Rc<dyn ElementaryStream>,
+        options: StreamOptions,
+    ) -> Result<Vec<Output>> {
+        let format_details_version_ordinal = get_ordinal(&mut self.format_details_ordinals);
+        let stream_lifetime_ordinal = get_ordinal(&mut self.stream_lifetime_ordinals);
+
+        vlog!(
+            2,
+            "Starting a stream with lifetime ordinal {} and format details ordinal {}",
+            stream_lifetime_ordinal,
+            format_details_version_ordinal
+        );
+
+        let mut codec = if let Some(codec) = self.current_codec.take() {
+            codec
+        } else {
+            // TODO(turnage): Accept parameters for using a decoder vs encoder,
+            // and their parameters.
+            await!(get_decoder(stream.as_ref(), format_details_version_ordinal))?
+        };
+        let mut events = codec.take_event_stream();
+
+        let output = {
+            let mut stream = Stream {
+                format_details_version_ordinal,
+                stream_lifetime_ordinal,
+                input_buffer_ordinals: &mut self.input_buffer_ordinals,
+                input_packet_stream: self.input_buffer_set.take().map(|buffer_set| {
+                    InputPacketStream::new(buffer_set, stream.stream(), stream_lifetime_ordinal)
+                }),
+                output_buffer_ordinals: &mut self.output_buffer_ordinals,
+                output_buffer_set: self.output_buffer_set.take(),
+                current_output_format: None,
+                codec: &mut codec,
+                stream: stream.as_ref(),
+                options,
+                output: vec![],
+            };
+
+            await!(stream.start())?;
+
+            let channel_closed = loop {
+                let event = if let Some(event) = await!(events.try_next())? {
+                    event
+                } else {
+                    break true;
+                };
+
+                let control_flow = await!(stream.handle_event(event))?;
+                match control_flow {
+                    StreamControlFlow::Continue => {}
+                    StreamControlFlow::Stop => break false,
+                };
+            };
+
+            let mut output = stream.output;
+            if channel_closed {
+                output.push(Output::CodecChannelClose);
+            }
+
+            self.input_buffer_set =
+                stream.input_packet_stream.map(|stream| stream.take_buffer_set());
+            self.output_buffer_set = stream.output_buffer_set;
+
+            output
+        };
+
+        self.current_codec = Some(codec);
+
+        if options.release_input_buffers_at_end {
+            self.input_buffer_set = None;
+        }
+
+        if options.release_output_buffers_at_end {
+            self.output_buffer_set = None;
+        }
+
+        Ok(output)
+    }
+}
+
+async fn get_decoder(
+    stream: &ElementaryStream,
+    format_details_version_ordinal: u64,
+) -> Result<StreamProcessorProxy> {
+    let factory = client::connect_to_service::<CodecFactoryMarker>()?;
+    let (decoder_client_end, decoder_request) = create_endpoints()?;
+    let decoder = decoder_client_end.into_proxy()?;
+    // TODO(turnage): Account for all error reporting methods in the runner options and output.
+    factory.create_decoder(
+        CreateDecoderParams {
+            input_details: Some(stream.format_details(format_details_version_ordinal)),
+            promise_separate_access_units_on_input: Some(stream.is_access_units()),
+            require_can_stream_bytes_input: Some(false),
+            require_can_find_start: Some(false),
+            require_can_re_sync: Some(false),
+            require_report_all_detected_errors: Some(false),
+            require_hw: Some(false),
+            permit_lack_of_split_header_handling: Some(true),
+        },
+        decoder_request,
+    )?;
+    Ok(decoder)
+}
diff --git a/src/media/codecs/stress/src/test_spec.rs b/src/media/codecs/stress/src/test_spec.rs
new file mode 100644
index 0000000..b74398d
--- /dev/null
+++ b/src/media/codecs/stress/src/test_spec.rs
@@ -0,0 +1,69 @@
+// Copyright 2019 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.
+
+use crate::{elementary_stream::*, output_validator::*, stream::*, stream_runner::*};
+use failure::{Error, ResultExt};
+use futures::{stream::FuturesUnordered, TryStreamExt};
+use std::rc::Rc;
+
+/// A test spec describes all the cases that will run and the circumstances in which
+/// they will run.
+pub struct TestSpec {
+    pub cases: Vec<TestCase>,
+    pub relation: CaseRelation,
+}
+
+/// A case relation describes the temporal relationship between two test cases.
+pub enum CaseRelation {
+    /// With serial relation, test cases will be run in sequence using the same codec server.
+    Serial,
+    /// With concurrent relation, test cases will run concurrently using two codec servers.
+    #[allow(unused)]
+    Concurrent,
+}
+
+/// A test cases describes a sequence of elementary stream chunks that should be fed into a codec
+/// server, and a set of validators to check the output. To pass, all validations must pass for all
+/// output from the stream.
+pub struct TestCase {
+    pub name: &'static str,
+    pub stream: Rc<dyn ElementaryStream>,
+    pub validators: Vec<Rc<dyn OutputValidator>>,
+    pub stream_options: Option<StreamOptions>,
+}
+
+impl TestSpec {
+    pub async fn run(self) -> Result<(), Error> {
+        match self.relation {
+            CaseRelation::Serial => await!(run_cases_serially(self.cases)),
+            CaseRelation::Concurrent => await!(run_cases_concurrently(self.cases)),
+        }
+    }
+}
+
+async fn run_cases_serially(cases: Vec<TestCase>) -> Result<(), Error> {
+    let mut stream_runner = StreamRunner::new();
+
+    for case in cases {
+        let output =
+            await!(stream_runner.run_stream(case.stream, case.stream_options.unwrap_or_default()))
+                .context(format!("Running case {}", case.name))?;
+        for validator in case.validators {
+            validator.validate(&output).context(format!("Validating case {}", case.name))?;
+        }
+    }
+
+    Ok(())
+}
+
+async fn run_cases_concurrently(cases: Vec<TestCase>) -> Result<(), Error> {
+    let mut unordered = FuturesUnordered::new();
+    for case in cases {
+        unordered.push(run_cases_serially(vec![case]))
+    }
+
+    while let Some(()) = await!(unordered.try_next())? {}
+
+    Ok(())
+}