feat: add protobuf/echo/ProtoMessageBuilder

infra/syntax/ProtoFormatter.cpp

@@ -130,6 +130,12 @@ namespace infra
         PutBytes(bytes);
     }
 
+    void ProtoFormatter::PutLengthDelimitedSize(std::size_t size, uint32_t fieldNumber)
+    {
+        PutVarInt((fieldNumber << 3) | 2);
+        PutVarInt(size);
+    }
+
     ProtoLengthDelimitedFormatter ProtoFormatter::LengthDelimitedFormatter(uint32_t fieldNumber)
     {
         return ProtoLengthDelimitedFormatter(*this, fieldNumber);

infra/syntax/ProtoFormatter.hpp

@@ -14,6 +14,7 @@ namespace infra
     {
     public:
         ProtoLengthDelimitedFormatter(ProtoFormatter& formatter, uint32_t fieldNumber);
+        ProtoLengthDelimitedFormatter(ProtoFormatter& formatter, uint32_t fieldNumber, std::size_t size);
         ProtoLengthDelimitedFormatter(const ProtoLengthDelimitedFormatter& other) = delete;
         ProtoLengthDelimitedFormatter(ProtoLengthDelimitedFormatter&& other) noexcept;
         ProtoLengthDelimitedFormatter& operator=(const ProtoLengthDelimitedFormatter& other) = delete;
@@ -43,6 +44,7 @@ namespace infra
         void PutLengthDelimitedField(infra::ConstByteRange range, uint32_t fieldNumber);
         void PutStringField(infra::BoundedConstString string, uint32_t fieldNumber);
         void PutBytesField(infra::ConstByteRange bytes, uint32_t fieldNumber);
+        void PutLengthDelimitedSize(std::size_t size, uint32_t fieldNumber);
         ProtoLengthDelimitedFormatter LengthDelimitedFormatter(uint32_t fieldNumber);
 
     private:

infra/syntax/ProtoParser.cpp

@@ -116,7 +116,40 @@ namespace infra
         return result;
     }
 
+    struct MakeFullField
+        : infra::StaticVisitor<ProtoParser::Field>
+    {
+        MakeFullField(infra::DataInputStream inputStream, infra::StreamErrorPolicy& formatErrorPolicy, uint32_t fieldNumber)
+            : inputStream(inputStream)
+            , formatErrorPolicy(formatErrorPolicy)
+            , fieldNumber(fieldNumber)
+        {}
+
+        template<class T>
+        ProtoParser::Field operator()(T value) const
+        {
+            return { value, fieldNumber };
+        }
+
+        ProtoParser::Field operator()(PartialProtoLengthDelimited value) const
+        {
+            return { ProtoLengthDelimited(inputStream, formatErrorPolicy, value.length), fieldNumber };
+        }
+
+    private:
+        infra::DataInputStream inputStream;
+        infra::StreamErrorPolicy& formatErrorPolicy;
+        uint32_t fieldNumber;
+    };
+
     ProtoParser::Field ProtoParser::GetField()
+    {
+        auto [value, fieldNumber] = GetPartialField();
+        MakeFullField visitor(input, formatErrorPolicy, fieldNumber);
+        return infra::ApplyVisitor(visitor, value);
+    }
+
+    ProtoParser::PartialField ProtoParser::GetPartialField()
     {
         uint32_t x = static_cast<uint32_t>(GetVarInt());
         uint8_t type = x & 7;
@@ -129,7 +162,7 @@ namespace infra
             case 1:
                 return std::make_pair(GetFixed64(), fieldNumber);
             case 2:
-                return std::make_pair(ProtoLengthDelimited(input, formatErrorPolicy, static_cast<uint32_t>(GetVarInt())), fieldNumber);
+                return std::make_pair(PartialProtoLengthDelimited{ static_cast<uint32_t>(GetVarInt()) }, fieldNumber);
             case 5:
                 return std::make_pair(GetFixed32(), fieldNumber);
             default:

infra/syntax/ProtoParser.hpp

@@ -34,10 +34,18 @@ namespace infra
         infra::StreamErrorPolicy& formatErrorPolicy;
     };
 
+    struct PartialProtoLengthDelimited
+    {
+        uint32_t length;
+    };
+
     class ProtoParser
     {
     public:
-        using Field = std::pair<infra::Variant<uint32_t, uint64_t, ProtoLengthDelimited>, uint32_t>;
+        using FieldVariant = infra::Variant<uint32_t, uint64_t, ProtoLengthDelimited>;
+        using Field = std::pair<FieldVariant, uint32_t>;
+        using PartialFieldVariant = infra::Variant<uint32_t, uint64_t, PartialProtoLengthDelimited>;
+        using PartialField = std::pair<PartialFieldVariant, uint32_t>;
 
         explicit ProtoParser(infra::DataInputStream inputStream);
         ProtoParser(infra::DataInputStream inputStream, infra::StreamErrorPolicy& formatErrorPolicy);
@@ -48,6 +56,7 @@ namespace infra
         uint64_t GetFixed64();
 
         Field GetField();
+        PartialField GetPartialField();
 
         void ReportFormatResult(bool ok);
         bool FormatFailed() const;

infra/syntax/test/TestProtoFormatter.cpp

@@ -76,6 +76,17 @@ TEST(ProtoFormatterTest, PutBytesField)
     EXPECT_EQ((std::array<uint8_t, 3>{ 4 << 3 | 2, 1, 5 }), stream.Writer().Processed());
 }
 
+TEST(ProtoFormatterTest, PutSubObjectOfKnownSize)
+{
+    infra::ByteOutputStream::WithStorage<20> stream;
+    infra::ProtoFormatter formatter(stream);
+
+    formatter.PutLengthDelimitedSize(2, 4);
+    formatter.PutVarIntField(2, 4);
+
+    EXPECT_EQ((std::array<uint8_t, 4>{ 4 << 3 | 2, 2, 4 << 3, 2 }), stream.Writer().Processed());
+}
+
 TEST(ProtoFormatterTest, PutSubObject)
 {
     infra::ByteOutputStream::WithStorage<20> stream;

infra/util/ConstructBin.cpp

@@ -32,6 +32,13 @@ namespace infra
         return *this;
     }
 
+    ConstructBin& ConstructBin::Repeat(std::size_t amount, std::initializer_list<uint8_t> v)
+    {
+        for (size_t i = 0; i != amount; ++i)
+            contents.insert(contents.end(), v.begin(), v.end());
+        return *this;
+    }
+
     ConstructBin& ConstructBin::RepeatString(std::size_t amount, const std::string& v)
     {
         for (size_t i = 0; i != amount; ++i)

infra/util/ConstructBin.hpp

@@ -17,6 +17,7 @@ namespace infra
         ConstructBin& operator()(std::initializer_list<uint8_t> v);
 
         ConstructBin& Repeat(std::size_t amount, uint8_t v);
+        ConstructBin& Repeat(std::size_t amount, std::initializer_list<uint8_t> v);
         ConstructBin& RepeatString(std::size_t amount, const std::string& v);
 
         template<class T>

protobuf/CMakeLists.txt

@@ -1,5 +1,5 @@
-add_subdirectory(echo)
 add_subdirectory(echo_attributes)
+add_subdirectory(echo)
 add_subdirectory(protoc_echo_plugin)
 add_subdirectory(protoc_echo_plugin_csharp)
 add_subdirectory(protoc_echo_plugin_java)

protobuf/echo/BufferingStreamReader.cpp

@@ -0,0 +1,109 @@
+#include "protobuf/echo/BufferingStreamReader.hpp"
+
+namespace services
+{
+    BufferingStreamReader::BufferingStreamReader(infra::BoundedDeque<uint8_t>& buffer, infra::ConstByteRange inputData)
+        : buffer(buffer)
+        , inputData(inputData)
+    {}
+
+    BufferingStreamReader::~BufferingStreamReader()
+    {
+        StoreRemainder();
+    }
+
+    void BufferingStreamReader::Extract(infra::ByteRange range, infra::StreamErrorPolicy& errorPolicy)
+    {
+        if (index != buffer.size())
+        {
+            auto from = infra::Head(buffer.contiguous_range(buffer.begin() + index), range.size());
+            infra::Copy(from, infra::Head(range, from.size()));
+            range.pop_front(from.size());
+            index += from.size();
+
+            // Perhaps the deque just wrapped around, try once more
+            from = infra::Head(buffer.contiguous_range(buffer.begin() + index), range.size());
+            infra::Copy(from, infra::Head(range, from.size()));
+            range.pop_front(from.size());
+            index += from.size();
+        }
+
+        if (!range.empty())
+        {
+            auto dataIndex = index - buffer.size();
+            auto from = infra::Head(infra::DiscardHead(inputData, dataIndex), range.size());
+            infra::Copy(from, infra::Head(range, from.size()));
+            range.pop_front(from.size());
+            index += from.size();
+        }
+
+        errorPolicy.ReportResult(range.empty());
+    }
+
+    uint8_t BufferingStreamReader::Peek(infra::StreamErrorPolicy& errorPolicy)
+    {
+        auto range = PeekContiguousRange(0);
+
+        errorPolicy.ReportResult(!range.empty());
+
+        if (range.empty())
+            return 0;
+        else
+            return range.front();
+    }
+
+    infra::ConstByteRange BufferingStreamReader::ExtractContiguousRange(std::size_t max)
+    {
+        if (index < buffer.size())
+        {
+            auto from = infra::Head(buffer.contiguous_range(buffer.begin() + index), max);
+            index += from.size();
+            return from;
+        }
+
+        auto dataIndex = index - buffer.size();
+        auto from = infra::Head(infra::DiscardHead(inputData, dataIndex), max);
+        index += from.size();
+        return from;
+    }
+
+    infra::ConstByteRange BufferingStreamReader::PeekContiguousRange(std::size_t start)
+    {
+        if (index + start < buffer.size())
+        {
+            auto from = buffer.contiguous_range(buffer.begin() + index + start);
+            return from;
+        }
+
+        auto dataIndex = index + start - buffer.size();
+        auto from = infra::DiscardHead(inputData, dataIndex);
+        return from;
+    }
+
+    bool BufferingStreamReader::Empty() const
+    {
+        return Available() == 0;
+    }
+
+    std::size_t BufferingStreamReader::Available() const
+    {
+        return buffer.size() + inputData.size() - index;
+    }
+
+    std::size_t BufferingStreamReader::ConstructSaveMarker() const
+    {
+        return index;
+    }
+
+    void BufferingStreamReader::Rewind(std::size_t marker)
+    {
+        index = marker;
+    }
+
+    void BufferingStreamReader::StoreRemainder()
+    {
+        std::size_t bufferDecrease = std::min(buffer.size(), index);
+        buffer.erase(buffer.begin(), buffer.begin() + bufferDecrease);
+        buffer.insert(buffer.end(), inputData.begin() + index - bufferDecrease, inputData.end());
+    }
+}

protobuf/echo/BufferingStreamReader.hpp

@@ -0,0 +1,38 @@
+#ifndef PROTOBUF_BUFFERING_STREAM_READER_HPP
+#define PROTOBUF_BUFFERING_STREAM_READER_HPP
+
+#include "infra/stream/InputStream.hpp"
+#include "infra/util/BoundedDeque.hpp"
+
+namespace services
+{
+    // Usage: Everything that is not read from the inputData is stored into the buffer upon destruction of the BufferingStreamReader
+    // Any data already present in the buffer is read first from the reader
+    class BufferingStreamReader
+        : public infra::StreamReaderWithRewinding
+    {
+    public:
+        BufferingStreamReader(infra::BoundedDeque<uint8_t>& buffer, infra::ConstByteRange inputData);
+        ~BufferingStreamReader();
+
+        // Implementation of StreamReaderWithRewinding
+        void Extract(infra::ByteRange range, infra::StreamErrorPolicy& errorPolicy) override;
+        uint8_t Peek(infra::StreamErrorPolicy& errorPolicy) override;
+        infra::ConstByteRange ExtractContiguousRange(std::size_t max) override;
+        infra::ConstByteRange PeekContiguousRange(std::size_t start) override;
+        bool Empty() const override;
+        std::size_t Available() const override;
+        std::size_t ConstructSaveMarker() const override;
+        void Rewind(std::size_t marker) override;
+
+    private:
+        void StoreRemainder();
+
+    private:
+        infra::BoundedDeque<uint8_t>& buffer;
+        infra::ConstByteRange inputData;
+        std::size_t index = 0;
+    };
+}
+
+#endif

protobuf/echo/BufferingStreamWriter.cpp

@@ -0,0 +1,65 @@
+#include "protobuf/echo/BufferingStreamWriter.hpp"
+
+namespace services
+{
+    BufferingStreamWriter::BufferingStreamWriter(infra::BoundedDeque<uint8_t>& buffer, infra::StreamWriter& output)
+        : buffer(buffer)
+        , output(output)
+    {
+        LoadRemainder();
+    }
+
+    void BufferingStreamWriter::Insert(infra::ConstByteRange range, infra::StreamErrorPolicy& errorPolicy)
+    {
+        auto first = infra::Head(range, output.Available());
+        output.Insert(first, errorPolicy);
+        index += first.size();
+        range.pop_front(first.size());
+
+        buffer.insert(buffer.end(), range.begin(), range.end());
+        index += range.size();
+    }
+
+    std::size_t BufferingStreamWriter::Available() const
+    {
+        return output.Available() + buffer.max_size() - buffer.size();
+    }
+
+    std::size_t BufferingStreamWriter::ConstructSaveMarker() const
+    {
+        return index;
+    }
+
+    std::size_t BufferingStreamWriter::GetProcessedBytesSince(std::size_t marker) const
+    {
+        return index - marker;
+    }
+
+    infra::ByteRange BufferingStreamWriter::SaveState(std::size_t marker)
+    {
+        std::abort();
+    }
+
+    void BufferingStreamWriter::RestoreState(infra::ByteRange range)
+    {
+        std::abort();
+    }
+
+    infra::ByteRange BufferingStreamWriter::Overwrite(std::size_t marker)
+    {
+        std::abort();
+    }
+
+    void BufferingStreamWriter::LoadRemainder()
+    {
+        infra::StreamErrorPolicy errorPolicy;
+        auto from = infra::Head(buffer.contiguous_range(buffer.begin()), output.Available());
+        output.Insert(from, errorPolicy);
+        buffer.erase(buffer.begin(), buffer.begin() + from.size());
+        from = infra::Head(buffer.contiguous_range(buffer.begin()), output.Available());
+        output.Insert(from, errorPolicy);
+        buffer.clear();
+
+        index = 0;
+    }
+}