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

using System;
using System.Collections.Generic;
using System.IO.Pipelines;
using System.Net.Http;
using System.Net.WebSockets;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.AspNetCore.Connections;
using Microsoft.AspNetCore.Http.Connections;
using Microsoft.AspNetCore.Http.Connections.Client;
using Microsoft.AspNetCore.Http.Connections.Client.Internal;
using Microsoft.AspNetCore.SignalR.Client;
using Microsoft.AspNetCore.Testing;
using Microsoft.AspNetCore.WebUtilities;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Logging.Testing;
using Moq;
using Moq.Protected;
using Xunit;
using HttpConnectionOptions = Microsoft.AspNetCore.Http.Connections.Client.HttpConnectionOptions;

namespace Microsoft.AspNetCore.SignalR.Tests;

public class EndToEndTestsCollection : ICollectionFixture<InProcessTestServer<Startup>>
{
    public const string Name = nameof(EndToEndTestsCollection);
}

[Collection(EndToEndTestsCollection.Name)]
public class EndToEndTests : FunctionalTestBase
{
    [Fact]
    public async Task CanStartAndStopConnectionUsingDefaultTransport()
    {
        await using (var server = await StartServer<Startup>())
        {
            var url = server.Url + "/echo";
            // The test should connect to the server using WebSockets transport on Windows 8 and newer.
            // On Windows 7/2008R2 it should use ServerSentEvents transport to connect to the server.
            var connection = new HttpConnection(new Uri(url), HttpTransports.All, LoggerFactory);
            await connection.StartAsync().DefaultTimeout();
            await connection.DisposeAsync().DefaultTimeout();
        }
    }

    [Fact]
    public async Task TransportThatFallsbackCreatesNewConnection()
    {
        bool ExpectedErrors(WriteContext writeContext)
        {
            return writeContext.LoggerName == typeof(HttpConnection).FullName &&
                   writeContext.EventId.Name == "ErrorStartingTransport";
        }

        await using (var server = await StartServer<Startup>(expectedErrorsFilter: ExpectedErrors))
        {
            var url = server.Url + "/echo";
            // The test should connect to the server using WebSockets transport on Windows 8 and newer.
            // On Windows 7/2008R2 it should use ServerSentEvents transport to connect to the server.

            // The test logic lives in the TestTransportFactory and FakeTransport.
            var connection = new HttpConnection(new HttpConnectionOptions { Url = new Uri(url), DefaultTransferFormat = TransferFormat.Text }, LoggerFactory, new TestTransportFactory());
            await connection.StartAsync().DefaultTimeout();
            await connection.DisposeAsync().DefaultTimeout();
        }
    }

    [Theory]
    [MemberData(nameof(TransportTypes))]
    [LogLevel(LogLevel.Trace)]
    [QuarantinedTest("https://github.com/dotnet/aspnetcore/issues/49315")]
    public async Task CanStartAndStopConnectionUsingGivenTransport(HttpTransportType transportType)
    {
        await using (var server = await StartServer<Startup>())
        {
            var url = server.Url + "/echo";
            var connection = new HttpConnection(new HttpConnectionOptions { Url = new Uri(url), Transports = transportType, DefaultTransferFormat = TransferFormat.Text }, LoggerFactory);
            await connection.StartAsync().DefaultTimeout();
            await connection.DisposeAsync().DefaultTimeout();
        }
    }

    [ConditionalFact]
    [WebSocketsSupportedCondition]
    public async Task WebSocketsTest()
    {
        await using (var server = await StartServer<Startup>())
        {
            var logger = LoggerFactory.CreateLogger<EndToEndTests>();

            const string message = "Hello, World!";
            using (var ws = new ClientWebSocket())
            {
                var socketUrl = server.WebSocketsUrl + "/echo";

                logger.LogInformation("Connecting WebSocket to {socketUrl}", socketUrl);
                await ws.ConnectAsync(new Uri(socketUrl), CancellationToken.None).DefaultTimeout();

                var bytes = Encoding.UTF8.GetBytes(message);
                logger.LogInformation("Sending {length} byte frame", bytes.Length);
                await ws.SendAsync(new ArraySegment<byte>(bytes), WebSocketMessageType.Binary, endOfMessage: true, CancellationToken.None).DefaultTimeout();

                logger.LogInformation("Receiving frame");
                var buffer = new ArraySegment<byte>(new byte[1024]);
                var result = await ws.ReceiveAsync(buffer, CancellationToken.None).DefaultTimeout();
                logger.LogInformation("Received {length} byte frame", result.Count);

                Assert.Equal(bytes, buffer.Array.AsSpan(0, result.Count).ToArray());

                logger.LogInformation("Closing socket");
                await ws.CloseOutputAsync(WebSocketCloseStatus.NormalClosure, "", CancellationToken.None).DefaultTimeout();
                logger.LogInformation("Waiting for close");
                result = await ws.ReceiveAsync(buffer, CancellationToken.None).DefaultTimeout();
                Assert.Equal(WebSocketMessageType.Close, result.MessageType);
                Assert.Equal(WebSocketCloseStatus.NormalClosure, result.CloseStatus);
                logger.LogInformation("Closed socket");
            }
        }
    }

