Prepare to SASL implementation

src/RabbitMQ.Next.Abstractions/Auth/PlainAuthMechanism.cs

@@ -1,3 +1,7 @@
+using System;
+using System.Text;
+using System.Threading.Tasks;
+
 namespace RabbitMQ.Next.Auth;
 
 public class PlainAuthMechanism : IAuthMechanism
@@ -9,8 +13,16 @@ public PlainAuthMechanism(string userName, string password)
     }
 
     public string Type => "PLAIN";
+    public ValueTask<ReadOnlyMemory<byte>> HandleChallengeAsync(ReadOnlySpan<byte> challenge)
+    {
+        if (!challenge.IsEmpty)
+        {
+            throw new NotSupportedException("PlainAuthMechanism does not support challenges.");
+        }
 
-    public string ToResponse() => $"\0{this.UserName}\0{this.Password}";
+        ReadOnlyMemory<byte> response = Encoding.UTF8.GetBytes($"\0{this.UserName}\0{this.Password}");
+        return ValueTask.FromResult(response);
+    }
 
     public string UserName { get; }
 

src/RabbitMQ.Next.Abstractions/IAuthMechanism.cs

@@ -1,8 +1,11 @@
+using System;
+using System.Threading.Tasks;
+
 namespace RabbitMQ.Next;
 
 public interface IAuthMechanism
 {
     string Type { get; }
 
-    string ToResponse();
+    ValueTask<ReadOnlyMemory<byte>> HandleChallengeAsync(ReadOnlySpan<byte> challenge);
 }

src/RabbitMQ.Next/Connection.cs

@@ -270,17 +270,16 @@ private static async Task<NegotiationResults> NegotiateConnectionAsync(IChannel
         // connection should be forcibly closed if negotiation phase take more then 10s.
         cancellation = new CancellationTokenSource(TimeSpan.FromSeconds(10)).Token.Combine(cancellation);
 
-        var startMethodTask = channel.WaitAsync<StartMethod>(cancellation);
-
-        var startMethod = await startMethodTask;
-
+        var startMethod = await channel.WaitAsync<StartMethod>(cancellation);
         if (!startMethod.Mechanisms.Contains(settings.Auth.Type))
         {
             throw new NotSupportedException("Provided auth mechanism does not supported by the server");
         }
 
+        var initialChallengeResponse = await settings.Auth.HandleChallengeAsync(Span<byte>.Empty);
+
         var tuneMethodTask = channel.WaitAsync<TuneMethod>(cancellation);
-        await channel.SendAsync(new StartOkMethod(settings.Auth.Type, settings.Auth.ToResponse(), settings.Locale, settings.ClientProperties), cancellation);
+        await channel.SendAsync(new StartOkMethod(settings.Auth.Type, initialChallengeResponse, settings.Locale, settings.ClientProperties), cancellation);
 
         var tuneMethod = await tuneMethodTask;
         var negotiationResult = new NegotiationResults(

src/RabbitMQ.Next/Transport/Methods/Connection/StartOkMethod.cs

@@ -1,11 +1,12 @@
+using System;
 using System.Collections.Generic;
 using RabbitMQ.Next.Methods;
 
 namespace RabbitMQ.Next.Transport.Methods.Connection;
 
 public readonly struct StartOkMethod : IOutgoingMethod
 {
-    public StartOkMethod(string mechanism, string response, string locale, IReadOnlyDictionary<string, object> clientProperties)
+    public StartOkMethod(string mechanism, ReadOnlyMemory<byte> response, string locale, IReadOnlyDictionary<string, object> clientProperties)
     {
         this.Mechanism = mechanism;
         this.Response = response;
@@ -19,7 +20,7 @@ public StartOkMethod(string mechanism, string response, string locale, IReadOnly
 
     public string Mechanism { get; }
 
-    public string Response { get; }
+    public ReadOnlyMemory<byte> Response { get; }
 
     public string Locale { get; }
 }

src/RabbitMQ.Next/Transport/Methods/Connection/StartOkMethodFormatter.cs

@@ -8,6 +8,6 @@ public Span<byte> Write(Span<byte> destination, StartOkMethod method)
         => destination
             .Write(method.ClientProperties)
             .Write(method.Mechanism)
-            .Write(method.Response, true)
+            .Write(method.Response.Span)
             .Write(method.Locale);
 }

tests/RabbitMQ.Next.Tests/Abstractions/Auth/PlainAuthMechanismTests.cs

@@ -1,3 +1,6 @@
+using System;
+using System.Text;
+using System.Threading.Tasks;
 using NSubstitute;
 using RabbitMQ.Next.Auth;
 using Xunit;
@@ -20,14 +23,16 @@ public void CtorTests()
     }
 
     [Fact]
-    public void ToResponse()
+    public async Task HandleChallengeAsync()
     {
         var user = "test";
         var password = "pwd";
+        var expected = "\0test\0pwd"u8.ToArray();
 
         var auth = new PlainAuthMechanism(user, password);
-
-        Assert.Equal($"\0{user}\0{password}", auth.ToResponse());
+        var response = await auth.HandleChallengeAsync(ReadOnlySpan<byte>.Empty);
+
+        Assert.Equal(expected, response.ToArray());
     }
 
     [Fact]

tests/RabbitMQ.Next.Tests/Tasks/TaskExtensionsTests.cs

@@ -60,7 +60,7 @@ public async Task WithCancellationReturnsResult()
 
         var wrapped = task.WithCancellation(cancellation.Token);
         tcs.SetResult(5);
-        await Task.Yield();
+        await Task.Delay(10);
 
         Assert.True(wrapped.IsCompleted);
         Assert.Equal(5, await wrapped);

tests/RabbitMQ.Next.Tests/Transport/Methods/Connection/ModelTests.cs

@@ -32,7 +32,7 @@ public void StartMethod()
     public void StartOkMethod()
     {
         var mechanism = "PLAIN";
-        var response = "test";
+        var response = "ab"u8.ToArray();
         var locale = "en_US";
         var clientProperties = new Dictionary<string, object>()
         {

tests/RabbitMQ.Next.Tests/Transport/Methods/Connection/SerializationTests.cs

@@ -49,7 +49,7 @@ public void StartOkMethodFormatter()
                 ["exchange_exchange_bindings"] = true
             }
         };
-        var method = new StartOkMethod("PLAIN", "\0test1\0test1", "en_US", clientProperties);
+        var method = new StartOkMethod("PLAIN",  "\0test1\0test1"u8.ToArray(), "en_US", clientProperties);
 
         this.TestFormatter(method);
     }