Fix setting headers in eventstream-rpc (#788)

Author: sfod

eventstream_rpc/include/aws/eventstreamrpc/EventStreamClient.h

@@ -101,6 +101,7 @@ namespace Aws
             MessageAmendment(const MessageAmendment &lhs);
             MessageAmendment(MessageAmendment &&rhs);
             MessageAmendment &operator=(const MessageAmendment &lhs);
+            MessageAmendment &operator=(MessageAmendment &&rhs);
             ~MessageAmendment() noexcept;
             explicit MessageAmendment(Crt::Allocator *allocator = Crt::g_allocator) noexcept;
             MessageAmendment(
@@ -114,10 +115,22 @@ namespace Aws
                 Crt::List<EventStreamHeader> &&headers,
                 Crt::Allocator *allocator = Crt::g_allocator) noexcept;
             MessageAmendment(const Crt::ByteBuf &payload, Crt::Allocator *allocator = Crt::g_allocator) noexcept;
+
+            /**
+             * Add a given header to the end of the header list.
+             */
             void AddHeader(EventStreamHeader &&header) noexcept;
+
+            /**
+             * Add given headers to the beginning of the header list.
+             */
+            void PrependHeaders(Crt::List<EventStreamHeader> &&headers);
             void SetPayload(const Crt::Optional<Crt::ByteBuf> &payload) noexcept;
-            const Crt::List<EventStreamHeader> &GetHeaders() const noexcept;
-            const Crt::Optional<Crt::ByteBuf> &GetPayload() const noexcept;
+            void SetPayload(Crt::Optional<Crt::ByteBuf> &&payload);
+            const Crt::List<EventStreamHeader> &GetHeaders() const &noexcept;
+            Crt::List<EventStreamHeader> &&GetHeaders() &&;
+            const Crt::Optional<Crt::ByteBuf> &GetPayload() const &noexcept;
+            Crt::Optional<Crt::ByteBuf> &&GetPayload() &&;
 
           private:
             Crt::List<EventStreamHeader> m_headers;

eventstream_rpc/source/EventStreamClient.cpp

@@ -51,7 +51,7 @@ namespace Aws
         }
 
         MessageAmendment::MessageAmendment(Crt::List<EventStreamHeader> &&headers, Crt::Allocator *allocator) noexcept
-            : m_headers(headers), m_payload(), m_allocator(allocator)
+            : m_headers(std::move(headers)), m_payload(), m_allocator(allocator)
         {
         }
 
@@ -79,14 +79,16 @@ namespace Aws
 
         MessageAmendment &MessageAmendment::operator=(const MessageAmendment &lhs)
         {
-            m_headers = lhs.m_headers;
-            if (lhs.m_payload.has_value())
+            if (this != &lhs)
             {
-                m_payload =
-                    Crt::ByteBufNewCopy(lhs.m_allocator, lhs.m_payload.value().buffer, lhs.m_payload.value().len);
+                m_headers = lhs.m_headers;
+                if (lhs.m_payload.has_value())
+                {
+                    m_payload =
+                        Crt::ByteBufNewCopy(lhs.m_allocator, lhs.m_payload.value().buffer, lhs.m_payload.value().len);
+                }
+                m_allocator = lhs.m_allocator;
             }
-            m_allocator = lhs.m_allocator;
-
             return *this;
         }
 
@@ -97,9 +99,37 @@ namespace Aws
             rhs.m_payload = Crt::Optional<Crt::ByteBuf>();
         }
 
-        const Crt::List<EventStreamHeader> &MessageAmendment::GetHeaders() const noexcept { return m_headers; }
+        MessageAmendment &MessageAmendment::operator=(MessageAmendment &&rhs)
+        {
+            if (this != &rhs)
+            {
+                m_headers = std::move(rhs.m_headers);
+                m_payload = std::move(rhs.m_payload);
+                m_allocator = rhs.m_allocator;
+
+                rhs.m_allocator = nullptr;
+                rhs.m_payload = Crt::Optional<Crt::ByteBuf>();
+            }
+            return *this;
+        }
 
-        const Crt::Optional<Crt::ByteBuf> &MessageAmendment::GetPayload() const noexcept { return m_payload; }
+        const Crt::List<EventStreamHeader> &MessageAmendment::GetHeaders() const &noexcept { return m_headers; }
+
+        Crt::List<EventStreamHeader> &&MessageAmendment::GetHeaders() && { return std::move(m_headers); }
+
+        const Crt::Optional<Crt::ByteBuf> &MessageAmendment::GetPayload() const &noexcept { return m_payload; }
+
+        Crt::Optional<Crt::ByteBuf> &&MessageAmendment::GetPayload() && { return std::move(m_payload); }
+
+        void MessageAmendment::AddHeader(EventStreamHeader &&header) noexcept
+        {
+            m_headers.emplace_back(std::move(header));
+        }
+
+        void MessageAmendment::PrependHeaders(Crt::List<EventStreamHeader> &&headers)
+        {
+            m_headers.splice(m_headers.begin(), std::move(headers));
+        }
 
         void MessageAmendment::SetPayload(const Crt::Optional<Crt::ByteBuf> &payload) noexcept
         {
@@ -109,6 +139,8 @@ namespace Aws
             }
         }
 
+        void MessageAmendment::SetPayload(Crt::Optional<Crt::ByteBuf> &&payload) { m_payload = std::move(payload); }
+
         MessageAmendment::~MessageAmendment() noexcept
         {
             if (m_payload.has_value())
@@ -663,23 +695,17 @@ namespace Aws
                 thisConnection->m_clientState = WAITING_FOR_CONNECT_ACK;
                 thisConnection->m_underlyingConnection = connection;
                 MessageAmendment messageAmendment;
-                Crt::List<EventStreamHeader> messageAmendmentHeaders = messageAmendment.GetHeaders();
 
                 if (thisConnection->m_connectMessageAmender)
                 {
                     MessageAmendment connectAmendment(thisConnection->m_connectMessageAmender());
-                    Crt::List<EventStreamHeader> amenderHeaderList = connectAmendment.GetHeaders();
                     /* The version header is necessary for establishing the connection. */
                     messageAmendment.AddHeader(EventStreamHeader(
                         Crt::String(EVENTSTREAM_VERSION_HEADER),
                         Crt::String(EVENTSTREAM_VERSION_STRING),
                         thisConnection->m_allocator));
-                    /* Note that we are prepending headers from the user-provided amender. */
-                    if (amenderHeaderList.size() > 0)
-                    {
-                        messageAmendmentHeaders.splice(messageAmendmentHeaders.end(), amenderHeaderList);
-                    }
-                    messageAmendment.SetPayload(connectAmendment.GetPayload());
+                    messageAmendment.PrependHeaders(std::move(connectAmendment).GetHeaders());
+                    messageAmendment.SetPayload(std::move(connectAmendment).GetPayload());
                 }
 
                 /* Send a CONNECT packet to the server. */
@@ -695,8 +721,6 @@ namespace Aws
             thisConnection->m_connectionSetupPromise.set_value();
         }
 
-        void MessageAmendment::AddHeader(EventStreamHeader &&header) noexcept { m_headers.emplace_back(header); }
-
         void ClientConnection::s_onConnectionShutdown(
             struct aws_event_stream_rpc_client_connection *connection,
             int errorCode,