// Licensed to the .NET Foundation under one or more agreements.
// The .NET Foundation licenses this file to you under the MIT license.

using System.Buffers;
using System.Diagnostics;
using System.Globalization;
using System.IO.Pipelines;
using System.Security.Claims;
using System.Text;
using System.Threading.Channels;
using MessagePack;
using MessagePack.Formatters;
using MessagePack.Resolvers;
using Microsoft.AspNetCore.Authorization;
using Microsoft.AspNetCore.Connections;
using Microsoft.AspNetCore.Connections.Features;
using Microsoft.AspNetCore.Http;
using Microsoft.AspNetCore.Http.Connections.Features;
using Microsoft.AspNetCore.SignalR.Internal;
using Microsoft.AspNetCore.SignalR.Protocol;
using Microsoft.AspNetCore.Testing;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Logging.Testing;
using Microsoft.Extensions.Options;
using Moq;
using Newtonsoft.Json;
using Newtonsoft.Json.Linq;
using Newtonsoft.Json.Serialization;

namespace Microsoft.AspNetCore.SignalR.Tests;

public partial class HubConnectionHandlerTests : VerifiableLoggedTest
{
    [Fact]
    [LogLevel(LogLevel.Trace)]
    public async Task HubsAreDisposed()
    {
        using (StartVerifiableLog())
        {
            var trackDispose = new TrackDispose();
            var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(s => s.AddSingleton(trackDispose), LoggerFactory);
            var connectionHandler = serviceProvider.GetService<HubConnectionHandler<DisposeTrackingHub>>();

            using (var client = new TestClient())
            {
                var connectionHandlerTask = await client.ConnectAsync(connectionHandler);

                // kill the connection
                client.Dispose();

                await connectionHandlerTask;

                Assert.Equal(2, trackDispose.DisposeCount);
            }
        }
    }

    [Fact]
    public async Task AsyncDisposablesInHubsAreSupported()
    {
        using (StartVerifiableLog())
        {
            var trackDispose = new TrackDispose();
            var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(s =>
            {
                s.AddScoped<AsyncDisposable>();
                s.AddSingleton(trackDispose);
            },
            LoggerFactory);
            var connectionHandler = serviceProvider.GetService<HubConnectionHandler<HubWithAsyncDisposable>>();

            using (var client = new TestClient())
            {
                var connectionHandlerTask = await client.ConnectAsync(connectionHandler);

                var result = (await client.InvokeAsync(nameof(HubWithAsyncDisposable.Test)).DefaultTimeout());
                Assert.NotNull(result);

                // kill the connection
                client.Dispose();

                await connectionHandlerTask;

                Assert.Equal(3, trackDispose.DisposeCount);
            }
        }
    }

    [Fact]
    public async Task ConnectionAbortedTokenTriggers()
    {
        using (StartVerifiableLog())
        {
            var state = new ConnectionLifetimeState();
            var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(s => s.AddSingleton(state), LoggerFactory);
            var connectionHandler = serviceProvider.GetService<HubConnectionHandler<ConnectionLifetimeHub>>();

            using (var client = new TestClient())
            {
                var connectionHandlerTask = await client.ConnectAsync(connectionHandler);

                // kill the connection
                client.Dispose();

                await connectionHandlerTask.DefaultTimeout();

                Assert.True(state.TokenCallbackTriggered);
                Assert.False(state.TokenStateInConnected);
                Assert.True(state.TokenStateInDisconnected);
            }
        }
    }

    [Fact]
    public async Task OnDisconnectedAsyncTriggersWhenAbortedTokenCallbackThrows()
    {
        using (StartVerifiableLog())
        {
            var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(null, LoggerFactory);
            var connectionHandler = serviceProvider.GetService<HubConnectionHandler<ErrorInAbortedTokenHub>>();

            using (var client = new TestClient())
            {
                var connectionHandlerTask = await client.ConnectAsync(connectionHandler);

                // kill the connection
                client.Dispose();

                await connectionHandlerTask.DefaultTimeout();

                var firedOnConnected = (bool)client.Connection.Items[nameof(ErrorInAbortedTokenHub.OnConnectedAsync)];
                var firedOnDisconnected = (bool)client.Connection.Items[nameof(ErrorInAbortedTokenHub.OnDisconnectedAsync)];

                Assert.True(firedOnConnected);
                Assert.True(firedOnDisconnected);
            }
        }
    }

    [Fact]
    public async Task AbortFromHubMethodForcesClientDisconnect()
    {
        using (StartVerifiableLog())
        {
            var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(null, LoggerFactory);
            var connectionHandler = serviceProvider.GetService<HubConnectionHandler<AbortHub>>();

            using (var client = new TestClient())
            {
                var connectionHandlerTask = await client.ConnectAsync(connectionHandler);

                await client.SendInvocationAsync(nameof(AbortHub.Kill)).DefaultTimeout();

                var close = Assert.IsType<CloseMessage>(await client.ReadAsync().DefaultTimeout());
                Assert.False(close.AllowReconnect);

                await connectionHandlerTask.DefaultTimeout();

                Assert.Null(client.TryRead());
            }
        }
    }

    [Fact]
    public async Task MissingHandshakeAndMessageSentFromHubConnectionCanBeDisposedCleanly()
    {
        using (StartVerifiableLog())
        {
            var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(null, LoggerFactory);
            var connectionHandler = serviceProvider.GetService<HubConnectionHandler<SimpleHub>>();

            using (var client = new TestClient())
            {
                var connectionHandlerTask = await client.ConnectAsync(connectionHandler, false, false);

                // kill the connection
                client.Dispose();

                await connectionHandlerTask;

                Assert.Null(client.HandshakeResponseMessage);
            }
        }
    }

    [Fact]
    public async Task HandshakeTimesOut()
    {
        using (StartVerifiableLog())
        {
            var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(services =>
            {
                services.Configure<HubOptions>(options =>
                {
                    options.HandshakeTimeout = TimeSpan.FromMilliseconds(5);
                });
            }, LoggerFactory);
            var connectionHandler = serviceProvider.GetService<HubConnectionHandler<SimpleHub>>();

            using (var client = new TestClient())
            {
                var connectionHandlerTask = await client.ConnectAsync(connectionHandler, false);

                Assert.NotNull(client.HandshakeResponseMessage);
                Assert.Equal("Handshake was canceled.", client.HandshakeResponseMessage.Error);

                await connectionHandlerTask.DefaultTimeout();
            }
        }
    }

    [Fact]
    public async Task CanLoadHubContext()
    {
        using (StartVerifiableLog())
        {
            var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(null, LoggerFactory);
            var context = serviceProvider.GetRequiredService<IHubContext<SimpleHub>>();
            await context.Clients.All.SendAsync("Send", "test");
        }
    }

    [Fact]
    public async Task CanLoadTypedHubContext()
    {
        using (StartVerifiableLog())
        {
            var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(null, LoggerFactory);
            var context = serviceProvider.GetRequiredService<IHubContext<SimpleTypedHub, ITypedHubClient>>();
            await context.Clients.All.Send("test");
        }
    }

    [Fact]
    public void FailsToLoadInvalidTypedHubClient()
    {
        using (StartVerifiableLog())
        {
            var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(null, LoggerFactory);
            var ex = Assert.Throws<InvalidOperationException>(() =>
                serviceProvider.GetRequiredService<IHubContext<SimpleVoidReturningTypedHub, IVoidReturningTypedHubClient>>());
            Assert.Equal($"Cannot generate proxy implementation for '{typeof(IVoidReturningTypedHubClient).FullName}.{nameof(IVoidReturningTypedHubClient.Send)}'. All client proxy methods must return '{typeof(Task).FullName}' or 'System.Threading.Tasks.Task<T>'.", ex.Message);
        }
    }

    [Fact]
    public async Task HandshakeFailureFromUnknownProtocolSendsResponseWithError()
    {
        using (StartVerifiableLog())
        {
            var hubProtocolMock = new Mock<IHubProtocol>();
            hubProtocolMock.Setup(m => m.Name).Returns("CustomProtocol");

            var connectionHandler = HubConnectionHandlerTestUtils.GetHubConnectionHandler(typeof(HubT), loggerFactory: LoggerFactory);

            using (var client = new TestClient(protocol: hubProtocolMock.Object))
            {
                var connectionHandlerTask = await client.ConnectAsync(connectionHandler);

                Assert.NotNull(client.HandshakeResponseMessage);
                Assert.Equal("The protocol 'CustomProtocol' is not supported.", client.HandshakeResponseMessage.Error);

                client.Dispose();

                await connectionHandlerTask.DefaultTimeout();
            }
        }
    }

    [Fact]
    public async Task HandshakeFailureFromUnsupportedFormatSendsResponseWithError()
    {
        using (StartVerifiableLog())
        {
            var hubProtocolMock = new Mock<IHubProtocol>();
            hubProtocolMock.Setup(m => m.Name).Returns("CustomProtocol");

            var connectionHandler = HubConnectionHandlerTestUtils.GetHubConnectionHandler(typeof(HubT), loggerFactory: LoggerFactory);

            using (var client = new TestClient(protocol: new MessagePackHubProtocol()))
            {
                client.SupportedFormats = TransferFormat.Text;

                var connectionHandlerTask = await client.ConnectAsync(connectionHandler);

                Assert.NotNull(client.HandshakeResponseMessage);
                Assert.Equal("Cannot use the 'messagepack' protocol on the current transport. The transport does not support 'Binary' transfer format.", client.HandshakeResponseMessage.Error);

                client.Dispose();

                await connectionHandlerTask.DefaultTimeout();
            }
        }
    }

    [Fact]
    public async Task ConnectionClosedWhenHandshakeLargerThanMaxMessageSize()
    {
        using (StartVerifiableLog())
        {
            var connectionHandler = HubConnectionHandlerTestUtils.GetHubConnectionHandler(typeof(HubT), loggerFactory: LoggerFactory,
                builder =>
                {
                    builder.AddSignalR(o =>
                    {
                        o.MaximumReceiveMessageSize = 1;
                    });
                });

            using (var client = new TestClient())
            {
                client.SupportedFormats = TransferFormat.Text;

                var connectionHandlerTask = await client.ConnectAsync(connectionHandler,
                                                        sendHandshakeRequestMessage: true,
                                                        expectedHandshakeResponseMessage: false);

                var message = await client.ReadAsync(isHandshake: true).DefaultTimeout();

                Assert.Equal("Handshake was canceled.", ((HandshakeResponseMessage)message).Error);

                // Connection closes
                await connectionHandlerTask.DefaultTimeout();

                client.Dispose();
            }
        }
    }

    [Fact]
    public async Task SendingHandshakeRequestInChunksWorks()
    {
        using (StartVerifiableLog())
        {
            var connectionHandler = HubConnectionHandlerTestUtils.GetHubConnectionHandler(typeof(HubT), loggerFactory: LoggerFactory);
            var part1 = Encoding.UTF8.GetBytes("{\"protocol\": \"json\"");
            var part2 = Encoding.UTF8.GetBytes(",\"version\": 1}");
            var part3 = Encoding.UTF8.GetBytes("\u001e");

            using (var client = new TestClient())
            {
                client.SupportedFormats = TransferFormat.Text;

                var connectionHandlerTask = await client.ConnectAsync(connectionHandler,
                                                        sendHandshakeRequestMessage: false,
                                                        expectedHandshakeResponseMessage: false);

                // Wait for the handshake response
                var task = client.ReadAsync(isHandshake: true);

                await client.Connection.Application.Output.WriteAsync(part1);

                Assert.False(task.IsCompleted);

                await client.Connection.Application.Output.WriteAsync(part2);

                Assert.False(task.IsCompleted);

                await client.Connection.Application.Output.WriteAsync(part3);

                Assert.True(task.IsCompleted);

                var response = (await task) as HandshakeResponseMessage;
                Assert.NotNull(response);

                client.Dispose();

                await connectionHandlerTask.DefaultTimeout();
            }
        }
    }

    [Fact]
    public async Task SendingInvocatonInChunksWorks()
    {
        using (StartVerifiableLog())
        {
            var connectionHandler = HubConnectionHandlerTestUtils.GetHubConnectionHandler(typeof(HubT), loggerFactory: LoggerFactory);
            var part1 = Encoding.UTF8.GetBytes("{\"type\":1, \"invocationId\":\"1\", ");
            var part2 = Encoding.UTF8.GetBytes("\"target\": \"Echo\", \"arguments\"");
            var part3 = Encoding.UTF8.GetBytes(":[\"hello\"]}\u001e");

            using (var client = new TestClient())
            {
                client.SupportedFormats = TransferFormat.Text;

                var connectionHandlerTask = await client.ConnectAsync(connectionHandler);

                // Wait for the hub completion
                var task = client.ReadAsync();

                await client.Connection.Application.Output.WriteAsync(part1);

                Assert.False(task.IsCompleted);

                await client.Connection.Application.Output.WriteAsync(part2);

                Assert.False(task.IsCompleted);

                await client.Connection.Application.Output.WriteAsync(part3);

                var completionMessage = await task.DefaultTimeout() as CompletionMessage;
                Assert.NotNull(completionMessage);
                Assert.Equal("hello", completionMessage.Result);
                Assert.Equal("1", completionMessage.InvocationId);

                client.Dispose();

                await connectionHandlerTask.DefaultTimeout();
            }
        }
    }

    [Fact]
    public async Task SendingHandshakeRequestAndInvocationInSamePayloadParsesHandshakeAndInvocation()
    {
        using (StartVerifiableLog())
        {
            var connectionHandler = HubConnectionHandlerTestUtils.GetHubConnectionHandler(typeof(HubT), loggerFactory: LoggerFactory);
            var payload = Encoding.UTF8.GetBytes("{\"protocol\": \"json\",\"version\": 1}\u001e{\"type\":1, \"invocationId\":\"1\", \"target\": \"Echo\", \"arguments\":[\"hello\"]}\u001e");

            using (var client = new TestClient())
            {
                client.SupportedFormats = TransferFormat.Text;

                var connectionHandlerTask = await client.ConnectAsync(connectionHandler,
                                                        sendHandshakeRequestMessage: false,
                                                        expectedHandshakeResponseMessage: false);

                // Wait for the handshake response
                var task = client.ReadAsync(isHandshake: true);

                await client.Connection.Application.Output.WriteAsync(payload);

                Assert.True(task.IsCompleted);

                var response = await task as HandshakeResponseMessage;
                Assert.NotNull(response);

                var completionMessage = await client.ReadAsync() as CompletionMessage;
                Assert.NotNull(completionMessage);
                Assert.Equal("hello", completionMessage.Result);
                Assert.Equal("1", completionMessage.InvocationId);

                client.Dispose();

                await connectionHandlerTask.DefaultTimeout();
            }
        }
    }

    [Fact]
    public async Task HandshakeSuccessSendsResponseWithoutError()
    {
        using (StartVerifiableLog())
        {
            var connectionHandler = HubConnectionHandlerTestUtils.GetHubConnectionHandler(typeof(HubT), loggerFactory: LoggerFactory);

            using (var client = new TestClient())
            {
                var connectionHandlerTask = await client.ConnectAsync(connectionHandler);

                Assert.NotNull(client.HandshakeResponseMessage);
                Assert.Null(client.HandshakeResponseMessage.Error);

                client.Dispose();

                await connectionHandlerTask.DefaultTimeout();
            }
        }
    }

    [Fact]
    public async Task HubMessageOverTheMaxMessageSizeThrows()
    {
        var payload = Encoding.UTF8.GetBytes("{\"type\":1, \"invocationId\":\"1\", \"target\": \"Echo\", \"arguments\":[\"hello\"]}\u001e");
        var maximumMessageSize = payload.Length - 10;

        using (StartVerifiableLog())
        {
            var connectionHandler = HubConnectionHandlerTestUtils.GetHubConnectionHandler(typeof(HubT), LoggerFactory,
                services => services.AddSignalR().AddHubOptions<HubT>(o => o.MaximumReceiveMessageSize = maximumMessageSize));

            using (var client = new TestClient())
            {
                var connectionHandlerTask = await client.ConnectAsync(connectionHandler);

                await client.Connection.Application.Output.WriteAsync(payload);

                client.Dispose();

                await connectionHandlerTask.DefaultTimeout();
            }
        }

        var exceptionLog = TestSink.Writes.Where(w => string.Equals(w.LoggerName, "Microsoft.AspNetCore.SignalR.HubConnectionHandler") &&
            (w.Exception is InvalidDataException ide));
        Assert.Single(exceptionLog);
        Assert.Equal(exceptionLog.First().Exception.Message, $"The maximum message size of {maximumMessageSize}B was exceeded. The message size can be configured in AddHubOptions.");
    }

    [Fact]
    public async Task ChunkedHubMessageOverTheMaxMessageSizeThrows()
    {
        var payload = Encoding.UTF8.GetBytes("{\"type\":1, \"invocationId\":\"1\", \"target\": \"Echo\", \"arguments\":[\"hello\"]}\u001e");
        var maximumMessageSize = payload.Length - 10;

        using (StartVerifiableLog())
        {
            var connectionHandler = HubConnectionHandlerTestUtils.GetHubConnectionHandler(typeof(HubT), LoggerFactory,
                services => services.AddSignalR().AddHubOptions<HubT>(o => o.MaximumReceiveMessageSize = maximumMessageSize));

            using (var client = new TestClient())
            {
                var connectionHandlerTask = await client.ConnectAsync(connectionHandler);

                await client.Connection.Application.Output.WriteAsync(payload.AsMemory(0, payload.Length / 2));
                await client.Connection.Application.Output.WriteAsync(payload.AsMemory(payload.Length / 2));

                client.Dispose();

                await connectionHandlerTask.DefaultTimeout();
            }
        }

        var exceptionLog = TestSink.Writes.Where(w => string.Equals(w.LoggerName, "Microsoft.AspNetCore.SignalR.HubConnectionHandler") &&
            (w.Exception is InvalidDataException ide));
        Assert.Single(exceptionLog);
        Assert.Equal(exceptionLog.First().Exception.Message, $"The maximum message size of {maximumMessageSize}B was exceeded. The message size can be configured in AddHubOptions.");
    }

    [Fact]
    public async Task ManyHubMessagesOneOverTheMaxMessageSizeThrows()
    {
        var payload1 = Encoding.UTF8.GetBytes("{\"type\":1, \"invocationId\":\"1\", \"target\": \"Echo\", \"arguments\":[\"one\"]}\u001e");
        var payload2 = Encoding.UTF8.GetBytes("{\"type\":1, \"invocationId\":\"2\", \"target\": \"Echo\", \"arguments\":[\"two\"]}\u001e");
        var payload3 = Encoding.UTF8.GetBytes("{\"type\":1, \"invocationId\":\"3\", \"target\": \"Echo\", \"arguments\":[\"three\"]}\u001e");

        // Between the first and the second payload so we'll end up slicing with some remaining in the slice for
        // the next message
        var maximumMessageSize = payload1.Length + 1;
        using (StartVerifiableLog())
        {
            var connectionHandler = HubConnectionHandlerTestUtils.GetHubConnectionHandler(typeof(HubT), LoggerFactory,
                services => services.AddSignalR().AddHubOptions<HubT>(o => o.MaximumReceiveMessageSize = maximumMessageSize));

            using (var client = new TestClient())
            {
                var connectionHandlerTask = await client.ConnectAsync(connectionHandler);

                client.Connection.Application.Output.Write(payload1);
                client.Connection.Application.Output.Write(payload2);
                client.Connection.Application.Output.Write(payload3);
                await client.Connection.Application.Output.FlushAsync();

                // 2 invocations should be processed
                var completionMessage = await client.ReadAsync().DefaultTimeout() as CompletionMessage;
                Assert.NotNull(completionMessage);
                Assert.Equal("1", completionMessage.InvocationId);
                Assert.Equal("one", completionMessage.Result);

                completionMessage = await client.ReadAsync().DefaultTimeout() as CompletionMessage;
                Assert.NotNull(completionMessage);
                Assert.Equal("2", completionMessage.InvocationId);
                Assert.Equal("two", completionMessage.Result);

                // We never receive the 3rd message since it was over the maximum message size
                CloseMessage closeMessage = await client.ReadAsync().DefaultTimeout() as CloseMessage;
                Assert.NotNull(closeMessage);

                client.Dispose();

                await connectionHandlerTask.DefaultTimeout();
            }
        }

        var exceptionLog = TestSink.Writes.Where(w => string.Equals(w.LoggerName, "Microsoft.AspNetCore.SignalR.HubConnectionHandler") &&
            (w.Exception is InvalidDataException ide));
        Assert.Single(exceptionLog);
        Assert.Equal(exceptionLog.First().Exception.Message, $"The maximum message size of {maximumMessageSize}B was exceeded. The message size can be configured in AddHubOptions.");
    }

    [Fact]
    public async Task ManyHubMessagesUnderTheMessageSizeButConfiguredWithMax()
    {
        var payload1 = Encoding.UTF8.GetBytes("{\"type\":1, \"invocationId\":\"1\", \"target\": \"Echo\", \"arguments\":[\"one\"]}\u001e");
        var payload2 = Encoding.UTF8.GetBytes("{\"type\":1, \"invocationId\":\"2\", \"target\": \"Echo\", \"arguments\":[\"two\"]}\u001e");
        var payload3 = Encoding.UTF8.GetBytes("{\"type\":1, \"invocationId\":\"3\", \"target\": \"Echo\", \"arguments\":[\"three\"]}\u001e");

        // Bigger than all 3 messages
        var maximumMessageSize = payload3.Length + 10;

        using (StartVerifiableLog())
        {
            var connectionHandler = HubConnectionHandlerTestUtils.GetHubConnectionHandler(typeof(HubT), LoggerFactory,
                services => services.AddSignalR().AddHubOptions<HubT>(o => o.MaximumReceiveMessageSize = maximumMessageSize));

            using (var client = new TestClient())
            {
                var connectionHandlerTask = await client.ConnectAsync(connectionHandler);

                client.Connection.Application.Output.Write(payload1);
                client.Connection.Application.Output.Write(payload2);
                client.Connection.Application.Output.Write(payload3);
                await client.Connection.Application.Output.FlushAsync();

                // 2 invocations should be processed
                var completionMessage = await client.ReadAsync().DefaultTimeout() as CompletionMessage;
                Assert.NotNull(completionMessage);
                Assert.Equal("1", completionMessage.InvocationId);
                Assert.Equal("one", completionMessage.Result);

                completionMessage = await client.ReadAsync().DefaultTimeout() as CompletionMessage;
                Assert.NotNull(completionMessage);
                Assert.Equal("2", completionMessage.InvocationId);
                Assert.Equal("two", completionMessage.Result);

                completionMessage = await client.ReadAsync().DefaultTimeout() as CompletionMessage;
                Assert.NotNull(completionMessage);
                Assert.Equal("3", completionMessage.InvocationId);
                Assert.Equal("three", completionMessage.Result);

                client.Dispose();

                await connectionHandlerTask.DefaultTimeout();
            }
        }
    }