    [ConditionalFact]
    [WebSocketsSupportedCondition]
    public async Task WebSocketsReceivesAndSendsPartialFramesTest()
    {
        await using (var server = await StartServer<Startup>())
        {
            var logger = LoggerFactory.CreateLogger<EndToEndTests>();

            const string message = "Hello, World!";
            using (var ws = new ClientWebSocket())
            {
                var socketUrl = server.WebSocketsUrl + "/echo";

                logger.LogInformation("Connecting WebSocket to {socketUrl}", socketUrl);
                await ws.ConnectAsync(new Uri(socketUrl), CancellationToken.None).DefaultTimeout();

                var bytes = Encoding.UTF8.GetBytes(message);
                logger.LogInformation("Sending {length} byte frame", bytes.Length);
                // We're sending a partial frame, we should still get the data
                await ws.SendAsync(new ArraySegment<byte>(bytes), WebSocketMessageType.Binary, endOfMessage: false, CancellationToken.None).DefaultTimeout();

                logger.LogInformation("Receiving frame");
                var buffer = new ArraySegment<byte>(new byte[1024]);
                var result = await ws.ReceiveAsync(buffer, CancellationToken.None).DefaultTimeout();
                logger.LogInformation("Received {length} byte frame", result.Count);

                Assert.Equal(bytes, buffer.Array.AsSpan(0, result.Count).ToArray());

                logger.LogInformation("Closing socket");
                await ws.CloseOutputAsync(WebSocketCloseStatus.NormalClosure, "", CancellationToken.None).DefaultTimeout();
                logger.LogInformation("Waiting for close");
                result = await ws.ReceiveAsync(buffer, CancellationToken.None).DefaultTimeout();
                Assert.Equal(WebSocketMessageType.Close, result.MessageType);
                Assert.Equal(WebSocketCloseStatus.NormalClosure, result.CloseStatus);
                logger.LogInformation("Closed socket");
            }
        }
    }

    [ConditionalFact]
    [WebSocketsSupportedCondition]
    public async Task HttpRequestsNotSentWhenWebSocketsTransportRequestedAndSkipNegotiationSet()
    {
        await using (var server = await StartServer<Startup>())
        {
            var logger = LoggerFactory.CreateLogger<EndToEndTests>();
            var url = server.Url + "/echo";

            var mockHttpHandler = new Mock<HttpMessageHandler>();
            mockHttpHandler.Protected()
                .Setup<Task<HttpResponseMessage>>("SendAsync", ItExpr.IsAny<HttpRequestMessage>(), ItExpr.IsAny<CancellationToken>())
                .Returns<HttpRequestMessage, CancellationToken>(
                    (request, cancellationToken) => Task.FromException<HttpResponseMessage>(new InvalidOperationException("HTTP requests should not be sent.")));

            var httpOptions = new HttpConnectionOptions
            {
                Url = new Uri(url),
                Transports = HttpTransportType.WebSockets,
                SkipNegotiation = true,
                HttpMessageHandlerFactory = (httpMessageHandler) => mockHttpHandler.Object
            };

            var connection = new HttpConnection(httpOptions, LoggerFactory);

            try
            {
                var message = new byte[] { 42 };
                await connection.StartAsync().DefaultTimeout();

                await connection.Transport.Output.WriteAsync(message).DefaultTimeout();

                var receivedData = await connection.Transport.Input.ReadAsync(1);
                Assert.Equal(message, receivedData);
            }
            catch (Exception ex)
            {
                logger.LogInformation(ex, "Test threw exception");
                throw;
            }
            finally
            {
                logger.LogInformation("Disposing Connection");
                await connection.DisposeAsync().DefaultTimeout();
                logger.LogInformation("Disposed Connection");
            }
        }
    }

