gzip filter: make sure the stream ends properly

source/common/compressor/zlib_compressor_impl.cc

@@ -82,5 +82,28 @@ void ZlibCompressorImpl::updateOutput(Buffer::Instance& output_buffer) {
   zstream_ptr_->next_out = chunk_char_ptr_.get();
 }
 
+void ZlibCompressorImpl::reset() {
+  const bool result = deflateReset(zstream_ptr_.get());
+  RELEASE_ASSERT(result == Z_OK);
+}
+
+void ZlibCompressorImpl::finish(Buffer::Instance& output_buffer) {
+  // Once the Z_FINISH parameter is provided, deflate() will begin to complete the compressed output
+  // stream. However depending on how much output space is provided, deflate() may have to be called
+  // several times until it has provided the complete compressed stream, even after it has consumed
+  // all of the input. The flush parameter must continue to be Z_FINISH for those subsequent calls.
+  // Ref: https://zlib.net/zlib_how.html.
+  const int result = deflate(zstream_ptr_.get(), Z_FINISH);
+  updateOutput(output_buffer);
+  if (result == Z_OK || result == Z_BUF_ERROR) {
+    finish(output_buffer);
+  } else {
+    // While Z_OK, Z_BUF_ERROR (covered in above case) and Z_STREAM_END are the acceptable values,
+    // the Z_STREAM_ERROR is only possible if the stream is not initialized properly. Ref:
+    // https://zlib.net/zlib_how.html.
+    RELEASE_ASSERT(result == Z_STREAM_END);
+  }
+}
+
 } // namespace Compressor
 } // namespace Envoy

source/common/compressor/zlib_compressor_impl.h

@@ -81,6 +81,18 @@ class ZlibCompressorImpl : public Compressor {
    */
   uint64_t checksum();
 
+  /**
+   * This is required to re-init the stream after calling finish(_).
+   */
+  void reset();
+
+  /**
+   * Completes the compressed output stream. To indicate that the currently pointed data is the last
+   * chunk of input data to compress.
+   * @param output_buffer supplies the buffer to output compressed data.
+   */
+  void finish(Buffer::Instance& output_buffer);
+
   // Compressor
   void compress(const Buffer::Instance& input_buffer, Buffer::Instance& output_buffer) override;
 

source/common/decompressor/zlib_decompressor_impl.cc

@@ -59,6 +59,13 @@ void ZlibDecompressorImpl::decompress(const Buffer::Instance& input_buffer,
 
 bool ZlibDecompressorImpl::inflateNext() {
   const int result = inflate(zstream_ptr_.get(), Z_NO_FLUSH);
+  if (result == Z_STREAM_END) {
+    // Z_FINISH informs inflate to not maintain a sliding window if the stream completes, which
+    // reduces inflate's memory footprint. Ref: https://www.zlib.net/manual.html.
+    inflate(zstream_ptr_.get(), Z_FINISH);
+    return false;
+  }
+
   if (result == Z_BUF_ERROR && zstream_ptr_->avail_in == 0) {
     return false; // This means that zlib needs more input, so stop here.
   }

source/common/http/filter/gzip_filter.cc

@@ -139,11 +139,15 @@ FilterDataStatus GzipFilter::encodeData(Buffer::Instance& data, bool end_stream)
 
   if (end_stream) {
     compressor_.flush(compressed_data_);
+    compressor_.finish(compressed_data_);
   }
 
   if (compressed_data_.length()) {
     data.drain(n_data);
     data.move(compressed_data_);
+
+    // Reset is required to re-init the compressor's z_stream.
+    compressor_.reset();
     return Http::FilterDataStatus::Continue;
   }
 

test/common/compressor/zlib_compressor_impl_test.cc

@@ -103,6 +103,9 @@ TEST_F(ZlibCompressorImplTest, CompressWithReducedInternalMemory) {
                   compressed_slices[num_comp_slices - 1].len_);
   // FOOTER four-byte sequence (sync flush)
   EXPECT_EQ("0000ffff", footer_hex_str.substr(footer_hex_str.size() - 8, 10));
+
+  compressor.finish(output_buffer);
+  EXPECT_EQ("0000ffff", footer_hex_str.substr(footer_hex_str.size() - 8, 10));
 }
 
 /**
@@ -142,6 +145,12 @@ TEST_F(ZlibCompressorImplTest, CompressWithContinuesFlush) {
                     compressed_slices[num_comp_slices - 1].len_);
     // FOOTER four-byte sequence (sync flush)
     EXPECT_EQ("0000ffff", footer_hex_str.substr(footer_hex_str.size() - 8, 10));
+
+    compressor.finish(output_buffer);
+    EXPECT_EQ("0000ffff", footer_hex_str.substr(footer_hex_str.size() - 8, 10));
+
+    // Re-init the compressor.
+    compressor.reset();
   }
 }
 
@@ -181,6 +190,9 @@ TEST_F(ZlibCompressorImplTest, CompressSmallInputMemory) {
                   compressed_slices[num_comp_slices - 1].len_);
   // FOOTER four-byte sequence (sync flush)
   EXPECT_EQ("0000ffff", footer_hex_str.substr(footer_hex_str.size() - 8, 10));
+
+  compressor.finish(output_buffer);
+  EXPECT_EQ("0000ffff", footer_hex_str.substr(footer_hex_str.size() - 8, 10));
 }
 
 /**
@@ -241,6 +253,9 @@ TEST_F(ZlibCompressorImplTest, CompressMoveFlushAndRepeat) {
                   compressed_slices[num_comp_slices - 1].len_);
   // FOOTER four-byte sequence (sync flush)
   EXPECT_EQ("0000ffff", footer_hex_str.substr(footer_hex_str.size() - 8, 10));
+
+  compressor.finish(output_buffer);
+  EXPECT_EQ("0000ffff", footer_hex_str.substr(footer_hex_str.size() - 8, 10));
 }
 
 /**
@@ -279,6 +294,9 @@ TEST_F(ZlibCompressorImplTest, CompressWithNotCommonParams) {
                   compressed_slices[num_comp_slices - 1].len_);
   // FOOTER four-byte sequence (sync flush)
   EXPECT_EQ("0000ffff", footer_hex_str.substr(footer_hex_str.size() - 8, 10));
+
+  compressor.finish(output_buffer);
+  EXPECT_EQ("0000ffff", footer_hex_str.substr(footer_hex_str.size() - 8, 10));
 }
 
 } // namespace