    [Fact]
    public async Task HandshakeFailureFromIncompatibleProtocolVersionSendsResponseWithError()
    {
        using (StartVerifiableLog())
        {
            var hubProtocolMock = new Mock<IHubProtocol>();
            hubProtocolMock.Setup(m => m.Name).Returns("json");
            hubProtocolMock.Setup(m => m.Version).Returns(9001);

            var connectionHandler = HubConnectionHandlerTestUtils.GetHubConnectionHandler(typeof(HubT), loggerFactory: LoggerFactory);

            using (var client = new TestClient(protocol: hubProtocolMock.Object))
            {
                var connectionHandlerTask = await client.ConnectAsync(connectionHandler);

                Assert.NotNull(client.HandshakeResponseMessage);
                Assert.Equal("The server does not support version 9001 of the 'json' protocol.", client.HandshakeResponseMessage.Error);

                client.Dispose();

                await connectionHandlerTask.DefaultTimeout();
            }
        }
    }

    [Fact]
    public async Task ConnectionClosesOnServerWithPartialHandshakeMessageAndCompletedPipe()
    {
        using (StartVerifiableLog())
        {
            var connectionHandler = HubConnectionHandlerTestUtils.GetHubConnectionHandler(typeof(HubT), loggerFactory: LoggerFactory);

            using (var client = new TestClient())
            {
                // partial handshake
                var payload = Encoding.UTF8.GetBytes("{\"protocol\": \"json\",\"ver");
                await client.Connection.Application.Output.WriteAsync(payload).DefaultTimeout();

                var connectionHandlerTask = await client.ConnectAsync(connectionHandler, sendHandshakeRequestMessage: false, expectedHandshakeResponseMessage: false);
                // Complete the pipe to 'close' the connection
                client.Connection.Application.Output.Complete();

                // This will never complete as the pipe was completed and nothing can be written to it
                var handshakeReadTask = client.ReadAsync(true);

                // Check that the connection was closed on the server
                await connectionHandlerTask.DefaultTimeout();
                Assert.False(handshakeReadTask.IsCompleted);

                client.Dispose();
            }
        }
    }

    [Fact]
    public async Task LifetimeManagerOnDisconnectedAsyncCalledIfLifetimeManagerOnConnectedAsyncThrows()
    {
        using (StartVerifiableLog())
        {
            var mockLifetimeManager = new Mock<HubLifetimeManager<Hub>>();
            mockLifetimeManager
                .Setup(m => m.OnConnectedAsync(It.IsAny<HubConnectionContext>()))
                .Throws(new InvalidOperationException("Lifetime manager OnConnectedAsync failed."));
            var mockHubActivator = new Mock<IHubActivator<Hub>>();

            var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(services =>
            {
                services.AddSingleton(mockLifetimeManager.Object);
                services.AddSingleton(mockHubActivator.Object);
            }, LoggerFactory);

            var connectionHandler = serviceProvider.GetService<HubConnectionHandler<Hub>>();

            using (var client = new TestClient())
            {
                var exception =
                    await Assert.ThrowsAsync<InvalidOperationException>(
                        async () =>
                        {
                            var connectionHandlerTask = await client.ConnectAsync(connectionHandler);
                            await connectionHandlerTask.DefaultTimeout();
                        });
                Assert.Equal("Lifetime manager OnConnectedAsync failed.", exception.Message);

                client.Dispose();

                mockLifetimeManager.Verify(m => m.OnConnectedAsync(It.IsAny<HubConnectionContext>()), Times.Once);
                mockLifetimeManager.Verify(m => m.OnDisconnectedAsync(It.IsAny<HubConnectionContext>()), Times.Once);
                // No hubs should be created since the connection is terminated
                mockHubActivator.Verify(m => m.Create(), Times.Never);
                mockHubActivator.Verify(m => m.Release(It.IsAny<Hub>()), Times.Never);
            }
        }
    }

    [Fact]
    public async Task HubOnDisconnectedAsyncCalledIfHubOnConnectedAsyncThrows()
    {
        bool ExpectedErrors(WriteContext writeContext)
        {
            return writeContext.LoggerName == "Microsoft.AspNetCore.SignalR.HubConnectionHandler" &&
                   writeContext.EventId.Name == "ErrorDispatchingHubEvent";
        }

        using (StartVerifiableLog(ExpectedErrors))
        {
            var mockLifetimeManager = new Mock<HubLifetimeManager<OnConnectedThrowsHub>>();
            var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(services =>
            {
                services.AddSingleton(mockLifetimeManager.Object);
            }, LoggerFactory);

            var connectionHandler = serviceProvider.GetService<HubConnectionHandler<OnConnectedThrowsHub>>();

            using (var client = new TestClient())
            {
                var connectionHandlerTask = await client.ConnectAsync(connectionHandler);
                client.Dispose();

                await connectionHandlerTask.DefaultTimeout();

                mockLifetimeManager.Verify(m => m.OnConnectedAsync(It.IsAny<HubConnectionContext>()), Times.Once);
                mockLifetimeManager.Verify(m => m.OnDisconnectedAsync(It.IsAny<HubConnectionContext>()), Times.Once);
            }
        }
    }

    [Fact]
    public async Task LifetimeManagerOnDisconnectedAsyncCalledIfHubOnDisconnectedAsyncThrows()
    {
        bool ExpectedErrors(WriteContext writeContext)
        {
            return writeContext.LoggerName == "Microsoft.AspNetCore.SignalR.HubConnectionHandler" &&
                   writeContext.EventId.Name == "ErrorDispatchingHubEvent";
        }

        using (StartVerifiableLog(ExpectedErrors))
        {
            var mockLifetimeManager = new Mock<HubLifetimeManager<OnDisconnectedThrowsHub>>();
            var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(services =>
            {
                services.AddSingleton(mockLifetimeManager.Object);
            }, LoggerFactory);

            var connectionHandler = serviceProvider.GetService<HubConnectionHandler<OnDisconnectedThrowsHub>>();

            using (var client = new TestClient())
            {
                var connectionHandlerTask = await client.ConnectAsync(connectionHandler);
                client.Dispose();

                var exception = await Assert.ThrowsAsync<InvalidOperationException>(async () => await connectionHandlerTask);
                Assert.Equal("Hub OnDisconnected failed.", exception.Message);

                mockLifetimeManager.Verify(m => m.OnConnectedAsync(It.IsAny<HubConnectionContext>()), Times.Once);
                mockLifetimeManager.Verify(m => m.OnDisconnectedAsync(It.IsAny<HubConnectionContext>()), Times.Once);
            }
        }
    }

    [Fact]
    public async Task HubMethodCanReturnValueFromTask()
    {
        using (StartVerifiableLog())
        {
            var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(null, LoggerFactory);

            var connectionHandler = serviceProvider.GetService<HubConnectionHandler<MethodHub>>();

            using (var client = new TestClient())
            {
                var connectionHandlerTask = await client.ConnectAsync(connectionHandler);

                var result = (await client.InvokeAsync(nameof(MethodHub.TaskValueMethod)).DefaultTimeout()).Result;

                // json serializer makes this a long
                Assert.Equal(42L, result);

                // kill the connection
                client.Dispose();

                await connectionHandlerTask.DefaultTimeout();
            }
        }
    }

    [Fact]
    public async Task HubMethodCanReturnValueFromValueTask()
    {
        using (StartVerifiableLog())
        {
            var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(null, LoggerFactory);

            var connectionHandler = serviceProvider.GetService<HubConnectionHandler<MethodHub>>();

            using (var client = new TestClient())
            {
                var connectionHandlerTask = await client.ConnectAsync(connectionHandler);

                var result = (await client.InvokeAsync(nameof(MethodHub.ValueTaskValueMethod)).DefaultTimeout()).Result;

                // json serializer makes this a long
                Assert.Equal(43L, result);

                // kill the connection
                client.Dispose();

                await connectionHandlerTask.DefaultTimeout();
            }
        }
    }

    [Fact]
    public async Task HubMethodCanReturnValueTask()
    {
        using (StartVerifiableLog())
        {
            var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(null, LoggerFactory);

            var connectionHandler = serviceProvider.GetService<HubConnectionHandler<MethodHub>>();

            using (var client = new TestClient())
            {
                var connectionHandlerTask = await client.ConnectAsync(connectionHandler);

                var result = (await client.InvokeAsync(nameof(MethodHub.ValueTaskMethod)).DefaultTimeout()).Result;

                Assert.Null(result);

                // kill the connection
                client.Dispose();

                await connectionHandlerTask.DefaultTimeout();
            }
        }
    }

    [Theory]
    [MemberData(nameof(HubTypes))]
    public async Task HubMethodsAreCaseInsensitive(Type hubType)
    {
        using (StartVerifiableLog())
        {
            var connectionHandler = HubConnectionHandlerTestUtils.GetHubConnectionHandler(hubType, loggerFactory: LoggerFactory);

            using (var client = new TestClient())
            {
                var connectionHandlerTask = (Task)await client.ConnectAsync(connectionHandler);

                var result = (await client.InvokeAsync("echo", "hello").DefaultTimeout()).Result;

                Assert.Equal("hello", result);

                // kill the connection
                client.Dispose();

                await connectionHandlerTask.DefaultTimeout();
            }
        }
    }

    [Theory]
    [InlineData(nameof(MethodHub.MethodThatThrows), true)]
    [InlineData(nameof(MethodHub.MethodThatYieldsFailedTask), false)]
    public async Task HubMethodCanThrowOrYieldFailedTask(string methodName, bool detailedErrors)
    {
        var hasErrorLog = false;
        bool ExpectedErrors(WriteContext writeContext)
        {
            var expected = writeContext.LoggerName == "Microsoft.AspNetCore.SignalR.Internal.DefaultHubDispatcher" &&
                   writeContext.EventId.Name == "FailedInvokingHubMethod";
            if (expected)
            {
                hasErrorLog = true;
                return true;
            }
            return false;
        }

        using (StartVerifiableLog(ExpectedErrors))
        {
            var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(builder =>
            {
                builder.AddSignalR(options =>
                {
                    options.EnableDetailedErrors = detailedErrors;
                });
            }, LoggerFactory);

            var connectionHandler = serviceProvider.GetService<HubConnectionHandler<MethodHub>>();

            using (var client = new TestClient())
            {
                var connectionHandlerTask = await client.ConnectAsync(connectionHandler);

                var message = await client.InvokeAsync(methodName).DefaultTimeout();

                if (detailedErrors)
                {
                    Assert.Equal($"An unexpected error occurred invoking '{methodName}' on the server. InvalidOperationException: BOOM!", message.Error);
                }
                else
                {
                    Assert.Equal($"An unexpected error occurred invoking '{methodName}' on the server.", message.Error);
                }

                // kill the connection
                client.Dispose();

                await connectionHandlerTask.DefaultTimeout();
            }
        }

        Assert.True(hasErrorLog);
    }

    [Fact]
    public async Task HubMethodListeningToConnectionAbortedClosesOnConnectionContextAbort()
    {
        using (StartVerifiableLog())
        {
            var connectionHandler = HubConnectionHandlerTestUtils.GetHubConnectionHandler(typeof(MethodHub), loggerFactory: LoggerFactory);

            using (var client = new TestClient())
            {
                var connectionHandlerTask = await client.ConnectAsync(connectionHandler);

                await client.SendInvocationAsync(nameof(MethodHub.BlockingMethod)).DefaultTimeout();

                client.Connection.Abort();

                var closeMessage = Assert.IsType<CloseMessage>(await client.ReadAsync().DefaultTimeout());
                Assert.False(closeMessage.AllowReconnect);

                // If this completes then the server has completed the connection
                await connectionHandlerTask.DefaultTimeout();

                // Nothing written to connection because it was closed
                Assert.Null(client.TryRead());
            }
        }
    }

    [Fact]
    public async Task DetailedExceptionEvenWhenNotExplicitlySet()
    {
        bool ExpectedErrors(WriteContext writeContext)
        {
            return writeContext.LoggerName == "Microsoft.AspNetCore.SignalR.Internal.DefaultHubDispatcher" &&
                   writeContext.EventId.Name == "FailedInvokingHubMethod";
        }

        using (StartVerifiableLog(ExpectedErrors))
        {
            var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(null, LoggerFactory);

            var methodName = nameof(MethodHub.ThrowHubException);

            var connectionHandler = serviceProvider.GetService<HubConnectionHandler<MethodHub>>();

            using (var client = new TestClient())
            {
                var connectionHandlerTask = await client.ConnectAsync(connectionHandler);

                var message = await client.InvokeAsync(methodName).DefaultTimeout();

                Assert.Equal($"An unexpected error occurred invoking '{methodName}' on the server. HubException: This is a hub exception", message.Error);

                // kill the connection
                client.Dispose();

                await connectionHandlerTask.DefaultTimeout();
            }
        }
    }

    [Fact]
    public async Task HubMethodDoesNotSendResultWhenInvocationIsNonBlocking()
    {
        using (StartVerifiableLog())
        {
            var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(null, LoggerFactory);

            var connectionHandler = serviceProvider.GetService<HubConnectionHandler<MethodHub>>();

            using (var client = new TestClient())
            {
                var connectionHandlerTask = await client.ConnectAsync(connectionHandler);

                await client.SendInvocationAsync(nameof(MethodHub.ValueMethod), nonBlocking: true).DefaultTimeout();

                // kill the connection
                client.Dispose();

                var message = Assert.IsType<CloseMessage>(client.TryRead());
                Assert.True(message.AllowReconnect);

                // Ensure the client channel is empty
                Assert.Null(client.TryRead());

                await connectionHandlerTask.DefaultTimeout();
            }
        }
    }

    [Fact]
    public async Task HubMethodCanBeVoid()
    {
        using (StartVerifiableLog())
        {
            var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(null, LoggerFactory);

            var connectionHandler = serviceProvider.GetService<HubConnectionHandler<MethodHub>>();

            using (var client = new TestClient())
            {
                var connectionHandlerTask = await client.ConnectAsync(connectionHandler);

                var result = (await client.InvokeAsync(nameof(MethodHub.VoidMethod)).DefaultTimeout()).Result;

                Assert.Null(result);

                // kill the connection
                client.Dispose();

                await connectionHandlerTask.DefaultTimeout();
            }
        }
    }

    [Fact]
    public async Task HubMethodCanBeRenamedWithAttribute()
    {
        using (StartVerifiableLog())
        {
            var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(null, LoggerFactory);

            var connectionHandler = serviceProvider.GetService<HubConnectionHandler<MethodHub>>();

            using (var client = new TestClient())
            {
                var connectionHandlerTask = await client.ConnectAsync(connectionHandler);

                var result = (await client.InvokeAsync("RenamedMethod").DefaultTimeout()).Result;

                // json serializer makes this a long
                Assert.Equal(43L, result);

                // kill the connection
                client.Dispose();

                await connectionHandlerTask.DefaultTimeout();
            }
        }
    }

    [Fact]
    public async Task HubMethodNameAttributeIsInherited()
    {
        using (StartVerifiableLog())
        {
            var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(null, LoggerFactory);

            var connectionHandler = serviceProvider.GetService<HubConnectionHandler<InheritedHub>>();

            using (var client = new TestClient())
            {
                var connectionHandlerTask = await client.ConnectAsync(connectionHandler);

                var result = (await client.InvokeAsync("RenamedVirtualMethod").DefaultTimeout()).Result;

                // json serializer makes this a long
                Assert.Equal(34L, result);

                // kill the connection
                client.Dispose();

                await connectionHandlerTask.DefaultTimeout();
            }
        }
    }

    [Theory]
    [InlineData(nameof(MethodHub.VoidMethod))]
    [InlineData(nameof(MethodHub.MethodThatThrows))]
    [InlineData(nameof(MethodHub.ValueMethod))]
    public async Task NonBlockingInvocationDoesNotSendCompletion(string methodName)
    {
        bool ExpectedErrors(WriteContext writeContext)
        {
            return methodName == nameof(MethodHub.MethodThatThrows) && writeContext.LoggerName == "Microsoft.AspNetCore.SignalR.Internal.DefaultHubDispatcher" &&
                   writeContext.EventId.Name == "FailedInvokingHubMethod";
        }

        using (StartVerifiableLog(ExpectedErrors))
        {
            var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(null, LoggerFactory);

            var connectionHandler = serviceProvider.GetService<HubConnectionHandler<MethodHub>>();

            using (var client = new TestClient())
            {
                var connectionHandlerTask = await client.ConnectAsync(connectionHandler);

                // This invocation should be completely synchronous
                await client.SendInvocationAsync(methodName, nonBlocking: true).DefaultTimeout();

                // kill the connection
                client.Dispose();

                // only thing written should be close message
                var closeMessage = await client.ReadAsync().DefaultTimeout();
                Assert.IsType<CloseMessage>(closeMessage);

                await connectionHandlerTask.DefaultTimeout();
            }
        }
    }

    [Fact]
    public async Task HubMethodWithMultiParam()
    {
        using (StartVerifiableLog())
        {
            var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(null, LoggerFactory);

            var connectionHandler = serviceProvider.GetService<HubConnectionHandler<MethodHub>>();

            using (var client = new TestClient())
            {
                var connectionHandlerTask = await client.ConnectAsync(connectionHandler);

                var result = (await client.InvokeAsync(nameof(MethodHub.ConcatString), (byte)32, 42, 'm', "string").DefaultTimeout()).Result;

                Assert.Equal("32, 42, m, string", result);

                // kill the connection
                client.Dispose();

                await connectionHandlerTask.DefaultTimeout();
            }
        }
    }

    [Fact]
    public async Task CanCallInheritedHubMethodFromInheritingHub()
    {
        using (StartVerifiableLog())
        {
            var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(null, LoggerFactory);

            var connectionHandler = serviceProvider.GetService<HubConnectionHandler<InheritedHub>>();

            using (var client = new TestClient())
            {
                var connectionHandlerTask = await client.ConnectAsync(connectionHandler);

                var result = (await client.InvokeAsync(nameof(InheritedHub.BaseMethod), "string").DefaultTimeout()).Result;

                Assert.Equal("string", result);

                // kill the connection
                client.Dispose();

                await connectionHandlerTask.DefaultTimeout();
            }
        }
    }

    [Fact]
    public async Task CanCallOverridenVirtualHubMethod()
    {
        using (StartVerifiableLog())
        {
            var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(null, LoggerFactory);

            var connectionHandler = serviceProvider.GetService<HubConnectionHandler<InheritedHub>>();

            using (var client = new TestClient())
            {
                var connectionHandlerTask = await client.ConnectAsync(connectionHandler);

                var result = (await client.InvokeAsync(nameof(InheritedHub.VirtualMethod), 10).DefaultTimeout()).Result;

                Assert.Equal(0L, result);

                // kill the connection
                client.Dispose();

                await connectionHandlerTask.DefaultTimeout();
            }
        }
    }

    [Fact]
    public async Task CannotCallOverriddenBaseHubMethod()
    {
        using (StartVerifiableLog())
        {
            var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(null, LoggerFactory);

            var connectionHandler = serviceProvider.GetService<HubConnectionHandler<MethodHub>>();

            using (var client = new TestClient())
            {
                var connectionHandlerTask = await client.ConnectAsync(connectionHandler);

                var result = await client.InvokeAsync(nameof(MethodHub.OnDisconnectedAsync)).DefaultTimeout();

                Assert.Equal("Failed to invoke 'OnDisconnectedAsync' due to an error on the server. HubException: Method does not exist.", result.Error);

                // kill the connection
                client.Dispose();

                await connectionHandlerTask.DefaultTimeout();
            }
        }
    }

    [Fact]
    public void HubsCannotHaveOverloadedMethods()
    {
        using (StartVerifiableLog())
        {
            var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(null, LoggerFactory);

            try
            {
                var connectionHandler = serviceProvider.GetService<HubConnectionHandler<InvalidHub>>();
                Assert.True(false);
            }
            catch (NotSupportedException ex)
            {
                Assert.Equal("Duplicate definitions of 'OverloadedMethod'. Overloading is not supported.", ex.Message);
            }
        }
    }

    [Fact]
    public async Task CannotCallStaticHubMethods()
    {
        using (StartVerifiableLog())
        {
            var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(null, LoggerFactory);

            var connectionHandler = serviceProvider.GetService<HubConnectionHandler<MethodHub>>();

            using (var client = new TestClient())
            {
                var connectionHandlerTask = await client.ConnectAsync(connectionHandler);

                var result = await client.InvokeAsync(nameof(MethodHub.StaticMethod)).DefaultTimeout();

                Assert.Equal("Failed to invoke 'StaticMethod' due to an error on the server. HubException: Method does not exist.", result.Error);

                // kill the connection
                client.Dispose();

                await connectionHandlerTask.DefaultTimeout();
            }
        }
    }

    [Fact]
    public async Task CannotCallObjectMethodsOnHub()
    {
        using (StartVerifiableLog())
        {
            var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(null, LoggerFactory);

            var connectionHandler = serviceProvider.GetService<HubConnectionHandler<MethodHub>>();

            using (var client = new TestClient())
            {
                var connectionHandlerTask = await client.ConnectAsync(connectionHandler);

                var result = await client.InvokeAsync(nameof(MethodHub.ToString)).DefaultTimeout();
                Assert.Equal("Failed to invoke 'ToString' due to an error on the server. HubException: Method does not exist.", result.Error);

                result = await client.InvokeAsync(nameof(MethodHub.GetHashCode)).DefaultTimeout();
                Assert.Equal("Failed to invoke 'GetHashCode' due to an error on the server. HubException: Method does not exist.", result.Error);

                result = await client.InvokeAsync(nameof(MethodHub.Equals)).DefaultTimeout();
                Assert.Equal("Failed to invoke 'Equals' due to an error on the server. HubException: Method does not exist.", result.Error);

                result = await client.InvokeAsync(nameof(MethodHub.ReferenceEquals)).DefaultTimeout();
                Assert.Equal("Failed to invoke 'ReferenceEquals' due to an error on the server. HubException: Method does not exist.", result.Error);

                // kill the connection
                client.Dispose();

                await connectionHandlerTask.DefaultTimeout();
            }
        }
    }

    [Fact]
    public async Task CannotCallDisposeMethodOnHub()
    {
        using (StartVerifiableLog())
        {
            var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(null, LoggerFactory);

            var connectionHandler = serviceProvider.GetService<HubConnectionHandler<MethodHub>>();

            using (var client = new TestClient())
            {
                var connectionHandlerTask = await client.ConnectAsync(connectionHandler);

                var result = await client.InvokeAsync(nameof(MethodHub.Dispose)).DefaultTimeout();

                Assert.Equal("Failed to invoke 'Dispose' due to an error on the server. HubException: Method does not exist.", result.Error);

                // kill the connection
                client.Dispose();

                await connectionHandlerTask.DefaultTimeout();
            }
        }
    }

