EventStreamHandler.h

 
#pragma once
 
#include <aws/core/Core_EXPORTS.h>
#include <aws/core/http/HttpTypes.h>
#include <aws/core/utils/event/EventHeader.h>
#include <aws/core/utils/event/EventMessage.h>
#include <aws/core/utils/event/EventStreamErrors.h>
#include <aws/core/utils/memory/stl/AWSString.h>
#include <aws/core/utils/memory/stl/AWSStreamFwd.h>
#include <aws/core/utils/memory/stl/AWSStringStream.h>
#include <cassert>
 
namespace Aws
{
    namespace Utils
    {
        namespace Event
        {
            enum class InitialResponseType
            {
                ON_EVENT,
                ON_RESPONSE
            };
 
            class AWS_CORE_API EventStreamHandler
            {
            public:
                EventStreamHandler() :
                    m_failure(false), m_internalError(EventStreamErrors::EVENT_STREAM_NO_ERROR), m_headersBytesReceived(0), m_payloadBytesReceived(0), m_message()
                {}
 
                virtual ~EventStreamHandler() = default;
            
                inline operator bool() const { return !m_failure; }
 
                inline void SetFailure() { m_failure = true; }
 
                inline virtual void Reset()
                {
                    m_failure = false;
                    m_internalError = EventStreamErrors::EVENT_STREAM_NO_ERROR;
                    m_headersBytesReceived = 0;
                    m_payloadBytesReceived = 0;
 
                    m_message.Reset();
                }
 
                inline void SetInternalError(int errorCode = 0)
                {
                    m_internalError = static_cast<EventStreamErrors>(errorCode);
                }
 
                inline EventStreamErrors GetInternalError()
                {
                    return m_internalError;
                }
 
                inline virtual bool IsMessageCompleted()
                {
                    return m_message.GetHeadersLength() == m_headersBytesReceived && m_message.GetPayloadLength() == m_payloadBytesReceived;
                }
 
                inline virtual void SetMessageMetadata(size_t totalLength, size_t headersLength, size_t payloadLength)
                {
                    m_message.SetTotalLength(totalLength);
                    m_message.SetHeadersLength(headersLength);
                    m_message.SetPayloadLength(payloadLength);
                    assert(totalLength == 12/*prelude length*/ + headersLength + payloadLength + 4/*message crc length*/);
                    if (totalLength != headersLength + payloadLength + 16)
                    {
                        AWS_LOG_WARN("EventStreamHandler", "Message total length mismatch.");
                    }
                }
 
                inline virtual void WriteMessageEventPayload(const unsigned char* data, size_t dataLength)
                {
                    m_message.WriteEventPayload(data, dataLength);
                    m_payloadBytesReceived += dataLength;
                }
                
                inline virtual Aws::Vector<unsigned char>&& GetEventPayloadWithOwnership() { return m_message.GetEventPayloadWithOwnership(); }
 
                inline virtual Aws::String GetEventPayloadAsString() { return m_message.GetEventPayloadAsString(); }
 
                inline virtual void InsertMessageEventHeader(const String& eventHeaderName, size_t eventHeaderLength, const Aws::Utils::Event::EventHeaderValue& eventHeaderValue)
                {
                    m_message.InsertEventHeader(eventHeaderName, eventHeaderValue);
                    m_headersBytesReceived += eventHeaderLength;
                }
 
                inline virtual const Aws::Utils::Event::EventHeaderValueCollection& GetEventHeaders() { return m_message.GetEventHeaders(); }
 
                inline virtual const Http::HeaderValueCollection GetEventHeadersAsHttpHeaders() const
                {
                    Http::HeaderValueCollection output;
                    using SrcT = Aws::Utils::Event::EventHeaderValueCollection::value_type;
                    using DstT = Http::HeaderValueCollection::value_type;
                    std::transform(m_message.GetEventHeaders().cbegin(), m_message.GetEventHeaders().cend(),
                                   std::inserter(output, output.end()),
                                   [](const SrcT& src)
                                   {
                                       return DstT(src.first, src.second.ToString());
                                   });
                    return output;
                }
 
                virtual void OnEvent() = 0;
 
            private:
                bool m_failure;
                EventStreamErrors m_internalError;
                size_t m_headersBytesReceived;
                size_t m_payloadBytesReceived;
                Aws::Utils::Event::Message m_message;
            };
        }
    }
}