NATS protocol parser

This is an experimental NATS Protocol Parser with zero allocations. Somewhat
inspired by the .NET Runtime System.Text.Json library's Utf8JsonReader.

| Method | Mean     | Error    | StdDev   | Allocated |
|------- |---------:|---------:|---------:|----------:|
| Parse  | 862.5 ns | 320.3 ns | 17.56 ns |         - |

NATS.Client.sln

@@ -107,6 +107,8 @@ Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Example.OpenTelemetry", "sa
 EndProject
 Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "NATS.Net.OpenTelemetry.Tests", "tests\NATS.Net.OpenTelemetry.Tests\NATS.Net.OpenTelemetry.Tests.csproj", "{B8554582-DE19-41A2-9784-9B27C9F22429}"
 EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "NatsProtocolParserProf", "NatsProtocolParserProf\NatsProtocolParserProf.csproj", "{1981B633-D522-4468-873D-5CC49B489159}"
+EndProject
 Global
 	GlobalSection(SolutionConfigurationPlatforms) = preSolution
 		Debug|Any CPU = Debug|Any CPU
@@ -285,6 +287,10 @@ Global
 		{B8554582-DE19-41A2-9784-9B27C9F22429}.Debug|Any CPU.Build.0 = Debug|Any CPU
 		{B8554582-DE19-41A2-9784-9B27C9F22429}.Release|Any CPU.ActiveCfg = Release|Any CPU
 		{B8554582-DE19-41A2-9784-9B27C9F22429}.Release|Any CPU.Build.0 = Release|Any CPU
+		{1981B633-D522-4468-873D-5CC49B489159}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{1981B633-D522-4468-873D-5CC49B489159}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{1981B633-D522-4468-873D-5CC49B489159}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{1981B633-D522-4468-873D-5CC49B489159}.Release|Any CPU.Build.0 = Release|Any CPU
 	EndGlobalSection
 	GlobalSection(SolutionProperties) = preSolution
 		HideSolutionNode = FALSE

NatsProtocolParserProf/NatsProtocolParserProf.csproj

@@ -0,0 +1,14 @@
+<Project Sdk="Microsoft.NET.Sdk">
+
+    <PropertyGroup>
+        <OutputType>Exe</OutputType>
+        <TargetFramework>net8.0</TargetFramework>
+        <ImplicitUsings>enable</ImplicitUsings>
+        <Nullable>enable</Nullable>
+    </PropertyGroup>
+
+    <ItemGroup>
+      <ProjectReference Include="..\src\NATS.Client.Core\NATS.Client.Core.csproj" />
+    </ItemGroup>
+
+</Project>

NatsProtocolParserProf/Program.cs

@@ -0,0 +1,137 @@
+// See https://aka.ms/new-console-template for more information
+
+using System.Buffers;
+using NATS.Client.Core;
+
+var bench = new NatsProtoParserBench();
+bench.Setup();
+
+Console.WriteLine("Setup completed");
+Console.ReadLine();
+
+var count = 0;
+for (var i = 0; i < 1_000_000; i++)
+{
+    count += bench.Parse();
+}
+
+Console.WriteLine($"count: {count}");
+Console.ReadLine();
+
+public class NatsProtoParserBench
+{
+    private List<ReadOnlySequence<byte>> _sequences;
+    private NatsProtocolParser _parser;
+
+    public void Setup()
+    {
+        _sequences =
+        [
+            new SequenceBuilder()
+                .Append("INFO {\"server_id\":\"nats-server\""u8.ToArray())
+                .Append("}\r"u8.ToArray())
+                .Append("\nPI"u8.ToArray())
+                .ReadOnlySequence,
+
+            new SequenceBuilder()
+                .Append("NG"u8.ToArray())
+                .Append("\r"u8.ToArray())
+                .Append("\n"u8.ToArray())
+                .Append("PO"u8.ToArray())
+                .ReadOnlySequence,
+
+            new SequenceBuilder()
+                .Append("NG\r\n"u8.ToArray())
+                .Append("+OK\r\n"u8.ToArray())
+                .Append("-ER"u8.ToArray())
+                .Append("R 'cra"u8.ToArray())
+                .Append("sh!'\r\nPI"u8.ToArray())
+                .Append("NG\r\n"u8.ToArray())
+                .ReadOnlySequence,
+
+            new SequenceBuilder()
+                .Append("MSG subject sid1 reply_to 1\r\nx\r\n"u8.ToArray())
+                .ReadOnlySequence,
+
+            new SequenceBuilder()
+                .Append("PING\r\n"u8.ToArray())
+                .ReadOnlySequence
+        ];
+
+        _parser = new NatsProtocolParser();
+    }
+
+    public int Parse()
+    {
+        var tokenizer = new NatsProtocolParser.NatsTokenizer();
+        var count = 0;
+
+        foreach (var sequence in _sequences)
+        {
+            var buffer = sequence;
+
+            while (_parser.TryRead(ref tokenizer, ref buffer))
+            {
+                switch (_parser.Command)
+                {
+                case NatsProtocolParser.NatsTokenizer.Command.INFO:
+                case NatsProtocolParser.NatsTokenizer.Command.PING:
+                case NatsProtocolParser.NatsTokenizer.Command.PONG:
+                case NatsProtocolParser.NatsTokenizer.Command.OK:
+                case NatsProtocolParser.NatsTokenizer.Command.ERR:
+                case NatsProtocolParser.NatsTokenizer.Command.MSG:
+                    count++;
+                    break;
+                }
+
+                _parser.Reset();
+            }
+        }
+
+        if (count != 8)
+            throw new Exception("Invalid count");
+
+        return count;
+    }
+
+    private class BufferSegment : ReadOnlySequenceSegment<byte>
+    {
+        public void SetMemory(ReadOnlyMemory<byte> memory) => Memory = memory;
+
+        public void SetNextSegment(BufferSegment? segment) => Next = segment;
+
+        public void SetRunningIndex(int index) => RunningIndex = index;
+    }
+
+    private class SequenceBuilder
+    {
+        private BufferSegment? _start;
+        private BufferSegment? _end;
+        private int _length;
+
+        public ReadOnlySequence<byte> ReadOnlySequence => new(_start!, 0, _end!, _end!.Memory.Length);
+
+        // Memory is only allowed rent from ArrayPool.
+        public SequenceBuilder Append(ReadOnlyMemory<byte> buffer)
+        {
+            var segment = new BufferSegment();
+            segment.SetMemory(buffer);
+
+            if (_start == null)
+            {
+                _start = segment;
+                _end = segment;
+            }
+            else
+            {
+                _end!.SetNextSegment(segment);
+                segment.SetRunningIndex(_length);
+                _end = segment;
+            }
+
+            _length += buffer.Length;
+
+            return this;
+        }
+    }
+}