    [Fact]
    public void CannotHaveGenericMethodOnHub()
    {
        using (StartVerifiableLog())
        {
            var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(null, LoggerFactory);

            var exception = Assert.Throws<NotSupportedException>(() => serviceProvider.GetService<HubConnectionHandler<GenericMethodHub>>());

            Assert.Equal("Method 'GenericMethod' is a generic method which is not supported on a Hub.", exception.Message);
        }
    }

    [Theory]
    [MemberData(nameof(HubTypes))]
    public async Task BroadcastHubMethodSendsToAllClients(Type hubType)
    {
        using (StartVerifiableLog())
        {
            var connectionHandler = HubConnectionHandlerTestUtils.GetHubConnectionHandler(hubType, loggerFactory: LoggerFactory);

            using (var firstClient = new TestClient())
            using (var secondClient = new TestClient())
            {
                var firstConnectionHandlerTask = await firstClient.ConnectAsync(connectionHandler);
                var secondConnectionHandlerTask = await secondClient.ConnectAsync(connectionHandler);

                await Task.WhenAll(firstClient.Connected, secondClient.Connected).DefaultTimeout();

                await firstClient.SendInvocationAsync(nameof(MethodHub.BroadcastMethod), "test").DefaultTimeout();

                foreach (var result in await Task.WhenAll(
                    firstClient.ReadAsync(),
                    secondClient.ReadAsync()).DefaultTimeout())
                {
                    var invocation = Assert.IsType<InvocationMessage>(result);
                    Assert.Equal("Broadcast", invocation.Target);
                    Assert.Single(invocation.Arguments);
                    Assert.Equal("test", invocation.Arguments[0]);
                }

                // kill the connections
                firstClient.Dispose();
                secondClient.Dispose();

                await Task.WhenAll(firstConnectionHandlerTask, secondConnectionHandlerTask).DefaultTimeout();
            }
        }
    }

    [Fact]
    public async Task SendArraySendsArrayToAllClients()
    {
        using (StartVerifiableLog())
        {
            var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(null, LoggerFactory);

            var connectionHandler = serviceProvider.GetService<HubConnectionHandler<MethodHub>>();

            using (var firstClient = new TestClient())
            using (var secondClient = new TestClient())
            {
                var firstConnectionHandlerTask = await firstClient.ConnectAsync(connectionHandler);
                var secondConnectionHandlerTask = await secondClient.ConnectAsync(connectionHandler);

                await Task.WhenAll(firstClient.Connected, secondClient.Connected).DefaultTimeout();

                await firstClient.SendInvocationAsync(nameof(MethodHub.SendArray)).DefaultTimeout();

                foreach (var result in await Task.WhenAll(
                    firstClient.ReadAsync(),
                    secondClient.ReadAsync()).DefaultTimeout())
                {
                    var invocation = Assert.IsType<InvocationMessage>(result);
                    Assert.Equal("Array", invocation.Target);
                    Assert.Single(invocation.Arguments);
                    var values = ((JArray)invocation.Arguments[0]).Select(t => t.Value<int>()).ToArray();
                    Assert.Equal(new[] { 1, 2, 3 }, values);
                }

                // kill the connections
                firstClient.Dispose();
                secondClient.Dispose();

                await Task.WhenAll(firstConnectionHandlerTask, secondConnectionHandlerTask).DefaultTimeout();
            }
        }
    }

    [Theory]
    [MemberData(nameof(HubTypes))]
    public async Task SendToOthers(Type hubType)
    {
        using (StartVerifiableLog())
        {
            var connectionHandler = HubConnectionHandlerTestUtils.GetHubConnectionHandler(hubType, loggerFactory: LoggerFactory);

            using (var firstClient = new TestClient())
            using (var secondClient = new TestClient())
            {
                var firstConnectionHandlerTask = await firstClient.ConnectAsync(connectionHandler);
                var secondConnectionHandlerTask = await secondClient.ConnectAsync(connectionHandler);

                await Task.WhenAll(firstClient.Connected, secondClient.Connected).DefaultTimeout();

                await firstClient.SendInvocationAsync("SendToOthers", "To others").DefaultTimeout();

                var secondClientResult = await secondClient.ReadAsync().DefaultTimeout();
                var invocation = Assert.IsType<InvocationMessage>(secondClientResult);
                Assert.Equal("Send", invocation.Target);
                Assert.Equal("To others", invocation.Arguments[0]);

                var firstClientResult = await firstClient.ReadAsync().DefaultTimeout();
                var completion = Assert.IsType<CompletionMessage>(firstClientResult);

                await secondClient.SendInvocationAsync("BroadcastMethod", "To everyone").DefaultTimeout();
                firstClientResult = await firstClient.ReadAsync().DefaultTimeout();
                invocation = Assert.IsType<InvocationMessage>(firstClientResult);
                Assert.Equal("Broadcast", invocation.Target);
                Assert.Equal("To everyone", invocation.Arguments[0]);

                // kill the connections
                firstClient.Dispose();
                secondClient.Dispose();

                await Task.WhenAll(firstConnectionHandlerTask, secondConnectionHandlerTask).DefaultTimeout();
            }
        }
    }

    [Theory]
    [MemberData(nameof(HubTypes))]
    public async Task SendToCaller(Type hubType)
    {
        using (StartVerifiableLog())
        {
            var connectionHandler = HubConnectionHandlerTestUtils.GetHubConnectionHandler(hubType, loggerFactory: LoggerFactory);

            using (var firstClient = new TestClient())
            using (var secondClient = new TestClient())
            {
                var firstConnectionHandlerTask = await firstClient.ConnectAsync(connectionHandler);
                var secondConnectionHandlerTask = await secondClient.ConnectAsync(connectionHandler);

                await Task.WhenAll(firstClient.Connected, secondClient.Connected).DefaultTimeout();

                await firstClient.SendInvocationAsync("SendToCaller", "To caller").DefaultTimeout();

                var firstClientResult = await firstClient.ReadAsync().DefaultTimeout();
                var invocation = Assert.IsType<InvocationMessage>(firstClientResult);
                Assert.Equal("Send", invocation.Target);
                Assert.Equal("To caller", invocation.Arguments[0]);

                await firstClient.SendInvocationAsync("BroadcastMethod", "To everyone").DefaultTimeout();
                var secondClientResult = await secondClient.ReadAsync().DefaultTimeout();
                invocation = Assert.IsType<InvocationMessage>(secondClientResult);
                Assert.Equal("Broadcast", invocation.Target);
                Assert.Equal("To everyone", invocation.Arguments[0]);

                // kill the connections
                firstClient.Dispose();
                secondClient.Dispose();

                await Task.WhenAll(firstConnectionHandlerTask, secondConnectionHandlerTask).DefaultTimeout();
            }
        }
    }

    [Fact]
    public async Task FailsToInitializeInvalidTypedHub()
    {
        bool ExpectedErrors(WriteContext writeContext)
        {
            return writeContext.LoggerName == "Microsoft.AspNetCore.SignalR.HubConnectionHandler" &&
                   writeContext.EventId.Name == "ErrorDispatchingHubEvent";
        }

        using (StartVerifiableLog(ExpectedErrors))
        {
            var connectionHandler = HubConnectionHandlerTestUtils.GetHubConnectionHandler(typeof(SimpleVoidReturningTypedHub), loggerFactory: LoggerFactory);

            using (var firstClient = new TestClient())
            {
                // ConnectAsync returns a Task<Task> and it's the INNER Task that will be faulted.
                var connectionTask = await firstClient.ConnectAsync(connectionHandler);

                // We should get a close frame now
                var close = Assert.IsType<CloseMessage>(await firstClient.ReadAsync());
                Assert.Equal("Connection closed with an error.", close.Error);
            }
        }
    }

    [Theory]
    [MemberData(nameof(HubTypes))]
    public async Task SendToAllExcept(Type hubType)
    {
        using (StartVerifiableLog())
        {
            var connectionHandler = HubConnectionHandlerTestUtils.GetHubConnectionHandler(hubType, loggerFactory: LoggerFactory);

            using (var firstClient = new TestClient())
            using (var secondClient = new TestClient())
            using (var thirdClient = new TestClient())
            {
                var firstConnectionHandlerTask = await firstClient.ConnectAsync(connectionHandler);
                var secondConnectionHandlerTask = await secondClient.ConnectAsync(connectionHandler);
                var thirdConnectionHandlerTask = await thirdClient.ConnectAsync(connectionHandler);

                await Task.WhenAll(firstClient.Connected, secondClient.Connected, thirdClient.Connected).DefaultTimeout();

                var excludeSecondClientId = new HashSet<string>();
                excludeSecondClientId.Add(secondClient.Connection.ConnectionId);
                var excludeThirdClientId = new HashSet<string>();
                excludeThirdClientId.Add(thirdClient.Connection.ConnectionId);

                await firstClient.SendInvocationAsync("SendToAllExcept", "To second", excludeThirdClientId).DefaultTimeout();
                await firstClient.SendInvocationAsync("SendToAllExcept", "To third", excludeSecondClientId).DefaultTimeout();

                var secondClientResult = await secondClient.ReadAsync().DefaultTimeout();
                var invocation = Assert.IsType<InvocationMessage>(secondClientResult);
                Assert.Equal("Send", invocation.Target);
                Assert.Equal("To second", invocation.Arguments[0]);

                var thirdClientResult = await thirdClient.ReadAsync().DefaultTimeout();
                invocation = Assert.IsType<InvocationMessage>(thirdClientResult);
                Assert.Equal("Send", invocation.Target);
                Assert.Equal("To third", invocation.Arguments[0]);

                // kill the connections
                firstClient.Dispose();
                secondClient.Dispose();
                thirdClient.Dispose();

                await Task.WhenAll(firstConnectionHandlerTask, secondConnectionHandlerTask, thirdConnectionHandlerTask).DefaultTimeout();
            }
        }
    }

    [Theory]
    [MemberData(nameof(HubTypes))]
    public async Task SendToMultipleClients(Type hubType)
    {
        using (StartVerifiableLog())
        {
            var connectionHandler = HubConnectionHandlerTestUtils.GetHubConnectionHandler(hubType, loggerFactory: LoggerFactory);

            using (var firstClient = new TestClient())
            using (var secondClient = new TestClient())
            using (var thirdClient = new TestClient())
            {
                var firstConnectionHandlerTask = await firstClient.ConnectAsync(connectionHandler);
                var secondConnectionHandlerTask = await secondClient.ConnectAsync(connectionHandler);
                var thirdConnectionHandlerTask = await thirdClient.ConnectAsync(connectionHandler);

                await Task.WhenAll(firstClient.Connected, secondClient.Connected, thirdClient.Connected).DefaultTimeout();

                var secondAndThirdClients = new HashSet<string> {secondClient.Connection.ConnectionId,
                    thirdClient.Connection.ConnectionId };

                await firstClient.SendInvocationAsync("SendToMultipleClients", "Second and Third", secondAndThirdClients).DefaultTimeout();

                var secondClientResult = await secondClient.ReadAsync().DefaultTimeout();
                var invocation = Assert.IsType<InvocationMessage>(secondClientResult);
                Assert.Equal("Send", invocation.Target);
                Assert.Equal("Second and Third", invocation.Arguments[0]);

                var thirdClientResult = await thirdClient.ReadAsync().DefaultTimeout();
                invocation = Assert.IsType<InvocationMessage>(thirdClientResult);
                Assert.Equal("Send", invocation.Target);
                Assert.Equal("Second and Third", invocation.Arguments[0]);

                // Check that first client only got the completion message
                var hubMessage = await firstClient.ReadAsync().DefaultTimeout();
                Assert.IsType<CompletionMessage>(hubMessage);
                Assert.Null(firstClient.TryRead());

                // kill the connections
                firstClient.Dispose();
                secondClient.Dispose();
                thirdClient.Dispose();

                await Task.WhenAll(firstConnectionHandlerTask, secondConnectionHandlerTask, thirdConnectionHandlerTask).DefaultTimeout();
            }
        }
    }

    [Theory]
    [MemberData(nameof(HubTypes))]
    public async Task SendToMultipleUsers(Type hubType)
    {
        using (StartVerifiableLog())
        {
            var connectionHandler = HubConnectionHandlerTestUtils.GetHubConnectionHandler(hubType, loggerFactory: LoggerFactory);

            using (var firstClient = new TestClient(userIdentifier: "userA"))
            using (var secondClient = new TestClient(userIdentifier: "userB"))
            using (var thirdClient = new TestClient(userIdentifier: "userC"))
            {
                var firstConnectionHandlerTask = await firstClient.ConnectAsync(connectionHandler);
                var secondConnectionHandlerTask = await secondClient.ConnectAsync(connectionHandler);
                var thirdConnectionHandlerTask = await thirdClient.ConnectAsync(connectionHandler);

                await Task.WhenAll(firstClient.Connected, secondClient.Connected, thirdClient.Connected).DefaultTimeout();

                await firstClient.SendInvocationAsync(nameof(MethodHub.SendToMultipleUsers), new[] { "userB", "userC" }, "Second and Third").DefaultTimeout();

                var secondClientResult = await secondClient.ReadAsync().DefaultTimeout();
                var invocation = Assert.IsType<InvocationMessage>(secondClientResult);
                Assert.Equal("Send", invocation.Target);
                Assert.Equal("Second and Third", invocation.Arguments[0]);

                var thirdClientResult = await thirdClient.ReadAsync().DefaultTimeout();
                invocation = Assert.IsType<InvocationMessage>(thirdClientResult);
                Assert.Equal("Send", invocation.Target);
                Assert.Equal("Second and Third", invocation.Arguments[0]);

                // Check that first client only got the completion message
                var hubMessage = await firstClient.ReadAsync().DefaultTimeout();
                Assert.IsType<CompletionMessage>(hubMessage);
                Assert.Null(firstClient.TryRead());

                // kill the connections
                firstClient.Dispose();
                secondClient.Dispose();
                thirdClient.Dispose();

                await Task.WhenAll(firstConnectionHandlerTask, secondConnectionHandlerTask, thirdConnectionHandlerTask).DefaultTimeout();
            }
        }
    }

    [Theory]
    [MemberData(nameof(HubTypes))]
    public async Task HubsCanAddAndSendToGroup(Type hubType)
    {
        using (StartVerifiableLog())
        {
            var connectionHandler = HubConnectionHandlerTestUtils.GetHubConnectionHandler(hubType, loggerFactory: LoggerFactory);

            using (var firstClient = new TestClient())
            using (var secondClient = new TestClient())
            {
                var firstConnectionHandlerTask = await firstClient.ConnectAsync(connectionHandler);
                var secondConnectionHandlerTask = await secondClient.ConnectAsync(connectionHandler);

                await Task.WhenAll(firstClient.Connected, secondClient.Connected).DefaultTimeout();

                var result = (await firstClient.InvokeAsync("GroupSendMethod", "testGroup", "test").DefaultTimeout()).Result;

                // check that 'firstConnection' hasn't received the group send
                Assert.Null(firstClient.TryRead());

                // check that 'secondConnection' hasn't received the group send
                Assert.Null(secondClient.TryRead());

                result = (await secondClient.InvokeAsync(nameof(MethodHub.GroupAddMethod), "testGroup").DefaultTimeout()).Result;

                await firstClient.SendInvocationAsync(nameof(MethodHub.GroupSendMethod), "testGroup", "test").DefaultTimeout();

                // check that 'secondConnection' has received the group send
                var hubMessage = await secondClient.ReadAsync().DefaultTimeout();
                var invocation = Assert.IsType<InvocationMessage>(hubMessage);
                Assert.Equal("Send", invocation.Target);
                Assert.Single(invocation.Arguments);
                Assert.Equal("test", invocation.Arguments[0]);

                // kill the connections
                firstClient.Dispose();
                secondClient.Dispose();

                await Task.WhenAll(firstConnectionHandlerTask, secondConnectionHandlerTask).DefaultTimeout();
            }
        }
    }

    [Theory]
    [MemberData(nameof(HubTypes))]
    public async Task SendToGroupExcept(Type hubType)
    {
        using (StartVerifiableLog())
        {
            var connectionHandler = HubConnectionHandlerTestUtils.GetHubConnectionHandler(hubType, loggerFactory: LoggerFactory);

            using (var firstClient = new TestClient())
            using (var secondClient = new TestClient())
            {
                var firstConnectionHandlerTask = await firstClient.ConnectAsync(connectionHandler);
                var secondConnectionHandlerTask = await secondClient.ConnectAsync(connectionHandler);

                await Task.WhenAll(firstClient.Connected, secondClient.Connected).DefaultTimeout();

                var result = (await firstClient.InvokeAsync("GroupSendMethod", "testGroup", "test").DefaultTimeout()).Result;

                // check that 'firstConnection' hasn't received the group send
                Assert.Null(firstClient.TryRead());

                // check that 'secondConnection' hasn't received the group send
                Assert.Null(secondClient.TryRead());

                await firstClient.InvokeAsync(nameof(MethodHub.GroupAddMethod), "testGroup").DefaultTimeout();
                await secondClient.InvokeAsync(nameof(MethodHub.GroupAddMethod), "testGroup").DefaultTimeout();

                var excludedConnectionIds = new List<string> { firstClient.Connection.ConnectionId };

                await firstClient.SendInvocationAsync("GroupExceptSendMethod", "testGroup", "test", excludedConnectionIds).DefaultTimeout();

                // check that 'secondConnection' has received the group send
                var hubMessage = await secondClient.ReadAsync().DefaultTimeout();
                var invocation = Assert.IsType<InvocationMessage>(hubMessage);
                Assert.Equal("Send", invocation.Target);
                Assert.Single(invocation.Arguments);
                Assert.Equal("test", invocation.Arguments[0]);

                // Check that first client only got the completion message
                hubMessage = await firstClient.ReadAsync().DefaultTimeout();
                Assert.IsType<CompletionMessage>(hubMessage);

                Assert.Null(firstClient.TryRead());

                // kill the connections
                firstClient.Dispose();
                secondClient.Dispose();

                await Task.WhenAll(firstConnectionHandlerTask, secondConnectionHandlerTask).DefaultTimeout();
            }
        }
    }

    [Theory]
    [MemberData(nameof(HubTypes))]
    public async Task SendToOthersInGroup(Type hubType)
    {
        using (StartVerifiableLog())
        {
            var connectionHandler = HubConnectionHandlerTestUtils.GetHubConnectionHandler(hubType, loggerFactory: LoggerFactory);

            using (var firstClient = new TestClient())
            using (var secondClient = new TestClient())
            {
                var firstConnectionHandlerTask = await firstClient.ConnectAsync(connectionHandler);
                var secondConnectionHandlerTask = await secondClient.ConnectAsync(connectionHandler);

                await Task.WhenAll(firstClient.Connected, secondClient.Connected).DefaultTimeout();

                var result = (await firstClient.InvokeAsync("GroupSendMethod", "testGroup", "test").DefaultTimeout()).Result;

                // check that 'firstConnection' hasn't received the group send
                Assert.Null(firstClient.TryRead());

                // check that 'secondConnection' hasn't received the group send
                Assert.Null(secondClient.TryRead());

                await firstClient.InvokeAsync(nameof(MethodHub.GroupAddMethod), "testGroup").DefaultTimeout();
                await secondClient.InvokeAsync(nameof(MethodHub.GroupAddMethod), "testGroup").DefaultTimeout();

                await firstClient.SendInvocationAsync("SendToOthersInGroup", "testGroup", "test").DefaultTimeout();

                // check that 'secondConnection' has received the group send
                var hubMessage = await secondClient.ReadAsync().DefaultTimeout();
                var invocation = Assert.IsType<InvocationMessage>(hubMessage);
                Assert.Equal("Send", invocation.Target);
                Assert.Single(invocation.Arguments);
                Assert.Equal("test", invocation.Arguments[0]);

                // Check that first client only got the completion message
                hubMessage = await firstClient.ReadAsync().DefaultTimeout();
                Assert.IsType<CompletionMessage>(hubMessage);

                Assert.Null(firstClient.TryRead());

                // kill the connections
                firstClient.Dispose();
                secondClient.Dispose();

                await Task.WhenAll(firstConnectionHandlerTask, secondConnectionHandlerTask).DefaultTimeout();
            }
        }
    }

    [Theory]
    [MemberData(nameof(HubTypes))]
    public async Task InvokeMultipleGroups(Type hubType)
    {
        using (StartVerifiableLog())
        {
            var connectionHandler = HubConnectionHandlerTestUtils.GetHubConnectionHandler(hubType, loggerFactory: LoggerFactory);

            using (var firstClient = new TestClient())
            using (var secondClient = new TestClient())
            {
                var firstConnectionHandlerTask = await firstClient.ConnectAsync(connectionHandler);
                var secondConnectionHandlerTask = await secondClient.ConnectAsync(connectionHandler);

                await Task.WhenAll(firstClient.Connected, secondClient.Connected).DefaultTimeout();

                await secondClient.InvokeAsync(nameof(MethodHub.GroupAddMethod), "GroupA").DefaultTimeout();
                await firstClient.InvokeAsync(nameof(MethodHub.GroupAddMethod), "GroupB").DefaultTimeout();

                var groupNames = new List<string> { "GroupA", "GroupB" };
                await firstClient.SendInvocationAsync(nameof(MethodHub.SendToMultipleGroups), "test", groupNames).DefaultTimeout();

                var hubMessage = await secondClient.ReadAsync().DefaultTimeout();
                var invocation = Assert.IsType<InvocationMessage>(hubMessage);
                Assert.Equal("Send", invocation.Target);
                Assert.Single(invocation.Arguments);
                Assert.Equal("test", invocation.Arguments[0]);

                hubMessage = await firstClient.ReadAsync().DefaultTimeout();
                invocation = Assert.IsType<InvocationMessage>(hubMessage);
                Assert.Equal("Send", invocation.Target);
                Assert.Single(invocation.Arguments);
                Assert.Equal("test", invocation.Arguments[0]);

                // kill the connections
                firstClient.Dispose();
                secondClient.Dispose();

                await Task.WhenAll(firstConnectionHandlerTask, secondConnectionHandlerTask).DefaultTimeout();
            }
        }
    }

