Implemented a C++ based stream that will interoperate with the low-le… (aws#89)

* Implemented a C++ based stream that will interoperate with the low-level aws_input_stream apis.

Author: JonathanHenson

include/aws/crt/http/HttpRequestResponse.h

@@ -16,6 +16,7 @@
 
 #include <aws/crt/Exports.h>
 #include <aws/crt/Types.h>
+#include <aws/crt/io/Stream.h>
 
 struct aws_http_header;
 struct aws_http_message;
@@ -49,13 +50,15 @@ namespace Aws
                 /**
                  * Gets the input stream representing the message body
                  */
-                std::shared_ptr<Aws::Crt::Io::IStream> GetBody() const noexcept;
+                std::shared_ptr<Aws::Crt::Io::InputStream> GetBody() const noexcept;
 
                 /**
                  * Sets the input stream representing the message body
                  */
                 bool SetBody(const std::shared_ptr<Aws::Crt::Io::IStream> &body) noexcept;
 
+                bool SetBody(const std::shared_ptr<Aws::Crt::Io::InputStream> &body) noexcept;
+
                 size_t GetHeaderCount() const noexcept;
                 Optional<HttpHeader> GetHeader(size_t index) const noexcept;
                 bool AddHeader(const HttpHeader &header) noexcept;
@@ -70,7 +73,7 @@ namespace Aws
 
                 Allocator *m_allocator;
                 struct aws_http_message *m_message;
-                std::shared_ptr<Aws::Crt::Io::IStream> m_bodyStream;
+                std::shared_ptr<Aws::Crt::Io::InputStream> m_bodyStream;
                 bool m_ownsMessage;
             };
 

include/aws/crt/io/Stream.h

@@ -16,21 +16,115 @@
 
 #include <aws/crt/Exports.h>
 #include <aws/crt/Types.h>
-
-struct aws_input_stream;
+#include <aws/io/stream.h>
 
 namespace Aws
 {
     namespace Crt
     {
         namespace Io
         {
-            /**
-             * Factory to create a aws-c-io input stream subclass from a C++ stream
+            using StreamStatus = aws_stream_status;
+            using OffsetType = aws_off_t;
+
+            enum class StreamSeekBasis
+            {
+                Begin = AWS_SSB_BEGIN,
+                End = AWS_SSB_END,
+            };
+
+            /***
+             * Interface for building an Object oriented stream that will be honored by the CRT's low-level
+             * aws_input_stream interface. To use, create a subclass of InputStream and define the abstract
+             * functions.
+             */
+            class AWS_CRT_CPP_API InputStream
+            {
+              public:
+                virtual ~InputStream();
+
+                InputStream(const InputStream &) = delete;
+                InputStream &operator=(const InputStream &) = delete;
+                InputStream(InputStream &&) = delete;
+                InputStream &operator=(InputStream &&) = delete;
+
+                explicit operator bool() const noexcept { return IsGood(); }
+
+                virtual bool IsGood() const noexcept = 0;
+
+                aws_input_stream *GetUnderlyingStream() noexcept { return &m_underlying_stream; }
+
+              protected:
+                Allocator *m_allocator;
+                aws_input_stream m_underlying_stream;
+
+                InputStream(Aws::Crt::Allocator *allocator = DefaultAllocator());
+
+                /***
+                 * Read up-to buffer::capacity - buffer::len into buffer::buffer
+                 * Increment buffer::len by the amount you read in.
+                 *
+                 * @return true on success, false otherwise. Return false, when there is nothing left to read.
+                 * You SHOULD raise an error via aws_raise_error()
+                 * if an actual failure condition occurs.
+                 * If you return false, GetStatusImpl() will be called to determine
+                 * the validity of the stream.
+                 */
+                virtual bool ReadImpl(ByteBuf &buffer) noexcept = 0;
+
+                /**
+                 * Returns the current status of the stream.
+                 */
+                virtual StreamStatus GetStatusImpl() const noexcept = 0;
+
+                /**
+                 * Returns the total length of the available data for the stream.
+                 * Returns -1 if not available.
+                 */
+                virtual int64_t GetLengthImpl() const noexcept = 0;
+
+                /**
+                 * Seek's the stream to seekBasis based offset bytes.
+                 *
+                 * It is expected, that if seeking to the beginning of a stream,
+                 * all error's are cleared if possible.
+                 *
+                 * @return true on success, false otherwise. You SHOULD raise an error via aws_raise_error()
+                 * if a failure occurs. If you return false, the m_good flag will be set to false.
+                 */
+                virtual bool SeekImpl(OffsetType offset, StreamSeekBasis seekBasis) noexcept = 0;
+
+              private:
+                static int s_Seek(aws_input_stream *stream, aws_off_t offset, enum aws_stream_seek_basis basis);
+                static int s_Read(aws_input_stream *stream, aws_byte_buf *dest);
+                static int s_GetStatus(aws_input_stream *stream, aws_stream_status *status);
+                static int s_GetLength(struct aws_input_stream *stream, int64_t *out_length);
+                static void s_Destroy(struct aws_input_stream *stream);
+
+                static aws_input_stream_vtable s_vtable;
+            };
+
+            /***
+             * Implementation of Aws::Crt::Io::InputStream that wraps a std::input_stream.
              */
-            AWS_CRT_CPP_API aws_input_stream *AwsInputStreamNewCpp(
-                const std::shared_ptr<Aws::Crt::Io::IStream> &stream,
-                Aws::Crt::Allocator *allocator = DefaultAllocator()) noexcept;
+            class AWS_CRT_CPP_API StdIOStreamInputStream : public InputStream
+            {
+              public:
+                StdIOStreamInputStream(
+                    std::shared_ptr<Aws::Crt::Io::IStream> stream,
+                    Aws::Crt::Allocator *allocator = DefaultAllocator()) noexcept;
+
+                bool IsGood() const noexcept override;
+
+              protected:
+                bool ReadImpl(ByteBuf &buffer) noexcept override;
+                StreamStatus GetStatusImpl() const noexcept override;
+                int64_t GetLengthImpl() const noexcept override;
+                bool SeekImpl(OffsetType offsetType, StreamSeekBasis seekBasis) noexcept override;
+
+              private:
+                std::shared_ptr<Aws::Crt::Io::IStream> m_stream;
+            };
         } // namespace Io
     }     // namespace Crt
 } // namespace Aws

