feat: add protobuf/echo/ProtoMessageBuilder

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/CMakeLists.txt

@@ -8,6 +8,8 @@ target_link_libraries(protobuf.echo PUBLIC
 target_sources(protobuf.echo PRIVATE
     Echo.cpp
     Echo.hpp
+    ProtoMessageBuilder.cpp
+    ProtoMessageBuilder.hpp
     ServiceForwarder.cpp
     ServiceForwarder.hpp
     TracingEcho.cpp

protobuf/echo/ProtoMessageBuilder.cpp

@@ -0,0 +1,102 @@
+#include "protobuf/echo/ProtoMessageBuilder.hpp"
+
+namespace services
+{
+    BufferingStreamReader::BufferingStreamReader(infra::BoundedDeque<uint8_t>& buffer, infra::ConstByteRange data)
+        : buffer(buffer)
+        , data(data)
+    {}
+
+    void BufferingStreamReader::ConsumeCurrent()
+    {
+        std::size_t bufferDecrease = std::min(buffer.size(), index);
+        buffer.erase(buffer.begin(), buffer.begin() + bufferDecrease);
+        index -= bufferDecrease;
+
+        data.pop_front(index);
+        index = 0;
+    }
+
+    void BufferingStreamReader::StoreRemainder()
+    {
+        buffer.insert(buffer.end(), data.begin(), data.end());
+    }
+
+    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(data, 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(data, 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(data, dataIndex);
+        return from;
+    }
+
+    bool BufferingStreamReader::Empty() const
+    {
+        return Available() == 0;
+    }
+
+    std::size_t BufferingStreamReader::Available() const
+    {
+        return buffer.size() + data.size() - index;
+    }
+}

protobuf/echo/ProtoMessageBuilder.hpp

@@ -0,0 +1,82 @@
+#ifndef PROTOBUF_PROTO_MESSAGE_BUILDER_HPP
+#define PROTOBUF_PROTO_MESSAGE_BUILDER_HPP
+
+#include "infra/stream/ByteInputStream.hpp"
+#include "infra/syntax/ProtoParser.hpp"
+#include "infra/util/BoundedDeque.hpp"
+
+namespace services
+{
+    class BufferingStreamReader
+        : public infra::StreamReader
+    {
+    public:
+        BufferingStreamReader(infra::BoundedDeque<uint8_t>& buffer, infra::ConstByteRange data);
+
+        void ConsumeCurrent();
+        void StoreRemainder();
+
+        // Implementation of StreamReader
+        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;
+
+    private:
+        infra::BoundedDeque<uint8_t>& buffer;
+        infra::ConstByteRange data;
+        std::size_t index = 0;
+    };
+
+    template<class Message>
+    class ProtoMessageBuilder
+    {
+    public:
+        void Feed(infra::ConstByteRange data);
+
+    public:
+        Message message;
+
+    private:
+        infra::BoundedDeque<uint8_t>::WithMaxSize<16> buffer;
+    };
+}
+
+//// Implementation ////
+
+namespace services
+{
+    template<class Message>
+    void ProtoMessageBuilder<Message>::Feed(infra::ConstByteRange data)
+    {
+        BufferingStreamReader reader{ buffer, data };
+        infra::DataInputStream::WithErrorPolicy stream{ reader, infra::softFail };
+        infra::ProtoParser parser{ stream };
+
+        while (true)
+        {
+            auto field = parser.GetField();
+            switch (field.second)
+            {
+                case Message::template fieldNumber<0>:
+                    DeserializeField(Message::template ProtoType<0>(), parser, field, message.value);
+                    break;
+                // default:
+                //     if (field.first.Is<infra::ProtoLengthDelimited>())
+                //         field.first.Get<infra::ProtoLengthDelimited>().SkipEverything();
+                //     break;
+            }
+
+            if (stream.Failed())
+                break;
+
+            reader.ConsumeCurrent();
+        }
+
+        reader.StoreRemainder();
+    }
+}
+
+#endif

protobuf/echo/test/CMakeLists.txt

@@ -2,8 +2,11 @@ add_executable(protobuf.echo_test)
 emil_build_for(protobuf.echo_test BOOL EMIL_BUILD_TESTS)
 emil_add_test(protobuf.echo_test)
 
+protocol_buffer_echo_cpp(protobuf.echo_test TestMessages.proto)
+
 target_sources(protobuf.echo_test PRIVATE
     TestEchoServiceResponseQueue.cpp
+    TestProtoMessageBuilder.cpp
     TestServiceForwarder.cpp
 )
 

protobuf/echo/test/TestMessages.proto

@@ -0,0 +1,170 @@
+syntax = "proto3";
+
+import "EchoAttributes.proto";
+
+package test_messages;
+option java_package = "com.philips.emil.protobufEcho";
+option java_outer_classname = "TestMessagesProto";
+
+enum Enumeration {
+  val0 = 0;
+  val1 = 1;
+}
+
+message TestEnum {
+  Enumeration e = 1;
+}
+
+message TestInt32 {
+  int32 value = 1;
+}
+
+message TestInt64 {
+  int64 value = 1;
+}
+
+message TestUInt32 {
+  uint32 value = 1;
+}
+
+message TestUInt64 {
+  uint64 value = 1;
+}
+
+message TestFixed32 {
+  fixed32 value = 1;
+}
+
+message TestFixed64 {
+  sfixed64 value = 1;
+}
+
+message TestSFixed64 {
+  fixed64 value = 1;
+}
+
+message TestSFixed32 {
+  sfixed32 value = 1;
+}
+
+message TestBool {
+  bool value = 1;
+}
+
+message TestString {
+  string value = 1 [(string_size) = 20];
+}
+
+message TestStdString {
+  string value = 1;
+}
+
+message TestRepeatedString {
+  repeated string value = 1 [(string_size) = 20, (array_size) = 10];
+}
+
+message TestBytes {
+  bytes value = 1 [(bytes_size) = 10];
+}
+
+message TestUnboundedBytes {
+  bytes value = 1;
+}
+
+message TestRepeatedUInt32 {
+  repeated uint32 value = 1 [(array_size) = 10];
+}
+
+message TestMessageWithMessageField {
+  TestUInt32 message = 1;
+}
+
+message TestRepeatedEverything {
+  repeated Enumeration v1 = 1 [(array_size) = 10];
+  repeated int32 v2 = 2 [(array_size) = 10];
+  repeated int64 v3 = 3 [(array_size) = 10];
+  repeated uint32 v4 = 4 [(array_size) = 10];
+  repeated uint64 v5 = 5 [(array_size) = 10];
+  repeated fixed32 v6 = 6 [(array_size) = 10];
+  repeated sfixed64 v7 = 7 [(array_size) = 10];
+  repeated fixed64 v8 = 8 [(array_size) = 10];
+  repeated sfixed32 v9 = 9 [(array_size) = 10];
+  repeated bool v10 = 10 [(array_size) = 10];
+  repeated string v11 = 11 [(string_size) = 20, (array_size) = 10];
+  repeated string v12 = 12 [(array_size) = 10];
+  repeated TestUInt32 v13 = 13 [(array_size) = 10];
+  repeated bytes v14 = 14 [(bytes_size) = 10, (array_size) = 10];
+  repeated bytes v15 = 15 [(array_size) = 10];
+}
+
+message TestUnboundedRepeatedEverything {
+  repeated Enumeration v1 = 1;
+  repeated int32 v2 = 2;
+  repeated int64 v3 = 3;
+  repeated uint32 v4 = 4;
+  repeated uint64 v5 = 5;
+  repeated fixed32 v6 = 6;
+  repeated sfixed64 v7 = 7;
+  repeated fixed64 v8 = 8;
+  repeated sfixed32 v9 = 9;
+  repeated bool v10 = 10;
+  repeated string v11 = 11 [(string_size) = 20];
+  repeated string v12 = 12;
+  repeated TestUInt32 v13 = 13;
+  repeated bytes v14 = 14 [(bytes_size) = 10];
+  repeated bytes v15 = 15;
+}
+
+message TestNestedMessage
+{
+    message NestedMessage
+    {
+        uint32 value = 1;
+    }
+
+    NestedMessage message = 1;
+}
+
+message TestMoreNestedMessage
+{
+    message NestedMessage1
+    {
+        uint32 value = 1;
+    }
+
+    message NestedMessage2
+    {
+        uint32 value = 2;
+    }
+
+    NestedMessage1 message1 = 1;
+    NestedMessage2 message2 = 2;
+}
+
+message TestNestedRepeatedMessage
+{
+    message NestedMessage
+    {
+        uint32 value = 1;
+    }
+
+    repeated NestedMessage message = 1 [(array_size) = 10];
+}
+
+message TestEmptyMessage
+{
+}
+
+service TestService1
+{
+  option (service_id) = 1;
+
+  rpc Method(TestUInt32) returns (Nothing) { option (method_id) = 1; }
+}
+
+service TestService2
+{
+  option (service_id) = 2;
+
+  rpc Search (TestString) returns (Nothing) { option (method_id) = 1; }
+}