    [Theory]
    [InlineData(HttpTransportType.LongPolling)]
    [InlineData(HttpTransportType.ServerSentEvents)]
    public async Task HttpConnectionThrowsIfSkipNegotiationSetAndTransportIsNotWebSockets(HttpTransportType transportType)
    {
        await using (var server = await StartServer<Startup>())
        {
            var logger = LoggerFactory.CreateLogger<EndToEndTests>();
            var url = server.Url + "/echo";

            var mockHttpHandler = new Mock<HttpMessageHandler>();
            mockHttpHandler.Protected()
                .Setup<Task<HttpResponseMessage>>("SendAsync", ItExpr.IsAny<HttpRequestMessage>(), ItExpr.IsAny<CancellationToken>())
                .Returns<HttpRequestMessage, CancellationToken>(
                    (request, cancellationToken) => Task.FromException<HttpResponseMessage>(new InvalidOperationException("HTTP requests should not be sent.")));

            var httpOptions = new HttpConnectionOptions
            {
                Url = new Uri(url),
                Transports = transportType,
                SkipNegotiation = true,
                HttpMessageHandlerFactory = (httpMessageHandler) => mockHttpHandler.Object
            };

            var connection = new HttpConnection(httpOptions, LoggerFactory);

            try
            {
                var exception = await Assert.ThrowsAsync<InvalidOperationException>(() => connection.StartAsync().DefaultTimeout());
                Assert.Equal("Negotiation can only be skipped when using the WebSocket transport directly.", exception.Message);
            }
            catch (Exception ex)
            {
                logger.LogInformation(ex, "Test threw exception");
                throw;
            }
            finally
            {
                await connection.DisposeAsync().DefaultTimeout();
            }
        }
    }

    [Theory]
    [MemberData(nameof(TransportTypesAndTransferFormats))]
    [LogLevel(LogLevel.Trace)]
    public async Task ConnectionCanSendAndReceiveMessages(HttpTransportType transportType, TransferFormat requestedTransferFormat)
    {
        await using (var server = await StartServer<Startup>())
        {
            var logger = LoggerFactory.CreateLogger<EndToEndTests>();

            const string message = "Major Key";

            var url = server.Url + "/echo";
            var connection = new HttpConnection(new HttpConnectionOptions { Url = new Uri(url), Transports = transportType, DefaultTransferFormat = requestedTransferFormat }, LoggerFactory);
            try
            {
                logger.LogInformation("Starting connection to {url}", url);
                await connection.StartAsync().DefaultTimeout();
                logger.LogInformation("Started connection to {url}", url);

                var bytes = Encoding.UTF8.GetBytes(message);

                logger.LogInformation("Sending {length} byte message", bytes.Length);
                try
                {
                    await connection.Transport.Output.WriteAsync(bytes).DefaultTimeout();
                }
                catch (OperationCanceledException)
                {
                    // Because the server and client are run in the same process there is a race where websocket.SendAsync
                    // can send a message but before returning be suspended allowing the server to run the EchoConnectionHandler and
                    // send a close frame which triggers a cancellation token on the client and cancels the websocket.SendAsync.
                    // Our solution to this is to just catch OperationCanceledException from the sent message if the race happens
                    // because we know the send went through, and its safe to check the response.
                }

                logger.LogInformation("Sent message");

                logger.LogInformation("Receiving message");
                Assert.Equal(message, Encoding.UTF8.GetString(await connection.Transport.Input.ReadAsync(bytes.Length).DefaultTimeout()));
                logger.LogInformation("Completed receive");
            }
            catch (Exception ex)
            {
                logger.LogInformation(ex, "Test threw exception");
                throw;
            }
            finally
            {
                logger.LogInformation("Disposing Connection");
                await connection.DisposeAsync().DefaultTimeout();
                logger.LogInformation("Disposed Connection");
            }
        }
    }