sandbox/MicroBenchmark/NatsProtoParserBench.cs

@@ -0,0 +1,128 @@
+using System.Buffers;
+using BenchmarkDotNet.Attributes;
+using NATS.Client.Core;
+
+namespace MicroBenchmark;
+
+[ShortRunJob]
+[MemoryDiagnoser]
+[PlainExporter]
+public class NatsProtoParserBench
+{
+    private List<ReadOnlySequence<byte>> _sequences;
+    private NatsProtocolParser _parser;
+
+    [GlobalSetup]
+    public void Setup()
+    {
+        _sequences =
+        [
+            new SequenceBuilder()
+                .Append("INFO {\"server_id\":\"nats-server\""u8.ToArray())
+                .Append("}\r"u8.ToArray())
+                .Append("\nPI"u8.ToArray())
+                .ReadOnlySequence,
+
+            new SequenceBuilder()
+                .Append("NG"u8.ToArray())
+                .Append("\r"u8.ToArray())
+                .Append("\n"u8.ToArray())
+                .Append("PO"u8.ToArray())
+                .ReadOnlySequence,
+
+            new SequenceBuilder()
+                .Append("NG\r\n"u8.ToArray())
+                .Append("+OK\r\n"u8.ToArray())
+                .Append("-ER"u8.ToArray())
+                .Append("R 'cra"u8.ToArray())
+                .Append("sh!'\r\nPI"u8.ToArray())
+                .Append("NG\r\n"u8.ToArray())
+                .ReadOnlySequence,
+
+            new SequenceBuilder()
+                .Append("MSG subject sid1 reply_to 1\r\nx\r\n"u8.ToArray())
+                .ReadOnlySequence,
+
+            new SequenceBuilder()
+                .Append("PING\r\n"u8.ToArray())
+                .ReadOnlySequence
+        ];
+
+        _parser = new NatsProtocolParser();
+    }
+
+    [Benchmark]
+    public int Parse()
+    {
+        var tokenizer = new NatsProtocolParser.NatsTokenizer();
+        var count = 0;
+
+        foreach (var sequence in _sequences)
+        {
+            var buffer = sequence;
+
+            while (_parser.TryRead(ref tokenizer, ref buffer))
+            {
+                switch (_parser.Command)
+                {
+                case NatsProtocolParser.NatsTokenizer.Command.INFO:
+                case NatsProtocolParser.NatsTokenizer.Command.PING:
+                case NatsProtocolParser.NatsTokenizer.Command.PONG:
+                case NatsProtocolParser.NatsTokenizer.Command.OK:
+                case NatsProtocolParser.NatsTokenizer.Command.ERR:
+                case NatsProtocolParser.NatsTokenizer.Command.MSG:
+                    count++;
+                    break;
+                }
+
+                _parser.Reset();
+            }
+        }
+
+        if (count != 8)
+            throw new Exception("Invalid count");
+
+        return count;
+    }
+
+    private class BufferSegment : ReadOnlySequenceSegment<byte>
+    {
+        public void SetMemory(ReadOnlyMemory<byte> memory) => Memory = memory;
+
+        public void SetNextSegment(BufferSegment? segment) => Next = segment;
+
+        public void SetRunningIndex(int index) => RunningIndex = index;
+    }
+
+    private class SequenceBuilder
+    {
+        private BufferSegment? _start;
+        private BufferSegment? _end;
+        private int _length;
+
+        public ReadOnlySequence<byte> ReadOnlySequence => new(_start!, 0, _end!, _end!.Memory.Length);
+
+        // Memory is only allowed rent from ArrayPool.
+        public SequenceBuilder Append(ReadOnlyMemory<byte> buffer)
+        {
+            var segment = new BufferSegment();
+            segment.SetMemory(buffer);
+
+            if (_start == null)
+            {
+                _start = segment;
+                _end = segment;
+            }
+            else
+            {
+                _end!.SetNextSegment(segment);
+                segment.SetRunningIndex(_length);
+                _end = segment;
+            }
+
+            _length += buffer.Length;
+
+            return this;
+        }
+    }
+}