    [Fact]
    public async Task RemoveFromGroupWhenNotInGroupDoesNotFail()
    {
        using (StartVerifiableLog())
        {
            var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(null, LoggerFactory);

            var connectionHandler = serviceProvider.GetService<HubConnectionHandler<MethodHub>>();

            using (var client = new TestClient())
            {
                var connectionHandlerTask = await client.ConnectAsync(connectionHandler);

                await client.SendInvocationAsync(nameof(MethodHub.GroupRemoveMethod), "testGroup").DefaultTimeout();

                // kill the connection
                client.Dispose();

                await connectionHandlerTask.DefaultTimeout();
            }
        }
    }

    [Theory]
    [MemberData(nameof(HubTypes))]
    public async Task HubsCanSendToUser(Type hubType)
    {
        using (StartVerifiableLog())
        {
            var connectionHandler = HubConnectionHandlerTestUtils.GetHubConnectionHandler(hubType, loggerFactory: LoggerFactory);

            using (var firstClient = new TestClient(userIdentifier: "userA"))
            using (var secondClient = new TestClient(userIdentifier: "userB"))
            {
                var firstConnectionHandlerTask = await firstClient.ConnectAsync(connectionHandler);
                var secondConnectionHandlerTask = await secondClient.ConnectAsync(connectionHandler);

                await Task.WhenAll(firstClient.Connected, secondClient.Connected).DefaultTimeout();

                await firstClient.SendInvocationAsync("ClientSendMethod", "userB", "test").DefaultTimeout();

                // check that 'secondConnection' has received the group send
                var hubMessage = await secondClient.ReadAsync().DefaultTimeout();
                var invocation = Assert.IsType<InvocationMessage>(hubMessage);
                Assert.Equal("Send", invocation.Target);
                Assert.Single(invocation.Arguments);
                Assert.Equal("test", invocation.Arguments[0]);

                // kill the connections
                firstClient.Dispose();
                secondClient.Dispose();

                await Task.WhenAll(firstConnectionHandlerTask, secondConnectionHandlerTask).DefaultTimeout();
            }
        }
    }

    [Theory]
    [MemberData(nameof(HubTypes))]
    public async Task HubsCanSendToConnection(Type hubType)
    {
        using (StartVerifiableLog())
        {
            var connectionHandler = HubConnectionHandlerTestUtils.GetHubConnectionHandler(hubType, loggerFactory: LoggerFactory);

            using (var firstClient = new TestClient())
            using (var secondClient = new TestClient())
            {
                var firstConnectionHandlerTask = await firstClient.ConnectAsync(connectionHandler);
                var secondConnectionHandlerTask = await secondClient.ConnectAsync(connectionHandler);

                await Task.WhenAll(firstClient.Connected, secondClient.Connected).DefaultTimeout();

                await firstClient.SendInvocationAsync("ConnectionSendMethod", secondClient.Connection.ConnectionId, "test").DefaultTimeout();

                // check that 'secondConnection' has received the group send
                var hubMessage = await secondClient.ReadAsync().DefaultTimeout();
                var invocation = Assert.IsType<InvocationMessage>(hubMessage);
                Assert.Equal("Send", invocation.Target);
                Assert.Single(invocation.Arguments);
                Assert.Equal("test", invocation.Arguments[0]);

                // kill the connections
                firstClient.Dispose();
                secondClient.Dispose();

                await Task.WhenAll(firstConnectionHandlerTask, secondConnectionHandlerTask).DefaultTimeout();
            }
        }
    }

    [Fact]
    public async Task DelayedSendTest()
    {
        using (StartVerifiableLog())
        {
            var connectionHandler = HubConnectionHandlerTestUtils.GetHubConnectionHandler(typeof(HubT), loggerFactory: LoggerFactory);

            using (var firstClient = new TestClient())
            using (var secondClient = new TestClient())
            {
                var firstConnectionHandlerTask = await firstClient.ConnectAsync(connectionHandler);
                var secondConnectionHandlerTask = await secondClient.ConnectAsync(connectionHandler);

                await Task.WhenAll(firstClient.Connected, secondClient.Connected).DefaultTimeout();

                await firstClient.SendInvocationAsync("DelayedSend", secondClient.Connection.ConnectionId, "test").DefaultTimeout();

                // check that 'secondConnection' has received the group send
                var hubMessage = await secondClient.ReadAsync().DefaultTimeout();
                var invocation = Assert.IsType<InvocationMessage>(hubMessage);
                Assert.Equal("Send", invocation.Target);
                Assert.Single(invocation.Arguments);
                Assert.Equal("test", invocation.Arguments[0]);

                // kill the connections
                firstClient.Dispose();
                secondClient.Dispose();

                await Task.WhenAll(firstConnectionHandlerTask, secondConnectionHandlerTask).DefaultTimeout();
            }
        }
    }

    [Theory]
    [MemberData(nameof(StreamingMethodAndHubProtocols))]
    public async Task HubsCanStreamResponses(string method, string protocolName)
    {
        using (StartVerifiableLog())
        {
            var protocol = HubProtocolHelpers.GetHubProtocol(protocolName);

            var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(null, LoggerFactory);
            var connectionHandler = serviceProvider.GetService<HubConnectionHandler<StreamingHub>>();
            var invocationBinder = new Mock<IInvocationBinder>();
            invocationBinder.Setup(b => b.GetStreamItemType(It.IsAny<string>())).Returns(typeof(string));

            using (var client = new TestClient(protocol: protocol, invocationBinder: invocationBinder.Object))
            {
                client.SupportedFormats = protocol.TransferFormat;

                var connectionHandlerTask = await client.ConnectAsync(connectionHandler);

                // Wait for a connection, or for the endpoint to fail.
                await client.Connected.OrThrowIfOtherFails(connectionHandlerTask).DefaultTimeout();

                var messages = await client.StreamAsync(method, 4).DefaultTimeout();

                Assert.Equal(5, messages.Count);
                HubConnectionHandlerTestUtils.AssertHubMessage(new StreamItemMessage(string.Empty, "0"), messages[0]);
                HubConnectionHandlerTestUtils.AssertHubMessage(new StreamItemMessage(string.Empty, "1"), messages[1]);
                HubConnectionHandlerTestUtils.AssertHubMessage(new StreamItemMessage(string.Empty, "2"), messages[2]);
                HubConnectionHandlerTestUtils.AssertHubMessage(new StreamItemMessage(string.Empty, "3"), messages[3]);
                HubConnectionHandlerTestUtils.AssertHubMessage(CompletionMessage.Empty(string.Empty), messages[4]);

                client.Dispose();

                await connectionHandlerTask.DefaultTimeout();
            }
        }
    }

    [Fact]
    public async Task NonErrorCompletionSentWhenStreamCanceledFromClient()
    {
        using (StartVerifiableLog())
        {
            var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(null, LoggerFactory);
            var connectionHandler = serviceProvider.GetService<HubConnectionHandler<StreamingHub>>();

            using (var client = new TestClient())
            {
                var connectionHandlerTask = await client.ConnectAsync(connectionHandler);

                await client.Connected.DefaultTimeout();

                var invocationId = Guid.NewGuid().ToString("N");
                await client.SendHubMessageAsync(new StreamInvocationMessage(invocationId, nameof(StreamingHub.BlockingStream), Array.Empty<object>()));

                // cancel the Streaming method
                await client.SendHubMessageAsync(new CancelInvocationMessage(invocationId)).DefaultTimeout();

                var hubMessage = Assert.IsType<CompletionMessage>(await client.ReadAsync().DefaultTimeout());
                Assert.Equal(invocationId, hubMessage.InvocationId);
                Assert.Null(hubMessage.Error);

                client.Dispose();

                await connectionHandlerTask.DefaultTimeout();
            }
        }
    }

    [Theory]
    [InlineData(nameof(StreamingHub.ExceptionAsyncEnumerable), "Exception: Exception from async enumerable")]
    [InlineData(nameof(StreamingHub.ExceptionAsyncEnumerable), null)]
    [InlineData(nameof(StreamingHub.ExceptionStream), "Exception: Exception from channel")]
    [InlineData(nameof(StreamingHub.ExceptionStream), null)]
    [InlineData(nameof(StreamingHub.ChannelClosedExceptionStream), "ChannelClosedException: ChannelClosedException from channel")]
    [InlineData(nameof(StreamingHub.ChannelClosedExceptionStream), null)]
    [InlineData(nameof(StreamingHub.ChannelClosedExceptionInnerExceptionStream), "Exception: ChannelClosedException from channel")]
    [InlineData(nameof(StreamingHub.ChannelClosedExceptionInnerExceptionStream), null)]
    public async Task ReceiveCorrectErrorFromStreamThrowing(string streamMethod, string detailedError)
    {
        bool ExpectedErrors(WriteContext writeContext)
        {
            return writeContext.LoggerName == "Microsoft.AspNetCore.SignalR.Internal.DefaultHubDispatcher" &&
                   writeContext.EventId.Name == "FailedStreaming";
        }

        using (StartVerifiableLog(ExpectedErrors))
        {
            var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(builder =>
            builder.AddSignalR(options =>
            {
                options.EnableDetailedErrors = detailedError != null;
            }), LoggerFactory);
            var connectionHandler = serviceProvider.GetService<HubConnectionHandler<StreamingHub>>();

            using (var client = new TestClient())
            {
                var connectionHandlerTask = await client.ConnectAsync(connectionHandler);

                await client.Connected.DefaultTimeout();

                var messages = await client.StreamAsync(streamMethod);

                Assert.Equal(1, messages.Count);
                var completion = messages[0] as CompletionMessage;
                Assert.NotNull(completion);
                if (detailedError != null)
                {
                    Assert.Equal($"An error occurred on the server while streaming results. {detailedError}", completion.Error);
                }
                else
                {
                    Assert.Equal("An error occurred on the server while streaming results.", completion.Error);
                }

                client.Dispose();

                await connectionHandlerTask.DefaultTimeout();
            }
        }
    }

    [Fact]
    public async Task CanSendToConnectionsWithDifferentProtocols()
    {
        using (StartVerifiableLog())
        {
            var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(null, LoggerFactory);
            var connectionHandler = serviceProvider.GetService<HubConnectionHandler<MethodHub>>();

            using (var client1 = new TestClient(protocol: new NewtonsoftJsonHubProtocol()))
            using (var client2 = new TestClient(protocol: new MessagePackHubProtocol()))
            {
                var firstConnectionHandlerTask = await client1.ConnectAsync(connectionHandler);
                var secondConnectionHandlerTask = await client2.ConnectAsync(connectionHandler);

                await client1.Connected.DefaultTimeout();
                await client2.Connected.DefaultTimeout();

                var sentMessage = "From Json";

                await client1.SendInvocationAsync(nameof(MethodHub.BroadcastMethod), sentMessage);
                var message1 = await client1.ReadAsync().DefaultTimeout();
                var message2 = await client2.ReadAsync().DefaultTimeout();

                var completion1 = message1 as InvocationMessage;
                Assert.NotNull(completion1);
                Assert.Equal(sentMessage, completion1.Arguments[0]);
                var completion2 = message2 as InvocationMessage;
                Assert.NotNull(completion2);
                // Argument[0] is a 'MsgPackObject' with a string internally, ToString to compare it
                Assert.Equal(sentMessage, completion2.Arguments[0].ToString());

                client1.Dispose();
                client2.Dispose();

                await firstConnectionHandlerTask.DefaultTimeout();
                await secondConnectionHandlerTask.DefaultTimeout();
            }
        }
    }

    public static IEnumerable<object[]> StreamingMethodAndHubProtocols
    {
        get
        {
            var methods = new[]
            {
                    nameof(StreamingHub.CounterChannel),
                    nameof(StreamingHub.CounterChannelAsync),
                    nameof(StreamingHub.CounterChannelValueTaskAsync),
                    nameof(StreamingHub.CounterAsyncEnumerable),
                    nameof(StreamingHub.CounterAsyncEnumerableAsync),
                    nameof(StreamingHub.CounterAsyncEnumerableImpl),
                    nameof(StreamingHub.AsyncEnumerableIsPreferredOverChannelReader),
                };

            foreach (var method in methods)
            {
                foreach (var protocolName in HubProtocolHelpers.AllProtocolNames)
                {
                    yield return new object[] { method, protocolName };
                }
            }
        }
    }

    [Fact]
    public async Task UnauthorizedConnectionCannotInvokeHubMethodWithAuthorization()
    {
        using (StartVerifiableLog())
        {
            var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(services =>
            {
                services.AddAuthorization(options =>
                {
                    options.AddPolicy("test", policy =>
                    {
                        policy.RequireClaim(ClaimTypes.NameIdentifier);
                        policy.AddAuthenticationSchemes("Default");
                    });
                });
            }, LoggerFactory);

            var connectionHandler = serviceProvider.GetService<HubConnectionHandler<MethodHub>>();

            using (var client = new TestClient())
            {
                var connectionHandlerTask = await client.ConnectAsync(connectionHandler);

                await client.Connected.DefaultTimeout();

                var message = await client.InvokeAsync(nameof(MethodHub.AuthMethod)).DefaultTimeout();

                Assert.NotNull(message.Error);

                client.Dispose();

                await connectionHandlerTask.DefaultTimeout();
            }
        }
    }

    [Fact]
    public async Task AuthorizedConnectionCanInvokeHubMethodWithAuthorization()
    {
        using (StartVerifiableLog())
        {
            var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(services =>
            {
                services.AddAuthorization(options =>
                {
                    options.AddPolicy("test", policy =>
                    {
                        policy.RequireClaim(ClaimTypes.NameIdentifier);
                        policy.AddAuthenticationSchemes("Default");
                    });
                });
            }, LoggerFactory);

            var connectionHandler = serviceProvider.GetService<HubConnectionHandler<MethodHub>>();

            using (var client = new TestClient())
            {
                client.Connection.User.AddIdentity(new ClaimsIdentity(new[] { new Claim(ClaimTypes.NameIdentifier, "name") }));
                var connectionHandlerTask = await client.ConnectAsync(connectionHandler);

                await client.Connected.DefaultTimeout();

                var message = await client.InvokeAsync(nameof(MethodHub.AuthMethod)).DefaultTimeout();

                Assert.Null(message.Error);

                client.Dispose();

                await connectionHandlerTask.DefaultTimeout();
            }
        }
    }

    private class TestConnectionLifetimeNotification : IConnectionLifetimeNotificationFeature
    {
        private readonly CancellationTokenSource _cts = new CancellationTokenSource();

        public CancellationToken ConnectionClosedRequested { get => _cts.Token; set => throw new NotImplementedException(); }

        public void RequestClose()
        {
            _cts.Cancel();
        }
    }

    [Fact]
    public async Task ConnectionLifetimeNotificationClosesConnectionWithReconnectAllowed()
    {
        using (StartVerifiableLog())
        {
            var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(loggerFactory: LoggerFactory);

            var connectionHandler = serviceProvider.GetService<HubConnectionHandler<MethodHub>>();

            using (var client = new TestClient())
            {
                client.Connection.Features.Set<IConnectionLifetimeNotificationFeature>(new TestConnectionLifetimeNotification());

                var connectionHandlerTask = await client.ConnectAsync(connectionHandler);

                await client.Connected.DefaultTimeout();

                client.Connection.Features.Get<IConnectionLifetimeNotificationFeature>().RequestClose();

                var close = Assert.IsType<CloseMessage>(await client.ReadAsync().DefaultTimeout());

                Assert.True(close.AllowReconnect);

                client.Dispose();

                await connectionHandlerTask.DefaultTimeout();
            }
        }
    }

    private class TestAuthHandler : IAuthorizationHandler
    {
        public Task HandleAsync(AuthorizationHandlerContext context)
        {
            Assert.NotNull(context.Resource);
            var resource = Assert.IsType<HubInvocationContext>(context.Resource);
            Assert.Equal(typeof(MethodHub), resource.Hub.GetType());
            Assert.Equal(nameof(MethodHub.MultiParamAuthMethod), resource.HubMethodName);
            Assert.Equal(2, resource.HubMethodArguments?.Count);
            Assert.Equal("Hello", resource.HubMethodArguments[0]);
            Assert.Equal("World!", resource.HubMethodArguments[1]);
            Assert.NotNull(resource.Context);
            Assert.Equal(context.User, resource.Context.User);
            Assert.NotNull(resource.Context.GetHttpContext());
            Assert.NotNull(resource.ServiceProvider);
            Assert.Equal(typeof(MethodHub).GetMethod(nameof(MethodHub.MultiParamAuthMethod)), resource.HubMethod);

            return Task.CompletedTask;
        }
    }

    [Fact]
    public async Task HubMethodWithAuthorizationProvidesResourceToAuthHandlers()
    {
        using (StartVerifiableLog())
        {
            var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(services =>
            {
                services.AddAuthorization(options =>
                {
                    options.AddPolicy("test", policy =>
                    {
                        policy.RequireClaim(ClaimTypes.NameIdentifier);
                        policy.AddAuthenticationSchemes("Default");
                    });
                });

                services.AddSingleton<IAuthorizationHandler, TestAuthHandler>();
            }, LoggerFactory);

            var connectionHandler = serviceProvider.GetService<HubConnectionHandler<MethodHub>>();

            using (var client = new TestClient())
            {
                client.Connection.User.AddIdentity(new ClaimsIdentity(new[] { new Claim(ClaimTypes.NameIdentifier, "name") }));

                // Setup a HttpContext to make sure it flows to the AuthHandler correctly
                var httpConnectionContext = new HttpContextFeatureImpl();
                httpConnectionContext.HttpContext = new DefaultHttpContext();
                client.Connection.Features.Set<IHttpContextFeature>(httpConnectionContext);

                var connectionHandlerTask = await client.ConnectAsync(connectionHandler);

                await client.Connected.DefaultTimeout();

                var message = await client.InvokeAsync(nameof(MethodHub.MultiParamAuthMethod), "Hello", "World!").DefaultTimeout();

                Assert.Null(message.Error);

                client.Dispose();

                await connectionHandlerTask.DefaultTimeout();
            }
        }
    }

    [Fact]
    public async Task HubOptionsCanUseCustomJsonSerializerSettings()
    {
        using (StartVerifiableLog())
        {
            var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(services =>
            {
                services
                    .AddSignalR()
                    .AddNewtonsoftJsonProtocol(o =>
                    {
                        o.PayloadSerializerSettings = new JsonSerializerSettings
                        {
                            ContractResolver = new DefaultContractResolver()
                        };
                    });
            }, LoggerFactory);

            var connectionHandler = serviceProvider.GetService<HubConnectionHandler<MethodHub>>();

            using (var client = new TestClient())
            {
                var connectionHandlerTask = await client.ConnectAsync(connectionHandler);

                await client.Connected.DefaultTimeout();

                await client.SendInvocationAsync(nameof(MethodHub.BroadcastItem)).DefaultTimeout();

                var message = (InvocationMessage)await client.ReadAsync().DefaultTimeout();

                var customItem = message.Arguments[0].ToString();
                // by default properties serialized by JsonHubProtocol are using camelCasing
                Assert.Contains("Message", customItem);
                Assert.Contains("paramName", customItem);

                client.Dispose();

                await connectionHandlerTask.DefaultTimeout();
            }
        }
    }

    [Fact]
    public async Task JsonHubProtocolUsesCamelCasingByDefault()
    {
        using (StartVerifiableLog())
        {
            var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(null, LoggerFactory);
            var connectionHandler = serviceProvider.GetService<HubConnectionHandler<MethodHub>>();

            using (var client = new TestClient())
            {
                var connectionHandlerTask = await client.ConnectAsync(connectionHandler);

                await client.Connected.DefaultTimeout();

                await client.SendInvocationAsync(nameof(MethodHub.BroadcastItem)).DefaultTimeout();

                var message = (InvocationMessage)await client.ReadAsync().DefaultTimeout();

                var customItem = message.Arguments[0].ToString();
                // originally Message, paramName
                Assert.Contains("message", customItem);
                Assert.Contains("paramName", customItem);

                client.Dispose();

                await connectionHandlerTask.DefaultTimeout();
            }
        }
    }

    [Fact]
    public async Task HubOptionsCanUseCustomMessagePackSettings()
    {
        using (StartVerifiableLog())
        {
            var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(services =>
            {
                services.AddSignalR()
                    .AddMessagePackProtocol(options =>
                    {
                        options.SerializerOptions = MessagePackSerializerOptions.Standard.WithResolver(CompositeResolver.Create(new CustomFormatter(), options.SerializerOptions.Resolver));
                    });
            }, LoggerFactory);

            var connectionHandler = serviceProvider.GetService<HubConnectionHandler<MethodHub>>();

            var msgPackOptions = serviceProvider.GetRequiredService<IOptions<MessagePackHubProtocolOptions>>();
            using (var client = new TestClient(protocol: new MessagePackHubProtocol(msgPackOptions)))
            {
                client.SupportedFormats = TransferFormat.Binary;
                var connectionHandlerTask = await client.ConnectAsync(connectionHandler);

                await client.Connected.DefaultTimeout();

                await client.SendInvocationAsync(nameof(MethodHub.BroadcastItem)).DefaultTimeout();

                var message = Assert.IsType<InvocationMessage>(await client.ReadAsync().DefaultTimeout());

                var result = message.Arguments[0] as Dictionary<object, object>;
                Assert.Equal("formattedString", result["Message"]);
                Assert.Equal("formattedString", result["paramName"]);

                client.Dispose();

                await connectionHandlerTask.DefaultTimeout();
            }
        }
    }

    [Fact]
    public async Task HubOptionsCanNotHaveNullSupportedProtocols()
    {
        using (StartVerifiableLog())
        {
            var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(services =>
            {
                services.AddSignalR(o =>
                {
                    o.SupportedProtocols = null;
                });
            }, LoggerFactory);

            var connectionHandler = serviceProvider.GetService<HubConnectionHandler<MethodHub>>();

            var msgPackOptions = serviceProvider.GetRequiredService<IOptions<MessagePackHubProtocolOptions>>();
            using (var client = new TestClient(protocol: new MessagePackHubProtocol(msgPackOptions)))
            {
                client.SupportedFormats = TransferFormat.Binary;
                await Assert.ThrowsAsync<InvalidOperationException>(async () => await await client.ConnectAsync(connectionHandler, expectedHandshakeResponseMessage: false)).DefaultTimeout();
            }
        }
    }

    [Fact]
    public async Task HubOptionsCanNotHaveEmptySupportedProtocols()
    {
        using (StartVerifiableLog())
        {
            var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(services =>
            {
                services.AddSignalR(o =>
                {
                    o.SupportedProtocols = new List<string>();
                });
            }, LoggerFactory);

            var connectionHandler = serviceProvider.GetService<HubConnectionHandler<MethodHub>>();

            var msgPackOptions = serviceProvider.GetRequiredService<IOptions<MessagePackHubProtocolOptions>>();
            using (var client = new TestClient(protocol: new MessagePackHubProtocol(msgPackOptions)))
            {
                client.SupportedFormats = TransferFormat.Binary;
                await Assert.ThrowsAsync<InvalidOperationException>(async () => await await client.ConnectAsync(connectionHandler, expectedHandshakeResponseMessage: false)).DefaultTimeout();
            }
        }
    }

