From 1ffb61beaafedb160a971732e599d5c6aa67a646 Mon Sep 17 00:00:00 2001
From: David Reiss <dreiss@apache.org>
Date: Tue, 8 Apr 2008 05:07:26 +0000
Subject: [PATCH] Fix a bug in TPipedTransport that broke pipelining.

Previously, TPipedTransport wrote it's full buffer to the "pipe" and
fully reset its buffer on a readEnd.  This assumed that the buffer was
fully read at that point.  This is not the case if requests are pipelined.
This change makes it only pipe out the portion that has been read and
copy the unread portion to the beginning of the buffer.

Also add a test that verifies the new functionality.


git-svn-id: https://svn.apache.org/repos/asf/incubator/thrift/trunk@665646 13f79535-47bb-0310-9956-ffa450edef68
---
 lib/cpp/src/transport/TTransportUtils.h |  9 ++++---
 test/Makefile.am                        | 11 ++++++++
 test/TPipedTransportTest.cpp            | 34 +++++++++++++++++++++++++
 3 files changed, 51 insertions(+), 3 deletions(-)
 create mode 100644 test/TPipedTransportTest.cpp

diff --git a/lib/cpp/src/transport/TTransportUtils.h b/lib/cpp/src/transport/TTransportUtils.h
index e54c1b8f..4360f86a 100644
--- a/lib/cpp/src/transport/TTransportUtils.h
+++ b/lib/cpp/src/transport/TTransportUtils.h
@@ -616,15 +616,18 @@ class TPipedTransport : virtual public TTransport {
   void readEnd() {
 
     if (pipeOnRead_) {
-      dstTrans_->write(rBuf_, rLen_);
+      dstTrans_->write(rBuf_, rPos_);
       dstTrans_->flush();
     }
 
     srcTrans_->readEnd();
 
-    // reset state
-    rLen_ = 0;
+    // If requests are being pipelined, copy down our read-ahead data,
+    // then reset our state.
+    int read_ahead = rLen_ - rPos_;
+    memcpy(rBuf_, rBuf_ + rPos_, read_ahead);
     rPos_ = 0;
+    rLen_ = read_ahead;
   }
 
   void write(const uint8_t* buf, uint32_t len);
diff --git a/test/Makefile.am b/test/Makefile.am
index 5490de84..f2ff45f4 100644
--- a/test/Makefile.am
+++ b/test/Makefile.am
@@ -28,6 +28,7 @@ Benchmark_LDADD = libtestgencpp.la
 
 check_PROGRAMS = \
 	TFDTransportTest \
+	TPipedTransportTest \
 	DebugProtoTest \
 	JSONProtoTest \
 	OptionalRequiredTest \
@@ -52,6 +53,16 @@ TFDTransportTest_SOURCES = \
 TFDTransportTest_LDADD = \
 	$(top_srcdir)/lib/cpp/libthrift.la
 
+
+#
+# TPipedTransportTest
+#
+TPipedTransportTest_SOURCES = \
+	TPipedTransportTest.cpp
+
+TPipedTransportTest_LDADD = \
+	$(top_srcdir)/lib/cpp/libthrift.la
+
 #
 # DebugProtoTest
 #
diff --git a/test/TPipedTransportTest.cpp b/test/TPipedTransportTest.cpp
new file mode 100644
index 00000000..c1b371da
--- /dev/null
+++ b/test/TPipedTransportTest.cpp
@@ -0,0 +1,34 @@
+#include <cstdlib>
+#include <stdexcept>
+#include <Thrift.h>
+#include <transport/TTransportUtils.h>
+using namespace std;
+using boost::shared_ptr;
+using facebook::thrift::transport::TTransportException;
+using facebook::thrift::transport::TPipedTransport;
+using facebook::thrift::transport::TMemoryBuffer;
+
+int main() {
+  shared_ptr<TMemoryBuffer> underlying(new TMemoryBuffer);
+  shared_ptr<TMemoryBuffer> pipe(new TMemoryBuffer);
+  shared_ptr<TPipedTransport> trans(new TPipedTransport(underlying, pipe));
+
+  uint8_t buffer[4];
+
+  underlying->write((uint8_t*)"abcd", 4);
+  trans->readAll(buffer, 2);
+  assert( string((char*)buffer, 2) == "ab" );
+  trans->readEnd();
+  assert( pipe->getBufferAsString() == "ab" );
+  pipe->resetBuffer();
+  underlying->write((uint8_t*)"ef", 2);
+  trans->readAll(buffer, 2);
+  assert( string((char*)buffer, 2) == "cd" );
+  trans->readAll(buffer, 2);
+  assert( string((char*)buffer, 2) == "ef" );
+  trans->readEnd();
+  assert( pipe->getBufferAsString() == "cdef" );
+
+  return 0;
+
+}
-- 
2.17.1