    [ConditionalTheory]
    [WebSocketsSupportedCondition]
    [InlineData(5 * 4096)]
    [InlineData(1000 * 4096 + 32)]
    [LogLevel(LogLevel.Trace)]
    public async Task ConnectionCanSendAndReceiveDifferentMessageSizesWebSocketsTransport(int length)
    {
        var message = new string('A', length);
        await using (var server = await StartServer<Startup>())
        {
            var logger = LoggerFactory.CreateLogger<EndToEndTests>();

            var url = server.Url + "/echo";
            var connection = new HttpConnection(new Uri(url), HttpTransportType.WebSockets, LoggerFactory);

            try
            {
                logger.LogInformation("Starting connection to {url}", url);
                await connection.StartAsync().DefaultTimeout();
                logger.LogInformation("Started connection to {url}", url);

                var bytes = Encoding.UTF8.GetBytes(message);

                async Task SendMessage()
                {
                    logger.LogInformation("Sending {length} byte message", bytes.Length);
                    await connection.Transport.Output.WriteAsync(bytes).DefaultTimeout();
                    logger.LogInformation("Sent message");
                }

                async Task ReceiveMessage()
                {
                    logger.LogInformation("Receiving message");
                    // Big timeout here because it can take a while to receive all the bytes
                    var receivedData = await connection.Transport.Input.ReadAsync(bytes.Length).DefaultTimeout();
                    Assert.Equal(message, Encoding.UTF8.GetString(receivedData));
                    logger.LogInformation("Completed receive");
                }

                // Send the receive concurrently so that back pressure is released
                // for server -> client sends
                var sendingTask = SendMessage();
                var receivingTask = ReceiveMessage();

                await sendingTask;
                await receivingTask;
            }
            catch (Exception ex)
            {
                logger.LogInformation(ex, "Test threw exception");
                throw;
            }
            finally
            {
                logger.LogInformation("Disposing Connection");
                await connection.DisposeAsync().DefaultTimeout();
                logger.LogInformation("Disposed Connection");
            }
        }
    }

    [ConditionalFact]
    [WebSocketsSupportedCondition]
    [LogLevel(LogLevel.Trace)]
    public async Task UnauthorizedWebSocketsConnectionDoesNotConnect()
    {
        bool ExpectedErrors(WriteContext writeContext)
        {
            return writeContext.LoggerName == typeof(HttpConnection).FullName &&
                   writeContext.EventId.Name == "ErrorWithNegotiation";
        }

        await using (var server = await StartServer<Startup>(ExpectedErrors))
        {
            var logger = LoggerFactory.CreateLogger<EndToEndTests>();

            var url = server.Url + "/auth";
            var connection = new HttpConnection(new Uri(url), HttpTransportType.WebSockets, LoggerFactory);

            var exception = await Assert.ThrowsAsync<HttpRequestException>(() => connection.StartAsync().DefaultTimeout());

            Assert.Contains("401", exception.Message);
        }
    }

    [ConditionalFact]
    [WebSocketsSupportedCondition]
    [LogLevel(LogLevel.Trace)]
    public async Task UnauthorizedDirectWebSocketsConnectionDoesNotConnect()
    {
        bool ExpectedErrors(WriteContext writeContext)
        {
            return writeContext.LoggerName == typeof(HttpConnection).FullName &&
                   writeContext.EventId.Name == "ErrorStartingTransport";
        }

        await using (var server = await StartServer<Startup>(ExpectedErrors))
        {
            var logger = LoggerFactory.CreateLogger<EndToEndTests>();

            var url = server.Url + "/auth";
            var options = new HttpConnectionOptions
            {
                Url = new Uri(url),
                Transports = HttpTransportType.WebSockets,
                SkipNegotiation = true
            };

            var connection = new HttpConnection(options, LoggerFactory);

            await Assert.ThrowsAsync<WebSocketException>(() => connection.StartAsync().DefaultTimeout());
        }
    }

    [Theory]
    [InlineData(HttpTransportType.LongPolling)]
    [InlineData(HttpTransportType.ServerSentEvents)]
    [LogLevel(LogLevel.Trace)]
    public async Task UnauthorizedConnectionDoesNotConnect(HttpTransportType transportType)
    {
        bool ExpectedErrors(WriteContext writeContext)
        {
            return writeContext.LoggerName == typeof(HttpConnection).FullName &&
                   writeContext.EventId.Name == "ErrorWithNegotiation";
        }

        await using (var server = await StartServer<Startup>(ExpectedErrors))
        {
            var logger = LoggerFactory.CreateLogger<EndToEndTests>();

            var url = server.Url + "/auth";
            var connection = new HttpConnection(new Uri(url), transportType, LoggerFactory);

            try
            {
                logger.LogInformation("Starting connection to {url}", url);
                await connection.StartAsync().DefaultTimeout();
                Assert.True(false);
            }
            catch (Exception ex)
            {
                Assert.Equal("Response status code does not indicate success: 401 (Unauthorized).", ex.Message);
            }
            finally
            {
                logger.LogInformation("Disposing Connection");
                await connection.DisposeAsync().DefaultTimeout();
                logger.LogInformation("Disposed Connection");
            }
        }
    }