    [Fact]
    public async Task ConnectionUserIdIsAssignedByUserIdProvider()
    {
        using (StartVerifiableLog())
        {
            var firstRequest = true;
            var userIdProvider = new TestUserIdProvider(c =>
            {
                if (firstRequest)
                {
                    firstRequest = false;
                    return "client1";
                }
                else
                {
                    return "client2";
                }
            });
            var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(services =>
            {
                services.AddSingleton<IUserIdProvider>(userIdProvider);
            }, LoggerFactory);
            var connectionHandler = serviceProvider.GetService<HubConnectionHandler<MethodHub>>();

            using (var client1 = new TestClient())
            using (var client2 = new TestClient())
            {
                var connectionHandlerTask1 = await client1.ConnectAsync(connectionHandler);
                var connectionHandlerTask2 = await client2.ConnectAsync(connectionHandler);

                await client1.Connected.DefaultTimeout();
                await client2.Connected.DefaultTimeout();

                await client2.SendInvocationAsync(nameof(MethodHub.SendToMultipleUsers), new[] { "client1" }, "Hi!").DefaultTimeout();

                var message = (InvocationMessage)await client1.ReadAsync().DefaultTimeout();

                Assert.Equal("Send", message.Target);
                Assert.Collection(message.Arguments, arg => Assert.Equal("Hi!", arg));

                client1.Dispose();
                client2.Dispose();

                await connectionHandlerTask1.DefaultTimeout();
                await connectionHandlerTask2.DefaultTimeout();

                // Read the completion, then we should have nothing left in client2's queue
                Assert.IsType<CompletionMessage>(client2.TryRead());
                Assert.IsType<CloseMessage>(client2.TryRead());
                Assert.Null(client2.TryRead());
            }
        }
    }

    private class CustomFormatter : IFormatterResolver
    {
        public IMessagePackFormatter<T> GetFormatter<T>()
        {
            if (typeof(T) == typeof(string))
            {
                return new StringFormatter<T>();
            }
            return null;
        }

        private class StringFormatter<T> : IMessagePackFormatter<T>
        {
            public T Deserialize(ref MessagePackReader reader, MessagePackSerializerOptions options)
            {
                return (T)(object)reader.ReadString();
            }

            public void Serialize(ref MessagePackWriter writer, T value, MessagePackSerializerOptions options)
            {
                writer.Write("formattedString");
            }
        }
    }

    [Fact]
    public async Task CanGetHttpContextFromHubConnectionContext()
    {
        using (StartVerifiableLog())
        {
            var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(null, LoggerFactory);

            var connectionHandler = serviceProvider.GetService<HubConnectionHandler<MethodHub>>();

            using (var client = new TestClient())
            {
                var httpContext = new DefaultHttpContext();
                var feature = new TestHttpContextFeature
                {
                    HttpContext = httpContext
                };
                client.Connection.Features.Set<IHttpContextFeature>(feature);
                var connectionHandlerTask = await client.ConnectAsync(connectionHandler);

                await client.Connected.DefaultTimeout();

                var result = (await client.InvokeAsync(nameof(MethodHub.HasHttpContext)).DefaultTimeout()).Result;
                Assert.True((bool)result);

                client.Dispose();

                await connectionHandlerTask.DefaultTimeout();
            }
        }
    }

    [Fact]
    public async Task GetHttpContextFromHubConnectionContextHandlesNull()
    {
        using (StartVerifiableLog())
        {
            var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(null, LoggerFactory);

            var connectionHandler = serviceProvider.GetService<HubConnectionHandler<MethodHub>>();

            using (var client = new TestClient())
            {
                var connectionHandlerTask = await client.ConnectAsync(connectionHandler);

                await client.Connected.DefaultTimeout();

                var result = (await client.InvokeAsync(nameof(MethodHub.HasHttpContext)).DefaultTimeout()).Result;
                Assert.False((bool)result);

                client.Dispose();

                await connectionHandlerTask.DefaultTimeout();
            }
        }
    }

    [Fact]
    public async Task AcceptsPingMessages()
    {
        using (StartVerifiableLog())
        {
            var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(null, LoggerFactory);
            var connectionHandler = serviceProvider.GetService<HubConnectionHandler<MethodHub>>();

            using (var client = new TestClient(new NewtonsoftJsonHubProtocol()))
            {
                var connectionHandlerTask = await client.ConnectAsync(connectionHandler);
                await client.Connected.DefaultTimeout();

                // Send a ping
                await client.SendHubMessageAsync(PingMessage.Instance).DefaultTimeout();

                // Now do an invocation to make sure we processed the ping message
                var completion = await client.InvokeAsync(nameof(MethodHub.ValueMethod)).DefaultTimeout();
                Assert.NotNull(completion);

                client.Dispose();

                await connectionHandlerTask.DefaultTimeout();
            }
        }
    }

    [Fact]
    public async Task DoesNotWritePingMessagesIfSufficientOtherMessagesAreSent()
    {
        using (StartVerifiableLog())
        {
            var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(services =>
                services.Configure<HubOptions>(options =>
                    options.KeepAliveInterval = TimeSpan.FromMilliseconds(100)), LoggerFactory);
            var connectionHandler = serviceProvider.GetService<HubConnectionHandler<MethodHub>>();

            using (var client = new TestClient(new NewtonsoftJsonHubProtocol()))
            {
                var connectionHandlerTask = await client.ConnectAsync(connectionHandler);

                await client.Connected.DefaultTimeout();

                // Echo a bunch of stuff, waiting 10ms between each, until 500ms have elapsed
                var start = DateTime.UtcNow;
                while ((DateTime.UtcNow - start).TotalMilliseconds <= 500.0)
                {
                    await client.SendInvocationAsync("Echo", "foo").DefaultTimeout();
                    await Task.Delay(10);
                }

                // Shut down
                client.Dispose();

                await connectionHandlerTask.DefaultTimeout();

                client.Connection.Transport.Output.Complete();

                // We shouldn't have any ping messages
                HubMessage message;
                var counter = 0;
                while ((message = await client.ReadAsync()) != null)
                {
                    counter += 1;
                    Assert.IsNotType<PingMessage>(message);
                }
                Assert.InRange(counter, 1, 50);
            }
        }
    }

    [Fact]
    public async Task WritesPingMessageIfNothingWrittenWhenKeepAliveIntervalElapses()
    {
        using (StartVerifiableLog())
        {
            var interval = TimeSpan.FromMilliseconds(100);
            var timeProvider = new MockTimeProvider();
            var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(services =>
                services.Configure<HubOptions>(options =>
                    options.KeepAliveInterval = interval), LoggerFactory);
            var connectionHandler = serviceProvider.GetService<HubConnectionHandler<MethodHub>>();
            connectionHandler.TimeProvider = timeProvider;

            using (var client = new TestClient(new NewtonsoftJsonHubProtocol()))
            {
                var connectionHandlerTask = await client.ConnectAsync(connectionHandler);
                await client.Connected.DefaultTimeout();

                // Trigger multiple keep alives
                var heartbeatCount = 5;
                for (var i = 0; i < heartbeatCount; i++)
                {
                    timeProvider.Advance(interval + TimeSpan.FromMilliseconds(1));
                    client.TickHeartbeat();
                }

                // Shut down
                client.Dispose();

                await connectionHandlerTask.DefaultTimeout();

                client.Connection.Transport.Output.Complete();

                // We should have all pings (and close message)
                HubMessage message;
                var pingCounter = 0;
                var hasCloseMessage = false;
                while ((message = await client.ReadAsync().DefaultTimeout()) != null)
                {
                    if (hasCloseMessage)
                    {
                        Assert.True(false, "Received message after close");
                    }

                    switch (message)
                    {
                        case PingMessage _:
                            pingCounter += 1;
                            break;
                        case CloseMessage _:
                            hasCloseMessage = true;
                            break;
                        default:
                            Assert.True(false, "Unexpected message type: " + message.GetType().Name);
                            break;
                    }
                }
                Assert.Equal(heartbeatCount, pingCounter);
            }
        }
    }

    [Fact]
    public async Task ConnectionNotTimedOutIfClientNeverPings()
    {
        using (StartVerifiableLog())
        {
            var timeout = TimeSpan.FromMilliseconds(100);
            var timeProvider = new MockTimeProvider();
            var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(services =>
                services.Configure<HubOptions>(options =>
                    options.ClientTimeoutInterval = timeout), LoggerFactory);
            var connectionHandler = serviceProvider.GetService<HubConnectionHandler<MethodHub>>();
            connectionHandler.TimeProvider = timeProvider;

            using (var client = new TestClient(new NewtonsoftJsonHubProtocol()))
            {
                var connectionHandlerTask = await client.ConnectAsync(connectionHandler);
                await client.Connected.DefaultTimeout();
                // This is a fake client -- it doesn't auto-ping to signal

                // We go over the 100 ms timeout interval multiple times
                for (var i = 0; i < 3; i++)
                {
                    timeProvider.Advance(timeout + TimeSpan.FromMilliseconds(1));
                    client.TickHeartbeat();
                }

                // Invoke a Hub method and wait for the result to reliably test if the connection is still active
                var id = await client.SendInvocationAsync(nameof(MethodHub.ValueMethod)).DefaultTimeout();
                var result = await client.ReadAsync().DefaultTimeout();

                // but client should still be open, since it never pinged to activate the timeout checking
                Assert.False(connectionHandlerTask.IsCompleted);
            }
        }
    }

    [Fact]
    public async Task ConnectionTimesOutIfInitialPingAndThenNoMessages()
    {
        using (StartVerifiableLog())
        {
            var timeout = TimeSpan.FromMilliseconds(100);
            var timeProvider = new MockTimeProvider();
            var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(services =>
                services.Configure<HubOptions>(options =>
                    options.ClientTimeoutInterval = timeout), LoggerFactory);
            var connectionHandler = serviceProvider.GetService<HubConnectionHandler<MethodHub>>();
            connectionHandler.TimeProvider = timeProvider;

            using (var client = new TestClient(new NewtonsoftJsonHubProtocol()))
            {
                var connectionHandlerTask = await client.ConnectAsync(connectionHandler);
                await client.Connected.DefaultTimeout();
                await client.SendHubMessageAsync(PingMessage.Instance);

                timeProvider.Advance(timeout + TimeSpan.FromMilliseconds(1));
                client.TickHeartbeat();

                await connectionHandlerTask.DefaultTimeout();
            }
        }
    }

    [Fact]
    public async Task ReceivingMessagesPreventsConnectionTimeoutFromOccuring()
    {
        using (StartVerifiableLog())
        {
            var timeout = TimeSpan.FromMilliseconds(300);
            var timeProvider = new MockTimeProvider();
            var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(services =>
                services.Configure<HubOptions>(options =>
                    options.ClientTimeoutInterval = timeout), LoggerFactory);
            var connectionHandler = serviceProvider.GetService<HubConnectionHandler<MethodHub>>();
            connectionHandler.TimeProvider = timeProvider;

            using (var client = new TestClient(new NewtonsoftJsonHubProtocol()))
            {
                var connectionHandlerTask = await client.ConnectAsync(connectionHandler);
                await client.Connected.DefaultTimeout();
                await client.SendHubMessageAsync(PingMessage.Instance);

                for (int i = 0; i < 10; i++)
                {
                    timeProvider.Advance(timeout - TimeSpan.FromMilliseconds(1));
                    client.TickHeartbeat();
                    await client.SendHubMessageAsync(PingMessage.Instance);
                }

                // Invoke a Hub method and wait for the result to reliably test if the connection is still active
                var id = await client.SendInvocationAsync(nameof(MethodHub.ValueMethod)).DefaultTimeout();
                var result = await client.ReadAsync().DefaultTimeout();

                Assert.IsType<CompletionMessage>(result);

                Assert.False(connectionHandlerTask.IsCompleted);
            }
        }
    }

    internal class PipeReaderWrapper : PipeReader
    {
        private readonly PipeReader _originalPipeReader;
        private TaskCompletionSource _waitForRead;
        private readonly object _lock = new object();

        public PipeReaderWrapper(PipeReader pipeReader)
        {
            _originalPipeReader = pipeReader;
            _waitForRead = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
        }

        public override void AdvanceTo(SequencePosition consumed) =>
            _originalPipeReader.AdvanceTo(consumed);

        public override void AdvanceTo(SequencePosition consumed, SequencePosition examined) =>
            _originalPipeReader.AdvanceTo(consumed, examined);

        public override void CancelPendingRead() =>
            _originalPipeReader.CancelPendingRead();

        public override void Complete(Exception exception = null) =>
            _originalPipeReader.Complete(exception);

        public override async ValueTask<ReadResult> ReadAsync(CancellationToken cancellationToken = default)
        {
            lock (_lock)
            {
                _waitForRead.SetResult();
            }

            try
            {
                return await _originalPipeReader.ReadAsync(cancellationToken);
            }
            finally
            {
                lock (_lock)
                {
                    _waitForRead = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
                }
            }
        }

        public override bool TryRead(out ReadResult result) =>
            _originalPipeReader.TryRead(out result);

        public Task WaitForReadStart()
        {
            lock (_lock)
            {
                return _waitForRead.Task;
            }
        }
    }

    internal class CustomDuplex : IDuplexPipe
    {
        private readonly IDuplexPipe _originalDuplexPipe;
        public readonly PipeReaderWrapper WrappedPipeReader;

        public CustomDuplex(IDuplexPipe duplexPipe)
        {
            _originalDuplexPipe = duplexPipe;
            WrappedPipeReader = new PipeReaderWrapper(_originalDuplexPipe.Input);
        }

        public PipeReader Input => WrappedPipeReader;

        public PipeWriter Output => _originalDuplexPipe.Output;
    }

    [Fact]
    public async Task HubMethodInvokeDoesNotCountTowardsClientTimeout()
    {
        using (StartVerifiableLog())
        {
            var tcsService = new TcsService();
            var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(services =>
            {
                services.Configure<HubOptions>(options =>
                {
                    options.ClientTimeoutInterval = TimeSpan.FromMilliseconds(0);
                    options.MaximumParallelInvocationsPerClient = 1;
                });
                services.AddSingleton(tcsService);
            }, LoggerFactory);
            var connectionHandler = serviceProvider.GetService<HubConnectionHandler<LongRunningHub>>();

            using (var client = new TestClient(new JsonHubProtocol()))
            {
                var customDuplex = new CustomDuplex(client.Connection.Transport);
                client.Connection.Transport = customDuplex;

                var connectionHandlerTask = await client.ConnectAsync(connectionHandler);
                // This starts the timeout logic
                await client.SendHubMessageAsync(PingMessage.Instance);

                // Call long running hub method
                var hubMethodTask1 = client.InvokeAsync(nameof(LongRunningHub.LongRunningMethod));
                await tcsService.StartedMethod.Task.DefaultTimeout();

                // Wait for server to start reading again
                await customDuplex.WrappedPipeReader.WaitForReadStart().DefaultTimeout();
                // Send another invocation to server, since we use Inline scheduling we know that once this call completes the server will have read and processed
                // the message, it should be stuck waiting for the in-progress invoke now
                _ = await client.SendInvocationAsync(nameof(LongRunningHub.LongRunningMethod)).DefaultTimeout();

                // Tick heartbeat while hub method is running to show that close isn't triggered
                client.TickHeartbeat();

                // Unblock long running hub method
                tcsService.EndMethod.SetResult(null);

                await hubMethodTask1.DefaultTimeout();
                await client.ReadAsync().DefaultTimeout();

                // There is a small window when the hub method finishes and the timer starts again
                // So we need to delay a little before ticking the heart beat.
                // We do this by waiting until we know the HubConnectionHandler code is in pipe.ReadAsync()
                await customDuplex.WrappedPipeReader.WaitForReadStart().DefaultTimeout();

                // Tick heartbeat again now that we're outside of the hub method
                client.TickHeartbeat();

                // Connection is closed
                await connectionHandlerTask.DefaultTimeout();
            }
        }
    }

    [Fact]
    public async Task HubMethodInvokeCountsTowardsClientTimeoutIfParallelNotMaxed()
    {
        using (StartVerifiableLog())
        {
            var tcsService = new TcsService();
            var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(services =>
            {
                services.Configure<HubOptions>(options =>
                {
                    options.ClientTimeoutInterval = TimeSpan.FromMilliseconds(0);
                    options.MaximumParallelInvocationsPerClient = 2;
                });
                services.AddSingleton(tcsService);
            }, LoggerFactory);
            var connectionHandler = serviceProvider.GetService<HubConnectionHandler<LongRunningHub>>();

            using (var client = new TestClient(new JsonHubProtocol()))
            {
                var connectionHandlerTask = await client.ConnectAsync(connectionHandler);
                // This starts the timeout logic
                await client.SendHubMessageAsync(PingMessage.Instance);

                // Call long running hub method
                var hubMethodTask = client.InvokeAsync(nameof(LongRunningHub.LongRunningMethod));
                await tcsService.StartedMethod.Task.DefaultTimeout();

                // Tick heartbeat while hub method is running
                client.TickHeartbeat();

                // Connection is closed
                await connectionHandlerTask.DefaultTimeout();
            }
        }
    }

    [Fact]
    public async Task EndingConnectionSendsCloseMessageWithNoError()
    {
        using (StartVerifiableLog())
        {
            var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(null, LoggerFactory);
            var connectionHandler = serviceProvider.GetService<HubConnectionHandler<MethodHub>>();

            using (var client = new TestClient(new NewtonsoftJsonHubProtocol()))
            {
                var connectionHandlerTask = await client.ConnectAsync(connectionHandler);

                await client.Connected.DefaultTimeout();

                // Shut down
                client.Dispose();

                await connectionHandlerTask.DefaultTimeout();

                client.Connection.Transport.Output.Complete();

                var message = await client.ReadAsync().DefaultTimeout();

                var closeMessage = Assert.IsType<CloseMessage>(message);
                Assert.Null(closeMessage.Error);
            }
        }
    }

    [Theory]
    [InlineData(true)]
    [InlineData(false)]
    public async Task ErrorInHubOnConnectSendsCloseMessageWithError(bool detailedErrors)
    {
        bool ExpectedErrors(WriteContext writeContext)
        {
            return writeContext.LoggerName == "Microsoft.AspNetCore.SignalR.HubConnectionHandler" &&
                   writeContext.EventId.Name == "ErrorDispatchingHubEvent";
        }

        using (StartVerifiableLog(ExpectedErrors))
        {
            var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(builder =>
            {
                builder.AddSignalR(options =>
                {
                    options.EnableDetailedErrors = detailedErrors;
                });
            }, LoggerFactory);
            var connectionHandler = serviceProvider.GetService<HubConnectionHandler<OnConnectedThrowsHub>>();

            using (var client = new TestClient(new NewtonsoftJsonHubProtocol()))
            {
                var connectionHandlerTask = await client.ConnectAsync(connectionHandler);

                var message = await client.ReadAsync().DefaultTimeout();

                var closeMessage = Assert.IsType<CloseMessage>(message);
                if (detailedErrors)
                {
                    Assert.Equal("Connection closed with an error. InvalidOperationException: Hub OnConnected failed.", closeMessage.Error);
                }
                else
                {
                    Assert.Equal("Connection closed with an error.", closeMessage.Error);
                }

                await connectionHandlerTask.DefaultTimeout();
            }
        }
    }

    [Fact]
    public async Task StreamingInvocationsDoNotBlockOtherInvocations()
    {
        using (StartVerifiableLog())
        {
            var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(services =>
            {
                services.AddSignalR(options =>
                {
                    options.MaximumParallelInvocationsPerClient = 1;
                });
            }, LoggerFactory);
            var connectionHandler = serviceProvider.GetService<HubConnectionHandler<StreamingHub>>();

            using (var client = new TestClient(new NewtonsoftJsonHubProtocol()))
            {
                var connectionHandlerTask = await client.ConnectAsync(connectionHandler).DefaultTimeout();

                // Blocking streaming invocation to test that other invocations can still run
                await client.SendHubMessageAsync(new StreamInvocationMessage("1", nameof(StreamingHub.BlockingStream), Array.Empty<object>())).DefaultTimeout();

                var completion = await client.InvokeAsync(nameof(StreamingHub.NonStream)).DefaultTimeout();
                Assert.Equal(42L, completion.Result);

                // Shut down
                client.Dispose();

                await connectionHandlerTask.DefaultTimeout();
            }
        }
    }

    [Fact]
    public async Task StreamMethodThatThrowsWillCleanup()
    {
        bool ExpectedErrors(WriteContext writeContext)
        {
            return writeContext.LoggerName == "Microsoft.AspNetCore.SignalR.Internal.DefaultHubDispatcher" &&
                   writeContext.EventId.Name == "FailedInvokingHubMethod";
        }

        using (StartVerifiableLog(ExpectedErrors))
        {
            var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(builder =>
            {
                builder.AddSingleton(typeof(IHubActivator<>), typeof(CustomHubActivator<>));
            }, LoggerFactory);
            var connectionHandler = serviceProvider.GetService<HubConnectionHandler<StreamingHub>>();

            using (var client = new TestClient())
            {
                var connectionHandlerTask = await client.ConnectAsync(connectionHandler);

                await client.Connected.DefaultTimeout();

                var messages = await client.StreamAsync(nameof(StreamingHub.ThrowStream));

                Assert.Equal(1, messages.Count);
                var completion = messages[0] as CompletionMessage;
                Assert.NotNull(completion);

                var hubActivator = serviceProvider.GetService<IHubActivator<StreamingHub>>() as CustomHubActivator<StreamingHub>;

                // OnConnectedAsync and ThrowStream hubs have been disposed
                Assert.Equal(2, hubActivator.ReleaseCount);

                client.Dispose();

                await connectionHandlerTask.DefaultTimeout();
            }
        }
    }

    [Fact]
    public async Task StreamMethodThatReturnsNullWillCleanup()
    {
        using (StartVerifiableLog())
        {
            var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(builder =>
            {
                builder.AddSingleton(typeof(IHubActivator<>), typeof(CustomHubActivator<>));
            }, LoggerFactory);
            var connectionHandler = serviceProvider.GetService<HubConnectionHandler<StreamingHub>>();

            using (var client = new TestClient())
            {
                var connectionHandlerTask = await client.ConnectAsync(connectionHandler);

                await client.Connected.DefaultTimeout();

                var messages = await client.StreamAsync(nameof(StreamingHub.NullStream));

                Assert.Equal(1, messages.Count);
                var completion = messages[0] as CompletionMessage;
                Assert.NotNull(completion);

                var hubActivator = serviceProvider.GetService<IHubActivator<StreamingHub>>() as CustomHubActivator<StreamingHub>;

                // OnConnectedAsync and NullStream hubs have been disposed
                Assert.Equal(2, hubActivator.ReleaseCount);

                client.Dispose();

                await connectionHandlerTask.DefaultTimeout();
            }
        }
    }

