feat: support streaming echo messages

examples/rpc/Server.cpp

@@ -8,8 +8,8 @@ class Console
     : public examples::Console
 {
 public:
-    Console(services::Echo& echo)
-        : examples::Console(echo)
+    Console(services::Echo& echo, services::MethodSerializerFactory& serializerFactory)
+        : examples::Console(echo, serializerFactory)
     {}
 
     void Write(infra::BoundedConstString message) override
@@ -23,8 +23,8 @@ class EchoConnection
     : public services::EchoOnConnection
 {
 public:
-    EchoConnection()
-        : console(*this)
+    EchoConnection(services::MethodSerializerFactory& serializerFactory)
+        : console(*this, serializerFactory)
     {}
 
 private:
@@ -45,12 +45,13 @@ class RpcServer
             connection->::services::ConnectionObserver::Subject().AbortAndDestroy();
 
         if (connection.Allocatable())
-            createdObserver(connection.Emplace());
+            createdObserver(connection.Emplace(serializerFactory));
     }
 
 private:
     infra::SharedPtr<void> listener;
     infra::SharedOptional<EchoConnection> connection;
+    services::MethodSerializerFactory::OnHeap serializerFactory;
 };
 
 int main(int argc, const char* argv[], const char* env[])

infra/stream/BufferingStreamReader.cpp

@@ -48,7 +48,9 @@ namespace infra
             return from;
         }
 
-        return input.ExtractContiguousRange(max);
+        auto result = input.ExtractContiguousRange(max);
+        index += result.size();
+        return result;
     }
 
     infra::ConstByteRange BufferingStreamReader::PeekContiguousRange(std::size_t start)

infra/stream/LimitedInputStream.cpp

@@ -58,7 +58,7 @@ namespace infra
     }
 
     LimitedStreamReaderWithRewinding::LimitedStreamReaderWithRewinding(StreamReaderWithRewinding& input, uint32_t length)
-        : input(input)
+        : input(&input)
         , length(length)
     {}
 
@@ -72,55 +72,65 @@ namespace infra
         length = newLength;
     }
 
+    bool LimitedStreamReaderWithRewinding::LimitReached() const
+    {
+        return length == 0;
+    }
+
+    void LimitedStreamReaderWithRewinding::SwitchInput(StreamReaderWithRewinding& newInput)
+    {
+        this->input = &newInput;
+    }
+
     void LimitedStreamReaderWithRewinding::Extract(ByteRange range, StreamErrorPolicy& errorPolicy)
     {
         errorPolicy.ReportResult(length >= range.size());
         range.shrink_from_back_to(length);
         length -= range.size();
-        input.Extract(range, errorPolicy);
+        input->Extract(range, errorPolicy);
     }
 
     uint8_t LimitedStreamReaderWithRewinding::Peek(StreamErrorPolicy& errorPolicy)
     {
         errorPolicy.ReportResult(length != 0);
 
         if (length != 0)
-            return input.Peek(errorPolicy);
+            return input->Peek(errorPolicy);
         else
             return 0;
     }
 
     ConstByteRange LimitedStreamReaderWithRewinding::ExtractContiguousRange(std::size_t max)
     {
-        ConstByteRange result = input.ExtractContiguousRange(std::min<std::size_t>(length, max));
+        ConstByteRange result = input->ExtractContiguousRange(std::min<std::size_t>(length, max));
         length -= result.size();
         return result;
     }
 
     ConstByteRange infra::LimitedStreamReaderWithRewinding::PeekContiguousRange(std::size_t start)
     {
-        return input.PeekContiguousRange(start);
+        return input->PeekContiguousRange(start);
     }
 
     bool LimitedStreamReaderWithRewinding::Empty() const
     {
-        return length == 0 || input.Empty();
+        return length == 0 || input->Empty();
     }
 
     std::size_t LimitedStreamReaderWithRewinding::Available() const
     {
-        return std::min<uint32_t>(length, input.Available());
+        return std::min<std::size_t>(length, input->Available());
     }
 
     std::size_t LimitedStreamReaderWithRewinding::ConstructSaveMarker() const
     {
-        return input.ConstructSaveMarker();
+        return input->ConstructSaveMarker();
     }
 
     void LimitedStreamReaderWithRewinding::Rewind(std::size_t marker)
     {
-        auto now = input.ConstructSaveMarker();
-        input.Rewind(marker);
+        auto now = input->ConstructSaveMarker();
+        input->Rewind(marker);
 
         length += now - marker;
     }

infra/stream/LimitedInputStream.hpp

@@ -44,6 +44,8 @@ namespace infra
         ~LimitedStreamReaderWithRewinding() = default;
 
         void ResetLength(uint32_t newLength);
+        bool LimitReached() const;
+        void SwitchInput(StreamReaderWithRewinding& newInput);
 
     public:
         void Extract(ByteRange range, StreamErrorPolicy& errorPolicy) override;
@@ -56,7 +58,7 @@ namespace infra
         void Rewind(std::size_t marker) override;
 
     private:
-        StreamReaderWithRewinding& input;
+        StreamReaderWithRewinding* input;
         uint32_t length;
     };
 

infra/stream/OutputStream.hpp

@@ -27,6 +27,11 @@ namespace infra
         virtual void Insert(ConstByteRange range, StreamErrorPolicy& errorPolicy) = 0;
         virtual std::size_t Available() const = 0;
 
+        bool Empty() const
+        {
+            return Available() == 0;
+        }
+
         virtual std::size_t ConstructSaveMarker() const;
         virtual std::size_t GetProcessedBytesSince(std::size_t marker) const;
         virtual infra::ByteRange SaveState(std::size_t marker);

infra/syntax/ProtoParser.cpp

@@ -90,14 +90,19 @@ namespace infra
         uint64_t result = 0;
         uint8_t byte = 0;
         uint8_t shift = 0;
+        uint8_t index = 0;
 
         do
         {
             input >> byte;
 
             result += static_cast<uint64_t>(byte & 0x7f) << shift;
             shift += 7;
-        } while (!input.Failed() && (byte & 0x80) != 0);
+            ++index;
+        } while (!input.Failed() && (byte & 0x80) != 0 && index != 10);
+
+        if (!input.Failed() && (byte & 0x80) != 0 && index == 10)
+            formatErrorPolicy.Failed();
 
         return result;
     }

protobuf/echo/CMakeLists.txt

@@ -8,11 +8,15 @@ target_link_libraries(protobuf.echo PUBLIC
 target_sources(protobuf.echo PRIVATE
     Echo.cpp
     Echo.hpp
+    EchoErrorPolicy.cpp
+    EchoErrorPolicy.hpp
     Proto.hpp
     ProtoMessageReceiver.cpp
     ProtoMessageReceiver.hpp
     ProtoMessageSender.cpp
     ProtoMessageSender.hpp
+    Serialization.cpp
+    Serialization.hpp
     ServiceForwarder.cpp
     ServiceForwarder.hpp
     TracingEcho.cpp