connection reroute test cases

.editorconfig

@@ -51,7 +51,7 @@ csharp_new_line_before_members_in_object_initializers = true
 csharp_new_line_before_members_in_anonymous_types = true
 
 # Namespace settings
-csharp_style_namespace_declarations = file_scoped:silent
+csharp_style_namespace_declarations = file_scoped:error
 
 # Brace settings
 csharp_prefer_braces = true:silent# Prefer curly braces even for one line of code
@@ -102,6 +102,7 @@ csharp_prefer_simple_default_expression = true:suggestion
 csharp_style_prefer_local_over_anonymous_function = true:suggestion
 csharp_style_prefer_index_operator = true:suggestion
 csharp_style_prefer_range_operator = true:suggestion
+csharp_space_around_binary_operators = before_and_after
 
 [*.{xml,config,*proj,nuspec,props,resx,targets,yml,tasks}]
 indent_size = 2

src/Microsoft.Azure.SignalR.Common/DefaultHeaderProvider.cs

@@ -0,0 +1,11 @@
+// Copyright (c) Microsoft. All rights reserved.
+// Licensed under the MIT license. See LICENSE file in the project root for full license information.
+
+using System.Collections.Generic;
+
+namespace Microsoft.Azure.SignalR;
+
+internal class DefaultHeaderProvider : ICustomHeaderProvider
+{
+    public IEnumerable<(string key, string val)> Headers => [];
+}

src/Microsoft.Azure.SignalR.Common/Interfaces/ICustomHeaderProvider.cs

@@ -0,0 +1,11 @@
+// Copyright (c) Microsoft. All rights reserved.
+// Licensed under the MIT license. See LICENSE file in the project root for full license information.
+
+using System.Collections.Generic;
+
+namespace Microsoft.Azure.SignalR;
+
+internal interface ICustomHeaderProvider
+{
+    IEnumerable<(string key, string val)> Headers { get; }
+}

test/Microsoft.Azure.SignalR.E2ETests/Common/ITestHubConnection.cs

@@ -0,0 +1,29 @@
+// Copyright (c) Microsoft. All rights reserved.
+// Licensed under the MIT license. See LICENSE file in the project root for full license information.
+
+using System.Threading.Tasks;
+
+namespace Microsoft.Azure.SignalR.E2ETests.Common;
+
+public interface ITestHubConnection
+{
+    public string ConnectionId { get; }
+
+    public string User { get; }
+
+    public int MessageCount { get; }
+
+    public void Listen(params string[] methods);
+
+    public Task StartAsync();
+
+    public Task StopAsync();
+
+    public Task SendAsync(string method, params string[] messages);
+
+    public void ResetInvoke(string method);
+
+    public Task ExpectInvokeAsync(string method, params string[] message);
+
+    public void ResetMessageCount();
+}

test/Microsoft.Azure.SignalR.E2ETests/Common/ITestHubConnectionGroup.cs

@@ -0,0 +1,10 @@
+// Copyright (c) Microsoft. All rights reserved.
+// Licensed under the MIT license. See LICENSE file in the project root for full license information.
+
+using System.Collections.Generic;
+
+namespace Microsoft.Azure.SignalR.E2ETests.Common;
+
+public interface ITestHubConnectionGroup : ITestHubConnection, IEnumerable<ITestHubConnection>
+{
+}

test/Microsoft.Azure.SignalR.E2ETests/Common/ITestServer.cs

@@ -0,0 +1,15 @@
+// Copyright (c) Microsoft. All rights reserved.
+// Licensed under the MIT license. See LICENSE file in the project root for full license information.
+
+using System;
+using System.Collections.Generic;
+using System.Threading.Tasks;
+
+namespace Microsoft.Azure.SignalR.E2ETests.Common;
+
+public interface ITestServer : IDisposable
+{
+    Task<string> StartAsync(Dictionary<string, string?>? configuration = null);
+
+    Task StopAsync();
+}

test/Microsoft.Azure.SignalR.E2ETests/Common/IngressHeaderProvider.cs

@@ -0,0 +1,29 @@
+// Copyright (c) Microsoft. All rights reserved.
+// Licensed under the MIT license. See LICENSE file in the project root for full license information.
+
+using System;
+using System.Collections.Generic;
+
+namespace Microsoft.Azure.SignalR.E2ETests.Common;
+
+internal sealed class IngressHeaderProvider : ICustomHeaderProvider
+{
+    public const string AsrsInternalForwardedBy = "X-ASRS-Internal-Forwarded-By";
+
+    public const string PodA = "podA";
+
+    public const string PodB = "podB";
+
+    private static readonly string[] Ingresses = [PodA, PodB];
+
+    private static readonly Random Random = new();
+
+    public IEnumerable<(string key, string val)> Headers
+    {
+        get
+        {
+            var podName = Ingresses[Random.Next() % Ingresses.Length];
+            yield return (AsrsInternalForwardedBy, podName);
+        }
+    }
+}

test/Microsoft.Azure.SignalR.E2ETests/Common/TestHubConnectionFactory.cs

