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);
check_PROGRAMS = \
TFDTransportTest \
+ TPipedTransportTest \
DebugProtoTest \
JSONProtoTest \
OptionalRequiredTest \
TFDTransportTest_LDADD = \
$(top_srcdir)/lib/cpp/libthrift.la
+
+#
+# TPipedTransportTest
+#
+TPipedTransportTest_SOURCES = \
+ TPipedTransportTest.cpp
+
+TPipedTransportTest_LDADD = \
+ $(top_srcdir)/lib/cpp/libthrift.la
+
#
# DebugProtoTest
#
--- /dev/null
+#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;
+
+}