diff --git a/src/ModelContextProtocol.Core/Client/StreamableHttpClientSessionTransport.cs b/src/ModelContextProtocol.Core/Client/StreamableHttpClientSessionTransport.cs index 5b294496d..37083a21d 100644 --- a/src/ModelContextProtocol.Core/Client/StreamableHttpClientSessionTransport.cs +++ b/src/ModelContextProtocol.Core/Client/StreamableHttpClientSessionTransport.cs @@ -185,19 +185,30 @@ internal async Task SendHttpRequestAsync(JsonRpcMessage mes var rpcRequest = message as JsonRpcRequest; JsonRpcMessageWithId? rpcResponseOrError = null; + // Records why the response could not be correlated with its request. The body itself is only ever logged at + // Trace level (see LogTransportReceivedMessageSensitive), so without this the no-reply failure below is opaque. + var correlationTrace = new ResponseCorrelationTrace + { + ResponseMediaType = response.Content.Headers.ContentType?.MediaType, + }; + if (response.Content.Headers.ContentType?.MediaType == "application/json") { var responseContent = await response.Content.ReadAsStringAsync(cancellationToken).ConfigureAwait(false); if (responseContent.Length > 0) { - rpcResponseOrError = await ProcessMessageAsync(responseContent, rpcRequest, cancellationToken).ConfigureAwait(false); + rpcResponseOrError = await ProcessMessageAsync(responseContent, rpcRequest, correlationTrace, cancellationToken).ConfigureAwait(false); + } + else + { + correlationTrace.RecordEmptyBody(); } } else if (response.Content.Headers.ContentType?.MediaType == "text/event-stream") { var sseState = new SseStreamState(); using var responseBodyStream = await response.Content.ReadAsStreamAsync(cancellationToken).ConfigureAwait(false); - var sseResponse = await ProcessSseResponseAsync(responseBodyStream, rpcRequest, sseState, cancellationToken).ConfigureAwait(false); + var sseResponse = await ProcessSseResponseAsync(responseBodyStream, rpcRequest, sseState, correlationTrace, cancellationToken).ConfigureAwait(false); rpcResponseOrError = sseResponse.Response; // Resumability: If POST SSE stream ended without a response but we have a Last-Event-ID (from priming), @@ -205,7 +216,7 @@ internal async Task SendHttpRequestAsync(JsonRpcMessage mes // events from the event store, allowing us to receive the pending response. if (rpcResponseOrError is null && rpcRequest is not null && sseState.LastEventId is not null) { - rpcResponseOrError = await SendGetSseRequestWithRetriesAsync(rpcRequest, sseState, cancellationToken).ConfigureAwait(false); + rpcResponseOrError = await SendGetSseRequestWithRetriesAsync(rpcRequest, sseState, cancellationToken, correlationTrace).ConfigureAwait(false); } } @@ -216,7 +227,9 @@ internal async Task SendHttpRequestAsync(JsonRpcMessage mes if (rpcResponseOrError is null) { - throw new McpException($"Streamable HTTP POST response completed without a reply to request with ID: {rpcRequest.Id}"); + throw new McpException( + $"Streamable HTTP POST response completed without a reply to request with ID: {rpcRequest.Id}. " + + correlationTrace.Describe()); } if (rpcRequest.Method == RequestMethods.Initialize && rpcResponseOrError is JsonRpcResponse initResponse) @@ -344,7 +357,8 @@ await SendGetSseRequestWithRetriesAsync( private async Task SendGetSseRequestWithRetriesAsync( JsonRpcRequest? relatedRpcRequest, SseStreamState state, - CancellationToken cancellationToken) + CancellationToken cancellationToken, + ResponseCorrelationTrace? correlationTrace = null) { // When LastEventId is null, the first attempt is the initial GET SSE connection (not a reconnection), // so we start at -1 to avoid counting it against MaxReconnectionAttempts. @@ -416,7 +430,7 @@ await SendGetSseRequestWithRetriesAsync( } using var responseStream = await response.Content.ReadAsStreamAsync(cancellationToken).ConfigureAwait(false); - var sseResponse = await ProcessSseResponseAsync(responseStream, relatedRpcRequest, state, cancellationToken).ConfigureAwait(false); + var sseResponse = await ProcessSseResponseAsync(responseStream, relatedRpcRequest, state, correlationTrace, cancellationToken).ConfigureAwait(false); if (sseResponse.Response is { } rpcResponseOrError) { @@ -446,6 +460,7 @@ private async Task ProcessSseResponseAsync( Stream responseStream, JsonRpcRequest? relatedRpcRequest, SseStreamState state, + ResponseCorrelationTrace? correlationTrace, CancellationToken cancellationToken) { try @@ -468,7 +483,7 @@ private async Task ProcessSseResponseAsync( continue; } - var rpcResponseOrError = await ProcessMessageAsync(sseEvent.Data, relatedRpcRequest, cancellationToken).ConfigureAwait(false); + var rpcResponseOrError = await ProcessMessageAsync(sseEvent.Data, relatedRpcRequest, correlationTrace, cancellationToken).ConfigureAwait(false); if (rpcResponseOrError is not null) { return new() { Response = rpcResponseOrError }; @@ -477,6 +492,7 @@ private async Task ProcessSseResponseAsync( } catch (Exception ex) when (ex is IOException or HttpRequestException) { + correlationTrace?.RecordNetworkError(); state.StreamEndedTimestamp = Stopwatch.GetTimestamp(); return new() { IsNetworkError = true }; } @@ -485,7 +501,11 @@ private async Task ProcessSseResponseAsync( return default; } - private async Task ProcessMessageAsync(string data, JsonRpcRequest? relatedRpcRequest, CancellationToken cancellationToken) + private async Task ProcessMessageAsync( + string data, + JsonRpcRequest? relatedRpcRequest, + ResponseCorrelationTrace? correlationTrace, + CancellationToken cancellationToken) { LogTransportReceivedMessageSensitive(Name, data); @@ -495,19 +515,30 @@ private async Task ProcessSseResponseAsync( if (message is null) { LogTransportMessageParseUnexpectedTypeSensitive(Name, data); + correlationTrace?.RecordNotAReply(); return null; } await WriteMessageAsync(message, cancellationToken).ConfigureAwait(false); if (message is JsonRpcResponse or JsonRpcError && - message is JsonRpcMessageWithId rpcResponseOrError && - rpcResponseOrError.Id == relatedRpcRequest?.Id) + message is JsonRpcMessageWithId rpcResponseOrError) { - return rpcResponseOrError; + if (rpcResponseOrError.Id == relatedRpcRequest?.Id) + { + return rpcResponseOrError; + } + + // The message was still delivered to the session above; record the mismatch so a subsequent + // no-reply failure can name the ID the peer actually replied with. + correlationTrace?.RecordMismatchedResponseId(rpcResponseOrError.Id); + return null; } + + correlationTrace?.RecordNotAReply(); } catch (JsonException ex) { + correlationTrace?.RecordParseFailure(); LogJsonException(ex, data); } @@ -665,6 +696,58 @@ private readonly struct SseResponse public bool IsNetworkError { get; init; } } + /// + /// Accumulates the facts needed to explain why an HTTP response could not be correlated with the request that + /// produced it. Only non-sensitive shape information is retained: the response media type, whether the body was + /// empty, and the outcome of deserializing and correlating the messages it contained. Message payloads are + /// deliberately excluded, since those are only logged through LogTransportReceivedMessageSensitive. + /// + private sealed class ResponseCorrelationTrace + { + private int _parseFailureCount; + private int _notAReplyCount; + + public string? ResponseMediaType { get; set; } + + public bool BodyWasEmpty { get; private set; } + + public bool IsNetworkError { get; private set; } + + public RequestId? MismatchedResponseId { get; private set; } + + public void RecordEmptyBody() => BodyWasEmpty = true; + + public void RecordNetworkError() => IsNetworkError = true; + + public void RecordParseFailure() => _parseFailureCount++; + + public void RecordNotAReply() => _notAReplyCount++; + + public void RecordMismatchedResponseId(RequestId id) => MismatchedResponseId = id; + + /// + /// Describes the most specific reason found for the missing reply, as a sentence suitable for appending to + /// the no-reply exception message. + /// + public string Describe() + { + string mediaTypeDescription = ResponseMediaType is { } mediaType + ? $"The response Content-Type was '{mediaType}'" + : "The response had no Content-Type"; + + string reason = + BodyWasEmpty ? "the response body was empty" : + _parseFailureCount > 0 ? $"{_parseFailureCount} JSON-RPC message(s) in the response failed to deserialize" : + MismatchedResponseId is { Id: null } ? "the response contained a JSON-RPC response or error with a null ID, which cannot be correlated with a request" : + MismatchedResponseId is { } mismatchedId ? $"the response contained a JSON-RPC response or error with ID '{mismatchedId}', which does not match the request ID" : + _notAReplyCount > 0 ? $"the response contained {_notAReplyCount} JSON-RPC message(s) that were not a response or error for this request" : + IsNetworkError ? "the event stream ended with a network error before a reply arrived" : + "no JSON-RPC response or error was present in the response"; + + return $"{mediaTypeDescription}, and {reason}."; + } + } + private static TimeSpan ElapsedSince(long stopwatchTimestamp) { #if NET diff --git a/tests/ModelContextProtocol.Tests/Transport/StreamableHttpResponseCorrelationTests.cs b/tests/ModelContextProtocol.Tests/Transport/StreamableHttpResponseCorrelationTests.cs new file mode 100644 index 000000000..a0a5e5cff --- /dev/null +++ b/tests/ModelContextProtocol.Tests/Transport/StreamableHttpResponseCorrelationTests.cs @@ -0,0 +1,497 @@ +using ModelContextProtocol.Client; +using ModelContextProtocol.Protocol; +using ModelContextProtocol.Tests.Utils; +using System.Net; +using System.Net.Sockets; +using System.Text; +using System.Text.Json.Nodes; + +namespace ModelContextProtocol.Tests.Transport; + +/// +/// Regression tests for how correlates an HTTP POST response +/// with the request that produced it, and for the diagnostics it reports when no reply can be correlated. +/// +/// +/// +/// The Streamable HTTP specification allows a server to answer a JSON-RPC request with either +/// text/event-stream or application/json, and clients MUST support both. Existing coverage of the +/// application/json case uses an in-memory response, which is already buffered and +/// carries a Content-Length. These tests exercise the production path instead: a real loopback socket that +/// sends Transfer-Encoding: chunked with no Content-Length, consumed through the transport's normal +/// ResponseHeadersRead flow. +/// +/// +/// The remaining tests pin the no-reply failure message, which previously only reported the request ID even though +/// the peer had clearly answered — with a different ID, an empty body, or something that wasn't a JSON-RPC reply. +/// See https://github.com/modelcontextprotocol/csharp-sdk/issues/1862. +/// +/// +public class StreamableHttpResponseCorrelationTests : LoggedTest +{ + public StreamableHttpResponseCorrelationTests(ITestOutputHelper testOutputHelper) + : base(testOutputHelper) + { + } + + [Fact] + public async Task ChunkedJsonResponse_WithCorrelatedError_FallsBackToInitialize() + { + var ct = TestContext.Current.CancellationToken; + var receivedMethods = new List(); + + await using var server = new LoopbackHttpServer + { + Handler = (_, body) => + { + string? method = null; + string? id = null; + try + { + if (JsonNode.Parse(body) is JsonObject request) + { + method = request["method"]?.GetValue(); + id = request["id"]?.ToJsonString(); + } + } + catch (Exception) + { + // Leave method/id null; the assertions below will report the unexpected request. + } + + lock (receivedMethods) + { + receivedMethods.Add(method ?? ""); + } + + switch (method) + { + case RequestMethods.ServerDiscover: + // HTTP 200, application/json with a charset parameter, Transfer-Encoding: chunked, and no + // Content-Length. The peer is an initialize-handshake server that rejects the + // server/discover probe with a correlated JSON-RPC error. + return LoopbackHttpServer.ChunkedJsonResponse(new JsonObject + { + ["jsonrpc"] = "2.0", + ["id"] = JsonNode.Parse(id!), + ["error"] = new JsonObject + { + ["code"] = -32600, + ["message"] = "Invalid Request", + }, + }.ToJsonString(), chunkSize: 16); + + case RequestMethods.Initialize: + return LoopbackHttpServer.ChunkedJsonResponse(new JsonObject + { + ["jsonrpc"] = "2.0", + ["id"] = JsonNode.Parse(id!), + ["result"] = new JsonObject + { + ["protocolVersion"] = McpProtocolVersions.June2025ProtocolVersion, + ["capabilities"] = new JsonObject(), + ["serverInfo"] = new JsonObject + { + ["name"] = "ChunkedLoopback", + ["version"] = "1.0.0", + }, + }, + }.ToJsonString(), chunkSize: 16); + + default: + // Notifications (e.g. notifications/initialized) have no id and expect no reply. + return LoopbackHttpServer.EmptyResponse(HttpStatusCode.Accepted); + } + }, + }; + + using var httpClient = new HttpClient(); + await using var transport = new HttpClientTransport(CreateOptions(server.Endpoint), httpClient, LoggerFactory); + + // Default options (ProtocolVersion = null) probe with server/discover and are expected to fall back to the + // initialize handshake when that probe is rejected. + await using var client = await McpClient.CreateAsync( + transport, + new McpClientOptions(), + loggerFactory: LoggerFactory, + cancellationToken: ct).WaitAsync(TestConstants.DefaultTimeout, ct); + + Assert.Equal(McpProtocolVersions.June2025ProtocolVersion, client.NegotiatedProtocolVersion); + + lock (receivedMethods) + { + Assert.Contains(RequestMethods.ServerDiscover, receivedMethods); + Assert.Contains(RequestMethods.Initialize, receivedMethods); + } + } + + [Fact] + public async Task ChunkedJsonResponse_WithCorrelatedError_IsDeliveredToTheSession() + { + var ct = TestContext.Current.CancellationToken; + + await using var server = new LoopbackHttpServer + { + Handler = (_, _) => LoopbackHttpServer.ChunkedJsonResponse( + """{"jsonrpc":"2.0","id":1,"error":{"code":-32600,"message":"Invalid Request"}}"""), + }; + + using var httpClient = new HttpClient(); + await using var transport = new HttpClientTransport(CreateOptions(server.Endpoint), httpClient, LoggerFactory); + await using var session = await transport.ConnectAsync(ct); + + // The correlated error must be surfaced as a message rather than turn into a no-reply failure. + await session.SendMessageAsync( + new JsonRpcRequest { Id = new RequestId(1), Method = RequestMethods.ServerDiscover, Params = new JsonObject() }, + ct).WaitAsync(TestConstants.DefaultTimeout, ct); + + Assert.True(session.MessageReader.TryRead(out var received)); + var error = Assert.IsType(received); + Assert.Equal(-32600, error.Error.Code); + Assert.Equal(1, Assert.IsType(error.Id.Id)); + } + + [Fact] + public async Task NoReply_ResponseWithDifferentId_ExceptionNamesTheObservedId() + { + var ct = TestContext.Current.CancellationToken; + + using var mockHttpHandler = new MockHttpHandler(); + using var httpClient = new HttpClient(mockHttpHandler); + await using var transport = new HttpClientTransport(CreateOptions(new Uri("http://localhost:8080")), httpClient, LoggerFactory); + await using var session = await transport.ConnectAsync(ct); + + mockHttpHandler.RequestHandler = _ => Task.FromResult(new HttpResponseMessage(HttpStatusCode.OK) + { + Content = new StringContent("""{"jsonrpc":"2.0","id":42,"result":{}}""", Encoding.UTF8, "application/json"), + }); + + var exception = await Assert.ThrowsAsync(() => session.SendMessageAsync( + new JsonRpcRequest { Id = new RequestId(1), Method = RequestMethods.ToolsList }, + ct)); + + Assert.Contains("with ID: 1", exception.Message); + Assert.Contains("The response Content-Type was 'application/json'", exception.Message); + Assert.Contains("ID '42'", exception.Message); + } + + [Fact] + public async Task NoReply_NullIdErrorResponse_ExceptionReportsTheNullId() + { + var ct = TestContext.Current.CancellationToken; + + using var mockHttpHandler = new MockHttpHandler(); + using var httpClient = new HttpClient(mockHttpHandler); + await using var transport = new HttpClientTransport(CreateOptions(new Uri("http://localhost:8080")), httpClient, LoggerFactory); + await using var session = await transport.ConnectAsync(ct); + + // JSON-RPC 2.0 permits an error response with a null ID when the peer could not determine the request ID + // (parse error / invalid request). Such a response can never be correlated with a request. + mockHttpHandler.RequestHandler = _ => Task.FromResult(new HttpResponseMessage(HttpStatusCode.OK) + { + Content = new StringContent( + """{"jsonrpc":"2.0","id":null,"error":{"code":-32700,"message":"Parse error"}}""", + Encoding.UTF8, + "application/json"), + }); + + var exception = await Assert.ThrowsAsync(() => session.SendMessageAsync( + new JsonRpcRequest { Id = new RequestId(1), Method = RequestMethods.ToolsList }, + ct)); + + Assert.Contains("with ID: 1", exception.Message); + Assert.Contains("with a null ID", exception.Message); + } + + [Fact] + public async Task NoReply_EmptyJsonBody_ExceptionReportsTheEmptyBody() + { + var ct = TestContext.Current.CancellationToken; + + using var mockHttpHandler = new MockHttpHandler(); + using var httpClient = new HttpClient(mockHttpHandler); + await using var transport = new HttpClientTransport(CreateOptions(new Uri("http://localhost:8080")), httpClient, LoggerFactory); + await using var session = await transport.ConnectAsync(ct); + + mockHttpHandler.RequestHandler = _ => Task.FromResult(new HttpResponseMessage(HttpStatusCode.OK) + { + Content = new StringContent("", Encoding.UTF8, "application/json"), + }); + + var exception = await Assert.ThrowsAsync(() => session.SendMessageAsync( + new JsonRpcRequest { Id = new RequestId(1), Method = RequestMethods.ToolsList }, + ct)); + + Assert.Contains("with ID: 1", exception.Message); + Assert.Contains("the response body was empty", exception.Message); + } + + [Fact] + public async Task NoReply_NonJsonContentType_ExceptionReportsTheMediaType() + { + var ct = TestContext.Current.CancellationToken; + + using var mockHttpHandler = new MockHttpHandler(); + using var httpClient = new HttpClient(mockHttpHandler); + await using var transport = new HttpClientTransport(CreateOptions(new Uri("http://localhost:8080")), httpClient, LoggerFactory); + await using var session = await transport.ConnectAsync(ct); + + mockHttpHandler.RequestHandler = _ => Task.FromResult(new HttpResponseMessage(HttpStatusCode.OK) + { + Content = new StringContent("not an MCP response", Encoding.UTF8, "text/plain"), + }); + + var exception = await Assert.ThrowsAsync(() => session.SendMessageAsync( + new JsonRpcRequest { Id = new RequestId(1), Method = RequestMethods.ToolsList }, + ct)); + + Assert.Contains("The response Content-Type was 'text/plain'", exception.Message); + Assert.Contains("no JSON-RPC response or error was present", exception.Message); + } + + private static HttpClientTransportOptions CreateOptions(Uri endpoint) => new() + { + Endpoint = endpoint, + TransportMode = HttpTransportMode.StreamableHttp, + // The transport opens a standalone GET SSE stream once it adopts the session; these tests only exercise POST. + EnableStandaloneGetStream = false, + }; + + /// + /// A minimal HTTP/1.1 loopback server that writes raw responses, so tests can control the framing + /// (notably Transfer-Encoding: chunked with no Content-Length) that a mocked + /// cannot produce. + /// + private sealed class LoopbackHttpServer : IAsyncDisposable + { + private readonly TcpListener _listener; + private readonly CancellationTokenSource _cts = new(); + private readonly Task _acceptLoop; + + public LoopbackHttpServer() + { + _listener = new TcpListener(IPAddress.Loopback, 0); + _listener.Start(); + _acceptLoop = Task.Run(AcceptLoopAsync); + } + + /// Receives the request method, the decoded request body, and returns the raw response to write. + public Func? Handler { get; set; } + + public Uri Endpoint => new($"http://127.0.0.1:{((IPEndPoint)_listener.LocalEndpoint).Port}/"); + + /// Builds an HTTP 200 application/json response sent with chunked transfer encoding. + public static string ChunkedJsonResponse(string body, int chunkSize = 12) + { + var response = new StringBuilder(); + response.Append("HTTP/1.1 200 OK\r\n"); + response.Append("Content-Type: application/json; charset=utf-8\r\n"); + response.Append("Transfer-Encoding: chunked\r\n"); + response.Append("Connection: close\r\n"); + response.Append("\r\n"); + + byte[] bytes = Encoding.UTF8.GetBytes(body); + for (int offset = 0; offset < bytes.Length; offset += chunkSize) + { + int length = Math.Min(chunkSize, bytes.Length - offset); + response.Append(length.ToString("x")).Append("\r\n"); + response.Append(Encoding.UTF8.GetString(bytes, offset, length)).Append("\r\n"); + } + + response.Append("0\r\n\r\n"); + return response.ToString(); + } + + /// Builds a response with no body, such as the 202 Accepted expected for a notification. + public static string EmptyResponse(HttpStatusCode statusCode) => + $"HTTP/1.1 {(int)statusCode} {statusCode}\r\nContent-Length: 0\r\nConnection: close\r\n\r\n"; + + private async Task AcceptLoopAsync() + { + var connections = new List(); + try + { + while (!_cts.IsCancellationRequested) + { + TcpClient client = await _listener.AcceptTcpClientAsync(_cts.Token); + connections.Add(Task.Run(() => HandleConnectionAsync(client))); + } + } + catch (OperationCanceledException) + { + } + catch (ObjectDisposedException) + { + } + } + + private async Task HandleConnectionAsync(TcpClient client) + { + using (client) + { + try + { + NetworkStream stream = client.GetStream(); + string headerText = await ReadHeadersAsync(stream); + + string[] headerLines = headerText.Split("\r\n"); + string method = headerLines[0].Split(' ')[0]; + + int contentLength = 0; + bool isChunked = false; + foreach (string line in headerLines) + { + if (line.StartsWith("Content-Length:", StringComparison.OrdinalIgnoreCase)) + { + contentLength = int.Parse(line["Content-Length:".Length..].Trim()); + } + else if (line.StartsWith("Transfer-Encoding:", StringComparison.OrdinalIgnoreCase) && + line.Contains("chunked", StringComparison.OrdinalIgnoreCase)) + { + isChunked = true; + } + } + + // The SDK sends JsonContent, whose length is unknown, so request bodies arrive chunked as well. + string body = isChunked + ? await ReadChunkedBodyAsync(stream) + : await ReadFixedLengthBodyAsync(stream, contentLength); + + string response = Handler?.Invoke(method, body) + ?? EmptyResponse(HttpStatusCode.NotFound); + + await stream.WriteAsync(Encoding.UTF8.GetBytes(response), _cts.Token); + await stream.FlushAsync(_cts.Token); + } + catch (Exception) when (_cts.IsCancellationRequested) + { + } + catch (IOException) + { + // The client closed the connection (e.g. after disposing the transport); nothing to do. + } + } + } + + private async Task ReadHeadersAsync(NetworkStream stream) + { + var bytes = new List(); + var single = new byte[1]; + while (true) + { + if (await stream.ReadAsync(single, _cts.Token) == 0) + { + break; + } + + bytes.Add(single[0]); + int count = bytes.Count; + if (count >= 4 && + bytes[count - 4] == '\r' && bytes[count - 3] == '\n' && + bytes[count - 2] == '\r' && bytes[count - 1] == '\n') + { + break; + } + } + + return Encoding.ASCII.GetString(bytes.ToArray()); + } + + private async Task ReadFixedLengthBodyAsync(NetworkStream stream, int contentLength) + { + if (contentLength == 0) + { + return ""; + } + + var buffer = new byte[contentLength]; + int read = 0; + while (read < contentLength) + { + int chunk = await stream.ReadAsync(buffer.AsMemory(read), _cts.Token); + if (chunk == 0) + { + break; + } + + read += chunk; + } + + return Encoding.UTF8.GetString(buffer, 0, read); + } + + private async Task ReadChunkedBodyAsync(NetworkStream stream) + { + var body = new StringBuilder(); + while (true) + { + int size = Convert.ToInt32((await ReadLineAsync(stream)).Trim(), 16); + if (size == 0) + { + await ReadLineAsync(stream); // Trailer-terminating CRLF. + break; + } + + var chunk = new byte[size]; + int read = 0; + while (read < size) + { + int count = await stream.ReadAsync(chunk.AsMemory(read), _cts.Token); + if (count == 0) + { + break; + } + + read += count; + } + + body.Append(Encoding.UTF8.GetString(chunk, 0, read)); + await ReadLineAsync(stream); // CRLF following the chunk data. + } + + return body.ToString(); + } + + private async Task ReadLineAsync(NetworkStream stream) + { + var line = new StringBuilder(); + var single = new byte[1]; + while (true) + { + if (await stream.ReadAsync(single, _cts.Token) == 0) + { + break; + } + + char c = (char)single[0]; + if (c == '\n') + { + break; + } + + if (c != '\r') + { + line.Append(c); + } + } + + return line.ToString(); + } + + public async ValueTask DisposeAsync() + { + await _cts.CancelAsync(); + _listener.Stop(); + + try + { + await _acceptLoop; + } + catch (Exception) + { + } + + _cts.Dispose(); + } + } +}