@@ -0,0 +1,195 @@
+// Copyright (c) Microsoft. All rights reserved.
+// Licensed under the MIT license. See LICENSE file in the project root for full license information.
+
+using System;
+using System.Collections;
+using System.Collections.Concurrent;
+using System.Collections.Generic;
+using System.Diagnostics.CodeAnalysis;
+using System.Linq;
+using System.Threading;
+using System.Threading.Tasks;
+
+using Microsoft.AspNetCore.SignalR.Client;
+
+using Xunit;
+
+namespace Microsoft.Azure.SignalR.E2ETests.Common;
+
+public class TestHubConnectionFactory
+{
+    public bool EnableStatefulReconnect { get; set; }
+
+    public ITestHubConnection NewConnection(string host, string? hub = null, string userId = "foo")
+    {
+        return new TestHubConnection(host, hub)
+        {
+            User = userId,
+            EnableStatefulReconnect = EnableStatefulReconnect
+        };
+    }
+
+    public ITestHubConnectionGroup NewConnectionGroup(string host, int count, string? hub = null, string userId = "foo")
+    {
+        return new TestHubConnectionGroup(host, count, hub)
+        {
+            User = userId,
+            EnableStatefulReconnect = EnableStatefulReconnect
+        };
+    }
+
+    private sealed class TestHubConnection(string host, string? hub = null) : ITestHubConnection
+    {
+        private readonly ConcurrentDictionary<string, TaskCompletionSource<string[]>> _expectedInvokes = new();
+
+        private HubConnection? _hubConnection;
+
+        private volatile int _messageCount;
+
+        public string User { get; init; } = "foo";
+
+        public int MessageCount => _messageCount;
+
+        public bool EnableStatefulReconnect { get; init; }
+
+        public string ConnectionId => _hubConnection?.ConnectionId ?? throw NotReady;
+
+        private static Exception NotReady { get; } = new InvalidOperationException("HubConnection is not in connected state.");
+
+        public Task StartAsync()
+        {
+            BuildConnectionIfNull();
+            return _hubConnection.StartAsync();
+        }
+
+        public Task StopAsync() => _hubConnection?.StopAsync() ?? Task.CompletedTask;
+
+        public Task SendAsync(string method, params string[] messages)
+        {
+            if (_hubConnection == null || _hubConnection.State != HubConnectionState.Connected)
+            {
+                throw NotReady;
+            }
+            return _hubConnection.SendCoreAsync(method, messages);
+        }
+
+        public void Listen(params string[] methods)
+        {
+            BuildConnectionIfNull();
+            foreach (var method in methods)
+            {
+                _expectedInvokes.TryAdd(method, new TaskCompletionSource<string[]>());
+                _hubConnection.On(method, (Action<string>)(message => Invoke(method, message)));
+            }
+        }
+
+        public void ResetInvoke(string method)
+        {
+            _expectedInvokes.AddOrUpdate(method,
+                                         new TaskCompletionSource<string[]>(),
+                                         (method, ov) => ov.Task.IsCompleted ? new TaskCompletionSource<string[]>() : ov);
+        }
+
+        public async Task ExpectInvokeAsync(string method, params string[] messages)
+        {
+            Assert.Equal(messages, await _expectedInvokes[method].Task);
+        }
+
+        public void ResetMessageCount()
+        {
+            _messageCount = 0;
+        }
+
+        private void Invoke(string method, params string[] messages)
+        {
+            Interlocked.Increment(ref _messageCount);
+            if (_expectedInvokes.TryGetValue(method, out var source))
+            {
+                source.TrySetResult(messages);
+            }
+        }
+
+        [MemberNotNull(nameof(_hubConnection))]
+        private void BuildConnectionIfNull()
+        {
+            if (_hubConnection == null)
+            {
+                hub ??= nameof(TestHub);
+                var url = $"{host}/{hub}?user={User}";
+                var builder = new HubConnectionBuilder().WithUrl(url);
+
+                if (EnableStatefulReconnect)
+                {
+                    builder = builder.WithStatefulReconnect();
+                }
+                _hubConnection = builder.Build();
+            }
+        }
+    }
+
+    private sealed class TestHubConnectionGroup(string host, int count, string? hub = null) : ITestHubConnectionGroup
+    {
+        private List<ITestHubConnection>? _connections;
+
+        public bool EnableStatefulReconnect { get; init; }
+
+        public IEnumerable<ITestHubConnection> Connections
+        {
+            get
+            {
+                _connections ??= (from i in Enumerable.Range(0, count)
+                                  select new TestHubConnection(host, hub)
+                                  {
+                                      User = User,
+                                      EnableStatefulReconnect = EnableStatefulReconnect,
+                                  } as ITestHubConnection).ToList();
+                return _connections;
+            }
+        }
+
+        public string User { get; init; } = string.Empty;
+
+        public int MessageCount => Connections.Select(x => x.MessageCount).Sum();
+
+        public string ConnectionId => throw new NotImplementedException("Connection group does not have ConnectionId.");
+
+        public void Listen(params string[] methods)
+        {
+            foreach (var connection in Connections)
+            {
+                connection.Listen(methods);
+            }
+        }
+
+        public async Task StartAsync() => await Task.WhenAll(Connections.Select(x => x.StartAsync()));
+
+        public IEnumerator<ITestHubConnection> GetEnumerator() => Connections.GetEnumerator();
+
+        IEnumerator IEnumerable.GetEnumerator() => GetEnumerator();
+
+        public async Task StopAsync() => await Task.WhenAll(Connections.Select(x => x.StopAsync()));
+
+        public void ResetInvoke(string method)
+        {
+            foreach (var connection in Connections)
+            {
+                connection.ResetInvoke(method);
+            }
+        }
+
+        public Task ExpectInvokeAsync(string method, params string[] messages)
+        {
+            return Task.WhenAll(Connections.Select(x => x.ExpectInvokeAsync(method, messages)));
+        }
+
+        public Task SendAsync(string method, params string[] messages) => Task.WhenAll(Connections.Select(x => x.SendAsync(method, messages)));
+
+        public void ResetMessageCount()
+        {
+            foreach (var connection in Connections)
+            {
+                connection.ResetMessageCount();
+            }
+        }
+    }
+}