EncoderZlib/DecoderZlib for chromoting

Encoder and decoder using zlib for chromoting. This implementation has zero
copy out of the decoder and encoder. The consequence is that we have to break
out the zlib stream into rect boundaries which requires the synchronication
flush feature in zlib. This feature will hurt compression ratio but the effect
still need to be measured.

This patch also provides tests for testing the Encoder and Decoder pair with
zlib.

TEST=remoting_unittests

Review URL: http://codereview.chromium.org/2868062

git-svn-id: svn://svn.chromium.org/chrome/trunk/src@53738 0039d316-1c4b-4281-b951-d872f2087c98

14 files changed, 550 insertions, 18 deletions

diff --git a/remoting/base/codec_test.cc b/remoting/base/codec_test.cc
index 26a3f65..4085b28 100644
--- a/remoting/base/codec_test.cc
+++ b/remoting/base/codec_test.cc
@@ -10,6 +10,7 @@
 #include "remoting/base/codec_test.h"
 #include "remoting/base/encoder.h"
 #include "remoting/base/mock_objects.h"
+#include "remoting/base/protocol_util.h"
 #include "testing/gtest/include/gtest/gtest.h"
 
 static const int kWidth = 320;
@@ -22,7 +23,7 @@ static const gfx::Rect kTestRects[] = {
   gfx::Rect(0, 0, kWidth / 2, kHeight / 2),
   gfx::Rect(kWidth / 2, kHeight / 2, kWidth / 2, kHeight / 2),
   gfx::Rect(16, 16, 16, 16),
-  gfx::Rect(128, 64, 32, 32)
+  gfx::Rect(128, 64, 32, 32),
 };
 
 namespace remoting {
@@ -70,6 +71,7 @@ class EncoderMessageTester {
       : begin_rect_(0),
         rect_data_(0),
         end_rect_(0),
+        added_rects_(0),
         state_(kWaitingForBeginRect),
         strict_(false) {
   }
@@ -77,6 +79,9 @@ class EncoderMessageTester {
   ~EncoderMessageTester() {
     EXPECT_EQ(begin_rect_, end_rect_);
     EXPECT_EQ(kWaitingForBeginRect, state_);
+    if (strict_){
+      EXPECT_EQ(begin_rect_, added_rects_);
+    }
   }
 
   // Test that we received the correct message.
@@ -123,6 +128,7 @@ class EncoderMessageTester {
 
   void AddRects(const gfx::Rect* rects, int count) {
     rects_.insert(rects_.begin() + rects_.size(), rects, rects + count);
+    added_rects_ += count;
   }
 
  private:
@@ -134,6 +140,7 @@ class EncoderMessageTester {
   int begin_rect_;
   int rect_data_;
   int end_rect_;
+  int added_rects_;
   State state_;
   bool strict_;
 
@@ -172,7 +179,6 @@ class DecoderTester {
       decoder_->EndDecode();
     }
     delete message;
-    return;
   }
 
   void set_strict(bool strict) {
@@ -194,11 +200,28 @@ class DecoderTester {
   void OnPartialDecodeDone() {
     if (!strict_)
       return;
+
+    // Test the content of the update rect.
     for (size_t i = 0; i < update_rects_.size(); ++i) {
+      LOG(INFO) << "Testing Rect " << i;
       EXPECT_FALSE(rects_.empty());
       gfx::Rect rect = rects_.front();
       rects_.pop_front();
       EXPECT_EQ(rect, update_rects_[i]);
+
+      EXPECT_EQ(frame_->stride(0), capture_data_->data_planes().strides[0]);
+      const int stride = frame_->stride(0);
+      const int offset =  stride * update_rects_[i].y() +
+          kBytesPerPixel * update_rects_[i].x();
+      const uint8* original = capture_data_->data_planes().data[0] + offset;
+      const uint8* decoded = frame_->data(0) + offset;
+      const int row_size = kBytesPerPixel * update_rects_[i].width();
+      for (int y = 0; y < update_rects_[i].height(); ++y) {
+        EXPECT_EQ(0, memcmp(original, decoded, row_size))
+            << "Row " << y << " is different";
+        original += stride;
+        decoded += stride;
+      }
     }
   }
 
@@ -354,6 +377,22 @@ static void TestEncodingRects(Encoder* encoder,
   decoder_tester->AddRects(rects, count);
   decoder_tester->reset_decode_done();
 
+  // Generate random data for the updated rects.
+  srand(0);
+  for (int i = 0; i < count; ++i) {
+    const gfx::Rect rect = rects[i];
+    const int bytes_per_pixel = GetBytesPerPixel(data->pixel_format());
+    const int row_size = bytes_per_pixel * rect.width();
+    uint8* memory = data->data_planes().data[0] +
+                    data->data_planes().strides[0] * rect.y() +
+                    bytes_per_pixel * rect.x();
+    for (int y = 0; y < rect.height(); ++y) {
+      for (int x = 0; x < row_size; ++x)
+        memory[x] = rand() % 256;
+      memory += data->data_planes().strides[0];
+    }
+  }
+
   encoder->Encode(data, true,
                   NewCallback(encoder_tester, &EncoderTester::DataAvailable));
   EXPECT_TRUE(decoder_tester->decode_done());
diff --git a/remoting/base/compressor.h b/remoting/base/compressor.h
index fd4c412..edb64f8 100644
--- a/remoting/base/compressor.h
+++ b/remoting/base/compressor.h
@@ -16,6 +16,13 @@ namespace remoting {
 // lifetime. This object should be destroyed after use.
 class Compressor {
  public:
+
+  // Defines the flush modes for a compressor.
+  enum CompressorFlush {
+    CompressorNoFlush,
+    CompressorSyncFlush,
+    CompressorFinish,
+  };
   virtual ~Compressor() {}
 
   // Compress |input_data| with |input_size| bytes.
@@ -23,7 +30,14 @@ class Compressor {
   // |output_data| is provided by the caller and |output_size| is the
   // size of |output_data|. |output_size| must be greater than 0.
   //
-  // |input_size| is set to 0 to indicate the end of input stream.
+  // |flush| is set to one of the three value:
+  // - CompressorNoFlush
+  //   No flushing is requested
+  // - CompressorSyncFlush
+  //   Write all pending output and write a synchronization point in the
+  //   output data stream.
+  // - CompressorFinish
+  //   Mark the end of stream.
   //
   // Compressed data is written to |output_data|. |consumed| will
   // contain the number of bytes consumed from the input. |written|
@@ -34,7 +48,7 @@ class Compressor {
   // useful for end of the compression stream.
   virtual bool Process(const uint8* input_data, int input_size,
                        uint8* output_data, int output_size,
-                       int* consumed, int* written) = 0;
+                       CompressorFlush flush, int* consumed, int* written) = 0;
 };
 
 }  // namespace remoting
diff --git a/remoting/base/compressor_zlib.cc b/remoting/base/compressor_zlib.cc
index ae5b388..6f700731 100644
--- a/remoting/base/compressor_zlib.cc
+++ b/remoting/base/compressor_zlib.cc
@@ -37,7 +37,8 @@ CompressorZlib::~CompressorZlib() {
 
 bool CompressorZlib::Process(const uint8* input_data, int input_size,
                              uint8* output_data, int output_size,
-                             int* consumed, int* written) {
+                             CompressorFlush flush, int* consumed,
+                             int* written) {
   DCHECK_GT(output_size, 0);
 
   // Setup I/O parameters.
@@ -46,7 +47,18 @@ bool CompressorZlib::Process(const uint8* input_data, int input_size,
   stream_->avail_out = output_size;
   stream_->next_out = (Bytef*)output_data;
 
-  int ret = deflate(stream_.get(), input_size ? Z_NO_FLUSH : Z_FINISH);
+  int z_flush = 0;
+  if (flush == CompressorSyncFlush) {
+    z_flush = Z_SYNC_FLUSH;
+  } else if (flush == CompressorFinish) {
+    z_flush = Z_FINISH;
+  } else if (flush == CompressorNoFlush) {
+    z_flush = Z_NO_FLUSH;
+  } else {
+    NOTREACHED() << "Unsupported flush mode";
+  }
+
+  int ret = deflate(stream_.get(), z_flush);
   if (ret == Z_STREAM_ERROR) {
     NOTREACHED() << "zlib compression failed";
   }
@@ -54,9 +66,19 @@ bool CompressorZlib::Process(const uint8* input_data, int input_size,
   *consumed = input_size - stream_->avail_in;
   *written = output_size - stream_->avail_out;
 
-  // Return true when we get Z_BUF_ERROR, this way we can feed more data
-  // to zlib.
-  return ret == Z_OK || ret == Z_BUF_ERROR;
+  // If |ret| equals Z_STREAM_END we have reached the end of stream.
+  // If |ret| equals Z_BUF_ERROR we have the end of the synchronication point.
+  // For these two cases we need to stop compressing.
+  if (ret == Z_OK) {
+    return true;
+  } else if (ret == Z_STREAM_END) {
+    return false;
+  } else if (ret == Z_BUF_ERROR) {
+    return stream_->avail_out == 0;
+  } else {
+    NOTREACHED() << "Unexpected zlib error: " << ret;
+    return false;
+  }
 }
 
 }  // namespace remoting
diff --git a/remoting/base/compressor_zlib.h b/remoting/base/compressor_zlib.h
index 5c8b91c..d2cc82b 100644
--- a/remoting/base/compressor_zlib.h
+++ b/remoting/base/compressor_zlib.h
@@ -21,7 +21,7 @@ class CompressorZlib : public Compressor {
   // Compressor implementations.
   virtual bool Process(const uint8* input_data, int input_size,
                        uint8* output_data, int output_size,
-                       int* consumed, int* written);
+                       CompressorFlush flush, int* consumed, int* written);
 
  private:
   scoped_ptr<z_stream> stream_;
diff --git a/remoting/base/compressor_zlib_unittest.cc b/remoting/base/compressor_zlib_unittest.cc
index 8127350..0b9eade 100644
--- a/remoting/base/compressor_zlib_unittest.cc
+++ b/remoting/base/compressor_zlib_unittest.cc
@@ -7,6 +7,8 @@
 #include "remoting/base/compressor_zlib.h"
 #include "testing/gtest/include/gtest/gtest.h"
 
+namespace remoting {
+
 static void GenerateTestData(uint8* data, int size, int seed) {
   srand(seed);
   for (int i = 0; i < size; ++i)
@@ -23,9 +25,11 @@ static void Compress(remoting::Compressor* compressor,
   // This loop will rewrite |output_data| continuously.
   int consumed = 0;
   int written = 0;
-  while (compressor->Process(input_data, input_size,
-                             output_data, output_size,
-                             &consumed, &written)) {
+  while (compressor->Process(
+             input_data, input_size, output_data, output_size,
+             input_size == 0 ?
+                 Compressor::CompressorFinish : Compressor::CompressorNoFlush,
+             &consumed, &written)) {
     input_data += consumed;
     input_size -= consumed;
   }
@@ -62,3 +66,5 @@ TEST(CompressorZlibTest, SmallOutputBuffer) {
   Compress(&compressor, raw_data, kRawDataSize,
            compressed_data, kCompressedDataSize);
 }
+
+}  // namespace remoting
diff --git a/remoting/base/decoder_zlib.cc b/remoting/base/decoder_zlib.cc
new file mode 100644
index 0000000..59f9852
--- /dev/null
+++ b/remoting/base/decoder_zlib.cc
@@ -0,0 +1,143 @@
+// Copyright (c) 2010 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#include "remoting/base/decoder_zlib.h"
+
+#include "remoting/base/decompressor_zlib.h"
+#include "remoting/base/protocol_util.h"
+
+namespace remoting {
+
+DecoderZlib::DecoderZlib()
+    : state_(kWaitingForBeginRect),
+      rect_x_(0),
+      rect_y_(0),
+      rect_width_(0),
+      rect_height_(0),
+      bytes_per_pixel_(0),
+      updated_rects_(NULL),
+      row_pos_(0),
+      row_y_(0) {
+}
+
+bool DecoderZlib::BeginDecode(scoped_refptr<media::VideoFrame> frame,
+                                  UpdatedRects* updated_rects,
+                                  Task* partial_decode_done,
+                                  Task* decode_done) {
+  DCHECK(!partial_decode_done_.get());
+  DCHECK(!decode_done_.get());
+  DCHECK(!updated_rects_);
+  DCHECK_EQ(kWaitingForBeginRect, state_);
+  CHECK(static_cast<PixelFormat>(frame->format()) == PixelFormatRgb32)
+      << "Only RGB32 is supported";
+
+  partial_decode_done_.reset(partial_decode_done);
+  decode_done_.reset(decode_done);
+  updated_rects_ = updated_rects;
+  frame_ = frame;
+
+  // Create the decompressor.
+  decompressor_.reset(new DecompressorZlib());
+  return true;
+}
+
+bool DecoderZlib::PartialDecode(HostMessage* message) {
+  scoped_ptr<HostMessage> msg_deleter(message);
+  DCHECK(message->has_update_stream_packet());
+
+  bool ret = true;
+  if (message->update_stream_packet().has_begin_rect())
+    ret = HandleBeginRect(message);
+  if (ret && message->update_stream_packet().has_rect_data())
+    ret = HandleRectData(message);
+  if (ret && message->update_stream_packet().has_end_rect())
+    ret = HandleEndRect(message);
+  return ret;
+}
+
+void DecoderZlib::EndDecode() {
+  DCHECK_EQ(kWaitingForBeginRect, state_);
+  decode_done_->Run();
+
+  partial_decode_done_.reset();
+  decode_done_.reset();
+  updated_rects_ = NULL;
+  frame_ = NULL;
+  decompressor_.reset();
+}
+
+bool DecoderZlib::HandleBeginRect(HostMessage* message) {
+  DCHECK_EQ(kWaitingForBeginRect, state_);
+  state_ = kWaitingForRectData;
+
+  rect_width_ = message->update_stream_packet().begin_rect().width();
+  rect_height_ = message->update_stream_packet().begin_rect().height();
+  rect_x_ = message->update_stream_packet().begin_rect().x();
+  rect_y_ = message->update_stream_packet().begin_rect().y();
+
+  PixelFormat pixel_format =
+      message->update_stream_packet().begin_rect().pixel_format();
+
+  if (static_cast<PixelFormat>(frame_->format()) != pixel_format) {
+    NOTREACHED() << "Pixel format of message doesn't match the video frame. "
+                    "Expected vs received = "
+                 << frame_->format() << " vs " << pixel_format
+                 << " Color space conversion required.";
+    return false;
+  }
+
+  bytes_per_pixel_ = GetBytesPerPixel(pixel_format);
+  row_pos_ = 0;
+  row_y_ = 0;
+  return true;
+}
+
+bool DecoderZlib::HandleRectData(HostMessage* message) {
+  DCHECK_EQ(kWaitingForRectData, state_);
+  DCHECK_EQ(0,
+            message->update_stream_packet().rect_data().sequence_number());
+
+  const uint8* in =
+      (const uint8*)message->update_stream_packet().rect_data().data().data();
+  const int in_size =
+      message->update_stream_packet().rect_data().data().size();
+  const int row_size = rect_width_ * bytes_per_pixel_;
+  const int stride = frame_->stride(media::VideoFrame::kRGBPlane);
+  uint8* out = frame_->data(media::VideoFrame::kRGBPlane) +
+      stride * (rect_y_ + row_y_) + bytes_per_pixel_ * rect_x_;
+
+  // Consume all the data in the message.
+  bool decompress_again = true;
+  int used = 0;
+  while (decompress_again && used < in_size) {
+    int written = 0;
+    int consumed = 0;
+    decompress_again = decompressor_->Process(
+        in + used, in_size - used, out + row_pos_, row_size - row_pos_,
+        &consumed, &written);
+    used += consumed;
+    row_pos_ += written;
+
+    // If this row is completely filled then move onto the next row.
+    if (row_pos_ == row_size) {
+      ++row_y_;
+      row_pos_ = 0;
+      out += stride;
+    }
+  }
+  return true;
+}
+
+bool DecoderZlib::HandleEndRect(HostMessage* message) {
+  DCHECK_EQ(kWaitingForRectData, state_);
+  state_ = kWaitingForBeginRect;
+
+  updated_rects_->clear();
+  updated_rects_->push_back(gfx::Rect(rect_x_, rect_y_,
+                                      rect_width_, rect_height_));
+  partial_decode_done_->Run();
+  return true;
+}
+
+}  // namespace remoting
diff --git a/remoting/base/decoder_zlib.h b/remoting/base/decoder_zlib.h
new file mode 100644
index 0000000..4d20c10
--- /dev/null
+++ b/remoting/base/decoder_zlib.h
@@ -0,0 +1,63 @@
+// Copyright (c) 2010 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#ifndef REMOTING_BASE_DECODER_ZLIB_H_
+#define REMOTING_BASE_DECODER_ZLIB_H_
+
+#include "remoting/base/decoder.h"
+
+namespace remoting {
+
+class DecompressorZlib;
+
+class DecoderZlib : public Decoder {
+ public:
+  DecoderZlib();
+
+  // Decoder implementations.
+  virtual bool BeginDecode(scoped_refptr<media::VideoFrame> frame,
+                           UpdatedRects* update_rects,
+                           Task* partial_decode_done,
+                           Task* decode_done);
+  virtual bool PartialDecode(HostMessage* message);
+  virtual void EndDecode();
+
+ private:
+  bool HandleBeginRect(HostMessage* message);
+  bool HandleRectData(HostMessage* message);
+  bool HandleEndRect(HostMessage* message);
+
+  // The internal state of the decoder.
+  State state_;
+
+  // Keeps track of the updating rect.
+  int rect_x_;
+  int rect_y_;
+  int rect_width_;
+  int rect_height_;
+  int bytes_per_pixel_;
+
+  // Tasks to call when decode is done.
+  scoped_ptr<Task> partial_decode_done_;
+  scoped_ptr<Task> decode_done_;
+
+  // The video frame to write to.
+  scoped_refptr<media::VideoFrame> frame_;
+  UpdatedRects* updated_rects_;
+
+  // A zlib decompressor used throughout one update sequence.
+  scoped_ptr<DecompressorZlib> decompressor_;
+
+  // The position in the row that we are updating.
+  int row_pos_;
+
+  // The row number in the rect that we are updaing.
+  int row_y_;
+
+  DISALLOW_COPY_AND_ASSIGN(DecoderZlib);
+};
+
+}  // namespace remoting
+
+#endif  // REMOTING_BASE_DECODER_ZLIB_H_
diff --git a/remoting/base/decoder_zlib_unittest.cc b/remoting/base/decoder_zlib_unittest.cc
new file mode 100644
index 0000000..8ebad1d
--- /dev/null
+++ b/remoting/base/decoder_zlib_unittest.cc
@@ -0,0 +1,28 @@
+// Copyright (c) 2010 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#include "media/base/video_frame.h"
+#include "remoting/base/codec_test.h"
+#include "remoting/base/decoder_zlib.h"
+#include "remoting/base/decompressor_zlib.h"
+#include "remoting/base/encoder_zlib.h"
+#include "remoting/client/mock_objects.h"
+#include "testing/gtest/include/gtest/gtest.h"
+
+
+namespace remoting {
+
+TEST(DecoderZlibTest, EncodeAndDecode) {
+  EncoderZlib encoder;
+  DecoderZlib decoder;
+  TestEncoderDecoder(&encoder, &decoder, true);
+}
+
+TEST(DecoderZlibTest, EncodeAndDecodeSmallOutputBuffer) {
+  EncoderZlib encoder(64);
+  DecoderZlib decoder;
+  TestEncoderDecoder(&encoder, &decoder, true);
+}
+
+}  // namespace remoting
diff --git a/remoting/base/decompressor_zlib_unittest.cc b/remoting/base/decompressor_zlib_unittest.cc
index d04a923..aebaa5a 100644
--- a/remoting/base/decompressor_zlib_unittest.cc
+++ b/remoting/base/decompressor_zlib_unittest.cc
@@ -8,6 +8,8 @@
 #include "remoting/base/decompressor_zlib.h"
 #include "testing/gtest/include/gtest/gtest.h"
 
+namespace remoting {
+
 static void GenerateTestData(uint8* data, int size, int seed) {
   srand(seed);
   for (int i = 0; i < size; ++i)
@@ -25,9 +27,11 @@ static void Compress(remoting::Compressor* compressor,
   *compressed_size = 0;
   while (true) {
     int consumed, written;
-    bool ret = compressor->Process(input_data, input_size,
-                                   output_data, output_size,
-                                   &consumed, &written);
+    bool ret = compressor->Process(
+        input_data, input_size, output_data, output_size,
+        input_size == 0 ?
+            Compressor::CompressorFinish : Compressor::CompressorNoFlush,
+        &consumed, &written);
     input_data += consumed;
     input_size -= consumed;
     output_data += written;
@@ -134,3 +138,5 @@ TEST(DecompressorZlibTest, SmallOutputBuffer) {
   }
   EXPECT_EQ(kRawDataSize, decompressed_size);
 }
+
+}  // namespace remoting
diff --git a/remoting/base/encoder_verbatim.cc b/remoting/base/encoder_verbatim.cc
index 509daf9..7f39f27 100644
--- a/remoting/base/encoder_verbatim.cc
+++ b/remoting/base/encoder_verbatim.cc
@@ -67,8 +67,7 @@ bool EncoderVerbatim::EncodeRect(
 
   // Resize the output data buffer.
   packet->mutable_rect_data()->mutable_data()->resize(output_size);
-  uint8* out = reinterpret_cast<uint8*>(
-      &((*packet->mutable_rect_data()->mutable_data())[0]));
+  uint8* out = (uint8*)packet->mutable_rect_data()->mutable_data()->data();
 
   for (int i = 0; i < DataPlanes::kPlaneCount; ++i) {
     // Skip over planes that don't have data.
diff --git a/remoting/base/encoder_zlib.cc b/remoting/base/encoder_zlib.cc
new file mode 100644
index 0000000..848804a
--- /dev/null
+++ b/remoting/base/encoder_zlib.cc
@@ -0,0 +1,138 @@
+// Copyright (c) 2010 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#include "remoting/base/encoder_zlib.h"
+
+#include "gfx/rect.h"
+#include "media/base/data_buffer.h"
+#include "remoting/base/capture_data.h"
+#include "remoting/base/compressor_zlib.h"
+#include "remoting/base/protocol_util.h"
+#include "remoting/base/protocol/chromotocol.pb.h"
+
+namespace remoting {
+
+static const int kPacketSize = 1024 * 1024;
+
+EncoderZlib::EncoderZlib() : packet_size_(kPacketSize) {
+}
+
+EncoderZlib::EncoderZlib(int packet_size) : packet_size_(packet_size) {
+}
+
+void EncoderZlib::Encode(scoped_refptr<CaptureData> capture_data,
+                         bool key_frame,
+                         DataAvailableCallback* data_available_callback) {
+  CHECK(capture_data->pixel_format() == PixelFormatRgb32)
+      << "Zlib Encoder only works with RGB32";
+  capture_data_ = capture_data;
+  callback_.reset(data_available_callback);
+
+  CompressorZlib compressor;
+  for (current_rect_ = 0; current_rect_ < capture_data->dirty_rects().size();
+       ++current_rect_) {
+    EncodeRect(&compressor);
+  }
+
+  capture_data_ = NULL;
+  callback_.reset();
+  current_rect_ = 0;
+}
+
+void EncoderZlib::EncodeRect(CompressorZlib* compressor) {
+  CHECK(capture_data_->data_planes().data[0]);
+  const gfx::Rect rect = capture_data_->dirty_rects()[current_rect_];
+  const int strides = capture_data_->data_planes().strides[0];
+  const int bytes_per_pixel = GetBytesPerPixel(capture_data_->pixel_format());
+  const int row_size = bytes_per_pixel * rect.width();
+
+  HostMessage* message = PrepareMessage(true);
+  const uint8 * in = capture_data_->data_planes().data[0] +
+                     rect.y() * strides +
+                     rect.x() * bytes_per_pixel;
+  // TODO(hclam): Fill in the sequence number.
+  uint8* out = (uint8*)message->mutable_update_stream_packet()->
+      mutable_rect_data()->mutable_data()->data();
+  int filled = 0;
+  int row_x = 0;
+  int row_y = 0;
+  bool compress_again = true;
+  while (compress_again) {
+    // Prepare a message for sending out.
+    if (!message) {
+      message = PrepareMessage(false);
+      out = (uint8*)(message->mutable_update_stream_packet()->
+                     mutable_rect_data()->mutable_data()->data());
+      filled = 0;
+    }
+
+    Compressor::CompressorFlush flush = Compressor::CompressorNoFlush;
+    if (row_y == rect.height() - 1) {
+      if (current_rect_ == capture_data_->dirty_rects().size() - 1)
+        flush = Compressor::CompressorFinish;
+      else
+        flush = Compressor::CompressorSyncFlush;
+    }
+
+    int consumed = 0;
+    int written = 0;
+    compress_again = compressor->Process(in + row_x, row_size - row_x,
+                                         out + filled, packet_size_ - filled,
+                                         flush, &consumed, &written);
+    row_x += consumed;
+    filled += written;
+
+    // We have reached the end of stream.
+    if (!compress_again) {
+      message->mutable_update_stream_packet()->mutable_end_rect();
+    }
+
+    // If we have filled the message or we have reached the end of stream.
+    if (filled == packet_size_ || !compress_again) {
+      message->mutable_update_stream_packet()->mutable_rect_data()->
+          mutable_data()->resize(filled);
+      SubmitMessage(message);
+      message = NULL;
+    }
+
+    // Reached the end of input row and we're not at the last row.
+    if (row_x == row_size && row_y < rect.height() - 1) {
+      row_x = 0;
+      in += strides;
+      ++row_y;
+    }
+  }
+}
+
+HostMessage* EncoderZlib::PrepareMessage(bool new_rect) {
+  HostMessage* message = new HostMessage();
+  UpdateStreamPacketMessage* packet = message->mutable_update_stream_packet();
+
+  // Prepare the begin rect content.
+  if (new_rect) {
+    gfx::Rect rect = capture_data_->dirty_rects()[current_rect_];
+    packet->mutable_begin_rect()->set_x(rect.x());
+    packet->mutable_begin_rect()->set_y(rect.y());
+    packet->mutable_begin_rect()->set_width(rect.width());
+    packet->mutable_begin_rect()->set_height(rect.height());
+    packet->mutable_begin_rect()->set_encoding(EncodingZlib);
+    packet->mutable_begin_rect()->set_pixel_format(
+        capture_data_->pixel_format());
+  }
+
+  packet->mutable_rect_data()->mutable_data()->resize(packet_size_);
+  return message;
+}
+
+void EncoderZlib::SubmitMessage(HostMessage* message) {
+  EncodingState state = EncodingInProgress;
+  if (current_rect_ == 0 && message->update_stream_packet().has_begin_rect())
+    state |= EncodingStarting;
+  if (current_rect_ == capture_data_->dirty_rects().size() - 1 &&
+      message->update_stream_packet().has_end_rect())
+    state |= EncodingEnded;
+  callback_->Run(message, state);
+}
+
+}  // namespace remoting
diff --git a/remoting/base/encoder_zlib.h b/remoting/base/encoder_zlib.h
new file mode 100644
index 0000000..9266e2b1
--- /dev/null
+++ b/remoting/base/encoder_zlib.h
@@ -0,0 +1,46 @@
+// Copyright (c) 2010 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#ifndef REMOTING_BASE_ENCODER_ZLIB_H_
+#define REMOTING_BASE_ENCODER_ZLIB_H_
+
+#include "remoting/base/encoder.h"
+
+namespace remoting {
+
+class CompressorZlib;
+class UpdateStreamPacket;
+
+// EncoderZlib implements an Encoder using Zlib for compression.
+class EncoderZlib : public Encoder {
+ public:
+  EncoderZlib();
+  EncoderZlib(int packet_size);
+
+  virtual ~EncoderZlib() {}
+
+  virtual void Encode(scoped_refptr<CaptureData> capture_data,
+                      bool key_frame,
+                      DataAvailableCallback* data_available_callback);
+
+ private:
+  // Encode a single dirty rect using compressor.
+  void EncodeRect(CompressorZlib* compressor);
+
+  // Create a new HostMessage with the right flag and attributes. The message
+  // can be used immediately for output of encoding.
+  HostMessage* PrepareMessage(bool new_rect);
+
+  // Submit |message| to |callback_|.
+  void SubmitMessage(HostMessage* message);
+
+  scoped_refptr<CaptureData> capture_data_;
+  scoped_ptr<DataAvailableCallback> callback_;
+  size_t current_rect_;
+  int packet_size_;
+};
+
+}  // namespace remoting
+
+#endif  // REMOTING_BASE_ENCODER_ZLIB_H_
diff --git a/remoting/base/encoder_zlib_unittest.cc b/remoting/base/encoder_zlib_unittest.cc
new file mode 100644
index 0000000..8f54e82
--- /dev/null
+++ b/remoting/base/encoder_zlib_unittest.cc
@@ -0,0 +1,22 @@
+// Copyright (c) 2010 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#include "remoting/base/codec_test.h"
+#include "remoting/base/encoder_zlib.h"
+#include "testing/gtest/include/gtest/gtest.h"
+
+namespace remoting {
+
+TEST(EncoderZlibTest, TestEncoder) {
+  EncoderZlib encoder;
+  TestEncoder(&encoder, true);
+}
+
+
+TEST(EncoderZlibTest, TestEncoderSmallOutputBuffer) {
+  EncoderZlib encoder(16);
+  TestEncoder(&encoder, true);
+}
+
+}  // namespace remoting
diff --git a/remoting/remoting.gyp b/remoting/remoting.gyp
index c7edcea..a0f3ea1 100644
--- a/remoting/remoting.gyp
+++ b/remoting/remoting.gyp
@@ -138,12 +138,16 @@
         'base/decoder.h',
         'base/decoder_verbatim.cc',
         'base/decoder_verbatim.h',
+        'base/decoder_zlib.cc',
+        'base/decoder_zlib.h',
         'base/decompressor.h',
         'base/decompressor_zlib.cc',
         'base/decompressor_zlib.h',
         'base/encoder.h',
         'base/encoder_verbatim.cc',
         'base/encoder_verbatim.h',
+        'base/encoder_zlib.cc',
+        'base/encoder_zlib.h',
         # TODO(hclam): Enable VP8 in the build.
         #'base/encoder_vp8.cc',
         #'base/encoder_vp8.h',
@@ -346,10 +350,12 @@
         'base/codec_test.h',
         'base/compressor_zlib_unittest.cc',
         'base/decoder_verbatim_unittest.cc',
+        'base/decoder_zlib_unittest.cc',
         'base/decompressor_zlib_unittest.cc',
         'base/encoder_verbatim_unittest.cc',
         # TODO(hclam): Enable VP8 in the build.
         #'base/encoder_vp8_unittest.cc',
+        'base/encoder_zlib_unittest.cc',
         'base/mock_objects.h',
         'base/multiple_array_input_stream_unittest.cc',
         'base/protocol_decoder_unittest.cc',