    [Fact]
    public async Task StreamMethodWithDuplicateIdFails()
    {
        using (StartVerifiableLog())
        {
            var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(builder =>
            {
                builder.AddSingleton(typeof(IHubActivator<>), typeof(CustomHubActivator<>));
            }, LoggerFactory);
            var connectionHandler = serviceProvider.GetService<HubConnectionHandler<StreamingHub>>();

            using (var client = new TestClient())
            {
                var connectionHandlerTask = await client.ConnectAsync(connectionHandler);

                await client.Connected.DefaultTimeout();

                await client.SendHubMessageAsync(new StreamInvocationMessage("123", nameof(StreamingHub.BlockingStream), Array.Empty<object>())).DefaultTimeout();

                await client.SendHubMessageAsync(new StreamInvocationMessage("123", nameof(StreamingHub.BlockingStream), Array.Empty<object>())).DefaultTimeout();

                var completion = Assert.IsType<CompletionMessage>(await client.ReadAsync().DefaultTimeout());
                Assert.Equal("Invocation ID '123' is already in use.", completion.Error);

                var hubActivator = serviceProvider.GetService<IHubActivator<StreamingHub>>() as CustomHubActivator<StreamingHub>;

                // OnConnectedAsync and BlockingStream hubs have been disposed
                Assert.Equal(2, hubActivator.ReleaseCount);

                client.Dispose();

                await connectionHandlerTask.DefaultTimeout();
            }
        }
    }

    [Fact]
    public async Task InvocationsRunInOrderWithNoParallelism()
    {
        using (StartVerifiableLog())
        {
            var tcsService = new TcsService();
            var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(builder =>
            {
                builder.AddSingleton(tcsService);

                builder.AddSignalR(options =>
                {
                    options.MaximumParallelInvocationsPerClient = 1;
                });
            }, LoggerFactory);
            var connectionHandler = serviceProvider.GetService<HubConnectionHandler<LongRunningHub>>();

            // Because we use PipeScheduler.Inline the hub invocations will run inline until they wait, which happens inside the LongRunningMethod call
            using (var client = new TestClient())
            {
                var connectionHandlerTask = await client.ConnectAsync(connectionHandler).DefaultTimeout();

                // Long running hub invocation to test that other invocations will not run until it is completed
                await client.SendInvocationAsync(nameof(LongRunningHub.LongRunningMethod), nonBlocking: false).DefaultTimeout();
                // Wait for the long running method to start
                await tcsService.StartedMethod.Task.DefaultTimeout();

                // Invoke another hub method which will wait for the first method to finish
                await client.SendInvocationAsync(nameof(LongRunningHub.SimpleMethod), nonBlocking: false).DefaultTimeout();
                // Both invocations should be waiting now
                Assert.Null(client.TryRead());

                // Release the long running hub method
                tcsService.EndMethod.TrySetResult(null);

                // Long running hub method result
                var firstResult = await client.ReadAsync().DefaultTimeout();

                var longRunningCompletion = Assert.IsType<CompletionMessage>(firstResult);
                Assert.Equal(12L, longRunningCompletion.Result);

                // simple hub method result
                var secondResult = await client.ReadAsync().DefaultTimeout();

                var simpleCompletion = Assert.IsType<CompletionMessage>(secondResult);
                Assert.Equal(21L, simpleCompletion.Result);

                // Shut down
                client.Dispose();

                await connectionHandlerTask.DefaultTimeout();
            }
        }
    }

    [Fact]
    public async Task InvocationsCanRunOutOfOrderWithParallelism()
    {
        using (StartVerifiableLog())
        {
            var tcsService = new TcsService();
            var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(builder =>
            {
                builder.AddSingleton(tcsService);

                builder.AddSignalR(options =>
                {
                    options.MaximumParallelInvocationsPerClient = 2;
                });
            }, LoggerFactory);
            var connectionHandler = serviceProvider.GetService<HubConnectionHandler<LongRunningHub>>();

            // Because we use PipeScheduler.Inline the hub invocations will run inline until they wait, which happens inside the LongRunningMethod call
            using (var client = new TestClient())
            {
                var connectionHandlerTask = await client.ConnectAsync(connectionHandler).DefaultTimeout();

                // Long running hub invocation to test that other invocations will not run until it is completed
                await client.SendInvocationAsync(nameof(LongRunningHub.LongRunningMethod), nonBlocking: false).DefaultTimeout();
                // Wait for the long running method to start
                await tcsService.StartedMethod.Task.DefaultTimeout();

                for (var i = 0; i < 5; i++)
                {
                    // Invoke another hub method which will wait for the first method to finish
                    await client.SendInvocationAsync(nameof(LongRunningHub.SimpleMethod), nonBlocking: false).DefaultTimeout();

                    // simple hub method result
                    var secondResult = await client.ReadAsync().DefaultTimeout();

                    var simpleCompletion = Assert.IsType<CompletionMessage>(secondResult);
                    Assert.Equal(21L, simpleCompletion.Result);
                }

                // Release the long running hub method
                tcsService.EndMethod.TrySetResult(null);

                // Long running hub method result
                var firstResult = await client.ReadAsync().DefaultTimeout();

                var longRunningCompletion = Assert.IsType<CompletionMessage>(firstResult);
                Assert.Equal(12L, longRunningCompletion.Result);

                // Shut down
                client.Dispose();

                await connectionHandlerTask.DefaultTimeout();
            }
        }
    }

    [Fact]
    public async Task PendingInvocationUnblockedWhenBlockingMethodCompletesWithParallelism()
    {
        using (StartVerifiableLog())
        {
            var tcsService = new TcsService();
            var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(builder =>
            {
                builder.AddSingleton(tcsService);

                builder.AddSignalR(options =>
                {
                    options.MaximumParallelInvocationsPerClient = 2;
                });
            }, LoggerFactory);
            var connectionHandler = serviceProvider.GetService<HubConnectionHandler<LongRunningHub>>();

            // Because we use PipeScheduler.Inline the hub invocations will run inline until they wait, which happens inside the LongRunningMethod call
            using (var client = new TestClient())
            {
                var connectionHandlerTask = await client.ConnectAsync(connectionHandler).DefaultTimeout();

                // Long running hub invocation to test that other invocations will not run until it is completed
                await client.SendInvocationAsync(nameof(LongRunningHub.LongRunningMethod), nonBlocking: false).DefaultTimeout();
                // Wait for the long running method to start
                await tcsService.StartedMethod.Task.DefaultTimeout();
                // Grab the tcs before resetting to use in the second long running method
                var endTcs = tcsService.EndMethod;
                tcsService.Reset();

                // Long running hub invocation to test that other invocations will not run until it is completed
                await client.SendInvocationAsync(nameof(LongRunningHub.LongRunningMethod), nonBlocking: false).DefaultTimeout();
                // Wait for the long running method to start
                await tcsService.StartedMethod.Task.DefaultTimeout();

                // Invoke another hub method which will wait for the first method to finish
                await client.SendInvocationAsync(nameof(LongRunningHub.SimpleMethod), nonBlocking: false).DefaultTimeout();
                // Both invocations should be waiting now
                Assert.Null(client.TryRead());

                // Release the second long running hub method
                tcsService.EndMethod.TrySetResult(null);

                // Long running hub method result
                var firstResult = await client.ReadAsync().DefaultTimeout();

                var longRunningCompletion = Assert.IsType<CompletionMessage>(firstResult);
                Assert.Equal(12L, longRunningCompletion.Result);

                // simple hub method result
                var secondResult = await client.ReadAsync().DefaultTimeout();

                var simpleCompletion = Assert.IsType<CompletionMessage>(secondResult);
                Assert.Equal(21L, simpleCompletion.Result);

                // Release the first long running hub method
                endTcs.TrySetResult(null);

                firstResult = await client.ReadAsync().DefaultTimeout();
                longRunningCompletion = Assert.IsType<CompletionMessage>(firstResult);
                Assert.Equal(12L, longRunningCompletion.Result);

                // Shut down
                client.Dispose();

                await connectionHandlerTask.DefaultTimeout();
            }
        }
    }

    [Fact]
    public async Task StreamInvocationsDoNotBlockOtherInvocations()
    {
        using (StartVerifiableLog())
        {
            var tcsService = new TcsService();
            var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(builder =>
            {
                builder.AddSingleton(tcsService);
                builder.AddSingleton(typeof(IHubActivator<>), typeof(CustomHubActivator<>));

                builder.AddSignalR(options =>
                {
                    options.MaximumParallelInvocationsPerClient = 1;
                });
            }, LoggerFactory);
            var connectionHandler = serviceProvider.GetService<HubConnectionHandler<LongRunningHub>>();

            // Because we use PipeScheduler.Inline the hub invocations will run inline until they go async
            using (var client = new TestClient())
            {
                var connectionHandlerTask = await client.ConnectAsync(connectionHandler).DefaultTimeout();

                // Long running stream invocation to test that other invocations can still run before it is completed
                var streamInvocationId = await client.SendStreamInvocationAsync(nameof(LongRunningHub.LongRunningStream), null).DefaultTimeout();
                // Wait for the long running method to start
                await tcsService.StartedMethod.Task.DefaultTimeout();

                // Invoke another hub method which will be able to run even though a streaming method is still running
                var completion = await client.InvokeAsync(nameof(LongRunningHub.SimpleMethod)).DefaultTimeout();
                Assert.Null(completion.Error);
                Assert.Equal(21L, completion.Result);

                // Release the long running hub method
                tcsService.EndMethod.TrySetResult(null);

                var hubActivator = serviceProvider.GetService<IHubActivator<LongRunningHub>>() as CustomHubActivator<LongRunningHub>;

                await client.SendHubMessageAsync(new CancelInvocationMessage(streamInvocationId)).DefaultTimeout();

                // Completion message for canceled Stream
                completion = Assert.IsType<CompletionMessage>(await client.ReadAsync().DefaultTimeout());
                Assert.Equal(streamInvocationId, completion.InvocationId);

                // Shut down
                client.Dispose();

                await connectionHandlerTask.DefaultTimeout();

                // OnConnectedAsync, SimpleMethod, LongRunningStream, OnDisconnectedAsync
                Assert.Equal(4, hubActivator.ReleaseCount);
            }
        }
    }

    [Fact]
    public async Task ServerSendsCloseWithErrorWhenConnectionClosedWithPartialMessage()
    {
        using (StartVerifiableLog())
        {
            var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(services =>
            {
                services.AddSignalR(options => options.EnableDetailedErrors = true);
            }, LoggerFactory);
            var connectionHandler = serviceProvider.GetService<HubConnectionHandler<SimpleHub>>();

            using (var client = new TestClient())
            {
                var connectionHandlerTask = await client.ConnectAsync(connectionHandler).DefaultTimeout();

                await client.Connection.Application.Output.WriteAsync(Encoding.UTF8.GetBytes(new[] { '{' })).DefaultTimeout();

                // Close connection
                client.Connection.Application.Output.Complete();

                // Ignore message from OnConnectedAsync
                await client.ReadAsync().DefaultTimeout();

                var closeMessage = Assert.IsType<CloseMessage>(await client.ReadAsync().DefaultTimeout());

                Assert.Equal("Connection closed with an error. InvalidDataException: Connection terminated while reading a message.", closeMessage.Error);

                // Shut down
                client.Dispose();

                await connectionHandlerTask.DefaultTimeout();
            }
        }
    }

    [Fact]
    public async Task StreamUploadBufferCapacityBlocksOtherInvocations()
    {
        var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(services =>
        {
            services.Configure<HubOptions>(options =>
            {
                options.StreamBufferCapacity = 1;
            });
        });

        var connectionHandler = serviceProvider.GetService<HubConnectionHandler<MethodHub>>();

        using (var client = new TestClient())
        {
            var connectionHandlerTask = await client.ConnectAsync(connectionHandler).DefaultTimeout();
            await client.BeginUploadStreamAsync("invocationId", nameof(MethodHub.StreamDontRead), new[] { "id" }, Array.Empty<object>()).DefaultTimeout();

            foreach (var letter in new[] { "A", "B", "C", "D", "E" })
            {
                await client.SendHubMessageAsync(new StreamItemMessage("id", letter)).DefaultTimeout();
            }

            var ex = await Assert.ThrowsAsync<TimeoutException>(async () =>
            {
                await client.SendInvocationAsync("Echo", "test");
                var result = (CompletionMessage)await client.ReadAsync().DefaultTimeout(5000);
            });
        }
    }

    [Fact]
    public async Task UploadStringsToConcat()
    {
        var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider();
        var connectionHandler = serviceProvider.GetService<HubConnectionHandler<MethodHub>>();

        using (var client = new TestClient())
        {
            var connectionHandlerTask = await client.ConnectAsync(connectionHandler).DefaultTimeout();
            await client.BeginUploadStreamAsync("invocation", nameof(MethodHub.StreamingConcat), new[] { "id" }, Array.Empty<object>());

            foreach (var letter in new[] { "B", "E", "A", "N", "E", "D" })
            {
                await client.SendHubMessageAsync(new StreamItemMessage("id", letter)).DefaultTimeout();
            }

            await client.SendHubMessageAsync(CompletionMessage.Empty("id")).DefaultTimeout();
            var result = (CompletionMessage)await client.ReadAsync().DefaultTimeout();

            Assert.Equal("BEANED", result.Result);
        }
    }

    [Fact]
    public async Task UploadStreamedObjects()
    {
        var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider();
        var connectionHandler = serviceProvider.GetService<HubConnectionHandler<MethodHub>>();

        using (var client = new TestClient())
        {
            var connectionHandlerTask = await client.ConnectAsync(connectionHandler).DefaultTimeout();
            await client.BeginUploadStreamAsync("invocation", nameof(MethodHub.UploadArray), new[] { "id" }, Array.Empty<object>());

            var objects = new[] { new SampleObject("solo", 322), new SampleObject("ggez", 3145) };
            foreach (var thing in objects)
            {
                await client.SendHubMessageAsync(new StreamItemMessage("id", thing)).DefaultTimeout();
            }

            await client.SendHubMessageAsync(CompletionMessage.Empty("id")).DefaultTimeout();
            var response = (CompletionMessage)await client.ReadAsync().DefaultTimeout();
            var result = ((JArray)response.Result).ToArray<object>();

            Assert.Equal(objects[0].Foo, ((JContainer)result[0])["foo"]);
            Assert.Equal(objects[0].Bar, ((JContainer)result[0])["bar"]);
            Assert.Equal(objects[1].Foo, ((JContainer)result[1])["foo"]);
            Assert.Equal(objects[1].Bar, ((JContainer)result[1])["bar"]);
        }
    }

    [Fact]
    public async Task UploadManyStreams()
    {
        var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider();
        var connectionHandler = serviceProvider.GetService<HubConnectionHandler<MethodHub>>();

        using (var client = new TestClient())
        {
            var connectionHandlerTask = await client.ConnectAsync(connectionHandler).DefaultTimeout();
            var ids = new[] { "0", "1", "2" };

            foreach (string id in ids)
            {
                await client.BeginUploadStreamAsync("invocation_" + id, nameof(MethodHub.StreamingConcat), new[] { id }, Array.Empty<object>());
            }

            var words = new[] { "zygapophyses", "qwerty", "abcd" };
            var pos = new[] { 0, 0, 0 };
            var order = new[] { 2, 2, 0, 2, 1, 0, 0, 0, 0, 0, 0, 2, 0, 1, 0, 1, 0, 0, 0, 1, 1, 1 };

            foreach (var spot in order)
            {
                await client.SendHubMessageAsync(new StreamItemMessage(spot.ToString(CultureInfo.InvariantCulture), words[spot][pos[spot]])).DefaultTimeout();
                pos[spot] += 1;
            }

            foreach (string id in new[] { "0", "2", "1" })
            {
                await client.SendHubMessageAsync(CompletionMessage.Empty(id)).DefaultTimeout();
                var response = await client.ReadAsync().DefaultTimeout();
                Debug.Write(response);
                Assert.Equal(words[int.Parse(id, CultureInfo.InvariantCulture)], ((CompletionMessage)response).Result);
            }
        }
    }

    private class DelayRequirement : AuthorizationHandler<DelayRequirement, HubInvocationContext>, IAuthorizationRequirement
    {
        private readonly TcsService _tcsService;
        public DelayRequirement(TcsService tcsService)
        {
            _tcsService = tcsService;
        }

        protected override async Task HandleRequirementAsync(AuthorizationHandlerContext context, DelayRequirement requirement, HubInvocationContext resource)
        {
            _tcsService.StartedMethod.SetResult(null);
            await _tcsService.EndMethod.Task;
            context.Succeed(requirement);
        }
    }

    [Fact]
    // Test to check if StreamItems can be processed before the Stream from the invocation is properly registered internally
    public async Task UploadStreamStreamItemsSentAsSoonAsPossible()
    {
        // Use Auth as the delay injection point because it is one of the first things to run after the invocation message has been parsed
        var tcsService = new TcsService();
        var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(services =>
        {
            services.AddAuthorization(options =>
            {
                options.AddPolicy("test", policy =>
                {
                    policy.Requirements.Add(new DelayRequirement(tcsService));
                });
            });
        });
        var connectionHandler = serviceProvider.GetService<HubConnectionHandler<MethodHub>>();

        using (var client = new TestClient())
        {
            var connectionHandlerTask = await client.ConnectAsync(connectionHandler).DefaultTimeout();
            await client.BeginUploadStreamAsync("invocation", nameof(MethodHub.UploadArrayAuth), new[] { "id" }, Array.Empty<object>());
            await tcsService.StartedMethod.Task.DefaultTimeout();

            var objects = new[] { new SampleObject("solo", 322), new SampleObject("ggez", 3145) };
            foreach (var thing in objects)
            {
                await client.SendHubMessageAsync(new StreamItemMessage("id", thing)).DefaultTimeout();
            }

            tcsService.EndMethod.SetResult(null);

            await client.SendHubMessageAsync(CompletionMessage.Empty("id")).DefaultTimeout();
            var response = (CompletionMessage)await client.ReadAsync().DefaultTimeout();
            var result = ((JArray)response.Result).ToArray<object>();

            Assert.Equal(objects[0].Foo, ((JContainer)result[0])["foo"]);
            Assert.Equal(objects[0].Bar, ((JContainer)result[0])["bar"]);
            Assert.Equal(objects[1].Foo, ((JContainer)result[1])["foo"]);
            Assert.Equal(objects[1].Bar, ((JContainer)result[1])["bar"]);
        }
    }

    [Fact]
    public async Task UploadStreamDoesNotCountTowardsMaxInvocationLimit()
    {
        var tcsService = new TcsService();
        var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(services =>
        {
            services.AddSignalR(options => options.MaximumParallelInvocationsPerClient = 1);
            services.AddSingleton(tcsService);
        });
        var connectionHandler = serviceProvider.GetService<HubConnectionHandler<LongRunningHub>>();

        using (var client = new TestClient())
        {
            var connectionHandlerTask = await client.ConnectAsync(connectionHandler).DefaultTimeout();
            await client.BeginUploadStreamAsync("invocation", nameof(LongRunningHub.Upload), new[] { "id" }, Array.Empty<object>());
            await tcsService.StartedMethod.Task.DefaultTimeout();

            var completion = await client.InvokeAsync(nameof(LongRunningHub.SimpleMethod)).DefaultTimeout();
            Assert.Null(completion.Error);
            Assert.Equal(21L, completion.Result);

            await client.SendHubMessageAsync(CompletionMessage.Empty("id")).DefaultTimeout();

            await tcsService.EndMethod.Task.DefaultTimeout();
            var response = (CompletionMessage)await client.ReadAsync().DefaultTimeout();
            Assert.Null(response.Result);
            Assert.Null(response.Error);
        }
    }

    [Fact]
    public async Task ConnectionAbortedIfSendFailsWithProtocolError()
    {
        using (StartVerifiableLog(write => write.EventId.Name == "FailedWritingMessage"))
        {
            var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(services =>
            {
                services.AddSignalR(options => options.EnableDetailedErrors = true);
            }, LoggerFactory);
            var connectionHandler = serviceProvider.GetService<HubConnectionHandler<MethodHub>>();

            using (var client = new TestClient())
            {
                var connectionHandlerTask = await client.ConnectAsync(connectionHandler).DefaultTimeout();

                await client.SendInvocationAsync(nameof(MethodHub.ProtocolError)).DefaultTimeout();
                await connectionHandlerTask.DefaultTimeout();
            }
        }
    }

    [Fact]
    public async Task SerializationExceptionsSendSelfArePassedToOnDisconnectedAsync()
    {
        using (StartVerifiableLog(write => write.EventId.Name == "FailedWritingMessage"))
        {
            var state = new ConnectionLifetimeState();
            var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(s => s.AddSingleton(state), LoggerFactory);
            var connectionHandler = serviceProvider.GetService<HubConnectionHandler<ConnectionLifetimeHub>>();

            using (var client = new TestClient())
            {
                var connectionHandlerTask = await client.ConnectAsync(connectionHandler);

                // Test HubConnectionContext.WriteCore(HubMessage) codepath
                await client.SendInvocationAsync(nameof(ConnectionLifetimeHub.ProtocolErrorSelf)).DefaultTimeout();

                await connectionHandlerTask.DefaultTimeout();

                Assert.IsType<System.Text.Json.JsonException>(state.DisconnectedException);
            }
        }
    }

    [Fact]
    public async Task SerializationExceptionsSendAllArePassedToOnDisconnectedAsync()
    {
        using (StartVerifiableLog(write => write.EventId.Name == "FailedWritingMessage"))
        {
            var state = new ConnectionLifetimeState();
            var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(s => s.AddSingleton(state), LoggerFactory);
            var connectionHandler = serviceProvider.GetService<HubConnectionHandler<ConnectionLifetimeHub>>();

            using (var client = new TestClient())
            {
                var connectionHandlerTask = await client.ConnectAsync(connectionHandler);

                // Test HubConnectionContext.WriteCore(SerializedHubMessage) codepath
                await client.SendInvocationAsync(nameof(ConnectionLifetimeHub.ProtocolErrorAll)).DefaultTimeout();

                await connectionHandlerTask.DefaultTimeout();

                Assert.IsType<System.Text.Json.JsonException>(state.DisconnectedException);
            }
        }
    }

