Repository: Kitware/kwiver (language: python)

Issue:
Add synthetic KLV test videos
This PR adds five 30-frame test videos with comprehensive ST0601 test data, and FFmpeg input tests to ensure there are no regressions as KLV development continues. Also included is a tweak to the JSON reader to make it report the stream index of the packet when reading, which is necessary for one of the tests.

The data file sizes are interesting here: each video is below 200KB, but the two JSON reference files are 2MB each. Reviewers may opine whether these sizes are a problem; the JSON files could be reduced significantly by removing whitespace and unnecessary field labels, albeit at the cost of readability.

@hdefazio 


Discussion / hints:
[Kwiver Linux build succeeded.](https://open.cdash.org/buildSummary.php?buildid=8303303)
[Kwiver CentOS build succeeded.](https://open.cdash.org/buildSummary.php?buildid=8303318)

Fix patch:
diff --git a/CMake/kwiver-depends-zlib.cmake b/CMake/kwiver-depends-zlib.cmake
new file mode 100644
index 0000000000..ef3120477d
--- /dev/null
+++ b/CMake/kwiver-depends-zlib.cmake
@@ -0,0 +1,10 @@
+# Optional find and configure ZLib dependency
+
+option( KWIVER_ENABLE_ZLIB
+  "Enable zlib dependent code and plugins (Arrows)"
+  ${fletch_ENABLED_ZLib}
+  )
+
+if( KWIVER_ENABLE_ZLIB )
+  find_package( ZLIB MODULE REQUIRED )
+endif()
diff --git a/CMake/kwiver-depends.cmake b/CMake/kwiver-depends.cmake
index 10e1aa15fd..9bdf1cb658 100644
--- a/CMake/kwiver-depends.cmake
+++ b/CMake/kwiver-depends.cmake
@@ -44,6 +44,7 @@ if(KWIVER_ENABLE_ARROWS)
   include( kwiver-depends-PDAL )
   include( kwiver-depends-PyTorch )
   include( kwiver-depends-COLMAP )
+  include( kwiver-depends-zlib )
 endif()
 
 include( kwiver-depends-ZeroMQ )
diff --git a/CMakeLists.txt b/CMakeLists.txt
index d911507edc..74b003b96e 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -112,6 +112,7 @@ OPTION(KWIVER_ENABLE_SPROKIT              "Enable building sprokit" OFF )
 if(KWIVER_ENABLE_ARROWS)
   OPTION(KWIVER_ENABLE_MVG                "Enable Multi-View Geometry Arrow" ON )
   OPTION(KWIVER_ENABLE_KLV                "Enable Key-Length-Value Metadata Arrow" ON)
+  OPTION(KWIVER_ENABLE_ZLIB               "Enable Zlib Arrow" ON)
 endif()
 
 CMAKE_DEPENDENT_OPTION(KWIVER_ENABLE_PROCESSES
diff --git a/arrows/CMakeLists.txt b/arrows/CMakeLists.txt
index 85bdeaabd3..eff23e14a7 100644
--- a/arrows/CMakeLists.txt
+++ b/arrows/CMakeLists.txt
@@ -120,3 +120,8 @@ add_subdirectory( serialize )
 if( KWIVER_ENABLE_COLMAP )
   add_subdirectory( colmap )
 endif()
+
+# If Zlib is enabled
+if( KWIVER_ENABLE_ZLIB )
+  add_subdirectory( zlib )
+endif()
diff --git a/arrows/core/applets/dump_klv.cxx b/arrows/core/applets/dump_klv.cxx
index 257721d055..151d4f4967 100644
--- a/arrows/core/applets/dump_klv.cxx
+++ b/arrows/core/applets/dump_klv.cxx
@@ -77,6 +77,7 @@ add_command_options()
     ( "multithread",
       "Use multithreading to accelerate encoding of frame images. "
       "Number of worker threads is not configurable at this time." )
+    ( "compress", "Compress output file. Only available for klv-json." )
 
     // positional parameters
     ( "video-file", "Video input file", cxxopts::value< std::string >() )
@@ -159,6 +160,11 @@ ::run()
     config->set_value( "metadata_serializer:type", serializer_type );
   }
 
+  if( cmd_args.count( "compress" ) )
+  {
+    config->set_value( "metadata_serializer:klv-json:compress", true );
+  }
+
   kva::video_input::set_nested_algo_configuration(
     "video_reader", config, video_reader );
   kva::video_input::get_nested_algo_configuration(
diff --git a/arrows/serialize/json/CMakeLists.txt b/arrows/serialize/json/CMakeLists.txt
index 09badeb95f..643ca91d64 100644
--- a/arrows/serialize/json/CMakeLists.txt
+++ b/arrows/serialize/json/CMakeLists.txt
@@ -1,6 +1,8 @@
 # Build / Install plugin for serialization
 
-find_package(ZLIB MODULE REQUIRED)
+if(NOT KWIVER_ENABLE_ZLIB)
+  message(FATAL_ERROR "JSON serialization requires zlib (KWIVER_ENABLE_ZLIB)")
+endif()
 include_directories( ${ZLIB_INCLUDE_DIRS} )
 
 set( headers_public
diff --git a/arrows/serialize/json/klv/CMakeLists.txt b/arrows/serialize/json/klv/CMakeLists.txt
index cca0f6c6cd..bc472a4a0a 100644
--- a/arrows/serialize/json/klv/CMakeLists.txt
+++ b/arrows/serialize/json/klv/CMakeLists.txt
@@ -1,5 +1,9 @@
 # Build / Install plugin for KLV serialization
 
+if(NOT KWIVER_ENABLE_ZLIB)
+  message(FATAL_ERROR "JSON KLV serialization requires the ZLib arrow (KWIVER_ENABLE_ZLIB)")
+endif()
+
 set( public_headers
   metadata_map_io.h
  )
@@ -30,7 +34,10 @@ kwiver_add_library( kwiver_serialize_json_klv
 
 target_link_libraries( kwiver_serialize_json_klv
   PUBLIC vital_algo
-  PRIVATE kwiver_algo_klv kwiver_serialize_json
+  PRIVATE
+    kwiver_algo_klv
+    kwiver_algo_zlib
+    kwiver_serialize_json
   )
 
 algorithms_create_plugin( kwiver_serialize_json_klv
diff --git a/arrows/serialize/json/klv/metadata_map_io.cxx b/arrows/serialize/json/klv/metadata_map_io.cxx
index ab7827fd41..544fd59a1c 100644
--- a/arrows/serialize/json/klv/metadata_map_io.cxx
+++ b/arrows/serialize/json/klv/metadata_map_io.cxx
@@ -10,6 +10,8 @@
 #include <arrows/klv/klv_metadata.h>
 #include <arrows/klv/klv_packet.h>
 
+#include <arrows/zlib/bytestream_compressor.h>
+
 #include <vital/internal/cereal/archives/json.hpp>
 #include <vital/internal/cereal/types/vector.hpp>
 
@@ -36,11 +38,24 @@ auto const output_options = cereal::JSONOutputArchive::Options{
 
 // ----------------------------------------------------------------------------
 class metadata_map_io_klv::priv
-{};
+{
+public:
+  priv();
+
+  bool compress;
+  vital::bytestream_compressor::compression_type_t compress_type;
+};
+
+// ----------------------------------------------------------------------------
+metadata_map_io_klv::priv
+::priv()
+  : compress{ false },
+    compress_type{ vital::bytestream_compressor::COMPRESSION_TYPE_DEFLATE }
+{}
 
 // ----------------------------------------------------------------------------
 metadata_map_io_klv
-::metadata_map_io_klv() : d{ nullptr } {}
+::metadata_map_io_klv() : d{ new priv } {}
 
 // ----------------------------------------------------------------------------
 metadata_map_io_klv
@@ -53,9 +68,23 @@ metadata_map_io_klv
 ::load_( std::istream& fin, VITAL_UNUSED std::string const& filename ) const
 {
   // Load KLV from JSON
-  cereal::JSONInputArchive archive( fin );
   std::vector< klv::klv_timed_packet > packets;
-  cereal::load( archive, packets );
+  if( d->compress )
+  {
+    vital::bytestream_compressor decompressor(
+      vital::bytestream_compressor::MODE_DECOMPRESS,
+      vital::bytestream_compressor::COMPRESSION_TYPE_DEFLATE,
+      vital::bytestream_compressor::DATA_TYPE_TEXT );
+    vital::compress_istream compress_is( fin, decompressor );
+
+    cereal::JSONInputArchive archive( compress_is );
+    cereal::load( archive, packets );
+  }
+  else
+  {
+    cereal::JSONInputArchive archive( fin );
+    cereal::load( archive, packets );
+  }
 
   // Group KLV by frame
   std::map< vital::frame_id_t, std::vector< klv::klv_packet > > packet_map;
@@ -138,8 +167,44 @@ ::save_( std::ostream& fout, vital::metadata_map_sptr data,
   }
 
   // Save KLV to JSON
-  cereal::JSONOutputArchive archive( fout, output_options );
-  cereal::save( archive, packets );
+  if( d->compress )
+  {
+    vital::bytestream_compressor compressor(
+      vital::bytestream_compressor::MODE_COMPRESS,
+      vital::bytestream_compressor::COMPRESSION_TYPE_DEFLATE,
+      vital::bytestream_compressor::DATA_TYPE_TEXT );
+    vital::compress_ostream compress_os( fout, compressor );
+
+    cereal::JSONOutputArchive archive( compress_os, output_options );
+    cereal::save( archive, packets );
+  }
+  else
+  {
+    cereal::JSONOutputArchive archive( fout, output_options );
+    cereal::save( archive, packets );
+  }
+}
+
+// ----------------------------------------------------------------------------
+vital::config_block_sptr
+metadata_map_io_klv
+::get_configuration() const
+{
+  auto config = algorithm::get_configuration();
+
+  config->set_value(
+    "compress", d->compress,
+    "Set to true to read and write compressed JSON instead." );
+
+  return config;
+}
+
+// ----------------------------------------------------------------------------
+void
+metadata_map_io_klv
+::set_configuration( vital::config_block_sptr config )
+{
+  d->compress = config->get_value< bool >( "compress", false );
 }
 
 } // namespace json
diff --git a/arrows/serialize/json/klv/metadata_map_io.h b/arrows/serialize/json/klv/metadata_map_io.h
index e7dc33555f..e50aec4501 100644
--- a/arrows/serialize/json/klv/metadata_map_io.h
+++ b/arrows/serialize/json/klv/metadata_map_io.h
@@ -35,6 +35,9 @@ class KWIVER_SERIALIZE_JSON_KLV_EXPORT metadata_map_io_klv
   save_( std::ostream& fout, vital::metadata_map_sptr data,
          std::string const& filename ) const override;
 
+  vital::config_block_sptr get_configuration() const override;
+  void set_configuration( vital::config_block_sptr config ) override;
+
 private:
   class priv;
 
diff --git a/arrows/serialize/protobuf/CMakeLists.txt b/arrows/serialize/protobuf/CMakeLists.txt
index ac4d0c5d7f..d5bf085348 100644
--- a/arrows/serialize/protobuf/CMakeLists.txt
+++ b/arrows/serialize/protobuf/CMakeLists.txt
@@ -1,7 +1,10 @@
 # Build / Install plugin for serialization
 
 find_package(Protobuf MODULE REQUIRED)
-find_package(ZLIB MODULE REQUIRED)
+
+if(NOT KWIVER_ENABLE_ZLIB)
+  message(FATAL_ERROR "Protobuf serialization requires zlib (KWIVER_ENABLE_ZLIB)")
+endif()
 include_directories( ${ZLIB_INCLUDE_DIRS} )
 
 if(NOT PROTOBUF_FOUND)
diff --git a/arrows/zlib/CMakeLists.txt b/arrows/zlib/CMakeLists.txt
new file mode 100644
index 0000000000..6fcf2e1388
--- /dev/null
+++ b/arrows/zlib/CMakeLists.txt
@@ -0,0 +1,40 @@
+# Build / Install Key-Length-Value Metadata Arrow
+
+set( CMAKE_FOLDER "Arrows/KLV" )
+
+set( sources
+  bytestream_compressor.cxx
+  )
+
+set( public_headers
+  bytestream_compressor.h
+  )
+
+set( private_headers )
+
+kwiver_install_headers(
+  ${public_headers}
+  SUBDIR arrows/zlib
+  )
+
+kwiver_install_headers(
+  ${CMAKE_CURRENT_BINARY_DIR}/kwiver_algo_zlib_export.h
+  NOPATH
+  SUBDIR arrows/zlib
+  )
+
+kwiver_add_library( kwiver_algo_zlib
+  ${sources}
+  ${public_headers}
+  ${private_headers}
+  )
+
+target_link_libraries( kwiver_algo_zlib
+  PRIVATE
+    vital
+    ${ZLIB_LIBRARIES}
+  )
+
+if (KWIVER_ENABLE_TESTS)
+  add_subdirectory( tests )
+endif()


Test patch:
diff --git a/arrows/zlib/bytestream_compressor.cxx b/arrows/zlib/bytestream_compressor.cxx
new file mode 100644
index 0000000000..8648d4ae40
--- /dev/null
+++ b/arrows/zlib/bytestream_compressor.cxx
@@ -0,0 +1,241 @@
+// This file is part of KWIVER, and is distributed under the
+// OSI-approved BSD 3-Clause License. See top-level LICENSE file or
+// https://github.com/Kitware/kwiver/blob/master/LICENSE for details.
+
+/// \file
+/// Implementation of bytestream (de)compressor.
+
+#include <arrows/zlib/bytestream_compressor.h>
+
+#define ZLIB_CONST
+#include <zlib.h>
+#undef ZLIB_CONST
+
+#include <array>
+#include <deque>
+#include <stdexcept>
+
+namespace kwiver {
+
+namespace vital {
+
+// ----------------------------------------------------------------------------
+class bytestream_compressor::impl
+{
+public:
+  impl( mode_t mode, compression_type_t compression_type,
+        data_type_t data_type );
+
+  ~impl();
+
+  mode_t mode;
+  compression_type_t compression_type;
+  data_type_t data_type;
+  z_stream_s stream;
+  std::deque< uint8_t > buffer;
+  bool flush;
+};
+
+// ----------------------------------------------------------------------------
+bytestream_compressor::impl
+::impl( mode_t mode, compression_type_t compression_type,
+        data_type_t data_type )
+  : mode{ mode }, compression_type{ compression_type }, data_type{ data_type },
+    stream{}, flush{ false }
+{
+  if( mode >= MODE_ENUM_END ||
+      compression_type >= COMPRESSION_TYPE_ENUM_END ||
+      data_type >= DATA_TYPE_ENUM_END )
+  {
+    throw std::runtime_error( "Invalid arguments" );
+  }
+
+  switch( data_type )
+  {
+    case DATA_TYPE_TEXT:
+      stream.data_type = Z_TEXT;
+      break;
+    case DATA_TYPE_BINARY:
+      stream.data_type = Z_BINARY;
+      break;
+    default:
+      throw std::logic_error( "Case not handled" );
+      break;
+  }
+
+  switch( mode )
+  {
+    case MODE_COMPRESS:
+      if( deflateInit( &stream, Z_BEST_COMPRESSION ) != Z_OK )
+      {
+        throw std::runtime_error{ "Initializing compression failed" };
+      }
+      break;
+    case MODE_DECOMPRESS:
+      if( inflateInit( &stream ) != Z_OK )
+      {
+        throw std::runtime_error{ "Initializing decompression failed" };
+      }
+      break;
+    default:
+      throw std::logic_error( "Case not handled" );
+      break;
+  }
+}
+
+// ----------------------------------------------------------------------------
+bytestream_compressor::impl
+::~impl()
+{
+  switch( mode )
+  {
+    case MODE_COMPRESS:
+      deflateEnd( &stream );
+      break;
+    case MODE_DECOMPRESS:
+      inflateEnd( &stream );
+      break;
+    default:
+      break;
+  }
+}
+
+// ----------------------------------------------------------------------------
+bytestream_compressor
+::bytestream_compressor(
+  mode_t mode, compression_type_t compression_type, data_type_t data_type )
+  : d{ new impl{ mode, compression_type, data_type } }
+{}
+
+// ----------------------------------------------------------------------------
+bytestream_compressor
+::bytestream_compressor( bytestream_compressor&& other )
+  : d{}
+{
+  *this = std::move( other );
+}
+
+// ----------------------------------------------------------------------------
+bytestream_compressor&
+bytestream_compressor
+::operator=( bytestream_compressor&& other )
+{
+  d = std::move( other.d );
+  return *this;
+}
+
+// ----------------------------------------------------------------------------
+bytestream_compressor
+::~bytestream_compressor()
+{}
+
+// ----------------------------------------------------------------------------
+void
+bytestream_compressor
+::write( void const* begin, void const* end )
+{
+  if( begin > end )
+  {
+    throw std::logic_error{ "Invalid range" };
+  }
+
+  int ( *fn )( z_streamp, int ) = nullptr;
+  switch( d->mode )
+  {
+    case MODE_COMPRESS:
+      fn = &deflate;
+      break;
+    case MODE_DECOMPRESS:
+      fn = &inflate;
+      break;
+    default:
+      throw std::logic_error( "Case not handled" );
+      break;
+  }
+
+  auto const begin_byte = static_cast< Bytef const* >( begin );
+  auto const end_byte = static_cast< Bytef const* >( end );
+  d->stream.next_in = begin_byte;
+  d->stream.avail_in =
+    static_cast< uInt >( std::distance( begin_byte, end_byte ) );
+
+  std::array< uint8_t, BUFSIZ > tmp_buffer;
+  do
+  {
+    d->stream.next_out = &*tmp_buffer.begin();
+    d->stream.avail_out = static_cast< uInt >( tmp_buffer.size() );
+
+    auto const err = fn( &d->stream, d->flush );
+    if( err != Z_OK && err != Z_STREAM_END && err != Z_BUF_ERROR )
+    {
+      throw std::runtime_error{ "(De)compression failed" };
+    }
+    d->buffer.insert(
+        d->buffer.end(), &*tmp_buffer.begin(), d->stream.next_out );
+  } while( d->stream.avail_out == 0 );
+
+  d->flush = false;
+}
+
+// ----------------------------------------------------------------------------
+void
+bytestream_compressor
+::write( std::vector< uint8_t > const& bytes )
+{
+  write( bytes.data(), bytes.data() + bytes.size() );
+}
+
+// ----------------------------------------------------------------------------
+void*
+bytestream_compressor
+::read( void* begin, void* end )
+{
+  if( begin > end )
+  {
+    throw std::logic_error{ "Invalid range" };
+  }
+
+  auto const begin_byte = static_cast< Bytef* >( begin );
+  auto const end_byte = static_cast< Bytef* >( end );
+
+  // Copy data, careful not to overflow bounds
+  auto const input_size = static_cast< size_t >( std::distance( begin_byte, end_byte ) );
+  auto const copy_size = std::min( input_size, d->buffer.size() );
+  auto const result = std::copy_n( d->buffer.begin(), copy_size, begin_byte );
+
+  // Remove copied data
+  d->buffer.erase( d->buffer.begin(), d->buffer.begin() + copy_size );
+
+  return result;
+}
+
+// ----------------------------------------------------------------------------
+std::vector< uint8_t >
+bytestream_compressor
+::read()
+{
+  std::vector< uint8_t > result{ d->buffer.begin(), d->buffer.end() };
+  d->buffer.clear();
+  return result;
+}
+
+// ----------------------------------------------------------------------------
+size_t
+bytestream_compressor
+::readable_bytes() const
+{
+  return d->buffer.size();
+}
+
+// ----------------------------------------------------------------------------
+void
+bytestream_compressor
+::flush()
+{
+  d->flush = true;
+  write( {} );
+}
+
+} // namespace vital
+
+} // namespace kwiver
diff --git a/arrows/zlib/bytestream_compressor.h b/arrows/zlib/bytestream_compressor.h
new file mode 100644
index 0000000000..c1ca33d1e0
--- /dev/null
+++ b/arrows/zlib/bytestream_compressor.h
@@ -0,0 +1,294 @@
+// This file is part of KWIVER, and is distributed under the
+// OSI-approved BSD 3-Clause License. See top-level LICENSE file or
+// https://github.com/Kitware/kwiver/blob/master/LICENSE for details.
+
+/// \file
+/// Declaration of bytestream (de)compressor.
+
+#ifndef KWIVER_ARROWS_ZLIB_BYTESTREAM_COMPRESSOR_H_
+#define KWIVER_ARROWS_ZLIB_BYTESTREAM_COMPRESSOR_H_
+
+#include <arrows/zlib/kwiver_algo_zlib_export.h>
+
+#include <algorithm>
+#include <array>
+#include <iostream>
+#include <memory>
+#include <streambuf>
+#include <vector>
+
+#include <cstdint>
+#include <cstdio>
+
+namespace kwiver {
+
+namespace vital {
+
+// ----------------------------------------------------------------------------
+/// Provides compression and decompression functionality.
+class KWIVER_ALGO_ZLIB_EXPORT bytestream_compressor
+{
+public:
+  // Operation to be performed on the input data.
+  enum mode_t
+  {
+    MODE_COMPRESS,
+    MODE_DECOMPRESS,
+    MODE_ENUM_END,
+  };
+
+  /// Compression algorithm to use.
+  enum compression_type_t
+  {
+    COMPRESSION_TYPE_DEFLATE,
+    // COMPRESSION_TYPE_GZIP, // TODO
+    COMPRESSION_TYPE_ENUM_END,
+  };
+
+  /// Nature of the uncompressed data, to help fine-tune the algorithm.
+  enum data_type_t
+  {
+    DATA_TYPE_BINARY,
+    DATA_TYPE_TEXT,
+    DATA_TYPE_ENUM_END,
+  };
+
+  /// \throw std::runtime_error If given configuration is not available.
+  bytestream_compressor(
+    mode_t mode, compression_type_t compression_type, data_type_t data_type );
+
+  bytestream_compressor( bytestream_compressor const& ) = delete;
+  bytestream_compressor& operator=( bytestream_compressor const& ) = delete;
+
+  bytestream_compressor( bytestream_compressor&& other );
+  bytestream_compressor& operator=( bytestream_compressor&& other );
+
+  ~bytestream_compressor();
+
+  /// Give the data between \p begin and \p end to be (de)compressed.
+  void write( void const* begin, void const* end );
+
+  /// Give \p bytes to be (de)compressed.
+  void write( std::vector< uint8_t > const& bytes );
+
+  /// Write (de)compressed data to the buffer between \p begin and \p end.
+  ///
+  /// \note Some data may remain buffered internally. Use \c flush() to force
+  //// all data to be readable.
+  ///
+  /// \return Pointer to byte one past the last one written.
+  void* read( void* begin, void* end );
+
+  /// Return all available (de)compressed data.
+  ///
+  /// \note Some data may remain buffered internally. Use \c flush() to force
+  //// all data to be readable.
+  std::vector< uint8_t > read();
+
+  /// Return the number of currently available (de)compressed bytes.
+  size_t readable_bytes() const;
+
+  /// Finish (de)compression on remaining buffered data.
+  ///
+  /// After calling, all (de)compressed data will be available via \c read().
+  ///
+  /// \warning
+  ///   Too-frequent use of this function may degrade quality of compression.
+  void flush();
+
+private:
+  class impl;
+
+  std::unique_ptr< impl > d;
+};
+
+// ----------------------------------------------------------------------------
+/// Wrapper around another std::istream which (de)compresses the data as it
+/// comes in.
+template< class CharT, class Traits = std::char_traits< CharT > >
+class basic_compress_istream
+  : private std::basic_streambuf< CharT, Traits >,
+    public std::basic_istream< CharT, Traits >
+{
+public:
+  using istream_t = std::basic_istream< CharT, Traits >;
+  using base_t = std::basic_streambuf< CharT, Traits >;
+
+  basic_compress_istream(
+    istream_t& source, bytestream_compressor& compressor )
+    : std::basic_streambuf< CharT, Traits >(),
+      std::basic_istream< CharT, Traits >(this),
+      m_source( source ),
+      m_compressor( compressor ),
+      m_flushed( false )
+  {}
+
+protected:
+  /// Called when the buffer of characters already decoded is empty, and the
+  /// user is requesting more characters.
+  int
+  underflow() override
+  {
+    for( void* ptr = &*m_buffer.begin(); true; )
+    {
+      // If the compressor has more output, use that
+      if( m_compressor.readable_bytes() )
+      {
+        ptr = m_compressor.read( ptr, &*m_buffer.end() );
+        base_t::setg(
+          &*m_buffer.begin(), &*m_buffer.begin(),
+          static_cast< CharT* >( ptr ) );
+
+        if( ptr == &*m_buffer.end() )
+        {
+          // The buffer is now full; return success.
+          return Traits::to_int_type( m_buffer.front() );
+        }
+      }
+
+      if( m_flushed )
+      {
+        // There's no more incoming data
+        if( ptr == &*m_buffer.begin() )
+        {
+          // The buffer's empty; we're done here
+          return Traits::eof();
+        }
+        else
+        {
+          // Return the final, partially-filled buffer
+          return Traits::to_int_type( m_buffer.front() );
+        }
+      }
+
+      // The compressor needs more input; pull it from the wrapped istream
+      std::array< CharT, buffer_size > tmp_buffer;
+      auto const count =
+        m_source.read( &*tmp_buffer.begin(), tmp_buffer.size() ).gcount();
+      if( count )
+      {
+        // We found more data to give to the compressor; do that
+        m_compressor.write(
+          &*tmp_buffer.begin(), &*tmp_buffer.begin() + count );
+      }
+      else
+      {
+        // The wrapped istream has no more data; tell the compressor to finish
+        m_compressor.flush();
+        m_flushed = true;
+      }
+    }
+  }
+
+private:
+  static constexpr size_t buffer_size = 512;
+  istream_t& m_source;
+  bytestream_compressor& m_compressor;
+  std::array< CharT, buffer_size > m_buffer;
+  bool m_flushed;
+};
+
+using compress_istream = basic_compress_istream< char >;
+
+// ----------------------------------------------------------------------------
+/// Wrapper around another std::ostream which (de)compresses the data as it
+/// goes out.
+///
+/// \warning
+///   Input is not guaranteed to write to the wrapped stream immediately;
+///   call \c flush() to guarantee this. Frequent use of \c flush()
+///   ( or \c std::endl ) will degrade the quality of the compression.
+template< class CharT, class Traits = std::char_traits< CharT > >
+class basic_compress_ostream
+  : private std::basic_streambuf< CharT, Traits >,
+    public std::basic_ostream< CharT, Traits >
+{
+public:
+  using ostream_t = std::basic_ostream< CharT, Traits >;
+  using base_t = std::basic_streambuf< CharT, Traits >;
+
+  basic_compress_ostream(
+    ostream_t& destination, bytestream_compressor& compressor )
+    : std::basic_streambuf< CharT, Traits >(),
+      std::basic_ostream< CharT, Traits >(this),
+      m_destination( destination ),
+      m_compressor( compressor )
+  {
+    // Input goes in the buffer
+    base_t::setp( &*m_buffer.begin(), &*m_buffer.end() );
+  }
+
+  ~basic_compress_ostream()
+  {
+    // Ensure all data is written out before deletion
+    sync();
+  }
+
+protected:
+  // Called when the input buffer is full, forcing writeout.
+  int
+  overflow( int ch )
+  {
+    // Give the buffer data to the compressor and clear it
+    m_compressor.write( base_t::pbase(), base_t::pptr() );
+    base_t::setp( &*m_buffer.begin(), &*m_buffer.end() );
+
+    if( ch == Traits::eof() )
+    {
+      // Special eof character means we flush all data to wrapped ostream
+      m_compressor.flush();
+    }
+    else
+    {
+      // Character passed in goes at the front of the newly-cleared buffer
+      m_buffer.front() = ch;
+      base_t::pbump( 1 );
+    }
+
+    // Write all data that the compressor gives us to wrapped ostream
+    std::array< CharT, buffer_size > tmp_buffer;
+    while( m_compressor.readable_bytes() )
+    {
+      auto const ptr =
+        m_compressor.read( &*tmp_buffer.begin(), &*tmp_buffer.end() );
+
+      m_destination.write(
+        &*tmp_buffer.begin(),
+        std::distance( &*tmp_buffer.begin(), static_cast< CharT* >( ptr ) ) );
+    }
+
+    return 0;
+  }
+
+  /// Called when \c flush() is called on this object. Forces all data to be
+  /// written out regardless if the buffer is full.
+  int
+  sync() override
+  {
+    // Write all our data to the wrapped ostream
+    if( overflow( Traits::eof() ) == Traits::eof() )
+    {
+      // An error occurred
+      return -1;
+    }
+
+    // Force the wrapped ostream to write out all of its data
+    m_destination.flush();
+
+    return m_destination.bad() ? -1 : 0;
+  }
+
+private:
+  static constexpr size_t buffer_size = 512;
+  ostream_t& m_destination;
+  bytestream_compressor& m_compressor;
+  std::array< CharT, buffer_size > m_buffer;
+};
+
+using compress_ostream = basic_compress_ostream< char >;
+
+} // namespace vital
+
+} // namespace kwiver
+
+#endif
diff --git a/arrows/zlib/tests/CMakeLists.txt b/arrows/zlib/tests/CMakeLists.txt
new file mode 100644
index 0000000000..d8e833cdfb
--- /dev/null
+++ b/arrows/zlib/tests/CMakeLists.txt
@@ -0,0 +1,13 @@
+project(arrows_test_zlib)
+
+set(CMAKE_FOLDER "Arrows/Zlib/Tests")
+
+include(kwiver-test-setup)
+
+set( test_libraries kwiver_algo_zlib vital )
+
+##############################
+# Zlib tests
+##############################
+
+kwiver_discover_gtests(zlib bytestream_compressor LIBRARIES ${test_libraries})
diff --git a/arrows/zlib/tests/test_bytestream_compressor.cxx b/arrows/zlib/tests/test_bytestream_compressor.cxx
new file mode 100644
index 0000000000..ef7d01b7bf
--- /dev/null
+++ b/arrows/zlib/tests/test_bytestream_compressor.cxx
@@ -0,0 +1,215 @@
+// This file is part of KWIVER, and is distributed under the
+// OSI-approved BSD 3-Clause License. See top-level LICENSE file or
+// https://github.com/Kitware/kwiver/blob/master/LICENSE for details.
+
+/// \file
+/// Test bytestream_compressor class.
+
+#include <gtest/gtest.h>
+
+#include <arrows/zlib/bytestream_compressor.h>
+
+#include <array>
+#include <random>
+
+using namespace kwiver::vital;
+
+// ----------------------------------------------------------------------------
+int
+main( int argc, char** argv )
+{
+  ::testing::InitGoogleTest( &argc, argv );
+  return RUN_ALL_TESTS();
+}
+
+// ----------------------------------------------------------------------------
+class test_bytestream_compressor : public ::testing::Test
+{
+private:
+
+  void
+  SetUp() override
+  {
+    text_data.resize( data_size );
+    binary_data.resize( data_size );
+
+    // Create some repeating, compressable data
+    for( size_t i = 0; i < data_size; ++i )
+    {
+      text_data[ i ] = 'A' + ( i % 16 ) + i / ( 1 << 14 );
+      binary_data[ i ] = i % 64 + i / ( 1 << 12 );
+    }
+  }
+
+public:
+  constexpr static size_t data_size = 1 << 16; // 64 KB
+  constexpr static size_t expected_binary_size = 343;
+  constexpr static size_t expected_text_size = 175;
+  std::vector< uint8_t > text_data;
+  std::vector< uint8_t > binary_data;
+};
+
+// ----------------------------------------------------------------------------
+TEST_F ( test_bytestream_compressor, round_trip_deflate_text )
+{
+  bytestream_compressor compressor(
+    bytestream_compressor::MODE_COMPRESS,
+    bytestream_compressor::COMPRESSION_TYPE_DEFLATE,
+    bytestream_compressor::DATA_TYPE_TEXT );
+
+  compressor.write( text_data );
+  compressor.flush();
+  auto const compressed_data = compressor.read();
+
+  EXPECT_EQ( expected_text_size, compressed_data.size() );
+
+  bytestream_compressor decompressor(
+    bytestream_compressor::MODE_DECOMPRESS,
+    bytestream_compressor::COMPRESSION_TYPE_DEFLATE,
+    bytestream_compressor::DATA_TYPE_TEXT );
+
+  decompressor.write( compressed_data );
+  decompressor.flush();
+  auto const decompressed_data = decompressor.read();
+
+  EXPECT_EQ( text_data, decompressed_data );
+}
+
+// ----------------------------------------------------------------------------
+TEST_F ( test_bytestream_compressor, round_trip_deflate_binary )
+{
+  bytestream_compressor compressor(
+    bytestream_compressor::MODE_COMPRESS,
+    bytestream_compressor::COMPRESSION_TYPE_DEFLATE,
+    bytestream_compressor::DATA_TYPE_BINARY );
+
+  compressor.write( binary_data );
+  compressor.flush();
+  auto const compressed_data = compressor.read();
+
+  EXPECT_EQ( expected_binary_size, compressed_data.size() );
+
+  bytestream_compressor decompressor(
+    bytestream_compressor::MODE_DECOMPRESS,
+    bytestream_compressor::COMPRESSION_TYPE_DEFLATE,
+    bytestream_compressor::DATA_TYPE_BINARY );
+
+  decompressor.write( compressed_data );
+  decompressor.flush();
+  auto const decompressed_data = decompressor.read();
+
+  EXPECT_EQ( binary_data, decompressed_data );
+}
+
+// ----------------------------------------------------------------------------
+TEST_F ( test_bytestream_compressor, round_trip_iostream_wrapper )
+{
+  bytestream_compressor compressor(
+    bytestream_compressor::MODE_COMPRESS,
+    bytestream_compressor::COMPRESSION_TYPE_DEFLATE,
+    bytestream_compressor::DATA_TYPE_TEXT );
+
+  bytestream_compressor decompressor(
+    bytestream_compressor::MODE_DECOMPRESS,
+    bytestream_compressor::COMPRESSION_TYPE_DEFLATE,
+    bytestream_compressor::DATA_TYPE_TEXT );
+
+  std::stringstream ss;
+  compress_ostream compress_os( ss, compressor );
+  compress_istream compress_is( ss, decompressor );
+
+  std::string text( text_data.begin(), text_data.end() );
+  compress_os << text << std::flush;
+
+  EXPECT_EQ( expected_text_size, ss.str().size() );
+
+  std::string out_text;
+  compress_is >> out_text;
+  std::vector< uint8_t > decompressed_data( out_text.begin(), out_text.end() );
+
+  EXPECT_EQ( text_data, decompressed_data );
+}
+
+// ----------------------------------------------------------------------------
+TEST_F ( test_bytestream_compressor, round_trip_deflate_piecemeal )
+{
+  // Feed the data to the (de)compressor in these small, prime-number-sized
+  // chunks, instead of all at once
+  constexpr size_t step = 37;
+
+  bytestream_compressor compressor(
+    bytestream_compressor::MODE_COMPRESS,
+    bytestream_compressor::COMPRESSION_TYPE_DEFLATE,
+    bytestream_compressor::DATA_TYPE_TEXT );
+
+  for( size_t i = 0; i < text_data.size(); i += step )
+  {
+    compressor.write(
+        text_data.data() + i,
+        text_data.data() + std::min( i + step, text_data.size() ) );
+  }
+  compressor.flush();
+  auto const compressed_data = compressor.read();
+
+  EXPECT_EQ( expected_text_size, compressed_data.size() );
+
+  bytestream_compressor decompressor(
+    bytestream_compressor::MODE_DECOMPRESS,
+    bytestream_compressor::COMPRESSION_TYPE_DEFLATE,
+    bytestream_compressor::DATA_TYPE_TEXT );
+
+  for( size_t i = 0; i < compressed_data.size(); i += step )
+  {
+    compressor.write(
+        compressed_data.data() + i,
+        compressed_data.data() +
+        std::min( i + step, compressed_data.size() ) );
+  }
+  decompressor.write( compressed_data );
+  decompressor.flush();
+  auto const decompressed_data = decompressor.read();
+
+  EXPECT_EQ( text_data, decompressed_data );
+}
+
+// ----------------------------------------------------------------------------
+TEST_F ( test_bytestream_compressor, round_trip_iostream_wrapper_piecemeal )
+{
+  // Feed the data to the (de)compressor in these small, prime-number-sized
+  // chunks, instead of all at once
+  constexpr size_t step = 37;
+
+  bytestream_compressor compressor(
+    bytestream_compressor::MODE_COMPRESS,
+    bytestream_compressor::COMPRESSION_TYPE_DEFLATE,
+    bytestream_compressor::DATA_TYPE_TEXT );
+
+  bytestream_compressor decompressor(
+    bytestream_compressor::MODE_DECOMPRESS,
+    bytestream_compressor::COMPRESSION_TYPE_DEFLATE,
+    bytestream_compressor::DATA_TYPE_TEXT );
+
+  std::stringstream ss;
+  compress_ostream compress_os( ss, compressor );
+  compress_istream compress_is( ss, decompressor );
+
+  std::string text( text_data.begin(), text_data.end() );
+  for( size_t i = 0; i < text.size(); i += step )
+  {
+    compress_os << text.substr( i, step );
+  }
+  compress_os << std::flush;
+
+  EXPECT_EQ( expected_text_size, ss.str().size() );
+
+  std::string out_text;
+  for( size_t i = 0; i < text.size(); i += step )
+  {
+    std::string tmp( std::min( step, text.size() - i ), '\0' );
+    compress_is.read( &tmp[ 0 ], tmp.size() );
+    out_text += tmp;
+  }
+  std::vector< uint8_t > decompressed_data( out_text.begin(), out_text.end() );
+
+  EXPECT_EQ( text_data, decompressed_data );
+}