    [Fact]
    [LogLevel(LogLevel.Trace)]
    public async Task AuthorizedConnectionCanConnect()
    {
        bool ExpectedErrors(WriteContext writeContext)
        {
            return writeContext.LoggerName == typeof(HttpConnection).FullName &&
                   writeContext.EventId.Name == "ErrorWithNegotiation";
        }

        await using (var server = await StartServer<Startup>(ExpectedErrors))
        {
            var logger = LoggerFactory.CreateLogger<EndToEndTests>();

            string token;
            using (var client = new HttpClient())
            {
                client.BaseAddress = new Uri(server.Url);

                var response = await client.GetAsync("generatetoken?user=bob");
                token = await response.Content.ReadAsStringAsync();
            }

            var url = server.Url + "/auth";
            var connection = new HttpConnection(
                new HttpConnectionOptions()
                {
                    Url = new Uri(url),
                    AccessTokenProvider = () => Task.FromResult(token),
                    Transports = HttpTransportType.ServerSentEvents,
                    DefaultTransferFormat = TransferFormat.Text,
                },
                LoggerFactory);

            try
            {
                logger.LogInformation("Starting connection to {url}", url);
                await connection.StartAsync().DefaultTimeout();
                logger.LogInformation("Connected to {url}", url);
            }
            finally
            {
                logger.LogInformation("Disposing Connection");
                await connection.DisposeAsync().DefaultTimeout();
                logger.LogInformation("Disposed Connection");
            }
        }
    }

    [ConditionalFact]
    [WebSocketsSupportedCondition]
    public async Task ServerClosesConnectionWithErrorIfHubCannotBeCreated_WebSocket()
    {
        try
        {
            await ServerClosesConnectionWithErrorIfHubCannotBeCreated(HttpTransportType.WebSockets);
            Assert.True(false, "Expected error was not thrown.");
        }
        catch
        {
            // error is expected
        }
    }

    [Fact]
    public async Task ServerClosesConnectionWithErrorIfHubCannotBeCreated_LongPolling()
    {
        try
        {
            await ServerClosesConnectionWithErrorIfHubCannotBeCreated(HttpTransportType.LongPolling);
            Assert.True(false, "Expected error was not thrown.");
        }
        catch
        {
            // error is expected
        }
    }

    private async Task ServerClosesConnectionWithErrorIfHubCannotBeCreated(HttpTransportType transportType)
    {
        await using (var server = await StartServer<Startup>())
        {
            var logger = LoggerFactory.CreateLogger<EndToEndTests>();

            var url = server.Url + "/uncreatable";
            var connection = new HubConnectionBuilder()
                    .WithLoggerFactory(LoggerFactory)
                    .WithUrl(url, transportType)
                    .Build();
            try
            {
                var closeTcs = new TaskCompletionSource();
                connection.Closed += e =>
                {
                    if (e != null)
                    {
                        closeTcs.SetException(e);
                    }
                    else
                    {
                        closeTcs.SetResult();
                    }
                    return Task.CompletedTask;
                };

                logger.LogInformation("Starting connection to {url}", url);

                try
                {
                    await connection.StartAsync().DefaultTimeout();
                }
                catch (OperationCanceledException)
                {
                    // Due to a race, this can fail with OperationCanceledException in the SendAsync
                    // call that HubConnection does to send the handshake message.
                    // This has only been happening on AppVeyor, likely due to a slower CI machine
                    // The closed event will still fire with the exception we care about.
                }

                await closeTcs.Task.DefaultTimeout();
            }
            catch (Exception ex)
            {
                logger.LogError(ex, "Test threw {exceptionType}: {message}", ex.GetType(), ex.Message);
                throw;
            }
            finally
            {
                logger.LogInformation("Disposing Connection");
                await connection.DisposeAsync().DefaultTimeout();
                logger.LogInformation("Disposed Connection");
            }
        }
    }

    [Fact]
    [LogLevel(LogLevel.Trace)]
    public async Task UnauthorizedHubConnectionDoesNotConnect()
    {
        bool ExpectedErrors(WriteContext writeContext)
        {
            return writeContext.LoggerName == typeof(HttpConnection).FullName &&
                   writeContext.EventId.Name == "ErrorWithNegotiation";
        }

        await using (var server = await StartServer<Startup>(ExpectedErrors))
        {
            var logger = LoggerFactory.CreateLogger<EndToEndTests>();

            var url = server.Url + "/authHub";
            var connection = new HubConnectionBuilder()
                    .WithLoggerFactory(LoggerFactory)
                    .WithUrl(url, HttpTransportType.LongPolling)
                    .Build();

            try
            {
                logger.LogInformation("Starting connection to {url}", url);
                await connection.StartAsync().DefaultTimeout();
                Assert.True(false);
            }
            catch (Exception ex)
            {
                Assert.Equal("Response status code does not indicate success: 401 (Unauthorized).", ex.Message);
            }
            finally
            {
                logger.LogInformation("Disposing Connection");
                await connection.DisposeAsync().DefaultTimeout();
                logger.LogInformation("Disposed Connection");
            }
        }
    }

