diff --git a/src/libraries/Common/src/System/Net/WebSockets/WebSocketValidate.cs b/src/libraries/Common/src/System/Net/WebSockets/WebSocketValidate.cs index 7c97c082d26f89..e087677be4608e 100644 --- a/src/libraries/Common/src/System/Net/WebSockets/WebSocketValidate.cs +++ b/src/libraries/Common/src/System/Net/WebSockets/WebSocketValidate.cs @@ -23,10 +23,15 @@ internal static partial class WebSocketValidate internal const int MaxDeflateWindowBits = 15; internal const int MaxControlFramePayloadLength = 123; +#if TARGET_BROWSER + private const int ValidCloseStatusCodesFrom = 3000; + private const int ValidCloseStatusCodesTo = 4999; +#else private const int CloseStatusCodeAbort = 1006; private const int CloseStatusCodeFailedTLSHandshake = 1015; private const int InvalidCloseStatusCodesFrom = 0; private const int InvalidCloseStatusCodesTo = 999; +#endif // [0x21, 0x7E] except separators "()<>@,;:\\\"/[]?={} ". private static readonly SearchValues s_validSubprotocolChars = @@ -84,11 +89,15 @@ internal static void ValidateCloseStatus(WebSocketCloseStatus closeStatus, strin } int closeStatusCode = (int)closeStatus; - +#if TARGET_BROWSER + // as defined in https://developer.mozilla.org/en-US/docs/Web/API/WebSocket/close#code + if (closeStatus != WebSocketCloseStatus.NormalClosure && (closeStatusCode < ValidCloseStatusCodesFrom || closeStatusCode > ValidCloseStatusCodesTo)) +#else if ((closeStatusCode >= InvalidCloseStatusCodesFrom && closeStatusCode <= InvalidCloseStatusCodesTo) || closeStatusCode == CloseStatusCodeAbort || closeStatusCode == CloseStatusCodeFailedTLSHandshake) +#endif { // CloseStatus 1006 means Aborted - this will never appear on the wire and is reflected by calling WebSocket.Abort throw new ArgumentException(SR.Format(SR.net_WebSockets_InvalidCloseStatusCode, diff --git a/src/libraries/Common/tests/System/Net/Prerequisites/NetCoreServer/Handlers/EchoWebSocketHandler.cs b/src/libraries/Common/tests/System/Net/Prerequisites/NetCoreServer/Handlers/EchoWebSocketHandler.cs index f4e5562600015d..8304f2d1156072 100644 --- a/src/libraries/Common/tests/System/Net/Prerequisites/NetCoreServer/Handlers/EchoWebSocketHandler.cs +++ b/src/libraries/Common/tests/System/Net/Prerequisites/NetCoreServer/Handlers/EchoWebSocketHandler.cs @@ -24,11 +24,11 @@ public static async Task InvokeAsync(HttpContext context) if (context.Request.QueryString.HasValue && context.Request.QueryString.Value.Contains("delay10sec")) { - Thread.Sleep(10000); + await Task.Delay(10000); } else if (context.Request.QueryString.HasValue && context.Request.QueryString.Value.Contains("delay20sec")) { - Thread.Sleep(20000); + await Task.Delay(20000); } try @@ -124,14 +124,15 @@ await socket.CloseAsync( } bool sendMessage = false; + string receivedMessage = null; if (receiveResult.MessageType == WebSocketMessageType.Text) { - string receivedMessage = Encoding.UTF8.GetString(receiveBuffer, 0, offset); + receivedMessage = Encoding.UTF8.GetString(receiveBuffer, 0, offset); if (receivedMessage == ".close") { await socket.CloseAsync(WebSocketCloseStatus.NormalClosure, receivedMessage, CancellationToken.None); } - if (receivedMessage == ".shutdown") + else if (receivedMessage == ".shutdown") { await socket.CloseOutputAsync(WebSocketCloseStatus.NormalClosure, receivedMessage, CancellationToken.None); } @@ -161,6 +162,14 @@ await socket.SendAsync( !replyWithPartialMessages, CancellationToken.None); } + if (receivedMessage == ".closeafter") + { + await socket.CloseAsync(WebSocketCloseStatus.NormalClosure, receivedMessage, CancellationToken.None); + } + else if (receivedMessage == ".shutdownafter") + { + await socket.CloseOutputAsync(WebSocketCloseStatus.NormalClosure, receivedMessage, CancellationToken.None); + } } } } diff --git a/src/libraries/System.Net.WebSockets.Client/src/System/Net/WebSockets/BrowserWebSockets/BrowserInterop.cs b/src/libraries/System.Net.WebSockets.Client/src/System/Net/WebSockets/BrowserWebSockets/BrowserInterop.cs index a03a0d6941190d..53f43c3c8592f0 100644 --- a/src/libraries/System.Net.WebSockets.Client/src/System/Net/WebSockets/BrowserWebSockets/BrowserInterop.cs +++ b/src/libraries/System.Net.WebSockets.Client/src/System/Net/WebSockets/BrowserWebSockets/BrowserInterop.cs @@ -11,16 +11,54 @@ internal static partial class BrowserInterop { public static string? GetProtocol(JSObject? webSocket) { - if (webSocket == null || webSocket.IsDisposed) return null; + if (webSocket == null || webSocket.IsDisposed) + { + return null; + } + string? protocol = webSocket.GetPropertyAsString("protocol"); return protocol; } + public static WebSocketCloseStatus? GetCloseStatus(JSObject? webSocket) + { + if (webSocket == null || webSocket.IsDisposed) + { + return null; + } + if (!webSocket.HasProperty("close_status")) + { + return null; + } + + int status = webSocket.GetPropertyAsInt32("close_status"); + return (WebSocketCloseStatus)status; + } + + public static string? GetCloseStatusDescription(JSObject? webSocket) + { + if (webSocket == null || webSocket.IsDisposed) + { + return null; + } + + string? description = webSocket.GetPropertyAsString("close_status_description"); + return description; + } + public static int GetReadyState(JSObject? webSocket) { - if (webSocket == null || webSocket.IsDisposed) return -1; + if (webSocket == null || webSocket.IsDisposed) + { + return -1; + } + int? readyState = webSocket.GetPropertyAsInt32("readyState"); - if (!readyState.HasValue) return -1; + if (!readyState.HasValue) + { + return -1; + } + return readyState.Value; } @@ -28,16 +66,14 @@ public static int GetReadyState(JSObject? webSocket) public static partial JSObject WebSocketCreate( string uri, string?[]? subProtocols, - IntPtr responseStatusPtr, - [JSMarshalAs>] Action onClosed); + IntPtr responseStatusPtr); public static unsafe JSObject UnsafeCreate( string uri, string?[]? subProtocols, - MemoryHandle responseHandle, - [JSMarshalAs>] Action onClosed) + MemoryHandle responseHandle) { - return WebSocketCreate(uri, subProtocols, (IntPtr)responseHandle.Pointer, onClosed); + return WebSocketCreate(uri, subProtocols, (IntPtr)responseHandle.Pointer); } [JSImport("INTERNAL.ws_wasm_open")] @@ -52,19 +88,9 @@ public static partial Task WebSocketOpen( int messageType, bool endOfMessage); - public static unsafe Task? UnsafeSendSync(JSObject jsWs, ArraySegment buffer, WebSocketMessageType messageType, bool endOfMessage) + public static unsafe Task? UnsafeSend(JSObject jsWs, MemoryHandle pinBuffer, int length, WebSocketMessageType messageType, bool endOfMessage) { - if (buffer.Count == 0) - { - return WebSocketSend(jsWs, IntPtr.Zero, 0, (int)messageType, endOfMessage); - } - - var span = buffer.AsSpan(); - // we can do this because the bytes in the buffer are always consumed synchronously (not later with Task resolution) - fixed (void* spanPtr = span) - { - return WebSocketSend(jsWs, (IntPtr)spanPtr, buffer.Count, (int)messageType, endOfMessage); - } + return WebSocketSend(jsWs, (IntPtr)pinBuffer.Pointer, length, (int)messageType, endOfMessage); } [JSImport("INTERNAL.ws_wasm_receive")] @@ -73,7 +99,7 @@ public static partial Task WebSocketOpen( IntPtr bufferPtr, int bufferLength); - public static unsafe Task? ReceiveUnsafeSync(JSObject jsWs, MemoryHandle pinBuffer, int length) + public static unsafe Task? ReceiveUnsafe(JSObject jsWs, MemoryHandle pinBuffer, int length) { return WebSocketReceive(jsWs, (IntPtr)pinBuffer.Pointer, length); } diff --git a/src/libraries/System.Net.WebSockets.Client/src/System/Net/WebSockets/BrowserWebSockets/BrowserWebSocket.cs b/src/libraries/System.Net.WebSockets.Client/src/System/Net/WebSockets/BrowserWebSockets/BrowserWebSocket.cs index 68120a5cf8f7ef..c635a3d48aad81 100644 --- a/src/libraries/System.Net.WebSockets.Client/src/System/Net/WebSockets/BrowserWebSockets/BrowserWebSocket.cs +++ b/src/libraries/System.Net.WebSockets.Client/src/System/Net/WebSockets/BrowserWebSockets/BrowserWebSocket.cs @@ -14,16 +14,18 @@ namespace System.Net.WebSockets /// internal sealed class BrowserWebSocket : WebSocket { -#if FEATURE_WASM_THREADS - private readonly object _thisLock = new object(); -#endif + private readonly object _lockObject = new object(); private WebSocketCloseStatus? _closeStatus; private string? _closeStatusDescription; private JSObject? _innerWebSocket; private WebSocketState _state; + private bool _closeSent; + private bool _closeReceived; private bool _disposed; private bool _aborted; + private bool _shouldAbort; + private bool _cancelled; private int[] responseStatus = new int[3]; private MemoryHandle? responseStatusHandle; @@ -33,29 +35,30 @@ public override WebSocketState State { get { -#if FEATURE_WASM_THREADS - lock (_thisLock) + lock (_lockObject) { -#endif - if (_innerWebSocket == null || _disposed || (_state != WebSocketState.Connecting && _state != WebSocketState.Open && _state != WebSocketState.CloseSent)) + if (_innerWebSocket == null || _disposed || _state == WebSocketState.Aborted || _state == WebSocketState.Closed) { return _state; } -#if FEATURE_WASM_THREADS - } //lock -#endif - -#if FEATURE_WASM_THREADS - return FastState = _innerWebSocket!.SynchronizationContext.Send(static (BrowserWebSocket self) => - { - lock (self._thisLock) + var st = GetReadyStateLocked(_innerWebSocket!); + if (st == WebSocketState.Closed || st == WebSocketState.CloseSent) { - return GetReadyState(self._innerWebSocket!); - } //lock - }, this); -#else - return FastState = GetReadyState(_innerWebSocket!); -#endif + if (_closeReceived && _closeSent) + { + st = WebSocketState.Closed; + } + else if (_closeReceived && !_closeSent) + { + st = WebSocketState.CloseReceived; + } + else if (!_closeReceived && _closeSent) + { + st = WebSocketState.CloseSent; + } + } + return FastState = st; + } // lock } } @@ -63,50 +66,68 @@ private WebSocketState FastState { get { -#if FEATURE_WASM_THREADS - lock (_thisLock) + lock (_lockObject) { -#endif return _state; -#if FEATURE_WASM_THREADS - } //lock -#endif + } // lock } set { -#if FEATURE_WASM_THREADS - lock (_thisLock) + lock (_lockObject) { -#endif _state = value; -#if FEATURE_WASM_THREADS - } //lock -#endif + } // lock } } - public override WebSocketCloseStatus? CloseStatus => _closeStatus; - public override string? CloseStatusDescription => _closeStatusDescription; - public override string? SubProtocol + public override WebSocketCloseStatus? CloseStatus { get { -#if FEATURE_WASM_THREADS - lock (_thisLock) + lock (_lockObject) { -#endif - ThrowIfDisposed(); - if (_innerWebSocket == null) throw new InvalidOperationException(SR.net_WebSockets_NotConnected); -#if FEATURE_WASM_THREADS - } //lock -#endif + if (_closeStatus != null) + { + return _closeStatus; + } + if (_disposed || _aborted || _cancelled) + { + return null; + } + return GetCloseStatusLocked(); + } + } + } -#if FEATURE_WASM_THREADS - return _innerWebSocket.SynchronizationContext.Send(BrowserInterop.GetProtocol, _innerWebSocket); -#else - return BrowserInterop.GetProtocol(_innerWebSocket); -#endif + public override string? CloseStatusDescription + { + get + { + lock (_lockObject) + { + if (_closeStatusDescription != null) + { + return _closeStatusDescription; + } + if (_disposed || _aborted || _cancelled) + { + return null; + } + return GetCloseStatusDescriptionLocked(); + } + } + } + public override string? SubProtocol + { + get + { + ThrowIfDisposed(); + lock (_lockObject) + { + if (_innerWebSocket == null) throw new InvalidOperationException(SR.net_WebSockets_NotConnected); + return BrowserInterop.GetProtocol(_innerWebSocket); + } // lock } } @@ -114,45 +135,24 @@ public override string? SubProtocol internal Task ConnectAsync(Uri uri, List? requestedSubProtocols, CancellationToken cancellationToken) { - ThrowIfDisposed(); - if (FastState != WebSocketState.None) + lock (_lockObject) { - throw new InvalidOperationException(SR.net_WebSockets_AlreadyStarted); - } -#if FEATURE_WASM_THREADS - JSHost.CurrentOrMainJSSynchronizationContext.Send(_ => - { - lock (_thisLock) + cancellationToken.ThrowIfCancellationRequested(); + ThrowIfDisposed(); + + if (FastState != WebSocketState.None) { - ThrowIfDisposed(); - FastState = WebSocketState.Connecting; - CreateCore(uri, requestedSubProtocols); + throw new InvalidOperationException(SR.net_WebSockets_AlreadyStarted); } - }, null); - - return JSHost.CurrentOrMainJSSynchronizationContext.Post(() => - { - return ConnectAsyncCore(cancellationToken); - }); -#else - FastState = WebSocketState.Connecting; + FastState = WebSocketState.Connecting; + } // lock CreateCore(uri, requestedSubProtocols); return ConnectAsyncCore(cancellationToken); -#endif } public override Task SendAsync(ArraySegment buffer, WebSocketMessageType messageType, bool endOfMessage, CancellationToken cancellationToken) { - cancellationToken.ThrowIfCancellationRequested(); - - ThrowIfDisposed(); - - // fast check of previous _state instead of GetReadyState(), the readyState would be validated on JS side - if (FastState != WebSocketState.Open) - { - throw new InvalidOperationException(SR.net_WebSockets_NotConnected); - } - + // this validation should be synchronous if (messageType != WebSocketMessageType.Binary && messageType != WebSocketMessageType.Text) { throw new ArgumentException(SR.Format(SR.net_WebSockets_Argument_InvalidMessageType, @@ -166,247 +166,135 @@ public override Task SendAsync(ArraySegment buffer, WebSocketMessageType m WebSocketValidate.ValidateArraySegment(buffer, nameof(buffer)); -#if FEATURE_WASM_THREADS - return _innerWebSocket!.SynchronizationContext.Post(() => - { - Task promise; - lock (_thisLock) - { - ThrowIfDisposed(); - promise = SendAsyncCore(buffer, messageType, endOfMessage, cancellationToken); - } //lock will unlock synchronously before promise is resolved! - - return promise; - }); -#else + ThrowIfDisposed(); return SendAsyncCore(buffer, messageType, endOfMessage, cancellationToken); -#endif } public override Task ReceiveAsync(ArraySegment buffer, CancellationToken cancellationToken) { - if (cancellationToken.IsCancellationRequested) - { - return Task.FromException(new OperationCanceledException(cancellationToken)); - } - ThrowIfDisposed(); - // fast check of previous _state instead of GetReadyState(), the readyState would be validated on JS side - var fastState = FastState; - if (fastState != WebSocketState.Open && fastState != WebSocketState.CloseSent) - { - throw new WebSocketException(WebSocketError.InvalidState, SR.Format(SR.net_WebSockets_InvalidState, fastState, "Open, CloseSent")); - } - + // this validation should be synchronous WebSocketValidate.ValidateArraySegment(buffer, nameof(buffer)); -#if FEATURE_WASM_THREADS - return _innerWebSocket!.SynchronizationContext.Post(() => - { - Task promise; - lock (_thisLock) - { - ThrowIfDisposed(); - promise = ReceiveAsyncCore(buffer, cancellationToken); - } //lock will unlock synchronously before task is resolved! - return promise; - }); -#else + ThrowIfDisposed(); return ReceiveAsyncCore(buffer, cancellationToken); -#endif } public override Task CloseOutputAsync(WebSocketCloseStatus closeStatus, string? statusDescription, CancellationToken cancellationToken) { + // this validation should be synchronous + WebSocketValidate.ValidateCloseStatus(closeStatus, statusDescription); + cancellationToken.ThrowIfCancellationRequested(); ThrowIfDisposed(); - WebSocketValidate.ValidateCloseStatus(closeStatus, statusDescription); - var fastState = FastState; - if (fastState == WebSocketState.None || fastState == WebSocketState.Closed) - { - throw new WebSocketException(WebSocketError.InvalidState, SR.Format(SR.net_WebSockets_InvalidState, fastState, "Connecting, Open, CloseSent, Aborted")); - } - -#if FEATURE_WASM_THREADS - return _innerWebSocket!.SynchronizationContext.Post(() => - { - Task promise; - lock (_thisLock) - { - ThrowIfDisposed(); -#endif - var state = State; - if (state == WebSocketState.None || state == WebSocketState.Closed) - { - throw new WebSocketException(WebSocketError.InvalidState, SR.Format(SR.net_WebSockets_InvalidState, state, "Connecting, Open, CloseSent, Aborted")); - } - if (state != WebSocketState.Open && state != WebSocketState.Connecting && state != WebSocketState.Aborted) - { - return Task.CompletedTask; - } -#if FEATURE_WASM_THREADS - promise = CloseAsyncCore(closeStatus, statusDescription, false, cancellationToken); - } //lock will unlock synchronously before task is resolved! - return promise; - }); -#else return CloseAsyncCore(closeStatus, statusDescription, false, cancellationToken); -#endif } public override Task CloseAsync(WebSocketCloseStatus closeStatus, string? statusDescription, CancellationToken cancellationToken) { - cancellationToken.ThrowIfCancellationRequested(); - ThrowIfDisposed(); - + // this validation should be synchronous WebSocketValidate.ValidateCloseStatus(closeStatus, statusDescription); - var fastState = FastState; - if (fastState == WebSocketState.None || fastState == WebSocketState.Closed) - { - throw new WebSocketException(WebSocketError.InvalidState, SR.Format(SR.net_WebSockets_InvalidState, fastState, "Connecting, Open, CloseSent, Aborted")); - } - -#if FEATURE_WASM_THREADS - return _innerWebSocket!.SynchronizationContext.Post(() => - { - Task promise; - lock (_thisLock) - { - ThrowIfDisposed(); -#endif - var state = State; - if (state == WebSocketState.None || state == WebSocketState.Closed) - { - throw new WebSocketException(WebSocketError.InvalidState, SR.Format(SR.net_WebSockets_InvalidState, state, "Connecting, Open, CloseSent, Aborted")); - } - if (state != WebSocketState.Open && state != WebSocketState.Connecting && state != WebSocketState.Aborted && state != WebSocketState.CloseSent) - { - return Task.CompletedTask; - } + cancellationToken.ThrowIfCancellationRequested(); + ThrowIfDisposed(); -#if FEATURE_WASM_THREADS - promise = CloseAsyncCore(closeStatus, statusDescription, state != WebSocketState.Aborted, cancellationToken); - } //lock will unlock synchronously before task is resolved! - return promise; - }); -#else - return CloseAsyncCore(closeStatus, statusDescription, state != WebSocketState.Aborted, cancellationToken); -#endif + return CloseAsyncCore(closeStatus, statusDescription, true, cancellationToken); } public override void Abort() { -#if FEATURE_WASM_THREADS - if (_disposed) - { - return; - } - _innerWebSocket?.SynchronizationContext.Send(static (BrowserWebSocket self) => + lock (_lockObject) { - lock (self._thisLock) + if (_disposed || _aborted) { - AbortCore(self, self.State); + return; } - }, this); -#else - AbortCore(this, this.State); -#endif + var fastState = FastState; + if (fastState == WebSocketState.Closed || fastState == WebSocketState.Aborted) + { + return; + } + + FastState = WebSocketState.Aborted; + _aborted = true; + + // We can call this cross-thread from inside the lock, because there are no callbacks which would lock the same lock + // This will reject/resolve some promises + BrowserInterop.WebSocketAbort(_innerWebSocket!); + } } public override void Dispose() { -#if FEATURE_WASM_THREADS - if (_disposed) + WebSocketState state; + lock (_lockObject) { - return; - } - _innerWebSocket?.SynchronizationContext.Send(static (BrowserWebSocket self) => - { - lock (self._thisLock) + if (_disposed) { - DisposeCore(self); + return; } - }, this); -#else - DisposeCore(this); -#endif - } + _disposed = true; + state = State; - private void ThrowIfDisposed() - { -#if FEATURE_WASM_THREADS - lock (_thisLock) - { -#endif - ObjectDisposedException.ThrowIf(_disposed, this); -#if FEATURE_WASM_THREADS - } //lock -#endif - } + if (state < WebSocketState.Closed && state != WebSocketState.None) + { + _shouldAbort = true; + FastState = WebSocketState.Aborted; + } + else if (state != WebSocketState.Aborted) + { + FastState = WebSocketState.Closed; + } - #region methods always called on one thread only, exclusively + } // lock - private static void DisposeCore(BrowserWebSocket self) - { - if (!self._disposed) + static void Cleanup(object? _state) { + var self = (BrowserWebSocket)_state!; var state = self.State; - self._disposed = true; - if (state < WebSocketState.Aborted && state != WebSocketState.None) - { - AbortCore(self, state); - } - if (self.FastState != WebSocketState.Aborted) + lock (self._lockObject) { - self.FastState = WebSocketState.Closed; + if (self._shouldAbort && !self._aborted) + { + self._aborted = true; + self._shouldAbort = false; + + // We can call this inside the lock, because there are no callbacks which would lock the same lock + // This will reject/resolve some promises + BrowserInterop.WebSocketAbort(self._innerWebSocket!); + } } self._innerWebSocket?.Dispose(); - self._innerWebSocket = null; self.responseStatusHandle?.Dispose(); } + +#if FEATURE_WASM_THREADS + // if this is finalizer thread, we need to postpone the abort -> dispose + _innerWebSocket?.SynchronizationContext.Post(Cleanup, this); +#else + Cleanup(this); +#endif } - private static void AbortCore(BrowserWebSocket self, WebSocketState currentState) + private void ThrowIfDisposed() { - if (!self._disposed && currentState != WebSocketState.Closed) + lock (_lockObject) { - self.FastState = WebSocketState.Aborted; - self._aborted = true; - if (self._innerWebSocket != null) - { - BrowserInterop.WebSocketAbort(self._innerWebSocket!); - } - } + ObjectDisposedException.ThrowIf(_disposed, this); + } // lock } + private void CreateCore(Uri uri, List? requestedSubProtocols) { try { string[]? subProtocols = requestedSubProtocols?.ToArray(); - var onClose = (int code, string reason) => - { - _closeStatus = (WebSocketCloseStatus)code; - _closeStatusDescription = reason; -#if FEATURE_WASM_THREADS - lock (_thisLock) - { -#endif - WebSocketState state = State; - if (state == WebSocketState.Connecting || state == WebSocketState.Open || state == WebSocketState.CloseSent) - { - FastState = WebSocketState.Closed; - } -#if FEATURE_WASM_THREADS - } //lock -#endif - }; Memory responseMemory = new Memory(responseStatus); responseStatusHandle = responseMemory.Pin(); - _innerWebSocket = BrowserInterop.UnsafeCreate(uri.ToString(), subProtocols, responseStatusHandle.Value, onClose); + _innerWebSocket = BrowserInterop.UnsafeCreate(uri.ToString(), subProtocols, responseStatusHandle.Value); } catch (Exception) { @@ -417,55 +305,52 @@ private void CreateCore(Uri uri, List? requestedSubProtocols) private async Task ConnectAsyncCore(CancellationToken cancellationToken) { -#if FEATURE_WASM_THREADS - lock (_thisLock) + Task openTask; + + lock (_lockObject) { -#endif if (_aborted) { FastState = WebSocketState.Closed; throw new WebSocketException(WebSocketError.Faulted, SR.net_webstatus_ConnectFailure); } ThrowIfDisposed(); -#if FEATURE_WASM_THREADS - } //lock -#endif + + openTask = BrowserInterop.WebSocketOpen(_innerWebSocket!); + } // lock try { - var openTask = BrowserInterop.WebSocketOpen(_innerWebSocket!); + await CancellationHelper(openTask!, cancellationToken, WebSocketState.Connecting).ConfigureAwait(false); - await CancelationHelper(openTask!, cancellationToken, FastState).ConfigureAwait(true); -#if FEATURE_WASM_THREADS - lock (_thisLock) + lock (_lockObject) { -#endif - if (State == WebSocketState.Connecting) + WebSocketState state = State; + if (state == WebSocketState.Connecting) { FastState = WebSocketState.Open; } -#if FEATURE_WASM_THREADS - } //lock -#endif + } // lock } catch (OperationCanceledException ex) { -#if FEATURE_WASM_THREADS - lock (_thisLock) + lock (_lockObject) { -#endif FastState = WebSocketState.Closed; if (_aborted) { throw new WebSocketException(WebSocketError.Faulted, SR.net_webstatus_ConnectFailure, ex); } -#if FEATURE_WASM_THREADS - } //lock -#endif + } // lock + throw; } catch (Exception) { + lock (_lockObject) + { + FastState = WebSocketState.Closed; + } Dispose(); throw; } @@ -473,161 +358,295 @@ private async Task ConnectAsyncCore(CancellationToken cancellationToken) private async Task SendAsyncCore(ArraySegment buffer, WebSocketMessageType messageType, bool endOfMessage, CancellationToken cancellationToken) { + WebSocketState previousState = WebSocketState.None; + Task? sendTask; + MemoryHandle? pinBuffer = null; + try { - var sendTask = BrowserInterop.UnsafeSendSync(_innerWebSocket!, buffer, messageType, endOfMessage); - if (sendTask != null) + lock (_lockObject) { - await CancelationHelper(sendTask, cancellationToken, FastState).ConfigureAwait(true); + cancellationToken.ThrowIfCancellationRequested(); + ThrowIfDisposed(); + + previousState = FastState; + if (previousState != WebSocketState.Open && previousState != WebSocketState.CloseReceived) + { + throw new InvalidOperationException(SR.net_WebSockets_NotConnected); + } + + if (buffer.Count == 0) + { + sendTask = BrowserInterop.WebSocketSend(_innerWebSocket!, IntPtr.Zero, 0, (int)messageType, endOfMessage); + } + else + { + Memory bufferMemory = buffer.AsMemory(); + pinBuffer = bufferMemory.Pin(); + sendTask = BrowserInterop.UnsafeSend(_innerWebSocket!, pinBuffer.Value, bufferMemory.Length, messageType, endOfMessage); + } } -#if FEATURE_WASM_THREADS - // return synchronously, not supported with MT - else + + if (sendTask != null) // this is optimization for single-threaded build, see resolvedPromise() in web-socket.ts. Null means synchronously resolved. { - Environment.FailFast("BrowserWebSocket.SendAsyncCore: Unexpected synchronous result"); + await CancellationHelper(sendTask, cancellationToken, previousState, pinBuffer).ConfigureAwait(false); } -#endif } catch (JSException ex) { if (ex.Message.StartsWith("InvalidState:")) { - throw new WebSocketException(WebSocketError.InvalidState, SR.Format(SR.net_WebSockets_InvalidState, State, "Open"), ex); + throw new WebSocketException(WebSocketError.InvalidState, SR.Format(SR.net_WebSockets_InvalidState, previousState, "Open"), ex); } throw new WebSocketException(WebSocketError.NativeError, ex); } + finally + { + // must be after await! + pinBuffer?.Dispose(); + } } private async Task ReceiveAsyncCore(ArraySegment buffer, CancellationToken cancellationToken) { + WebSocketState previousState = WebSocketState.None; + Task? receiveTask; + MemoryHandle? pinBuffer = null; try { - Memory bufferMemory = buffer.AsMemory(); - using (MemoryHandle pinBuffer = bufferMemory.Pin()) + lock (_lockObject) { - var receiveTask = BrowserInterop.ReceiveUnsafeSync(_innerWebSocket!, pinBuffer, bufferMemory.Length); - if (receiveTask != null) - { - await CancelationHelper(receiveTask, cancellationToken, FastState).ConfigureAwait(true); - } -#if FEATURE_WASM_THREADS - // return synchronously, not supported with MT - else + cancellationToken.ThrowIfCancellationRequested(); + ThrowIfDisposed(); + + previousState = FastState; + if (previousState != WebSocketState.Open && previousState != WebSocketState.CloseSent) { - Environment.FailFast("BrowserWebSocket.ReceiveAsyncCore: Unexpected synchronous result"); + throw new WebSocketException(WebSocketError.InvalidState, SR.Format(SR.net_WebSockets_InvalidState, previousState, "Open, CloseSent")); } -#endif + Memory bufferMemory = buffer.AsMemory(); + pinBuffer = bufferMemory.Pin(); + receiveTask = BrowserInterop.ReceiveUnsafe(_innerWebSocket!, pinBuffer.Value, bufferMemory.Length); + } -#if FEATURE_WASM_THREADS - lock (_thisLock) - { -#endif - return ConvertResponse(this); -#if FEATURE_WASM_THREADS - } //lock -#endif + if (receiveTask != null) // this is optimization for single-threaded build, see resolvedPromise() in web-socket.ts. Null means synchronously resolved. + { + await CancellationHelper(receiveTask, cancellationToken, previousState, pinBuffer).ConfigureAwait(false); } + + return ConvertResponse(); } catch (JSException ex) { if (ex.Message.StartsWith("InvalidState:")) { - throw new WebSocketException(WebSocketError.InvalidState, SR.Format(SR.net_WebSockets_InvalidState, State, "Open, CloseSent"), ex); + throw new WebSocketException(WebSocketError.InvalidState, SR.Format(SR.net_WebSockets_InvalidState, previousState, "Open, CloseSent"), ex); } throw new WebSocketException(WebSocketError.NativeError, ex); } + finally + { + // must be after await! + pinBuffer?.Dispose(); + } } - private static WebSocketReceiveResult ConvertResponse(BrowserWebSocket self) + private WebSocketReceiveResult ConvertResponse() { const int countIndex = 0; const int typeIndex = 1; const int endIndex = 2; - WebSocketMessageType messageType = (WebSocketMessageType)self.responseStatus[typeIndex]; + int count; + WebSocketMessageType messageType; + bool isEnd = responseStatus[endIndex] != 0; + lock (_lockObject) + { + messageType = (WebSocketMessageType)responseStatus[typeIndex]; + count = responseStatus[countIndex]; + if (messageType == WebSocketMessageType.Close) + { + _closeReceived = true; + FastState = _closeSent ? WebSocketState.Closed : WebSocketState.CloseReceived; + ForceReadCloseStatusLocked(); + } + } // lock + if (messageType == WebSocketMessageType.Close) { - return new WebSocketReceiveResult(self.responseStatus[countIndex], messageType, self.responseStatus[endIndex] != 0, self.CloseStatus, self.CloseStatusDescription); + switch (_closeStatus ?? WebSocketCloseStatus.NormalClosure) + { + case WebSocketCloseStatus.NormalClosure: + case WebSocketCloseStatus.Empty: + return new WebSocketReceiveResult(count, messageType, isEnd, _closeStatus, _closeStatusDescription); + case WebSocketCloseStatus.InvalidMessageType: + case WebSocketCloseStatus.InvalidPayloadData: + throw new WebSocketException(WebSocketError.InvalidMessageType, _closeStatusDescription); + case WebSocketCloseStatus.EndpointUnavailable: + throw new WebSocketException(WebSocketError.NotAWebSocket, _closeStatusDescription); + case WebSocketCloseStatus.ProtocolError: + throw new WebSocketException(WebSocketError.UnsupportedProtocol, _closeStatusDescription); + case WebSocketCloseStatus.InternalServerError: + throw new WebSocketException(WebSocketError.Faulted, _closeStatusDescription); + default: + throw new WebSocketException(WebSocketError.NativeError, (int)_closeStatus!.Value, _closeStatusDescription); + } } - return new WebSocketReceiveResult(self.responseStatus[countIndex], messageType, self.responseStatus[endIndex] != 0); + return new WebSocketReceiveResult(count, messageType, isEnd); } - private async Task CloseAsyncCore(WebSocketCloseStatus closeStatus, string? statusDescription, bool waitForCloseReceived, CancellationToken cancellationToken) + private async Task CloseAsyncCore(WebSocketCloseStatus closeStatus, string? statusDescription, bool fullClose, CancellationToken cancellationToken) { - _closeStatus = closeStatus; - _closeStatusDescription = statusDescription; - - var closeTask = BrowserInterop.WebSocketClose(_innerWebSocket!, (int)closeStatus, statusDescription, waitForCloseReceived); - if (closeTask != null) + Task? closeTask; + WebSocketState previousState; + lock (_lockObject) { - await CancelationHelper(closeTask, cancellationToken, FastState).ConfigureAwait(true); + cancellationToken.ThrowIfCancellationRequested(); + + previousState = FastState; + if (_aborted) + { + return; + } + if (!_closeReceived) + { + _closeStatus = closeStatus; + _closeStatusDescription = statusDescription; + } + if (previousState == WebSocketState.None || previousState == WebSocketState.Closed) + { + throw new WebSocketException(WebSocketError.InvalidState, SR.Format(SR.net_WebSockets_InvalidState, previousState, "Connecting, Open, CloseSent, Aborted")); + } + + _closeSent = true; + + closeTask = BrowserInterop.WebSocketClose(_innerWebSocket!, (int)closeStatus, statusDescription, fullClose); } -#if FEATURE_WASM_THREADS - // return synchronously, not supported with MT - else + + if (closeTask != null) // this is optimization for single-threaded build, see resolvedPromise() in web-socket.ts. Null means synchronously resolved. { - Environment.FailFast("BrowserWebSocket.CloseAsyncCore: Unexpected synchronous result"); + await CancellationHelper(closeTask, cancellationToken, previousState).ConfigureAwait(false); } -#endif -#if FEATURE_WASM_THREADS - lock (_thisLock) + if (fullClose) { -#endif - var state = State; - if (state == WebSocketState.Open || state == WebSocketState.Connecting || state == WebSocketState.CloseSent) + lock (_lockObject) { - FastState = waitForCloseReceived ? WebSocketState.Closed : WebSocketState.CloseSent; + _closeReceived = true; + ForceReadCloseStatusLocked(); + _ = State; } -#if FEATURE_WASM_THREADS - } //lock -#endif + } } - private async Task CancelationHelper(Task jsTask, CancellationToken cancellationToken, WebSocketState previousState) + private async Task CancellationHelper(Task promise, CancellationToken cancellationToken, WebSocketState previousState, IDisposable? disposable = null) { - if (jsTask.IsCompletedSuccessfully) - { - return; - } try { - using (var receiveRegistration = cancellationToken.Register(static s => + cancellationToken.ThrowIfCancellationRequested(); + if (promise.IsCompletedSuccessfully) { - CancelablePromise.CancelPromise((Task)s!); - }, jsTask)) - { - await jsTask.ConfigureAwait(true); + disposable?.Dispose(); return; } - } - catch (JSException ex) - { - if (State == WebSocketState.Aborted) + if (promise.IsCompleted) { - throw new OperationCanceledException(nameof(WebSocketState.Aborted), ex); + // don't have to register for cancelation + await promise.ConfigureAwait(false); + return; } - if (cancellationToken.IsCancellationRequested) + using (var receiveRegistration = cancellationToken.Register(static s => { - FastState = WebSocketState.Aborted; - throw new OperationCanceledException(cancellationToken); - } - if (ex.Message == "Error: OperationCanceledException") + CancelablePromise.CancelPromise((Task)s!); + }, promise)) { - FastState = WebSocketState.Aborted; - throw new OperationCanceledException("The operation was cancelled.", ex, cancellationToken); + await promise.ConfigureAwait(false); + return; } - if (previousState == WebSocketState.Connecting) + } + catch (Exception ex) + { + lock (_lockObject) { - throw new WebSocketException(WebSocketError.Faulted, SR.net_webstatus_ConnectFailure, ex); + var state = State; + if (state == WebSocketState.Aborted) + { + ForceReadCloseStatusLocked(); + throw new OperationCanceledException(nameof(WebSocketState.Aborted), ex); + } + if (ex is OperationCanceledException) + { + if(state != WebSocketState.Closed) + { + FastState = WebSocketState.Aborted; + } + _cancelled = true; + throw; + } + if (state != WebSocketState.Closed && cancellationToken.IsCancellationRequested) + { + FastState = WebSocketState.Aborted; + _cancelled = true; + throw new OperationCanceledException(cancellationToken); + } + if (state != WebSocketState.Closed && ex.Message == "Error: OperationCanceledException") + { + FastState = WebSocketState.Aborted; + _cancelled = true; + throw new OperationCanceledException("The operation was cancelled.", ex, cancellationToken); + } + if (previousState == WebSocketState.Connecting) + { + ForceReadCloseStatusLocked(); + throw new WebSocketException(WebSocketError.Faulted, SR.net_webstatus_ConnectFailure, ex); + } + throw new WebSocketException(WebSocketError.NativeError, ex); } - throw new WebSocketException(WebSocketError.NativeError, ex); } + finally + { + disposable?.Dispose(); + } + } + + // needs to be called with locked _lockObject + private void ForceReadCloseStatusLocked() + { + if (!_disposed && _closeStatus == null) + { + GetCloseStatusLocked(); + GetCloseStatusDescriptionLocked(); + } + } + + // needs to be called with locked _lockObject + private WebSocketCloseStatus? GetCloseStatusLocked() + { + ThrowIfDisposed(); + var closeStatus = BrowserInterop.GetCloseStatus(_innerWebSocket); + if (closeStatus != null && _closeStatus == null) + { + _closeStatus = closeStatus; + } + return _closeStatus; + } + + // needs to be called with locked _lockObject + private string? GetCloseStatusDescriptionLocked() + { + ThrowIfDisposed(); + var closeStatusDescription = BrowserInterop.GetCloseStatusDescription(_innerWebSocket); + if (closeStatusDescription != null && _closeStatusDescription == null) + { + _closeStatusDescription = closeStatusDescription; + } + return _closeStatusDescription; } - private static WebSocketState GetReadyState(JSObject innerWebSocket) + // needs to be called with locked _lockObject + private static WebSocketState GetReadyStateLocked(JSObject innerWebSocket) { var readyState = BrowserInterop.GetReadyState(innerWebSocket); // https://developer.mozilla.org/en-US/docs/Web/API/WebSocket/readyState @@ -641,6 +660,5 @@ private static WebSocketState GetReadyState(JSObject innerWebSocket) }; } - #endregion } } diff --git a/src/libraries/System.Net.WebSockets.Client/tests/CloseTest.cs b/src/libraries/System.Net.WebSockets.Client/tests/CloseTest.cs index 26dfd9a4fd01ba..81f80a1e647bdb 100644 --- a/src/libraries/System.Net.WebSockets.Client/tests/CloseTest.cs +++ b/src/libraries/System.Net.WebSockets.Client/tests/CloseTest.cs @@ -33,12 +33,12 @@ public class CloseTest : ClientWebSocketTestBase public CloseTest(ITestOutputHelper output) : base(output) { } - [ActiveIssue("https://github.com/dotnet/runtime/issues/28957")] + [ActiveIssue("https://github.com/dotnet/runtime/issues/28957", typeof(PlatformDetection), nameof(PlatformDetection.IsNotBrowser))] [OuterLoop("Uses external servers", typeof(PlatformDetection), nameof(PlatformDetection.LocalEchoServerIsNotAvailable))] [ConditionalTheory(nameof(WebSocketsSupported)), MemberData(nameof(EchoServersAndBoolean))] public async Task CloseAsync_ServerInitiatedClose_Success(Uri server, bool useCloseOutputAsync) { - const string closeWebSocketMetaCommand = ".close"; + const string shutdownWebSocketMetaCommand = ".shutdown"; using (ClientWebSocket cws = await GetConnectedWebSocket(server, TimeOutMilliseconds, _output)) { @@ -46,7 +46,7 @@ public async Task CloseAsync_ServerInitiatedClose_Success(Uri server, bool useCl _output.WriteLine("SendAsync starting."); await cws.SendAsync( - WebSocketData.GetBufferFromText(closeWebSocketMetaCommand), + WebSocketData.GetBufferFromText(shutdownWebSocketMetaCommand), WebSocketMessageType.Text, true, cts.Token); @@ -59,26 +59,27 @@ await cws.SendAsync( // Verify received server-initiated close message. Assert.Equal(WebSocketCloseStatus.NormalClosure, recvResult.CloseStatus); - Assert.Equal(closeWebSocketMetaCommand, recvResult.CloseStatusDescription); + Assert.Equal(shutdownWebSocketMetaCommand, recvResult.CloseStatusDescription); Assert.Equal(WebSocketMessageType.Close, recvResult.MessageType); // Verify current websocket state as CloseReceived which indicates only partial close. Assert.Equal(WebSocketState.CloseReceived, cws.State); Assert.Equal(WebSocketCloseStatus.NormalClosure, cws.CloseStatus); - Assert.Equal(closeWebSocketMetaCommand, cws.CloseStatusDescription); + Assert.Equal(shutdownWebSocketMetaCommand, cws.CloseStatusDescription); // Send back close message to acknowledge server-initiated close. _output.WriteLine("Close starting."); + var closeStatus = PlatformDetection.IsNotBrowser ? WebSocketCloseStatus.InvalidMessageType : (WebSocketCloseStatus)3210; await (useCloseOutputAsync ? - cws.CloseOutputAsync(WebSocketCloseStatus.InvalidMessageType, string.Empty, cts.Token) : - cws.CloseAsync(WebSocketCloseStatus.InvalidMessageType, string.Empty, cts.Token)); + cws.CloseOutputAsync(closeStatus, string.Empty, cts.Token) : + cws.CloseAsync(closeStatus, string.Empty, cts.Token)); _output.WriteLine("Close done."); Assert.Equal(WebSocketState.Closed, cws.State); // Verify that there is no follow-up echo close message back from the server by // making sure the close code and message are the same as from the first server close message. Assert.Equal(WebSocketCloseStatus.NormalClosure, cws.CloseStatus); - Assert.Equal(closeWebSocketMetaCommand, cws.CloseStatusDescription); + Assert.Equal(shutdownWebSocketMetaCommand, cws.CloseStatusDescription); } } @@ -224,7 +225,6 @@ await Assert.ThrowsAnyAsync(async () => [OuterLoop("Uses external servers", typeof(PlatformDetection), nameof(PlatformDetection.LocalEchoServerIsNotAvailable))] [ConditionalTheory(nameof(WebSocketsSupported)), MemberData(nameof(EchoServers))] - [SkipOnPlatform(TestPlatforms.Browser, "This never really worked for browser, it was just lucky timing that browser's `close` event was executed in next browser tick, for this test. See also https://github.com/dotnet/runtime/issues/45538")] public async Task CloseOutputAsync_ClientInitiated_CanReceive_CanClose(Uri server) { string message = "Hello WebSockets!"; @@ -233,7 +233,7 @@ public async Task CloseOutputAsync_ClientInitiated_CanReceive_CanClose(Uri serve { var cts = new CancellationTokenSource(TimeOutMilliseconds); - var closeStatus = WebSocketCloseStatus.InvalidPayloadData; + var closeStatus = PlatformDetection.IsNotBrowser ? WebSocketCloseStatus.InvalidPayloadData : (WebSocketCloseStatus)3210; string closeDescription = "CloseOutputAsync_Client_InvalidPayloadData"; await cws.SendAsync(WebSocketData.GetBufferFromText(message), WebSocketMessageType.Text, true, cts.Token); @@ -261,7 +261,64 @@ public async Task CloseOutputAsync_ClientInitiated_CanReceive_CanClose(Uri serve } } - [ActiveIssue("https://github.com/dotnet/runtime/issues/28957")] + [ActiveIssue("https://github.com/dotnet/runtime/issues/28957", typeof(PlatformDetection), nameof(PlatformDetection.IsNotBrowser))] + [OuterLoop("Uses external servers", typeof(PlatformDetection), nameof(PlatformDetection.LocalEchoServerIsNotAvailable))] + [ConditionalTheory(nameof(WebSocketsSupported)), MemberData(nameof(EchoServers))] + public async Task CloseOutputAsync_ServerInitiated_CanReceive(Uri server) + { + string message = "Hello WebSockets!"; + var expectedCloseStatus = WebSocketCloseStatus.NormalClosure; + var expectedCloseDescription = ".shutdownafter"; + + using (ClientWebSocket cws = await GetConnectedWebSocket(server, TimeOutMilliseconds, _output)) + { + var cts = new CancellationTokenSource(TimeOutMilliseconds); + + await cws.SendAsync( + WebSocketData.GetBufferFromText(expectedCloseDescription), + WebSocketMessageType.Text, + true, + cts.Token); + + // Should be able to receive the message echoed by the server. + var recvBuffer = new byte[100]; + var segmentRecv = new ArraySegment(recvBuffer); + WebSocketReceiveResult recvResult = await cws.ReceiveAsync(segmentRecv, cts.Token); + Assert.Equal(expectedCloseDescription.Length, recvResult.Count); + segmentRecv = new ArraySegment(segmentRecv.Array, 0, recvResult.Count); + Assert.Equal(expectedCloseDescription, WebSocketData.GetTextFromBuffer(segmentRecv)); + Assert.Null(recvResult.CloseStatus); + Assert.Null(recvResult.CloseStatusDescription); + + // Should be able to receive a shutdown message. + segmentRecv = new ArraySegment(recvBuffer); + recvResult = await cws.ReceiveAsync(segmentRecv, cts.Token); + Assert.Equal(0, recvResult.Count); + Assert.Equal(expectedCloseStatus, recvResult.CloseStatus); + Assert.Equal(expectedCloseDescription, recvResult.CloseStatusDescription); + + // Verify WebSocket state + Assert.Equal(expectedCloseStatus, cws.CloseStatus); + Assert.Equal(expectedCloseDescription, cws.CloseStatusDescription); + + Assert.Equal(WebSocketState.CloseReceived, cws.State); + + // Should be able to send. + await cws.SendAsync(WebSocketData.GetBufferFromText(message), WebSocketMessageType.Text, true, cts.Token); + + // Cannot change the close status/description with the final close. + var closeStatus = PlatformDetection.IsNotBrowser ? WebSocketCloseStatus.InvalidPayloadData : (WebSocketCloseStatus)3210; + var closeDescription = "CloseOutputAsync_Client_Description"; + + await cws.CloseAsync(closeStatus, closeDescription, cts.Token); + + Assert.Equal(expectedCloseStatus, cws.CloseStatus); + Assert.Equal(expectedCloseDescription, cws.CloseStatusDescription); + Assert.Equal(WebSocketState.Closed, cws.State); + } + } + + [ActiveIssue("https://github.com/dotnet/runtime/issues/28957", typeof(PlatformDetection), nameof(PlatformDetection.IsNotBrowser))] [OuterLoop("Uses external servers", typeof(PlatformDetection), nameof(PlatformDetection.LocalEchoServerIsNotAvailable))] [ConditionalTheory(nameof(WebSocketsSupported)), MemberData(nameof(EchoServers))] public async Task CloseOutputAsync_ServerInitiated_CanSend(Uri server) @@ -298,7 +355,7 @@ await cws.SendAsync( await cws.SendAsync(WebSocketData.GetBufferFromText(message), WebSocketMessageType.Text, true, cts.Token); // Cannot change the close status/description with the final close. - var closeStatus = WebSocketCloseStatus.InvalidPayloadData; + var closeStatus = PlatformDetection.IsNotBrowser ? WebSocketCloseStatus.InvalidPayloadData : (WebSocketCloseStatus)3210; var closeDescription = "CloseOutputAsync_Client_Description"; await cws.CloseAsync(closeStatus, closeDescription, cts.Token); diff --git a/src/libraries/System.Net.WebSockets.Client/tests/SendReceiveTest.cs b/src/libraries/System.Net.WebSockets.Client/tests/SendReceiveTest.cs index 2016f4f7285a87..357dcb0945d665 100644 --- a/src/libraries/System.Net.WebSockets.Client/tests/SendReceiveTest.cs +++ b/src/libraries/System.Net.WebSockets.Client/tests/SendReceiveTest.cs @@ -249,12 +249,12 @@ public async Task SendAsync_MultipleOutstandingSendOperations_Throws(Uri server) [OuterLoop("Uses external servers", typeof(PlatformDetection), nameof(PlatformDetection.LocalEchoServerIsNotAvailable))] [ConditionalTheory(nameof(WebSocketsSupported)), MemberData(nameof(EchoServers))] // This will also pass when no exception is thrown. Current implementation doesn't throw. - [ActiveIssue("https://github.com/dotnet/runtime/issues/83517", typeof(PlatformDetection), nameof(PlatformDetection.IsBrowser))] + [ActiveIssue("https://github.com/dotnet/runtime/issues/83517", typeof(PlatformDetection), nameof(PlatformDetection.IsNodeJS))] public async Task ReceiveAsync_MultipleOutstandingReceiveOperations_Throws(Uri server) { using (ClientWebSocket cws = await GetConnectedWebSocket(server, TimeOutMilliseconds, _output)) { - var cts = new CancellationTokenSource(TimeOutMilliseconds); + var cts = new CancellationTokenSource(PlatformDetection.LocalEchoServerIsNotAvailable ? TimeOutMilliseconds : 200); Task[] tasks = new Task[2]; diff --git a/src/libraries/System.Net.WebSockets.Client/tests/System.Net.WebSockets.Client.Tests.csproj b/src/libraries/System.Net.WebSockets.Client/tests/System.Net.WebSockets.Client.Tests.csproj index 54be463b694e12..b797fcf1894599 100644 --- a/src/libraries/System.Net.WebSockets.Client/tests/System.Net.WebSockets.Client.Tests.csproj +++ b/src/libraries/System.Net.WebSockets.Client/tests/System.Net.WebSockets.Client.Tests.csproj @@ -17,9 +17,6 @@ 01:15:00 - - - <_XUnitBackgroundExec>false diff --git a/src/libraries/System.Runtime.InteropServices.JavaScript/tests/System.Runtime.InteropServices.JavaScript.UnitTests/System.Runtime.InteropServices.JavaScript.Tests.csproj b/src/libraries/System.Runtime.InteropServices.JavaScript/tests/System.Runtime.InteropServices.JavaScript.UnitTests/System.Runtime.InteropServices.JavaScript.Tests.csproj index 3e6c61c71a1419..ab0e7ac77251ef 100644 --- a/src/libraries/System.Runtime.InteropServices.JavaScript/tests/System.Runtime.InteropServices.JavaScript.UnitTests/System.Runtime.InteropServices.JavaScript.Tests.csproj +++ b/src/libraries/System.Runtime.InteropServices.JavaScript/tests/System.Runtime.InteropServices.JavaScript.UnitTests/System.Runtime.InteropServices.JavaScript.Tests.csproj @@ -1,4 +1,5 @@ + $([MSBuild]::GetTargetPlatformIdentifier('$(TargetFramework)')) true diff --git a/src/libraries/System.Runtime.InteropServices.JavaScript/tests/System.Runtime.InteropServices.JavaScript.UnitTests/System/Runtime/InteropServices/JavaScript/WebWorkerTest.cs b/src/libraries/System.Runtime.InteropServices.JavaScript/tests/System.Runtime.InteropServices.JavaScript.UnitTests/System/Runtime/InteropServices/JavaScript/WebWorkerTest.cs index 31352864753af6..cd8e24d6343e70 100644 --- a/src/libraries/System.Runtime.InteropServices.JavaScript/tests/System.Runtime.InteropServices.JavaScript.UnitTests/System/Runtime/InteropServices/JavaScript/WebWorkerTest.cs +++ b/src/libraries/System.Runtime.InteropServices.JavaScript/tests/System.Runtime.InteropServices.JavaScript.UnitTests/System/Runtime/InteropServices/JavaScript/WebWorkerTest.cs @@ -503,5 +503,102 @@ public async Task JSObject_CapturesAffinity(Executor executor1, Executor executo } #endregion + + #region WebSocket + + [Theory, MemberData(nameof(GetTargetThreads))] + public async Task WebSocketClient_ContentInSameThread(Executor executor) + { + var cts = new CancellationTokenSource(TimeoutMilliseconds); + + var uri = new Uri(WebWorkerTestHelper.LocalWsEcho + "?guid=" + Guid.NewGuid()); + var message = "hello"; + var send = Encoding.UTF8.GetBytes(message); + var receive = new byte[100]; + + await executor.Execute(async () => + { + using var client = new ClientWebSocket(); + await client.ConnectAsync(uri, CancellationToken.None); + await client.SendAsync(send, WebSocketMessageType.Text, true, CancellationToken.None); + + var res = await client.ReceiveAsync(receive, CancellationToken.None); + Assert.Equal(WebSocketMessageType.Text, res.MessageType); + Assert.True(res.EndOfMessage); + Assert.Equal(send.Length, res.Count); + Assert.Equal(message, Encoding.UTF8.GetString(receive, 0, res.Count)); + }, cts.Token); + } + + + [Theory, MemberData(nameof(GetTargetThreads2x))] + public Task WebSocketClient_ResponseCloseInDifferentThread(Executor executor1, Executor executor2) + { + var cts = new CancellationTokenSource(TimeoutMilliseconds); + + var uri = new Uri(WebWorkerTestHelper.LocalWsEcho + "?guid=" + Guid.NewGuid()); + var message = "hello"; + var send = Encoding.UTF8.GetBytes(message); + var receive = new byte[100]; + + var e1Job = async (Task e2done, TaskCompletionSource e1State) => + { + using var client = new ClientWebSocket(); + await client.ConnectAsync(uri, CancellationToken.None); + await client.SendAsync(send, WebSocketMessageType.Text, true, CancellationToken.None); + + // share the state with the E2 continuation + e1State.SetResult(client); + await e2done; + }; + + var e2Job = async (ClientWebSocket client) => + { + var res = await client.ReceiveAsync(receive, CancellationToken.None); + Assert.Equal(WebSocketMessageType.Text, res.MessageType); + Assert.True(res.EndOfMessage); + Assert.Equal(send.Length, res.Count); + Assert.Equal(message, Encoding.UTF8.GetString(receive, 0, res.Count)); + + await client.CloseAsync(WebSocketCloseStatus.NormalClosure, "bye", CancellationToken.None); + }; + + return ActionsInDifferentThreads(executor1, executor2, e1Job, e2Job, cts); + } + + [Theory, MemberData(nameof(GetTargetThreads2x))] + public Task WebSocketClient_CancelInDifferentThread(Executor executor1, Executor executor2) + { + var cts = new CancellationTokenSource(TimeoutMilliseconds); + + var uri = new Uri(WebWorkerTestHelper.LocalWsEcho + "?guid=" + Guid.NewGuid()); + var message = ".delay5sec"; // this will make the loopback server slower + var send = Encoding.UTF8.GetBytes(message); + var receive = new byte[100]; + + var e1Job = async (Task e2done, TaskCompletionSource e1State) => + { + using var client = new ClientWebSocket(); + await client.ConnectAsync(uri, CancellationToken.None); + await client.SendAsync(send, WebSocketMessageType.Text, true, CancellationToken.None); + + // share the state with the E2 continuation + e1State.SetResult(client); + await e2done; + }; + + var e2Job = async (ClientWebSocket client) => + { + CancellationTokenSource cts2 = new CancellationTokenSource(); + var resTask = client.ReceiveAsync(receive, cts2.Token); + cts2.Cancel(); + var ex = await Assert.ThrowsAsync(() => resTask); + Assert.Equal(cts2.Token, ex.CancellationToken); + }; + + return ActionsInDifferentThreads(executor1, executor2, e1Job, e2Job, cts); + } + + #endregion } } diff --git a/src/mono/browser/runtime/web-socket.ts b/src/mono/browser/runtime/web-socket.ts index ec88b12d9ff14e..bbcac93a1bd790 100644 --- a/src/mono/browser/runtime/web-socket.ts +++ b/src/mono/browser/runtime/web-socket.ts @@ -11,7 +11,6 @@ import { VoidPtr } from "./types/emscripten"; import { PromiseController } from "./types/internal"; import { mono_log_warn } from "./logging"; import { viewOrCopy, utf8ToStringRelaxed, stringToUTF8 } from "./strings"; -import { IDisposable } from "./marshal"; import { wrap_as_cancelable } from "./cancelable-promise"; import { assert_js_interop } from "./invoke-js"; @@ -25,9 +24,10 @@ const wasm_ws_pending_open_promise_used = Symbol.for("wasm wasm_ws_pending_open_ const wasm_ws_pending_close_promises = Symbol.for("wasm ws_pending_close_promises"); const wasm_ws_pending_send_promises = Symbol.for("wasm ws_pending_send_promises"); const wasm_ws_is_aborted = Symbol.for("wasm ws_is_aborted"); -const wasm_ws_on_closed = Symbol.for("wasm ws_on_closed"); +const wasm_ws_close_sent = Symbol.for("wasm wasm_ws_close_sent"); +const wasm_ws_close_received = Symbol.for("wasm wasm_ws_close_received"); const wasm_ws_receive_status_ptr = Symbol.for("wasm ws_receive_status_ptr"); -let mono_wasm_web_socket_close_warning = false; + const ws_send_buffer_blocking_threshold = 65536; const emptyBuffer = new Uint8Array(); @@ -43,11 +43,10 @@ function verifyEnvironment() { } } -export function ws_wasm_create(uri: string, sub_protocols: string[] | null, receive_status_ptr: VoidPtr, onClosed: (code: number, reason: string) => void): WebSocketExtension { +export function ws_wasm_create(uri: string, sub_protocols: string[] | null, receive_status_ptr: VoidPtr): WebSocketExtension { verifyEnvironment(); assert_js_interop(); mono_assert(uri && typeof uri === "string", () => `ERR12: Invalid uri ${typeof uri}`); - mono_assert(typeof onClosed === "function", () => `ERR12: Invalid onClosed ${typeof onClosed}`); const ws = new globalThis.WebSocket(uri, sub_protocols || undefined) as WebSocketExtension; const { promise_control: open_promise_control } = createPromiseController(); @@ -58,7 +57,6 @@ export function ws_wasm_create(uri: string, sub_protocols: string[] | null, rece ws[wasm_ws_pending_send_promises] = []; ws[wasm_ws_pending_close_promises] = []; ws[wasm_ws_receive_status_ptr] = receive_status_ptr; - ws[wasm_ws_on_closed] = onClosed as any; ws.binaryType = "arraybuffer"; const local_on_open = () => { if (ws[wasm_ws_is_aborted]) return; @@ -77,7 +75,9 @@ export function ws_wasm_create(uri: string, sub_protocols: string[] | null, rece if (ws[wasm_ws_is_aborted]) return; if (loaderHelpers.is_exited()) return; - onClosed(ev.code, ev.reason); + ws[wasm_ws_close_received] = true; + ws["close_status"] = ev.code; + ws["close_status_description"] = ev.reason; // this reject would not do anything if there was already "open" before it. open_promise_control.reject(new Error(ev.reason)); @@ -94,9 +94,6 @@ export function ws_wasm_create(uri: string, sub_protocols: string[] | null, rece setI32(receive_status_ptr + 8, 1);// end_of_message: true receive_promise_control.resolve(); }); - - // cleanup the delegate proxy - ws[wasm_ws_on_closed].dispose(); }; const local_on_error = (ev: any) => { if (ws[wasm_ws_is_aborted]) return; @@ -131,6 +128,10 @@ export function ws_wasm_open(ws: WebSocketExtension): Promise | null { mono_assert(!!ws, "ERR17: expected ws instance"); + if (ws[wasm_ws_is_aborted] || ws[wasm_ws_close_sent]) { + return rejectedPromise("InvalidState: The WebSocket is not connected."); + } + const buffer_view = new Uint8Array(localHeapViewU8().buffer, buffer_ptr, buffer_length); const whole_buffer = _mono_wasm_web_socket_send_buffering(ws, buffer_view, message_type, end_of_message); @@ -144,14 +145,18 @@ export function ws_wasm_send(ws: WebSocketExtension, buffer_ptr: VoidPtr, buffer export function ws_wasm_receive(ws: WebSocketExtension, buffer_ptr: VoidPtr, buffer_length: number): Promise | null { mono_assert(!!ws, "ERR18: expected ws instance"); + // we can't quickly return if wasm_ws_close_received==true, because there could be pending messages + if (ws[wasm_ws_is_aborted]) { + const receive_status_ptr = ws[wasm_ws_receive_status_ptr]; + setI32(receive_status_ptr, 0); // count + setI32(receive_status_ptr + 4, 2); // type:close + setI32(receive_status_ptr + 8, 1);// end_of_message: true + return resolvedPromise(); + } + const receive_event_queue = ws[wasm_ws_pending_receive_event_queue]; const receive_promise_queue = ws[wasm_ws_pending_receive_promise_queue]; - const readyState = ws.readyState; - if (readyState != WebSocket.OPEN && readyState != WebSocket.CLOSING) { - throw new Error(`InvalidState: ${readyState} The WebSocket is not connected.`); - } - if (receive_event_queue.getLength()) { mono_assert(receive_promise_queue.getLength() == 0, "ERR20: Invalid WS state"); @@ -159,6 +164,16 @@ export function ws_wasm_receive(ws: WebSocketExtension, buffer_ptr: VoidPtr, buf return resolvedPromise(); } + + const readyState = ws.readyState; + if (readyState == WebSocket.CLOSED) { + const receive_status_ptr = ws[wasm_ws_receive_status_ptr]; + setI32(receive_status_ptr, 0); // count + setI32(receive_status_ptr + 4, 2); // type:close + setI32(receive_status_ptr + 8, 1);// end_of_message: true + return resolvedPromise(); + } + const { promise, promise_control } = createPromiseController(); const receive_promise_control = promise_control as ReceivePromiseControl; receive_promise_control.buffer_ptr = buffer_ptr; @@ -171,10 +186,10 @@ export function ws_wasm_receive(ws: WebSocketExtension, buffer_ptr: VoidPtr, buf export function ws_wasm_close(ws: WebSocketExtension, code: number, reason: string | null, wait_for_close_received: boolean): Promise | null { mono_assert(!!ws, "ERR19: expected ws instance"); - if (ws.readyState == WebSocket.CLOSED) { + if (ws[wasm_ws_is_aborted] || ws[wasm_ws_close_sent] || ws.readyState == WebSocket.CLOSED) { return resolvedPromise(); } - + ws[wasm_ws_close_sent] = true; if (wait_for_close_received) { const { promise, promise_control } = createPromiseController(); ws[wasm_ws_pending_close_promises].push(promise_control); @@ -187,10 +202,6 @@ export function ws_wasm_close(ws: WebSocketExtension, code: number, reason: stri return promise; } else { - if (!mono_wasm_web_socket_close_warning) { - mono_wasm_web_socket_close_warning = true; - mono_log_warn("WARNING: Web browsers do not support closing the output side of a WebSocket. CloseOutputAsync has closed the socket and discarded any incoming messages."); - } if (typeof reason === "string") { ws.close(code, reason); } else { @@ -203,12 +214,13 @@ export function ws_wasm_close(ws: WebSocketExtension, code: number, reason: stri export function ws_wasm_abort(ws: WebSocketExtension): void { mono_assert(!!ws, "ERR18: expected ws instance"); + if (ws[wasm_ws_is_aborted] || ws[wasm_ws_close_sent]) { + return; + } + ws[wasm_ws_is_aborted] = true; reject_promises(ws, new Error("OperationCanceledException")); - // cleanup the delegate proxy - ws[wasm_ws_on_closed]?.dispose(); - try { // this is different from Managed implementation ws.close(1000, "Connection was aborted."); @@ -416,11 +428,14 @@ type WebSocketExtension = WebSocket & { [wasm_ws_pending_send_promises]: PromiseController[] [wasm_ws_pending_close_promises]: PromiseController[] [wasm_ws_is_aborted]: boolean - [wasm_ws_on_closed]: IDisposable + [wasm_ws_close_received]: boolean + [wasm_ws_close_sent]: boolean [wasm_ws_receive_status_ptr]: VoidPtr [wasm_ws_pending_send_buffer_offset]: number [wasm_ws_pending_send_buffer_type]: number [wasm_ws_pending_send_buffer]: Uint8Array | null + ["close_status"]: number | undefined + ["close_status_description"]: string | undefined dispose(): void } @@ -448,4 +463,9 @@ function resolvedPromise(): Promise | null { // in practice the `resolve()` callback would arrive before the `reject()` of the cancelation. return wrap_as_cancelable(resolved); } -} \ No newline at end of file +} + +function rejectedPromise(message: string): Promise | null { + const resolved = Promise.reject(new Error(message)); + return wrap_as_cancelable(resolved); +}