    [Fact(Skip = "Magic auto cast not supported")]
    public async Task UploadStreamItemInvalidTypeAutoCasts()
    {
        using (StartVerifiableLog())
        {
            // NOTE -- json.net is flexible here, and casts for us

            var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(loggerFactory: LoggerFactory);
            var connectionHandler = serviceProvider.GetService<HubConnectionHandler<MethodHub>>();

            using (var client = new TestClient())
            {
                var connectionHandlerTask = await client.ConnectAsync(connectionHandler).DefaultTimeout();
                await client.BeginUploadStreamAsync("invocation", nameof(MethodHub.StreamingConcat), streamIds: new[] { "id" }, Array.Empty<object>()).DefaultTimeout();

                // send integers that are then cast to strings
                await client.SendHubMessageAsync(new StreamItemMessage("id", 5)).DefaultTimeout();
                await client.SendHubMessageAsync(new StreamItemMessage("id", 10)).DefaultTimeout();

                await client.SendHubMessageAsync(CompletionMessage.Empty("id")).DefaultTimeout();
                var response = (CompletionMessage)await client.ReadAsync().DefaultTimeout();

                Assert.Null(response.Error);
                Assert.Equal("510", response.Result);
            }
        }
    }

    [Fact]
    public async Task ServerReportsProtocolMinorVersion()
    {
        using (StartVerifiableLog())
        {
            var testProtocol = new Mock<IHubProtocol>();
            testProtocol.Setup(m => m.Name).Returns("CustomProtocol");
            testProtocol.Setup(m => m.IsVersionSupported(It.IsAny<int>())).Returns(true);
            testProtocol.Setup(m => m.TransferFormat).Returns(TransferFormat.Binary);

            var connectionHandler = HubConnectionHandlerTestUtils.GetHubConnectionHandler(typeof(HubT),
                LoggerFactory, (services) => services.AddSingleton<IHubProtocol>(testProtocol.Object));

            using (var client = new TestClient(protocol: testProtocol.Object))
            {
                var connectionHandlerTask = await client.ConnectAsync(connectionHandler).DefaultTimeout();

                Assert.NotNull(client.HandshakeResponseMessage);

                client.Dispose();
                await connectionHandlerTask.DefaultTimeout();
            }
        }
    }

    [Fact]
    public async Task UploadStreamItemInvalidType()
    {
        using (StartVerifiableLog())
        {
            var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(loggerFactory: LoggerFactory);
            var connectionHandler = serviceProvider.GetService<HubConnectionHandler<MethodHub>>();

            using (var client = new TestClient())
            {
                var connectionHandlerTask = await client.ConnectAsync(connectionHandler).DefaultTimeout();
                await client.BeginUploadStreamAsync("invocationId", nameof(MethodHub.TestTypeCastingErrors), new[] { "channelId" }, Array.Empty<object>()).DefaultTimeout();

                // client is running wild, sending strings not ints.
                // this error should be propogated to the user's HubMethod code
                await client.SendHubMessageAsync(new StreamItemMessage("channelId", "not a number")).DefaultTimeout();
                var response = await client.ReadAsync().DefaultTimeout();

                Assert.Equal(typeof(CompletionMessage), response.GetType());
                Assert.Equal("error identified and caught", (string)((CompletionMessage)response).Result);
            }
        }
    }

    [Fact]
    public async Task UploadStreamItemInvalidId()
    {
        using (StartVerifiableLog())
        {
            var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(services =>
            {
                services.AddSignalR(options => options.EnableDetailedErrors = true);
            }, loggerFactory: LoggerFactory);
            var connectionHandler = serviceProvider.GetService<HubConnectionHandler<MethodHub>>();

            using (var client = new TestClient())
            {
                var connectionHandlerTask = await client.ConnectAsync(connectionHandler).DefaultTimeout();
                await client.SendHubMessageAsync(new StreamItemMessage("fake_id", "not a number")).DefaultTimeout();

                var message = client.TryRead();
                Assert.Null(message);
            }
        }

        Assert.Single(TestSink.Writes.Where(w => w.LoggerName == "Microsoft.AspNetCore.SignalR.Internal.DefaultHubDispatcher" &&
            w.EventId.Name == "ClosingStreamWithBindingError"));
    }

    [Fact]
    public async Task UploadStreamCompleteInvalidId()
    {
        using (StartVerifiableLog())
        {
            var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(services =>
            {
                services.AddSignalR(options => options.EnableDetailedErrors = true);
            }, loggerFactory: LoggerFactory);
            var connectionHandler = serviceProvider.GetService<HubConnectionHandler<MethodHub>>();

            using (var client = new TestClient())
            {
                var connectionHandlerTask = await client.ConnectAsync(connectionHandler).DefaultTimeout();
                await client.SendHubMessageAsync(CompletionMessage.Empty("fake_id")).DefaultTimeout();

                var message = client.TryRead();
                Assert.Null(message);
            }
        }

        Assert.Single(TestSink.Writes.Where(w => w.LoggerName == "Microsoft.AspNetCore.SignalR.Internal.DefaultHubDispatcher" &&
            w.EventId.Name == "UnexpectedCompletion"));
    }

    public static string CustomErrorMessage = "custom error for testing ::::)";

    [Fact]
    public async Task UploadStreamCompleteWithError()
    {
        using (StartVerifiableLog())
        {
            var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(loggerFactory: LoggerFactory);
            var connectionHandler = serviceProvider.GetService<HubConnectionHandler<MethodHub>>();

            using (var client = new TestClient())
            {
                await client.ConnectAsync(connectionHandler).DefaultTimeout();
                await client.BeginUploadStreamAsync("invocation", nameof(MethodHub.TestCustomErrorPassing), streamIds: new[] { "id" }, args: Array.Empty<object>()).DefaultTimeout();
                await client.SendHubMessageAsync(CompletionMessage.WithError("id", CustomErrorMessage)).DefaultTimeout();

                var response = (CompletionMessage)await client.ReadAsync().DefaultTimeout();
                Assert.True((bool)response.Result);
            }
        }
    }

    [Fact]
    public async Task UploadStreamWithTooManyStreamsFails()
    {
        using (StartVerifiableLog())
        {
            var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(loggerFactory: LoggerFactory);
            var connectionHandler = serviceProvider.GetService<HubConnectionHandler<MethodHub>>();

            using (var client = new TestClient())
            {
                await client.ConnectAsync(connectionHandler).DefaultTimeout();
                await client.BeginUploadStreamAsync("invocation", nameof(MethodHub.StreamingConcat), streamIds: new[] { "id", "id2" }, args: Array.Empty<object>()).DefaultTimeout();

                var response = (CompletionMessage)await client.ReadAsync().DefaultTimeout();
                Assert.Equal("An unexpected error occurred invoking 'StreamingConcat' on the server. HubException: Client sent 2 stream(s), Hub method expects 1.", response.Error);
            }
        }
    }

    [Fact]
    public async Task UploadStreamWithTooFewStreamsFails()
    {
        using (StartVerifiableLog())
        {
            var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(loggerFactory: LoggerFactory);
            var connectionHandler = serviceProvider.GetService<HubConnectionHandler<MethodHub>>();

            using (var client = new TestClient())
            {
                await client.ConnectAsync(connectionHandler).DefaultTimeout();
                await client.BeginUploadStreamAsync("invocation", nameof(MethodHub.StreamingConcat), streamIds: Array.Empty<string>(), args: Array.Empty<object>()).DefaultTimeout();

                var response = (CompletionMessage)await client.ReadAsync().DefaultTimeout();
                Assert.Equal("An unexpected error occurred invoking 'StreamingConcat' on the server. HubException: Client sent 0 stream(s), Hub method expects 1.", response.Error);
            }
        }
    }

    [Fact]
    public async Task UploadStreamReleasesHubActivatorOnceComplete()
    {
        using (StartVerifiableLog())
        {
            var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(builder =>
            {
                builder.AddSingleton(typeof(IHubActivator<>), typeof(CustomHubActivator<>));
            }, LoggerFactory);
            var connectionHandler = serviceProvider.GetService<HubConnectionHandler<MethodHub>>();

            using (var client = new TestClient())
            {
                var connectionHandlerTask = await client.ConnectAsync(connectionHandler).DefaultTimeout();

                await client.BeginUploadStreamAsync("invocation", nameof(MethodHub.StreamingConcat), streamIds: new[] { "id" }, args: Array.Empty<object>()).DefaultTimeout();

                await client.SendHubMessageAsync(new StreamItemMessage("id", "hello")).DefaultTimeout();
                await client.SendHubMessageAsync(new StreamItemMessage("id", " world")).DefaultTimeout();
                await client.SendHubMessageAsync(CompletionMessage.Empty("id")).DefaultTimeout();
                var result = await client.ReadAsync().DefaultTimeout();

                var simpleCompletion = Assert.IsType<CompletionMessage>(result);
                Assert.Equal("hello world", simpleCompletion.Result);

                var hubActivator = serviceProvider.GetService<IHubActivator<MethodHub>>() as CustomHubActivator<MethodHub>;

                // OnConnectedAsync and StreamingConcat hubs have been disposed
                Assert.Equal(2, hubActivator.ReleaseCount);

                // Shut down
                client.Dispose();

                await connectionHandlerTask.DefaultTimeout();
            }
        }
    }

    [Fact]
    public async Task UploadStreamFromSendReleasesHubActivatorOnceComplete()
    {
        using (StartVerifiableLog())
        {
            var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(builder =>
            {
                builder.AddSingleton(typeof(IHubActivator<>), typeof(CustomHubActivator<>));
            }, LoggerFactory);
            var connectionHandler = serviceProvider.GetService<HubConnectionHandler<MethodHub>>();

            using (var client = new TestClient())
            {
                var connectionHandlerTask = await client.ConnectAsync(connectionHandler).DefaultTimeout();

                var hubActivator = serviceProvider.GetService<IHubActivator<MethodHub>>() as CustomHubActivator<MethodHub>;
                var createTask = hubActivator.CreateTask.Task;

                // null ID means we're doing a Send and not an Invoke
                await client.BeginUploadStreamAsync(invocationId: null, nameof(MethodHub.StreamingConcat), streamIds: new[] { "id" }, args: Array.Empty<object>()).DefaultTimeout();
                await client.SendHubMessageAsync(new StreamItemMessage("id", "hello")).DefaultTimeout();
                await client.SendHubMessageAsync(new StreamItemMessage("id", " world")).DefaultTimeout();

                await createTask.DefaultTimeout();
                var tcs = hubActivator.ReleaseTask;
                await client.SendHubMessageAsync(CompletionMessage.Empty("id")).DefaultTimeout();

                await tcs.Task.DefaultTimeout();

                // OnConnectedAsync and StreamingConcat hubs have been disposed
                Assert.Equal(2, hubActivator.ReleaseCount);

                // Shut down
                client.Dispose();

                await connectionHandlerTask.DefaultTimeout();
            }
        }
    }

    [Fact]
    public async Task UploadStreamClosesStreamsOnServerWhenMethodCompletes()
    {
        using (StartVerifiableLog())
        {
            var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(loggerFactory: LoggerFactory);
            var connectionHandler = serviceProvider.GetService<HubConnectionHandler<MethodHub>>();

            using (var client = new TestClient())
            {
                var connectionHandlerTask = await client.ConnectAsync(connectionHandler).DefaultTimeout();

                await client.BeginUploadStreamAsync("invocation", nameof(MethodHub.UploadIgnoreItems), streamIds: new[] { "id" }, args: Array.Empty<object>()).DefaultTimeout();

                await client.SendHubMessageAsync(new StreamItemMessage("id", "ignored")).DefaultTimeout();
                var result = await client.ReadAsync().DefaultTimeout();

                var simpleCompletion = Assert.IsType<CompletionMessage>(result);
                Assert.Null(simpleCompletion.Result);

                // This will log a warning on the server as the hub method has completed and will complete all associated streams
                await client.SendHubMessageAsync(new StreamItemMessage("id", "error!")).DefaultTimeout();

                // Check that the connection hasn't been closed
                await client.SendInvocationAsync("VoidMethod").DefaultTimeout();

                // Shut down
                client.Dispose();

                await connectionHandlerTask.DefaultTimeout();
            }
        }

        Assert.Single(TestSink.Writes.Where(w => w.LoggerName == "Microsoft.AspNetCore.SignalR.Internal.DefaultHubDispatcher" &&
            w.EventId.Name == "ClosingStreamWithBindingError"));
    }

    [Fact]
    public async Task UploadStreamAndStreamingMethodClosesStreamsOnServerWhenMethodCompletes()
    {
        using (StartVerifiableLog())
        {
            var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(loggerFactory: LoggerFactory);
            var connectionHandler = serviceProvider.GetService<HubConnectionHandler<MethodHub>>();

            using (var client = new TestClient())
            {
                var connectionHandlerTask = await client.ConnectAsync(connectionHandler).DefaultTimeout();

                await client.SendStreamInvocationAsync(nameof(MethodHub.StreamAndUploadIgnoreItems), streamIds: new[] { "id" }, args: Array.Empty<object>()).DefaultTimeout();

                await client.SendHubMessageAsync(new StreamItemMessage("id", "ignored")).DefaultTimeout();
                var result = await client.ReadAsync().DefaultTimeout();

                var simpleCompletion = Assert.IsType<CompletionMessage>(result);
                Assert.Null(simpleCompletion.Result);

                // This will log a warning on the server as the hub method has completed and will complete all associated streams
                await client.SendHubMessageAsync(new StreamItemMessage("id", "error!")).DefaultTimeout();

                // Check that the connection hasn't been closed
                await client.SendInvocationAsync("VoidMethod").DefaultTimeout();

                // Shut down
                client.Dispose();

                await connectionHandlerTask.DefaultTimeout();
            }
        }

        Assert.Single(TestSink.Writes.Where(w => w.LoggerName == "Microsoft.AspNetCore.SignalR.Internal.DefaultHubDispatcher" &&
            w.EventId.Name == "ClosingStreamWithBindingError"));
    }

    [Theory]
    [InlineData(nameof(LongRunningHub.CancelableStreamSingleParameter))]
    [InlineData(nameof(LongRunningHub.CancelableStreamMultiParameter), 1, 2)]
    [InlineData(nameof(LongRunningHub.CancelableStreamMiddleParameter), 1, 2)]
    [InlineData(nameof(LongRunningHub.CancelableStreamGeneratedAsyncEnumerable))]
    [InlineData(nameof(LongRunningHub.CancelableStreamCustomAsyncEnumerable))]
    public async Task StreamHubMethodCanBeTriggeredOnCancellation(string methodName, params object[] args)
    {
        using (StartVerifiableLog())
        {
            var tcsService = new TcsService();
            var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(builder =>
            {
                builder.AddSingleton(tcsService);
            }, LoggerFactory);
            var connectionHandler = serviceProvider.GetService<HubConnectionHandler<LongRunningHub>>();

            using (var client = new TestClient())
            {
                var connectionHandlerTask = await client.ConnectAsync(connectionHandler).DefaultTimeout();

                var streamInvocationId = await client.SendStreamInvocationAsync(methodName, args).DefaultTimeout();
                // Wait for the stream method to start
                await tcsService.StartedMethod.Task.DefaultTimeout();

                // Cancel the stream which should trigger the CancellationToken in the hub method
                await client.SendHubMessageAsync(new CancelInvocationMessage(streamInvocationId)).DefaultTimeout();

                var result = await client.ReadAsync().DefaultTimeout();

                var simpleCompletion = Assert.IsType<CompletionMessage>(result);
                Assert.Null(simpleCompletion.Result);

                // CancellationToken passed to hub method will allow EndMethod to be triggered if it is canceled.
                await tcsService.EndMethod.Task.DefaultTimeout();

                // Shut down
                client.Dispose();

                await connectionHandlerTask.DefaultTimeout();
            }
        }
    }

    [Theory]
    [InlineData(nameof(LongRunningHub.CountingCancelableStreamGeneratedAsyncEnumerable), 2)]
    [InlineData(nameof(LongRunningHub.CountingCancelableStreamGeneratedChannel), 2)]
    public async Task CancellationAfterGivenMessagesEndsStreaming(string methodName, int count)
    {
        using (StartVerifiableLog())
        {
            var tcsService = new TcsService();
            var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(builder =>
            {
                builder.AddSingleton(tcsService);
            }, LoggerFactory);
            var connectionHandler = serviceProvider.GetService<HubConnectionHandler<LongRunningHub>>();
            var invocationBinder = new Mock<IInvocationBinder>();
            invocationBinder.Setup(b => b.GetStreamItemType(It.IsAny<string>())).Returns(typeof(string));

            using (var client = new TestClient(invocationBinder: invocationBinder.Object))
            {
                var connectionHandlerTask = await client.ConnectAsync(connectionHandler).DefaultTimeout();

                // Start streaming count number of messages.
                var invocationId = await client.SendStreamInvocationAsync(methodName, count).DefaultTimeout();

                // Listening on incoming messages
                var listeningMessages = client.ListenAsync(invocationId);

                // Wait for the number of messages expected to be received. This point the sender just waits forever or until cancellation.
                await listeningMessages.ReadAsync(count).DefaultTimeout();

                // Send cancellation.
                await client.SendHubMessageAsync(new CancelInvocationMessage(invocationId)).DefaultTimeout();

                // Wait for the completion message.
                var messages = await listeningMessages.ReadAllAsync().DefaultTimeout();
                Assert.Single(messages);

                // CancellationToken passed to hub method will allow EndMethod to be triggered if it is canceled.
                await tcsService.EndMethod.Task.DefaultTimeout();

                // Shut down
                client.Dispose();

                await connectionHandlerTask.DefaultTimeout();
            }
        }
    }

    [Fact]
    public async Task StreamHubMethodCanAcceptCancellationTokenAsArgumentAndBeTriggeredOnConnectionAborted()
    {
        using (StartVerifiableLog())
        {
            var tcsService = new TcsService();
            var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(builder =>
            {
                builder.AddSingleton(tcsService);
            }, LoggerFactory);
            var connectionHandler = serviceProvider.GetService<HubConnectionHandler<LongRunningHub>>();

            using (var client = new TestClient())
            {
                var connectionHandlerTask = await client.ConnectAsync(connectionHandler).DefaultTimeout();

                var streamInvocationId = await client.SendStreamInvocationAsync(nameof(LongRunningHub.CancelableStreamSingleParameter)).DefaultTimeout();
                // Wait for the stream method to start
                await tcsService.StartedMethod.Task.DefaultTimeout();

                // Shut down the client which should trigger the CancellationToken in the hub method
                client.Dispose();

                // CancellationToken passed to hub method will allow EndMethod to be triggered if it is canceled.
                await tcsService.EndMethod.Task.DefaultTimeout();

                await connectionHandlerTask.DefaultTimeout();
            }
        }
    }

    [Fact]
    public async Task StreamHubMethodCanAcceptNullableParameter()
    {
        using (StartVerifiableLog())
        {
            var tcsService = new TcsService();
            var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(builder =>
            {
                builder.AddSingleton(tcsService);
            }, LoggerFactory);
            var connectionHandler = serviceProvider.GetService<HubConnectionHandler<LongRunningHub>>();

            using (var client = new TestClient())
            {
                var connectionHandlerTask = await client.ConnectAsync(connectionHandler).DefaultTimeout();

                var streamInvocationId = await client.SendStreamInvocationAsync(nameof(LongRunningHub.StreamNullableParameter), 5, null).DefaultTimeout();
                // Wait for the stream method to start
                var firstArgument = await tcsService.StartedMethod.Task.DefaultTimeout();
                Assert.Equal(5, firstArgument);

                var secondArgument = await tcsService.EndMethod.Task.DefaultTimeout();
                Assert.Null(secondArgument);
            }
        }
    }

    [Fact]
    public async Task StreamHubMethodCanAcceptNullableParameterWithCancellationToken()
    {
        using (StartVerifiableLog())
        {
            var tcsService = new TcsService();
            var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(builder =>
            {
                builder.AddSingleton(tcsService);
            }, LoggerFactory);
            var connectionHandler = serviceProvider.GetService<HubConnectionHandler<LongRunningHub>>();

            using (var client = new TestClient())
            {
                var connectionHandlerTask = await client.ConnectAsync(connectionHandler).DefaultTimeout();

                var streamInvocationId = await client.SendStreamInvocationAsync(nameof(LongRunningHub.CancelableStreamNullableParameter), 5, null).DefaultTimeout();
                // Wait for the stream method to start
                var firstArgument = await tcsService.StartedMethod.Task.DefaultTimeout();
                Assert.Equal(5, firstArgument);

                // Cancel the stream which should trigger the CancellationToken in the hub method
                await client.SendHubMessageAsync(new CancelInvocationMessage(streamInvocationId)).DefaultTimeout();

                var secondArgument = await tcsService.EndMethod.Task.DefaultTimeout();
                Assert.Null(secondArgument);
            }
        }
    }

    [Fact]
    public async Task InvokeHubMethodCannotAcceptCancellationTokenAsArgument()
    {
        using (StartVerifiableLog())
        {
            var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(null, LoggerFactory);
            var connectionHandler = serviceProvider.GetService<HubConnectionHandler<MethodHub>>();

            using (var client = new TestClient())
            {
                var connectionHandlerTask = await client.ConnectAsync(connectionHandler).DefaultTimeout();

                var invocationId = await client.SendInvocationAsync(nameof(MethodHub.InvalidArgument)).DefaultTimeout();

                var completion = Assert.IsType<CompletionMessage>(await client.ReadAsync().DefaultTimeout());

                Assert.Equal("Failed to invoke 'InvalidArgument' due to an error on the server.", completion.Error);

                client.Dispose();

                await connectionHandlerTask.DefaultTimeout();
            }
        }
    }

    [Fact]
    public async Task CanPassStreamingParameterToStreamHubMethod()
    {
        using (StartVerifiableLog())
        {
            IServiceProvider serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(loggerFactory: LoggerFactory);
            HubConnectionHandler<StreamingHub> connectionHandler = serviceProvider.GetService<HubConnectionHandler<StreamingHub>>();
            Mock<IInvocationBinder> invocationBinder = new Mock<IInvocationBinder>();
            invocationBinder.Setup(b => b.GetStreamItemType(It.IsAny<string>())).Returns(typeof(string));

            using (TestClient client = new TestClient(invocationBinder: invocationBinder.Object))
            {
                Task connectionHandlerTask = await client.ConnectAsync(connectionHandler);

                // Wait for a connection, or for the endpoint to fail.
                await client.Connected.OrThrowIfOtherFails(connectionHandlerTask).DefaultTimeout();

                var streamId = "sample_id";
                var messagePromise = client.StreamAsync(nameof(StreamingHub.StreamEcho), new[] { streamId }, Array.Empty<object>()).DefaultTimeout();

                var phrases = new[] { "asdf", "qwer", "zxcv" };
                foreach (var phrase in phrases)
                {
                    await client.SendHubMessageAsync(new StreamItemMessage(streamId, phrase));
                }
                await client.SendHubMessageAsync(CompletionMessage.Empty(streamId));

                var messages = await messagePromise;

                // add one because this includes the completion
                Assert.Equal(phrases.Length + 1, messages.Count);
                for (var i = 0; i < phrases.Length; i++)
                {
                    Assert.Equal("echo:" + phrases[i], ((StreamItemMessage)messages[i]).Item);
                }

                client.Dispose();

                await connectionHandlerTask.DefaultTimeout();
            }
        }
    }

