lang/c++/api/Stream.hh - avro - Git at Google

 /*
  * 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
  *
  *     https://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 avro_Stream_hh__
 #define avro_Stream_hh__

 #include <memory>
 #include <string.h>
 #include <stdint.h>

 #include "boost/utility.hpp"

 #include "Config.hh"
 #include "Exception.hh"

 namespace avro {

 /**
  * A no-copy input stream.
  */
 class AVRO_DECL InputStream : boost::noncopyable {
 protected:

     /**
      * An empty constuctor.
      */
     InputStream() { }

 public:
     /**
      * Destructor.
      */
     virtual ~InputStream() { }

     /**
      * Returns some of available data.
      *
      * Returns true if some data is available, false if no more data is
      * available or an error has occurred.
      */
     virtual bool next(const uint8_t** data, size_t* len) = 0;

     /**
      * "Returns" back some of the data to the stream. The returned
      * data must be less than what was obtained in the last call to
      * next().
      */
     virtual void backup(size_t len) = 0;

     /**
      * Skips number of bytes specified by len.
      */
     virtual void skip(size_t len) = 0;

     /**
      * Returns the number of bytes read from this stream so far.
      * All the bytes made available through next are considered
      * to be used unless, retutned back using backup.
      */
     virtual size_t byteCount() const = 0;
 };

 typedef std::unique_ptr<InputStream> InputStreamPtr;

 /**
  * An InputStream which also supports seeking to a specific offset.
  */
 class AVRO_DECL SeekableInputStream : public InputStream {
 protected:

     /**
      * An empty constuctor.
      */
     SeekableInputStream() { }

 public:
     /**
      * Destructor.
      */
     virtual ~SeekableInputStream() { }

     /**
      * Seek to a specific position in the stream. This may invalidate pointers
      * returned from next(). This will also reset byteCount() to the given
      * position.
      */
     virtual void seek(int64_t position) = 0;
 };

 typedef std::unique_ptr<SeekableInputStream> SeekableInputStreamPtr;

 /**
  * A no-copy output stream.
  */
 class AVRO_DECL OutputStream : boost::noncopyable {
 protected:

     /**
      * An empty constuctor.
      */
     OutputStream() { }
 public:

     /**
      * Destructor.
      */
     virtual ~OutputStream() { }

     /**
      * Returns a buffer that can be written into.
      * On successful return, data has the pointer to the buffer
      * and len has the number of bytes available at data.
      */
     virtual bool next(uint8_t** data, size_t* len) = 0;

     /**
      * "Returns" back to the stream some of the buffer obtained
      * from in the last call to next().
      */
     virtual void backup(size_t len) = 0;

     /**
      * Number of bytes written so far into this stream. The whole buffer
      * returned by next() is assumed to be written unless some of
      * it was retutned using backup().
      */
     virtual uint64_t byteCount() const = 0;

     /**
      * Flushes any data remaining in the buffer to the stream's underlying
      * store, if any.
      */
     virtual void flush() = 0;
 };

 typedef std::unique_ptr<OutputStream> OutputStreamPtr;

 /**
  * Returns a new OutputStream, which grows in memory chunks of specified size.
  */
 AVRO_DECL OutputStreamPtr memoryOutputStream(size_t chunkSize = 4 * 1024);

 /**
  * Returns a new InputStream, with the data from the given byte array.
  * It does not copy the data, the byte array should remain valid
  * until the InputStream is used.
  */
 AVRO_DECL InputStreamPtr memoryInputStream(const uint8_t* data, size_t len);

 /**
  * Returns a new InputStream with the contents written into an
  * outputstream. The output stream must have been returned by
  * an earlier call to memoryOutputStream(). The contents for the new
  * input stream are the snapshot of the outputstream. One can construct
  * any number of memory input stream from a single memory output stream.
  */
 AVRO_DECL InputStreamPtr memoryInputStream(const OutputStream& source);

 /**
  * Returns the contents written so far into the output stream, which should
  * be a memory output stream. That is it must have been returned by a pervious
  * call to memoryOutputStream().
  */
 AVRO_DECL std::shared_ptr<std::vector<uint8_t> > snapshot(const OutputStream& source);

 /**
  * Returns a new OutputStream whose contents would be stored in a file.
  * Data is written in chunks of given buffer size.
  *
  * If there is a file with the given name, it is truncated and overwritten.
  * If there is no file with the given name, it is created.
  */
 AVRO_DECL OutputStreamPtr fileOutputStream(const char* filename,
     size_t bufferSize = 8 * 1024);

 /**
  * Returns a new InputStream whose contents come from the given file.
  * Data is read in chunks of given buffer size.
  */
 AVRO_DECL InputStreamPtr fileInputStream(
     const char *filename, size_t bufferSize = 8 * 1024);
 AVRO_DECL SeekableInputStreamPtr fileSeekableInputStream(
     const char *filename, size_t bufferSize = 8 * 1024);

 /**
  * Returns a new OutputStream whose contents will be sent to the given
  * std::ostream. The std::ostream object should outlive the returned
  * OutputStream.
  */
 AVRO_DECL OutputStreamPtr ostreamOutputStream(std::ostream& os,
     size_t bufferSize = 8 * 1024);

 /**
  * Returns a new InputStream whose contents come from the given
  * std::istream. The std::istream object should outlive the returned
  * InputStream.
  */
 AVRO_DECL InputStreamPtr istreamInputStream(
     std::istream &in, size_t bufferSize = 8 * 1024);

 /**
  * Returns a new InputStream whose contents come from the given
  * std::istream. Use this instead of istreamInputStream if
  * the istream does not support seekg (e.g. compressed streams).
  * The returned InputStream would read off bytes instead of seeking.
  * Of, course it has a performance penalty when reading instead of seeking;
  * So, use this only when seekg does not work.
  * The std::istream object should outlive the returned
  * InputStream.
  */
 AVRO_DECL InputStreamPtr nonSeekableIstreamInputStream(
     std::istream& is, size_t bufferSize = 8 * 1024);


 /** A convenience class for reading from an InputStream */
 struct StreamReader {
     /**
      * The underlying input stream.
      */
     InputStream* in_;

     /**
      * The next location to read from.
      */
     const uint8_t* next_;

     /**
      * One past the last valid location.
      */
     const uint8_t* end_;

     /**
      * Constructs an empty reader.
      */
     StreamReader() : in_(0), next_(0), end_(0) { }

     /**
      * Constructs a reader with the given underlying stream.
      */
     StreamReader(InputStream& in) : in_(0), next_(0), end_(0) { reset(in); }

     /**
      * Replaces the current input stream with the given one after backing up
      * the original one if required.
      */
     void reset(InputStream& is) {
         if (in_ != 0 && end_ != next_) {
             in_->backup(end_ - next_);
         }
         in_ = &is;
         next_ = end_ = 0;
     }

     /**
      * Read just one byte from the underlying stream. If there are no
      * more data, throws an exception.
      */
     uint8_t read() {
         if (next_ == end_) {
             more();
         }
         return *next_++;
     }

     /**
      * Reads the given number of bytes from the underlying stream.
      * If there are not that many bytes, throws an exception.
      */
     void readBytes(uint8_t* b, size_t n) {
         while (n > 0) {
             if (next_ == end_) {
                 more();
             }
             size_t q = end_ - next_;
             if (q > n) {
                 q = n;
             }
             ::memcpy(b, next_, q);
             next_ += q;
             b += q;
             n -= q;
         }
     }

     /**
      * Skips the given number of bytes. Of there are not so that many
      * bytes, throws an exception.
      */
     void skipBytes(size_t n) {
         if (n > static_cast<size_t>(end_ - next_)) {
             n -= end_ - next_;
             next_ = end_;
             in_->skip(n);
         } else {
             next_ += n;
         }
     }

     /**
      * Get as many byes from the underlying stream as possible in a single
      * chunk.
      * \return true if some data could be obtained. False is no more
      * data is available on the stream.
      */
     bool fill() {
         size_t n = 0;
         while (in_->next(&next_, &n)) {
             if (n != 0) {
                 end_ = next_ + n;
                 return true;
             }
         }
         return false;
     }

     /**
      * Tries to get more data and if it cannot, throws an exception.
      */
     void more() {
         if (! fill()) {
             throw Exception("EOF reached");
         }
     }

     /**
      * Returns true if and only if the end of stream is not reached.
      */
     bool hasMore() {
         return (next_ == end_) ? fill() : true;
     }

     /**
      * Returns unused bytes back to the underlying stream.
      * If unRead is true the last byte read is also pushed back.
      */
     void drain(bool unRead) {
         if (unRead) {
             --next_;
         }
         in_->backup(end_ - next_);
         end_ = next_;
     }
 };

 /**
  * A convinience class to write data into an OutputStream.
  */
 struct StreamWriter {
     /**
      * The underlying output stream for this writer.
      */
     OutputStream* out_;

     /**
      * The next location to write to.
      */
     uint8_t* next_;

     /**
      * One past the last location one can write to.
      */
     uint8_t* end_;

     /**
      * Constructs a writer with no underlying stream.
      */
     StreamWriter() : out_(0), next_(0), end_(0) { }

     /**
      * Constructs a new writer with the given underlying stream.
      */
     StreamWriter(OutputStream& out) : out_(0), next_(0), end_(0) { reset(out); }

     /**
      * Replaces the current underlying stream with a new one.
      * If required, it backs up unused bytes in the previous stream.
      */
     void reset(OutputStream& os) {
         if (out_ != 0 && end_ != next_) {
             out_->backup(end_ - next_);
         }
         out_ = &os;
         next_ = end_;
     }

     /**
      * Writes a single byte.
      */
     void write(uint8_t c) {
         if (next_ == end_) {
             more();
         }
         *next_++ = c;
     }

     /**
      * Writes the specified number of bytes starting at \p b.
      */
     void writeBytes(const uint8_t* b, size_t n) {
         while (n > 0) {
             if (next_ == end_) {
                 more();
             }
             size_t q = end_ - next_;
             if (q > n) {
                 q = n;
             }
             ::memcpy(next_, b, q);
             next_ += q;
             b += q;
             n -= q;
         }
     }

     /**
      * backs up upto the currently written data and flushes the
      * underlying stream.
      */
     void flush() {
         if (next_ != end_) {
             out_->backup(end_ - next_);
             next_ = end_;
         }
         out_->flush();
     }

     /**
      * Return the number of bytes written so far. For a meaningful
      * result, call this after a flush().
      */
     int64_t byteCount() const {
         return out_->byteCount();
     }

     /**
      * Gets more space to write to. Throws an exception it cannot.
      */
     void more() {
         size_t n = 0;
         while (out_->next(&next_, &n)) {
             if (n != 0) {
                 end_ = next_ + n;
                 return;
             }
         }
         throw Exception("EOF reached");
     }
 };

 /**
  * A convenience function to copy all the contents of an input stream into
  * an output stream.
  */
 inline void copy(InputStream& in, OutputStream& out)
 {
     const uint8_t *p = 0;
     size_t n = 0;
     StreamWriter w(out);
     while (in.next(&p, &n)) {
         w.writeBytes(p, n);
     }
     w.flush();
 }

 }   // namespace avro
 #endif
	/*
	* 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
	*
	* https://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 avro_Stream_hh__
	#define avro_Stream_hh__

	#include <memory>
	#include <string.h>
	#include <stdint.h>

	#include "boost/utility.hpp"

	#include "Config.hh"
	#include "Exception.hh"

	namespace avro {

	/**
	* A no-copy input stream.
	*/
	class AVRO_DECL InputStream : boost::noncopyable {
	protected:

	/**
	* An empty constuctor.
	*/
	InputStream() { }

	public:
	/**
	* Destructor.
	*/
	virtual ~InputStream() { }

	/**
	* Returns some of available data.
	*
	* Returns true if some data is available, false if no more data is
	* available or an error has occurred.
	*/
	virtual bool next(const uint8_t** data, size_t* len) = 0;

	/**
	* "Returns" back some of the data to the stream. The returned
	* data must be less than what was obtained in the last call to
	* next().
	*/
	virtual void backup(size_t len) = 0;

	/**
	* Skips number of bytes specified by len.
	*/
	virtual void skip(size_t len) = 0;

	/**
	* Returns the number of bytes read from this stream so far.
	* All the bytes made available through next are considered
	* to be used unless, retutned back using backup.
	*/
	virtual size_t byteCount() const = 0;
	};

	typedef std::unique_ptr<InputStream> InputStreamPtr;

	/**
	* An InputStream which also supports seeking to a specific offset.
	*/
	class AVRO_DECL SeekableInputStream : public InputStream {
	protected:

	/**
	* An empty constuctor.
	*/
	SeekableInputStream() { }

	public:
	/**
	* Destructor.
	*/
	virtual ~SeekableInputStream() { }

	/**
	* Seek to a specific position in the stream. This may invalidate pointers
	* returned from next(). This will also reset byteCount() to the given
	* position.
	*/
	virtual void seek(int64_t position) = 0;
	};

	typedef std::unique_ptr<SeekableInputStream> SeekableInputStreamPtr;

	/**
	* A no-copy output stream.
	*/
	class AVRO_DECL OutputStream : boost::noncopyable {
	protected:

	/**
	* An empty constuctor.
	*/
	OutputStream() { }
	public:

	/**
	* Destructor.
	*/
	virtual ~OutputStream() { }

	/**
	* Returns a buffer that can be written into.
	* On successful return, data has the pointer to the buffer
	* and len has the number of bytes available at data.
	*/
	virtual bool next(uint8_t** data, size_t* len) = 0;

	/**
	* "Returns" back to the stream some of the buffer obtained
	* from in the last call to next().
	*/
	virtual void backup(size_t len) = 0;

	/**
	* Number of bytes written so far into this stream. The whole buffer
	* returned by next() is assumed to be written unless some of
	* it was retutned using backup().
	*/
	virtual uint64_t byteCount() const = 0;

	/**
	* Flushes any data remaining in the buffer to the stream's underlying
	* store, if any.
	*/
	virtual void flush() = 0;
	};

	typedef std::unique_ptr<OutputStream> OutputStreamPtr;

	/**
	* Returns a new OutputStream, which grows in memory chunks of specified size.
	*/
	AVRO_DECL OutputStreamPtr memoryOutputStream(size_t chunkSize = 4 * 1024);

	/**
	* Returns a new InputStream, with the data from the given byte array.
	* It does not copy the data, the byte array should remain valid
	* until the InputStream is used.
	*/
	AVRO_DECL InputStreamPtr memoryInputStream(const uint8_t* data, size_t len);

	/**
	* Returns a new InputStream with the contents written into an
	* outputstream. The output stream must have been returned by
	* an earlier call to memoryOutputStream(). The contents for the new
	* input stream are the snapshot of the outputstream. One can construct
	* any number of memory input stream from a single memory output stream.
	*/
	AVRO_DECL InputStreamPtr memoryInputStream(const OutputStream& source);

	/**
	* Returns the contents written so far into the output stream, which should
	* be a memory output stream. That is it must have been returned by a pervious
	* call to memoryOutputStream().
	*/
	AVRO_DECL std::shared_ptr<std::vector<uint8_t> > snapshot(const OutputStream& source);

	/**
	* Returns a new OutputStream whose contents would be stored in a file.
	* Data is written in chunks of given buffer size.
	*
	* If there is a file with the given name, it is truncated and overwritten.
	* If there is no file with the given name, it is created.
	*/
	AVRO_DECL OutputStreamPtr fileOutputStream(const char* filename,
	size_t bufferSize = 8 * 1024);

	/**
	* Returns a new InputStream whose contents come from the given file.
	* Data is read in chunks of given buffer size.
	*/
	AVRO_DECL InputStreamPtr fileInputStream(
	const char filename, size_t bufferSize = 8 1024);
	AVRO_DECL SeekableInputStreamPtr fileSeekableInputStream(
	const char filename, size_t bufferSize = 8 1024);

	/**
	* Returns a new OutputStream whose contents will be sent to the given
	* std::ostream. The std::ostream object should outlive the returned
	* OutputStream.
	*/
	AVRO_DECL OutputStreamPtr ostreamOutputStream(std::ostream& os,
	size_t bufferSize = 8 * 1024);

	/**
	* Returns a new InputStream whose contents come from the given
	* std::istream. The std::istream object should outlive the returned
	* InputStream.
	*/
	AVRO_DECL InputStreamPtr istreamInputStream(
	std::istream &in, size_t bufferSize = 8 * 1024);

	/**
	* Returns a new InputStream whose contents come from the given
	* std::istream. Use this instead of istreamInputStream if
	* the istream does not support seekg (e.g. compressed streams).
	* The returned InputStream would read off bytes instead of seeking.
	* Of, course it has a performance penalty when reading instead of seeking;
	* So, use this only when seekg does not work.
	* The std::istream object should outlive the returned
	* InputStream.
	*/
	AVRO_DECL InputStreamPtr nonSeekableIstreamInputStream(
	std::istream& is, size_t bufferSize = 8 * 1024);


	/** A convenience class for reading from an InputStream */
	struct StreamReader {
	/**
	* The underlying input stream.
	*/
	InputStream* in_;

	/**
	* The next location to read from.
	*/
	const uint8_t* next_;

	/**
	* One past the last valid location.
	*/
	const uint8_t* end_;

	/**
	* Constructs an empty reader.
	*/
	StreamReader() : in_(0), next_(0), end_(0) { }

	/**
	* Constructs a reader with the given underlying stream.
	*/
	StreamReader(InputStream& in) : in_(0), next_(0), end_(0) { reset(in); }

	/**
	* Replaces the current input stream with the given one after backing up
	* the original one if required.
	*/
	void reset(InputStream& is) {
	if (in_ != 0 && end_ != next_) {
	in_->backup(end_ - next_);
	}
	in_ = &is;
	next_ = end_ = 0;
	}

	/**
	* Read just one byte from the underlying stream. If there are no
	* more data, throws an exception.
	*/
	uint8_t read() {
	if (next_ == end_) {
	more();
	}
	return *next_++;
	}

	/**
	* Reads the given number of bytes from the underlying stream.
	* If there are not that many bytes, throws an exception.
	*/
	void readBytes(uint8_t* b, size_t n) {
	while (n > 0) {
	if (next_ == end_) {
	more();
	}
	size_t q = end_ - next_;
	if (q > n) {
	q = n;
	}
	::memcpy(b, next_, q);
	next_ += q;
	b += q;
	n -= q;
	}
	}

	/**
	* Skips the given number of bytes. Of there are not so that many
	* bytes, throws an exception.
	*/
	void skipBytes(size_t n) {
	if (n > static_cast<size_t>(end_ - next_)) {
	n -= end_ - next_;
	next_ = end_;
	in_->skip(n);
	} else {
	next_ += n;
	}
	}

	/**
	* Get as many byes from the underlying stream as possible in a single
	* chunk.
	* \return true if some data could be obtained. False is no more
	* data is available on the stream.
	*/
	bool fill() {
	size_t n = 0;
	while (in_->next(&next_, &n)) {
	if (n != 0) {
	end_ = next_ + n;
	return true;
	}
	}
	return false;
	}

	/**
	* Tries to get more data and if it cannot, throws an exception.
	*/
	void more() {
	if (! fill()) {
	throw Exception("EOF reached");
	}
	}

	/**
	* Returns true if and only if the end of stream is not reached.
	*/
	bool hasMore() {
	return (next_ == end_) ? fill() : true;
	}

	/**
	* Returns unused bytes back to the underlying stream.
	* If unRead is true the last byte read is also pushed back.
	*/
	void drain(bool unRead) {
	if (unRead) {
	--next_;
	}
	in_->backup(end_ - next_);
	end_ = next_;
	}
	};

	/**
	* A convinience class to write data into an OutputStream.
	*/
	struct StreamWriter {
	/**
	* The underlying output stream for this writer.
	*/
	OutputStream* out_;

	/**
	* The next location to write to.
	*/
	uint8_t* next_;

	/**
	* One past the last location one can write to.
	*/
	uint8_t* end_;

	/**
	* Constructs a writer with no underlying stream.
	*/
	StreamWriter() : out_(0), next_(0), end_(0) { }

	/**
	* Constructs a new writer with the given underlying stream.
	*/
	StreamWriter(OutputStream& out) : out_(0), next_(0), end_(0) { reset(out); }

	/**
	* Replaces the current underlying stream with a new one.
	* If required, it backs up unused bytes in the previous stream.
	*/
	void reset(OutputStream& os) {
	if (out_ != 0 && end_ != next_) {
	out_->backup(end_ - next_);
	}
	out_ = &os;
	next_ = end_;
	}

	/**
	* Writes a single byte.
	*/
	void write(uint8_t c) {
	if (next_ == end_) {
	more();
	}
	*next_++ = c;
	}

	/**
	* Writes the specified number of bytes starting at \p b.
	*/
	void writeBytes(const uint8_t* b, size_t n) {
	while (n > 0) {
	if (next_ == end_) {
	more();
	}
	size_t q = end_ - next_;
	if (q > n) {
	q = n;
	}
	::memcpy(next_, b, q);
	next_ += q;
	b += q;
	n -= q;
	}
	}

	/**
	* backs up upto the currently written data and flushes the
	* underlying stream.
	*/
	void flush() {
	if (next_ != end_) {
	out_->backup(end_ - next_);
	next_ = end_;
	}
	out_->flush();
	}

	/**
	* Return the number of bytes written so far. For a meaningful
	* result, call this after a flush().
	*/
	int64_t byteCount() const {
	return out_->byteCount();
	}

	/**
	* Gets more space to write to. Throws an exception it cannot.
	*/
	void more() {
	size_t n = 0;
	while (out_->next(&next_, &n)) {
	if (n != 0) {
	end_ = next_ + n;
	return;
	}
	}
	throw Exception("EOF reached");
	}
	};

	/**
	* A convenience function to copy all the contents of an input stream into
	* an output stream.
	*/
	inline void copy(InputStream& in, OutputStream& out)
	{
	const uint8_t *p = 0;
	size_t n = 0;
	StreamWriter w(out);
	while (in.next(&p, &n)) {
	w.writeBytes(p, n);
	}
	w.flush();
	}

	} // namespace avro
	#endif