    [Fact]
    [LogLevel(LogLevel.Trace)]
    public async Task AuthorizedHubConnectionCanConnect()
    {
        bool ExpectedErrors(WriteContext writeContext)
        {
            return writeContext.LoggerName == typeof(HttpConnection).FullName &&
                   writeContext.EventId.Name == "ErrorWithNegotiation";
        }

        await using (var server = await StartServer<Startup>(ExpectedErrors))
        {
            var logger = LoggerFactory.CreateLogger<EndToEndTests>();

            string token;
            using (var client = new HttpClient())
            {
                client.BaseAddress = new Uri(server.Url);

                var response = await client.GetAsync("generatetoken?user=bob");
                token = await response.Content.ReadAsStringAsync();
            }

            var url = server.Url + "/authHub";
            var connection = new HubConnectionBuilder()
                    .WithLoggerFactory(LoggerFactory)
                    .WithUrl(url, HttpTransportType.LongPolling, o =>
                    {
                        o.AccessTokenProvider = () => Task.FromResult(token);
                    })
                    .Build();

            try
            {
                logger.LogInformation("Starting connection to {url}", url);
                await connection.StartAsync().DefaultTimeout();
                logger.LogInformation("Connected to {url}", url);
            }
            finally
            {
                logger.LogInformation("Disposing Connection");
                await connection.DisposeAsync().DefaultTimeout();
                logger.LogInformation("Disposed Connection");
            }
        }
    }

    // Serves a fake transport that lets us verify fallback behavior
    private class TestTransportFactory : ITransportFactory
    {
        private ITransport _transport;

        public ITransport CreateTransport(HttpTransportType availableServerTransports, bool useStatefulReconnect)
        {
            if (_transport == null)
            {
                _transport = new FakeTransport();
            }

            return _transport;
        }
    }

    private class FakeTransport : ITransport
    {
        private int _tries;
        private string _prevConnectionId = null;
        private IDuplexPipe _application;
        private IDuplexPipe _transport;
        private readonly int availableTransports = 3;

        public PipeReader Input => _transport.Input;

        public PipeWriter Output => _transport.Output;

        public FakeTransport()
        {
            if (!TestHelpers.IsWebSocketsSupported())
            {
                availableTransports -= 1;
            }
        }

        public Task StartAsync(Uri url, TransferFormat transferFormat, CancellationToken cancellationToken = default)
        {
            var options = ClientPipeOptions.DefaultOptions;
            var pair = DuplexPipe.CreateConnectionPair(options, options);

            _transport = pair.Transport;
            _application = pair.Application;
            _tries++;
            Assert.True(QueryHelpers.ParseQuery(url.Query).TryGetValue("id", out var id));
            if (_prevConnectionId == null)
            {
                _prevConnectionId = id;
            }
            else
            {
                Assert.True(_prevConnectionId != id);
                _prevConnectionId = id;
            }

            if (_tries < availableTransports)
            {
                return Task.FromException(new Exception());
            }
            else
            {
                return Task.CompletedTask;
            }
        }

        public Task StopAsync()
        {
            _application.Output.Complete();
            _application.Input.Complete();
            return Task.CompletedTask;
        }
    }

    public static IEnumerable<object[]> TransportTypes
    {
        get
        {
            if (TestHelpers.IsWebSocketsSupported())
            {
                yield return new object[] { HttpTransportType.WebSockets };
            }
            yield return new object[] { HttpTransportType.ServerSentEvents };
            yield return new object[] { HttpTransportType.LongPolling };
        }
    }

    public static IEnumerable<object[]> TransportTypesAndTransferFormats
    {
        get
        {
            foreach (var transport in TransportTypes)
            {
                yield return new[] { transport[0], TransferFormat.Text };

                if ((HttpTransportType)transport[0] != HttpTransportType.ServerSentEvents)
                {
                    yield return new[] { transport[0], TransferFormat.Binary };
                }
            }
        }
    }
}
