Github user ben-craig commented on a diff in the pull request:
https://github.com/apache/thrift/pull/357#discussion_r23307653
--- Diff: lib/cpp/src/thrift/transport/THeaderTransport.h ---
@@ -0,0 +1,308 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#ifndef THRIFT_TRANSPORT_THEADERTRANSPORT_H_
+#define THRIFT_TRANSPORT_THEADERTRANSPORT_H_ 1
+
+#include <tr1/functional>
+
+#include <thrift/protocol/TBinaryProtocol.h>
+#include <thrift/protocol/TProtocolTypes.h>
+#include <thrift/transport/TBufferTransports.h>
+#include <thrift/transport/TTransport.h>
+#include <thrift/transport/TVirtualTransport.h>
+
+#include <bitset>
+#include "boost/scoped_array.hpp"
+#include <pwd.h>
+#include <unistd.h>
+
+// Don't include the unknown client.
+#define CLIENT_TYPES_LEN 4
+
+enum CLIENT_TYPE {
+ THRIFT_HEADER_CLIENT_TYPE = 0,
+ THRIFT_FRAMED_DEPRECATED = 1,
+ THRIFT_UNFRAMED_DEPRECATED = 2,
+ THRIFT_UNKNOWN_CLIENT_TYPE = 4,
+};
+
+namespace apache { namespace thrift { namespace transport {
+
+using apache::thrift::protocol::T_COMPACT_PROTOCOL;
+
+/**
+ * Header transport. All writes go into an in-memory buffer until flush is
+ * called, at which point the transport writes the length of the entire
+ * binary chunk followed by the data payload. This allows the receiver on
the
+ * other end to always do fixed-length reads.
+ *
+ * Subclass TFramedTransport because most of the read/write methods are
similar
+ * and need similar buffers. Major changes are readFrame & flush.
+ */
+class THeaderTransport
+ : public TVirtualTransport<THeaderTransport, TFramedTransport> {
+ public:
+
+ static const int DEFAULT_BUFFER_SIZE = 512u;
+
+ /// Use default buffer sizes.
+ explicit THeaderTransport(const boost::shared_ptr<TTransport> transport)
+ : transport_(transport)
+ , outTransport_(transport)
+ , protoId(T_COMPACT_PROTOCOL)
+ , clientType(THRIFT_HEADER_CLIENT_TYPE)
+ , seqId(0)
+ , flags(0)
+ , tBufSize_(0)
+ , tBuf_(NULL)
+ {
+ initBuffers();
+ }
+
+ THeaderTransport(const boost::shared_ptr<TTransport> inTransport,
+ const boost::shared_ptr<TTransport> outTransport)
+ : transport_(inTransport)
+ , outTransport_(outTransport)
+ , protoId(T_COMPACT_PROTOCOL)
+ , clientType(THRIFT_HEADER_CLIENT_TYPE)
+ , seqId(0)
+ , flags(0)
+ , tBufSize_(0)
+ , tBuf_(NULL)
+ {
+ initBuffers();
+ }
+
+ void open() {
+ transport_->open();
+ }
+
+ bool isOpen() {
+ return transport_->isOpen();
+ }
+
+ bool peek() {
+ return (this->rBase_ < this->rBound_) || transport_->peek();
+ }
+
+ void close() {
+ flush();
+ transport_->close();
+ }
+
+ virtual uint32_t readSlow(uint8_t* buf, uint32_t len);
+ virtual uint32_t readAll(uint8_t* buf, uint32_t len);
+ virtual void flush();
+
+ void resizeTransformBuffer();
+
+ boost::shared_ptr<TTransport> getUnderlyingTransport() {
+ return transport_;
+ }
+
+ /*
+ * TVirtualTransport provides a default implementation of readAll().
+ * We want to use the TBufferBase version instead.
+ */
+ using TBufferBase::readAll;
+
+ uint16_t getProtocolId() const;
+ void setProtocolId(uint16_t protoId) { this->protoId = protoId; }
+
+ void resetProtocol();
+
+ /**
+ * We know we got a packet in header format here, try to parse the header
+ *
+ * @param headerSize size of the header portion
+ * @param sz Size of the whole message, including header
+ */
+ void readHeaderFormat(uint16_t headerSize, uint32_t sz);
+
+ /**
+ * Untransform the data based on the recieved header flags
+ * On conclusion of function, setReadBuffer is called with the
+ * untransformed data.
+ *
+ * @param ptr ptr to data
+ * @param size of data
+ */
+ void untransform(uint8_t* ptr, uint32_t sz);
+
+ /**
+ * Transform the data based on our write transform flags
+ * At conclusion of function the write buffer is set to the
+ * transformed data.
+ *
+ * @param ptr Ptr to data to transform
+ * @param sz Size of data buffer
+ */
+ void transform(uint8_t* ptr, uint32_t sz);
+
+ uint16_t getNumTransforms() const {
+ int trans = writeTrans_.size();
+ return trans;
+ }
+
+ void setTransform(uint16_t transId) { writeTrans_.push_back(transId); }
+
+ // Info headers
+
+ typedef std::map<std::string, std::string> StringToStringMap;
+
+ // these work with write headers
+ void setHeader(const std::string& key, const std::string& value);
+
+ void clearHeaders();
+
+ StringToStringMap& getWriteHeaders() {
+ return writeHeaders_;
+ }
+
+ // these work with read headers
+ const StringToStringMap& getHeaders() const {
+ return readHeaders_;
+ }
+
+ // accessors for seqId
+ int32_t getSequenceNumber() const { return seqId; }
+ void setSequenceNumber(int32_t seqId) { this->seqId = seqId; }
+
+ enum TRANSFORMS {
+ ZLIB_TRANSFORM = 0x01,
+ };
+
+ protected:
+
+ std::bitset<CLIENT_TYPES_LEN> supported_clients;
+
+ void initSupportedClients(std::bitset<CLIENT_TYPES_LEN> const*);
+
+ /**
+ * Reads a frame of input from the underlying stream.
+ *
+ * Returns true if a frame was read successfully, or false on EOF.
+ * (Raises a TTransportException if EOF occurs after a partial frame.)
+ */
+ bool readFrame(uint32_t minFrameSize);
+
+ void allocateReadBuffer(uint32_t sz);
+ uint32_t getWriteBytes();
+
+ void initBuffers() {
+ setReadBuffer(NULL, 0);
+ setWriteBuffer(wBuf_.get(), wBufSize_);
+ }
+
+ boost::shared_ptr<TTransport> transport_;
+ boost::shared_ptr<TTransport> outTransport_;
+
+ // 0 and 16th bits must be 0 to differentiate from framed & unframed
+ static const uint32_t HEADER_MAGIC = 0x0FFF0000;
+ static const uint32_t HEADER_MASK = 0xFFFF0000;
+ static const uint32_t FLAGS_MASK = 0x0000FFFF;
+
+ static const uint32_t MAX_FRAME_SIZE = 0x3FFFFFFF;
+
+ int16_t protoId;
+ uint16_t clientType;
+ uint32_t seqId;
+ uint16_t flags;
+
+ std::vector<uint16_t> readTrans_;
+ std::vector<uint16_t> writeTrans_;
+
+ // Map to use for headers
+ StringToStringMap readHeaders_;
+ StringToStringMap writeHeaders_;
+
+ /**
+ * Returns the maximum number of bytes that write k/v headers can take
+ */
+ size_t getMaxWriteHeadersSize() const;
+
+ struct infoIdType {
+ enum idType {
+ // start at 1 to avoid confusing header padding for an infoId
+ KEYVALUE = 1,
+ END // signal the end of infoIds we can handle
+ };
+ };
+
+ // Buffers to use for transform processing
+ uint32_t tBufSize_;
+ boost::scoped_array<uint8_t> tBuf_;
+
+ void readString(uint8_t* &ptr, /* out */ std::string &str,
+ uint8_t const* headerBoundary);
+
+ void writeString(uint8_t* &ptr, const std::string& str);
+
+ // Varint utils
+ /**
+ * Read an i16 from the wire as a varint. The MSB of each byte is set
+ * if there is another byte to follow. This can read up to 3 bytes.
+ */
+ uint32_t readVarint16(uint8_t const* ptr, int16_t* i16,
+ uint8_t const* boundary);
+
+ /**
+ * Read an i32 from the wire as a varint. The MSB of each byte is set
+ * if there is another byte to follow. This can read up to 5 bytes.
+ */
+ uint32_t readVarint32(uint8_t const* ptr, int32_t* i32,
+ uint8_t const* boundary);
+
+ /**
+ * Write an i32 as a varint. Results in 1-5 bytes on the wire.
+ */
+ uint32_t writeVarint32(uint32_t n, uint8_t* pkt);
--- End diff --
If this is writing an i32... 'n' should be an int32_t instead of a uint32_t.
Similar commends on writeVarint16.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