    /// <summary>
    /// Hub methods might be written by users in a way that accepts an interface or base class as a parameter
    /// and deserialization could supply a derived class.
    /// This test ensures implementation and subclass arguments are correctly bound for dispatch.
    /// </summary>
    [Theory]
    [InlineData(nameof(StreamingHub.DerivedParameterInterfaceAsyncEnumerable))]
    [InlineData(nameof(StreamingHub.DerivedParameterBaseClassAsyncEnumerable))]
    [InlineData(nameof(StreamingHub.DerivedParameterInterfaceAsyncEnumerableWithCancellation))]
    [InlineData(nameof(StreamingHub.DerivedParameterBaseClassAsyncEnumerableWithCancellation))]
    public async Task CanPassDerivedParameterToStreamHubMethod(string method)
    {
        using (StartVerifiableLog())
        {
            var argument = new StreamingHub.DerivedParameterTestObject { Value = "test" };
            var protocolOptions = new NewtonsoftJsonHubProtocolOptions
            {
                PayloadSerializerSettings = new JsonSerializerSettings()
                {
                    // The usage of TypeNameHandling.All is a security risk.
                    // If you're implementing this in your own application instead use your own 'type' field and a custom JsonConverter
                    // or ensure you're restricting to only known types with a custom SerializationBinder like we are here.
                    // See https://github.com/dotnet/aspnetcore/issues/11495#issuecomment-505047422
                    TypeNameHandling = TypeNameHandling.All,
                    SerializationBinder = StreamingHub.DerivedParameterKnownTypesBinder.Instance
                }
            };
            var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(
                services => services.AddSignalR()
                    .AddNewtonsoftJsonProtocol(o => o.PayloadSerializerSettings = protocolOptions.PayloadSerializerSettings),
                LoggerFactory);
            var connectionHandler = serviceProvider.GetService<HubConnectionHandler<StreamingHub>>();
            var invocationBinder = new Mock<IInvocationBinder>();
            invocationBinder.Setup(b => b.GetStreamItemType(It.IsAny<string>())).Returns(typeof(string));

            using (var client = new TestClient(
                protocol: new NewtonsoftJsonHubProtocol(Options.Create(protocolOptions)),
                invocationBinder: invocationBinder.Object))
            {
                var connectionHandlerTask = await client.ConnectAsync(connectionHandler);

                // Wait for a connection, or for the endpoint to fail.
                await client.Connected.OrThrowIfOtherFails(connectionHandlerTask).DefaultTimeout();

                var messages = await client.StreamAsync(method, argument).DefaultTimeout();

                Assert.Equal(2, messages.Count);
                HubConnectionHandlerTestUtils.AssertHubMessage(new StreamItemMessage(string.Empty, argument.Value), messages[0]);
                HubConnectionHandlerTestUtils.AssertHubMessage(CompletionMessage.Empty(string.Empty), messages[1]);

                client.Dispose();

                await connectionHandlerTask.DefaultTimeout();
            }
        }
    }

    [Fact]
    public async Task ClientsCallerPropertyCanBeUsedOutsideOfHub()
    {
        CallerService callerService = new CallerService();
        var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(services =>
        {
            services.AddSingleton(callerService);
        });
        var connectionHandler = serviceProvider.GetService<HubConnectionHandler<CallerServiceHub>>();

        using (StartVerifiableLog())
        {
            using (var client = new TestClient())
            {
                var connectionHandlerTask = await client.ConnectAsync(connectionHandler);

                // Wait for a connection, or for the endpoint to fail.
                await client.Connected.OrThrowIfOtherFails(connectionHandlerTask).DefaultTimeout();

                await callerService.Caller.SendAsync("Echo", "message").DefaultTimeout();

                var message = Assert.IsType<InvocationMessage>(await client.ReadAsync().DefaultTimeout());

                Assert.Equal("Echo", message.Target);
                Assert.Equal("message", message.Arguments[0]);
            }
        }
    }

    [Fact]
    public async Task CanSendThroughIHubContext()
    {
        using (StartVerifiableLog())
        {
            var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(null, LoggerFactory);
            var connectionHandler = serviceProvider.GetService<HubConnectionHandler<MethodHub>>();

            using var client = new TestClient();

            var connectionHandlerTask = await client.ConnectAsync(connectionHandler);

            // Wait for a connection, or for the endpoint to fail.
            await client.Connected.OrThrowIfOtherFails(connectionHandlerTask).DefaultTimeout();

            IHubContext context = (IHubContext)serviceProvider.GetRequiredService<IHubContext<MethodHub>>();
            await context.Clients.All.SendAsync("Send", "test");

            var message = await client.ReadAsync().DefaultTimeout();
            var invocation = Assert.IsType<InvocationMessage>(message);

            Assert.Single(invocation.Arguments);
            Assert.Equal("test", invocation.Arguments[0]);
            Assert.Equal("Send", invocation.Target);
        }
    }

    [Fact]
    public async Task ConnectionCloseCleansUploadStreams()
    {
        var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider();
        var connectionHandler = serviceProvider.GetService<HubConnectionHandler<MethodHub>>();

        using (StartVerifiableLog())
        {
            using var client = new TestClient();

            var connectionHandlerTask = await client.ConnectAsync(connectionHandler);

            // Wait for a connection, or for the endpoint to fail.
            await client.Connected.OrThrowIfOtherFails(connectionHandlerTask).DefaultTimeout();

            await client.BeginUploadStreamAsync("invocation", nameof(MethodHub.UploadDoesWorkOnComplete), streamIds: new[] { "id" }, args: Array.Empty<object>()).DefaultTimeout();

            await client.SendHubMessageAsync(new StreamItemMessage("id", "hello")).DefaultTimeout();

            await client.DisposeAsync().DefaultTimeout();

            await connectionHandlerTask.DefaultTimeout();

            // This task completes if the upload stream is completed, via closing the connection
            var task = (Task<int>)client.Connection.Items[nameof(MethodHub.UploadDoesWorkOnComplete)];

            var exception = await Assert.ThrowsAsync<OperationCanceledException>(() => task).DefaultTimeout();
            Assert.Equal("The underlying connection was closed.", exception.Message);
        }
    }

    [Fact]
    public async Task SpecificHubOptionForMaximumReceiveMessageSizeIsUsedOverGlobalHubOption()
    {
        var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(serviceBuilder =>
        {
            serviceBuilder.AddSignalR(o =>
            {
                // ConnectAsync would fail if this value was used
                o.MaximumReceiveMessageSize = 1;
            }).AddHubOptions<MethodHub>(o =>
            {
                // null is treated as both no-limit and not set, this test verifies that we track if the user explicitly sets the value
                o.MaximumReceiveMessageSize = null;
            });
        });
        var connectionHandler = serviceProvider.GetService<HubConnectionHandler<MethodHub>>();

        using (StartVerifiableLog())
        {
            using var client = new TestClient();

            var connectionHandlerTask = await client.ConnectAsync(connectionHandler);

            // Wait for a connection, or for the endpoint to fail.
            await client.Connected.OrThrowIfOtherFails(connectionHandlerTask).DefaultTimeout();

            await client.DisposeAsync().DefaultTimeout();

            await connectionHandlerTask.DefaultTimeout();
        }
    }

    [Fact]
    public async Task CanSendThroughIHubContextBaseHub()
    {
        using (StartVerifiableLog())
        {
            var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(null, LoggerFactory);
            var connectionHandler = serviceProvider.GetService<HubConnectionHandler<MethodHub>>();

            using var client = new TestClient();

            var connectionHandlerTask = await client.ConnectAsync(connectionHandler);

            // Wait for a connection, or for the endpoint to fail.
            await client.Connected.OrThrowIfOtherFails(connectionHandlerTask).DefaultTimeout();

            IHubContext<TestHub> context = serviceProvider.GetRequiredService<IHubContext<MethodHub>>();
            await context.Clients.All.SendAsync("Send", "test");

            var message = await client.ReadAsync().DefaultTimeout();
            var invocation = Assert.IsType<InvocationMessage>(message);

            Assert.Single(invocation.Arguments);
            Assert.Equal("test", invocation.Arguments[0]);
            Assert.Equal("Send", invocation.Target);
        }
    }

    [Fact]
    public async Task HubMethodFailsIfServiceNotFound()
    {
        var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(provider =>
        {
            provider.AddSignalR(o => o.EnableDetailedErrors = true);
        });
        var connectionHandler = serviceProvider.GetService<HubConnectionHandler<ServicesHub>>();

        using (var client = new TestClient())
        {
            var connectionHandlerTask = await client.ConnectAsync(connectionHandler).DefaultTimeout();
            var res = await client.InvokeAsync(nameof(ServicesHub.SingleService)).DefaultTimeout();
            Assert.Equal("An unexpected error occurred invoking 'SingleService' on the server. InvalidOperationException: No service for type 'Microsoft.AspNetCore.SignalR.Tests.Service1' has been registered.", res.Error);
        }
    }

    [Fact]
    public async Task HubMethodCanInjectService()
    {
        var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(provider =>
        {
            provider.AddSingleton<Service1>();
        });
        var connectionHandler = serviceProvider.GetService<HubConnectionHandler<ServicesHub>>();

        using (var client = new TestClient())
        {
            var connectionHandlerTask = await client.ConnectAsync(connectionHandler).DefaultTimeout();
            var res = await client.InvokeAsync(nameof(ServicesHub.SingleService)).DefaultTimeout();
            Assert.True(Assert.IsType<bool>(res.Result));
        }
    }

    [Fact]
    public async Task HubMethodCanInjectMultipleServices()
    {
        var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(provider =>
        {
            provider.AddSingleton<Service1>();
            provider.AddSingleton<Service2>();
            provider.AddSingleton<Service3>();
        });
        var connectionHandler = serviceProvider.GetService<HubConnectionHandler<ServicesHub>>();

        using (var client = new TestClient())
        {
            var connectionHandlerTask = await client.ConnectAsync(connectionHandler).DefaultTimeout();
            var res = await client.InvokeAsync(nameof(ServicesHub.MultipleServices)).DefaultTimeout();
            Assert.True(Assert.IsType<bool>(res.Result));
        }
    }

    [Fact]
    public async Task HubMethodCanInjectServicesWithOtherParameters()
    {
        var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(provider =>
        {
            provider.AddSingleton<Service1>();
            provider.AddSingleton<Service2>();
            provider.AddSingleton<Service3>();
        });
        var connectionHandler = serviceProvider.GetService<HubConnectionHandler<ServicesHub>>();

        using (var client = new TestClient())
        {
            var connectionHandlerTask = await client.ConnectAsync(connectionHandler).DefaultTimeout();
            await client.BeginUploadStreamAsync("0", nameof(ServicesHub.ServicesAndParams), new string[] { "1" }, 10, true).DefaultTimeout();

            await client.SendHubMessageAsync(new StreamItemMessage("1", 1)).DefaultTimeout();
            await client.SendHubMessageAsync(new StreamItemMessage("1", 14)).DefaultTimeout();

            await client.SendHubMessageAsync(CompletionMessage.Empty("1")).DefaultTimeout();

            var response = Assert.IsType<CompletionMessage>(await client.ReadAsync().DefaultTimeout());
            Assert.Equal(25L, response.Result);
        }
    }

    [Fact]
    public async Task StreamFromServiceDoesNotWork()
    {
        var channel = Channel.CreateBounded<int>(10);
        var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(provider =>
        {
            provider.AddSingleton(channel.Reader);
        });
        var connectionHandler = serviceProvider.GetService<HubConnectionHandler<ServicesHub>>();

        using (var client = new TestClient())
        {
            var connectionHandlerTask = await client.ConnectAsync(connectionHandler).DefaultTimeout();
            var res = await client.InvokeAsync(nameof(ServicesHub.Stream)).DefaultTimeout();
            Assert.Equal("An unexpected error occurred invoking 'Stream' on the server. HubException: Client sent 0 stream(s), Hub method expects 1.", res.Error);
        }
    }

    [Fact]
    public async Task ServiceNotResolvedWithoutAttribute_WithSettingDisabledGlobally()
    {
        var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(provider =>
        {
            provider.AddSignalR(options =>
            {
                options.EnableDetailedErrors = true;
                options.DisableImplicitFromServicesParameters = true;
            });
            provider.AddSingleton<Service1>();
        });
        var connectionHandler = serviceProvider.GetService<HubConnectionHandler<ServicesHub>>();

        using (var client = new TestClient())
        {
            var connectionHandlerTask = await client.ConnectAsync(connectionHandler).DefaultTimeout();
            var res = await client.InvokeAsync(nameof(ServicesHub.ServiceWithoutAttribute)).DefaultTimeout();
            Assert.Equal("Failed to invoke 'ServiceWithoutAttribute' due to an error on the server. InvalidDataException: Invocation provides 0 argument(s) but target expects 1.", res.Error);
        }
    }

    [Fact]
    public async Task ServiceResolvedWithoutAttribute()
    {
        var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(provider =>
        {
            provider.AddSignalR(options =>
            {
                options.EnableDetailedErrors = true;
            });
            provider.AddSingleton<Service1>();
        });
        var connectionHandler = serviceProvider.GetService<HubConnectionHandler<ServicesHub>>();

        using (var client = new TestClient())
        {
            var connectionHandlerTask = await client.ConnectAsync(connectionHandler).DefaultTimeout();
            var res = await client.InvokeAsync(nameof(ServicesHub.ServiceWithoutAttribute)).DefaultTimeout();
            Assert.Equal(1L, res.Result);
        }
    }

    [Fact]
    public async Task ServiceResolvedForIEnumerableParameter()
    {
        var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(provider =>
        {
            provider.AddSignalR(options =>
            {
                options.EnableDetailedErrors = true;
            });
            provider.AddSingleton<Service1>();
        });
        var connectionHandler = serviceProvider.GetService<HubConnectionHandler<ServicesHub>>();

        using (var client = new TestClient())
        {
            var connectionHandlerTask = await client.ConnectAsync(connectionHandler).DefaultTimeout();
            var res = await client.InvokeAsync(nameof(ServicesHub.IEnumerableOfServiceWithoutAttribute)).DefaultTimeout();
            Assert.Equal(1L, res.Result);
        }
    }

    [Fact]
    public async Task ServiceResolvedWithoutAttribute_WithHubSpecificSettingEnabled()
    {
        var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(provider =>
        {
            provider.AddSignalR(options =>
            {
                options.EnableDetailedErrors = true;
                options.DisableImplicitFromServicesParameters = true;
            }).AddHubOptions<ServicesHub>(options =>
            {
                options.DisableImplicitFromServicesParameters = false;
            });
            provider.AddSingleton<Service1>();
        });
        var connectionHandler = serviceProvider.GetService<HubConnectionHandler<ServicesHub>>();

        using (var client = new TestClient())
        {
            var connectionHandlerTask = await client.ConnectAsync(connectionHandler).DefaultTimeout();
            var res = await client.InvokeAsync(nameof(ServicesHub.ServiceWithoutAttribute)).DefaultTimeout();
            Assert.Equal(1L, res.Result);
        }
    }

    [Fact]
    public async Task ServiceNotResolvedWithAndWithoutAttribute_WithOptionDisabled()
    {
        var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(provider =>
        {
            provider.AddSignalR(options =>
            {
                options.EnableDetailedErrors = true;
                options.DisableImplicitFromServicesParameters = true;
            });
            provider.AddSingleton<Service1>();
            provider.AddSingleton<Service2>();
        });
        var connectionHandler = serviceProvider.GetService<HubConnectionHandler<ServicesHub>>();

        using (var client = new TestClient())
        {
            var connectionHandlerTask = await client.ConnectAsync(connectionHandler).DefaultTimeout();
            var res = await client.InvokeAsync(nameof(ServicesHub.ServiceWithAndWithoutAttribute)).DefaultTimeout();
            Assert.Equal("Failed to invoke 'ServiceWithAndWithoutAttribute' due to an error on the server. InvalidDataException: Invocation provides 0 argument(s) but target expects 1.", res.Error);
        }
    }

    [Fact]
    public async Task ServiceResolvedWithAndWithoutAttribute()
    {
        var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(provider =>
        {
            provider.AddSignalR(options =>
            {
                options.EnableDetailedErrors = true;
            });
            provider.AddSingleton<Service1>();
            provider.AddSingleton<Service2>();
        });
        var connectionHandler = serviceProvider.GetService<HubConnectionHandler<ServicesHub>>();

        using (var client = new TestClient())
        {
            var connectionHandlerTask = await client.ConnectAsync(connectionHandler).DefaultTimeout();
            var res = await client.InvokeAsync(nameof(ServicesHub.ServiceWithAndWithoutAttribute)).DefaultTimeout();
            Assert.Equal(1L, res.Result);
        }
    }

    [Fact]
    public async Task ServiceNotResolvedIfNotInDI()
    {
        var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(provider =>
        {
            provider.AddSignalR(options =>
            {
                options.EnableDetailedErrors = true;
            });
        });
        var connectionHandler = serviceProvider.GetService<HubConnectionHandler<ServicesHub>>();

        using (var client = new TestClient())
        {
            var connectionHandlerTask = await client.ConnectAsync(connectionHandler).DefaultTimeout();
            var res = await client.InvokeAsync(nameof(ServicesHub.ServiceWithoutAttribute)).DefaultTimeout();
            Assert.Equal("Failed to invoke 'ServiceWithoutAttribute' due to an error on the server. InvalidDataException: Invocation provides 0 argument(s) but target expects 1.", res.Error);
        }
    }

    [Fact]
    public async Task ServiceNotResolvedForIEnumerableParameterIfNotInDI()
    {
        var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(provider =>
        {
            provider.AddSignalR(options =>
            {
                options.EnableDetailedErrors = true;
            });
        });
        var connectionHandler = serviceProvider.GetService<HubConnectionHandler<ServicesHub>>();

        using (var client = new TestClient())
        {
            var connectionHandlerTask = await client.ConnectAsync(connectionHandler).DefaultTimeout();
            var res = await client.InvokeAsync(nameof(ServicesHub.IEnumerableOfServiceWithoutAttribute)).DefaultTimeout();
            Assert.Equal("Failed to invoke 'IEnumerableOfServiceWithoutAttribute' due to an error on the server. InvalidDataException: Invocation provides 0 argument(s) but target expects 1.", res.Error);
        }
    }

    [Fact]
    public void TooManyParametersWithServiceThrows()
    {
        var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(provider =>
        {
            provider.AddSingleton<Service1>();
        });
        Assert.Throws<InvalidOperationException>(
            () => serviceProvider.GetService<HubConnectionHandler<TooManyParamsHub>>());
    }

    [Fact]
    public async Task SendToAnotherClientFromOnConnectedAsync()
    {
        var serviceProvider = HubConnectionHandlerTestUtils.CreateServiceProvider(provider =>
        {
            provider.AddSignalR(options =>
            {
                options.EnableDetailedErrors = true;
            });
        });
        var connectionHandler = serviceProvider.GetService<HubConnectionHandler<OnConnectedSendToClientHub>>();

        using (var client1 = new TestClient())
        using (var client2 = new TestClient())
        {
            var httpContext = new DefaultHttpContext();
            httpContext.Request.QueryString = new QueryString($"?client={client1.Connection.ConnectionId}");
            var feature = new TestHttpContextFeature
            {
                HttpContext = httpContext
            };
            client2.Connection.Features.Set<IHttpContextFeature>(feature);

            var connectionHandlerTask = await client1.ConnectAsync(connectionHandler).DefaultTimeout();
            _ = await client2.ConnectAsync(connectionHandler).DefaultTimeout();

            var message = Assert.IsType<InvocationMessage>(await client1.ReadAsync().DefaultTimeout());
            Assert.Single(message.Arguments);
            Assert.Equal(1L, message.Arguments[0]);
            Assert.Equal("Test", message.Target);
        }
    }

    private class CustomHubActivator<THub> : IHubActivator<THub> where THub : Hub
    {
        public int ReleaseCount;
        private readonly IServiceProvider _serviceProvider;
        public TaskCompletionSource ReleaseTask = new TaskCompletionSource();
        public TaskCompletionSource CreateTask = new TaskCompletionSource();

        public CustomHubActivator(IServiceProvider serviceProvider)
        {
            _serviceProvider = serviceProvider;
        }

        public THub Create()
        {
            ReleaseTask = new TaskCompletionSource();
            var hub = new DefaultHubActivator<THub>(_serviceProvider).Create();
            CreateTask.TrySetResult();
            return hub;
        }

        public void Release(THub hub)
        {
            ReleaseCount++;
            hub.Dispose();
            ReleaseTask.TrySetResult();
            CreateTask = new TaskCompletionSource();
        }
    }

    public static IEnumerable<object[]> HubTypes()
    {
        yield return new[] { typeof(DynamicTestHub) };
        yield return new[] { typeof(MethodHub) };
        yield return new[] { typeof(HubT) };
    }

    public class TestHttpContextFeature : IHttpContextFeature
    {
        public HttpContext HttpContext { get; set; }
    }

    private class TestUserIdProvider : IUserIdProvider
    {
        private readonly Func<HubConnectionContext, string> _getUserId;

        public TestUserIdProvider(Func<HubConnectionContext, string> getUserId)
        {
            _getUserId = getUserId;
        }

        public string GetUserId(HubConnectionContext connection) => _getUserId(connection);
    }

    private class SampleObject
    {
        public SampleObject(string foo, int bar)
        {
            Bar = bar;
            Foo = foo;
        }
        public int Bar { get; }
        public string Foo { get; }
    }

    private class HttpContextFeatureImpl : IHttpContextFeature
    {
        public HttpContext HttpContext { get; set; }
    }
}

public static class IAsyncEnumerableExtension
{
    public static async Task<IEnumerable<T>> ReadAsync<T>(this IAsyncEnumerable<T> enumerable, int count)
    {
        if (count <= 0)
        {
            throw new ArgumentException("Input must be greater than zero.", nameof(count));
        }

        var result = new List<T>();
        await foreach (var item in enumerable)
        {
            result.Add(item);
            if (result.Count == count)
            {
                break;
            }
        }
        return result;
    }

    public static async Task<IEnumerable<T>> ReadAllAsync<T>(this IAsyncEnumerable<T> enumerable)
    {
        var result = new List<T>();
        await foreach (var item in enumerable)
        {
            result.Add(item);
        }

        return result;
    }
}