Add Exchange version (#277)

* Add Exchange version
Implement #153
Signed-off-by: Gabriele Santomaggio <G.santomaggio@gmail.com>

Author: Gsantomaggio

RabbitMQ.Stream.Client/Client.cs

@@ -364,6 +364,12 @@ public async Task<RouteQueryResponse> QueryRoute(string superStream, string rout
                 new RouteQueryRequest(corr, superStream, routingKey)).ConfigureAwait(false);
         }
 
+        public async Task<CommandVersionsResponse> ExchangeVersions()
+        {
+            return await Request<CommandVersionsRequest, CommandVersionsResponse>(corr =>
+                new CommandVersionsRequest(corr)).ConfigureAwait(false);
+        }
+
         private async ValueTask<TOut> Request<TIn, TOut>(Func<uint, TIn> request, TimeSpan? timeout = null)
             where TIn : struct, ICommand where TOut : struct, ICommand
         {
@@ -553,6 +559,10 @@ private void HandleCorrelatedCommand(ushort tag, ref ReadOnlySequence<byte> fram
                     RouteQueryResponse.Read(frame, out var routeQueryResponse);
                     HandleCorrelatedResponse(routeQueryResponse);
                     break;
+                case CommandVersionsResponse.Key:
+                    CommandVersionsResponse.Read(frame, out var commandVersionsResponse);
+                    HandleCorrelatedResponse(commandVersionsResponse);
+                    break;
                 default:
                     if (MemoryMarshal.TryGetArray(frame.First, out var segment))
                     {

RabbitMQ.Stream.Client/CommandVersionsRequest.cs

@@ -0,0 +1,48 @@
+// This source code is dual-licensed under the Apache License, version
+// 2.0, and the Mozilla Public License, version 2.0.
+// Copyright (c) 2007-2023 VMware, Inc.
+
+using System;
+
+namespace RabbitMQ.Stream.Client;
+
+public readonly struct CommandVersionsRequest : ICommand
+{
+    private const ushort Key = 0x001b;
+    private readonly uint _correlationId;
+    private readonly ICommandVersions[] _commands = { };
+
+    public CommandVersionsRequest(uint correlationId)
+    {
+        _correlationId = correlationId;
+    }
+
+    public int SizeNeeded
+    {
+        get
+        {
+            var size = 2 + 2 + 4
+                       + 4 + // _commands.Length
+                          _commands.Length * (2 + 2 + 2);
+            return size;
+        }
+    }
+
+    public int Write(Span<byte> span)
+    {
+        var offset = WireFormatting.WriteUInt16(span, Key);
+        offset += WireFormatting.WriteUInt16(span[offset..], ((ICommand)this).Version);
+        offset += WireFormatting.WriteUInt32(span[offset..], _correlationId);
+        offset += WireFormatting.WriteInt32(span[offset..], _commands.Length);
+
+        foreach (var iCommandVersions in _commands)
+        {
+            offset += WireFormatting.WriteUInt16(span[offset..], iCommandVersions.Command);
+            offset += WireFormatting.WriteUInt16(span[offset..], iCommandVersions.MinVersion);
+            offset += WireFormatting.WriteUInt16(span[offset..], iCommandVersions.MaxVersion);
+        }
+
+        return offset;
+    }
+}
+

RabbitMQ.Stream.Client/CommandVersionsResponse.cs

@@ -0,0 +1,50 @@
+// This source code is dual-licensed under the Apache License, version
+// 2.0, and the Mozilla Public License, version 2.0.
+// Copyright (c) 2007-2023 VMware, Inc.
+
+using System;
+using System.Buffers;
+using System.Collections.Generic;
+
+namespace RabbitMQ.Stream.Client;
+
+public struct CommandVersionsResponse : ICommand
+{
+    internal const ushort Key = 0x001b;
+
+    private CommandVersionsResponse(uint correlationId, ResponseCode responseCode, List<ICommandVersions> commands)
+    {
+        CorrelationId = correlationId;
+        ResponseCode = responseCode;
+        Commands = commands;
+    }
+
+    public int SizeNeeded { get => throw new NotImplementedException(); }
+    public int Write(Span<byte> span) => throw new NotImplementedException();
+
+    public uint CorrelationId { get; }
+    public ResponseCode ResponseCode { get; }
+
+    public List<ICommandVersions> Commands { get; }
+
+    internal static int Read(ReadOnlySequence<byte> frame, out CommandVersionsResponse command)
+    {
+        var offset = WireFormatting.ReadUInt16(frame, out _);
+        offset += WireFormatting.ReadUInt16(frame.Slice(offset), out _);
+        offset += WireFormatting.ReadUInt32(frame.Slice(offset), out var correlation);
+        offset += WireFormatting.ReadUInt16(frame.Slice(offset), out var responseCode);
+        offset += WireFormatting.ReadUInt32(frame.Slice(offset), out var commandLen);
+        var commands = new List<ICommandVersions>();
+
+        for (var i = 0; i < commandLen; i++)
+        {
+            offset += WireFormatting.ReadUInt16(frame.Slice(offset), out var commandKey);
+            offset += WireFormatting.ReadUInt16(frame.Slice(offset), out var minVersion);
+            offset += WireFormatting.ReadUInt16(frame.Slice(offset), out var maxVersion);
+            commands.Add(new CommandVersions(commandKey, minVersion, maxVersion));
+        }
+
+        command = new CommandVersionsResponse(correlation, (ResponseCode)responseCode, commands);
+        return offset;
+    }
+}

RabbitMQ.Stream.Client/ICommandVersions.cs

@@ -0,0 +1,26 @@
+// This source code is dual-licensed under the Apache License, version
+// 2.0, and the Mozilla Public License, version 2.0.
+// Copyright (c) 2007-2023 VMware, Inc.
+
+namespace RabbitMQ.Stream.Client;
+
+public interface ICommandVersions
+{
+    public ushort MaxVersion { get; }
+    public ushort MinVersion { get; }
+    public ushort Command { get; }
+}
+
+public class CommandVersions : ICommandVersions
+{
+    public CommandVersions(ushort command, ushort minVersion, ushort maxVersion)
+    {
+        Command = command;
+        MinVersion = minVersion;
+        MaxVersion = maxVersion;
+    }
+
+    public ushort MaxVersion { get; }
+    public ushort MinVersion { get; }
+    public ushort Command { get; }
+}

RabbitMQ.Stream.Client/PublicAPI.Unshipped.txt

@@ -9,11 +9,33 @@ RabbitMQ.Stream.Client.AuthMechanismNotSupportedException
 RabbitMQ.Stream.Client.AuthMechanismNotSupportedException.AuthMechanismNotSupportedException(string s) -> void
 RabbitMQ.Stream.Client.Chunk.Data.get -> System.Memory<byte>
 RabbitMQ.Stream.Client.Chunk.MagicVersion.get -> byte
+RabbitMQ.Stream.Client.Client.ExchangeVersions() -> System.Threading.Tasks.Task<RabbitMQ.Stream.Client.CommandVersionsResponse>
 RabbitMQ.Stream.Client.Client.QueryRoute(string superStream, string routingKey) -> System.Threading.Tasks.Task<RabbitMQ.Stream.Client.RouteQueryResponse>
 RabbitMQ.Stream.Client.Client.StreamStats(string stream) -> System.Threading.Tasks.ValueTask<RabbitMQ.Stream.Client.StreamStatsResponse>
 RabbitMQ.Stream.Client.ClientParameters.AuthMechanism.get -> RabbitMQ.Stream.Client.AuthMechanism
 RabbitMQ.Stream.Client.ClientParameters.AuthMechanism.set -> void
+RabbitMQ.Stream.Client.CommandVersions
+RabbitMQ.Stream.Client.CommandVersions.Command.get -> ushort
+RabbitMQ.Stream.Client.CommandVersions.CommandVersions(ushort command, ushort minVersion, ushort maxVersion) -> void
+RabbitMQ.Stream.Client.CommandVersions.MaxVersion.get -> ushort
+RabbitMQ.Stream.Client.CommandVersions.MinVersion.get -> ushort
+RabbitMQ.Stream.Client.CommandVersionsRequest
+RabbitMQ.Stream.Client.CommandVersionsRequest.CommandVersionsRequest() -> void
+RabbitMQ.Stream.Client.CommandVersionsRequest.CommandVersionsRequest(uint correlationId) -> void
+RabbitMQ.Stream.Client.CommandVersionsRequest.SizeNeeded.get -> int
+RabbitMQ.Stream.Client.CommandVersionsRequest.Write(System.Span<byte> span) -> int
+RabbitMQ.Stream.Client.CommandVersionsResponse
+RabbitMQ.Stream.Client.CommandVersionsResponse.Commands.get -> System.Collections.Generic.List<RabbitMQ.Stream.Client.ICommandVersions>
+RabbitMQ.Stream.Client.CommandVersionsResponse.CommandVersionsResponse() -> void
+RabbitMQ.Stream.Client.CommandVersionsResponse.CorrelationId.get -> uint
+RabbitMQ.Stream.Client.CommandVersionsResponse.ResponseCode.get -> RabbitMQ.Stream.Client.ResponseCode
+RabbitMQ.Stream.Client.CommandVersionsResponse.SizeNeeded.get -> int
+RabbitMQ.Stream.Client.CommandVersionsResponse.Write(System.Span<byte> span) -> int
 RabbitMQ.Stream.Client.HashRoutingMurmurStrategy.Route(RabbitMQ.Stream.Client.Message message, System.Collections.Generic.List<string> partitions) -> System.Threading.Tasks.Task<System.Collections.Generic.List<string>>
+RabbitMQ.Stream.Client.ICommandVersions
+RabbitMQ.Stream.Client.ICommandVersions.Command.get -> ushort
+RabbitMQ.Stream.Client.ICommandVersions.MaxVersion.get -> ushort
+RabbitMQ.Stream.Client.ICommandVersions.MinVersion.get -> ushort
 RabbitMQ.Stream.Client.IConsumerConfig.InitialCredits.get -> ushort
 RabbitMQ.Stream.Client.IConsumerConfig.InitialCredits.set -> void
 RabbitMQ.Stream.Client.IRoutingStrategy.Route(RabbitMQ.Stream.Client.Message message, System.Collections.Generic.List<string> partitions) -> System.Threading.Tasks.Task<System.Collections.Generic.List<string>>

Tests/ClientTests.cs

@@ -119,10 +119,7 @@ public async void PublishShouldError()
             await client.CreateStream(stream, new Dictionary<string, string>());
             var testPassed = new TaskCompletionSource<bool>();
 
-            Action<ReadOnlyMemory<ulong>> confirmed = (pubIds) =>
-            {
-                testPassed.SetResult(false);
-            };
+            Action<ReadOnlyMemory<ulong>> confirmed = (pubIds) => { testPassed.SetResult(false); };
 
             Action<(ulong, ResponseCode)[]> errored = (errors) =>
             {
@@ -415,5 +412,16 @@ await Assert.ThrowsAsync<VirtualHostAccessFailureException>(
                 async () => { await Client.Create(clientParameters); }
             );
         }
+
+        [Fact]
+        public async void ExchangeVersionCommandsShouldNotBeEmpty()
+        {
+            var clientParameters = new ClientParameters { };
+            var client = await Client.Create(clientParameters);
+            var response = await client.ExchangeVersions();
+            Assert.Equal(ResponseCode.Ok, response.ResponseCode);
+            Assert.True(response.Commands.Count > 0);
+            await client.Close("done");
+        }
     }
 }