source/http/HttpRequestResponse.cpp

@@ -49,32 +49,31 @@ namespace Aws
                 }
             }
 
-            std::shared_ptr<Aws::Crt::Io::IStream> HttpMessage::GetBody() const noexcept { return m_bodyStream; }
+            std::shared_ptr<Aws::Crt::Io::InputStream> HttpMessage::GetBody() const noexcept { return m_bodyStream; }
 
             bool HttpMessage::SetBody(const std::shared_ptr<Aws::Crt::Io::IStream> &body) noexcept
             {
-                aws_input_stream *stream = nullptr;
+                aws_http_message_set_body_stream(m_message, nullptr);
+                m_bodyStream = nullptr;
+
                 if (body != nullptr)
                 {
-                    stream = Aws::Crt::Io::AwsInputStreamNewCpp(body, m_allocator);
-                    if (stream == nullptr)
+                    m_bodyStream = MakeShared<Io::StdIOStreamInputStream>(m_allocator, body, m_allocator);
+                    if (m_bodyStream == nullptr || !m_bodyStream)
                     {
                         return false;
                     }
+                    aws_http_message_set_body_stream(m_message, m_bodyStream->GetUnderlyingStream());
                 }
 
-                /*
-                 * clean up the old stream before setting the new
-                 */
-                aws_input_stream *old_stream = aws_http_message_get_body_stream(m_message);
-                if (old_stream != nullptr)
-                {
-                    aws_input_stream_destroy(old_stream);
-                }
-
-                aws_http_message_set_body_stream(m_message, stream);
+                return true;
+            }
 
-                m_bodyStream = (stream) ? body : nullptr;
+            bool HttpMessage::SetBody(const std::shared_ptr<Aws::Crt::Io::InputStream> &body) noexcept
+            {
+                m_bodyStream = body;
+                aws_http_message_set_body_stream(
+                    m_message, m_bodyStream && *m_bodyStream ? m_bodyStream->GetUnderlyingStream() : nullptr);
 
                 return true;
             }