< Summary - Jellyfin

Information
Class: Emby.Server.Implementations.HttpServer.WebSocketConnection
Assembly: Emby.Server.Implementations
File(s): /srv/git/jellyfin/Emby.Server.Implementations/HttpServer/WebSocketConnection.cs
Line coverage
64%
Covered lines: 70
Uncovered lines: 38
Coverable lines: 108
Total lines: 323
Line coverage: 64.8%
Branch coverage
50%
Covered branches: 21
Total branches: 42
Branch coverage: 50%
Method coverage

Feature is only available for sponsors

Upgrade to PRO version

Coverage history

Coverage history 0 25 50 75 100 5/12/2026 - 12:15:57 AM Line coverage: 9.4% (9/95) Branch coverage: 0% (0/34) Total lines: 2835/15/2026 - 12:15:55 AM Line coverage: 9% (9/99) Branch coverage: 0% (0/36) Total lines: 3007/3/2026 - 12:15:32 AM Line coverage: 41.4% (41/99) Branch coverage: 36.1% (13/36) Total lines: 30410/29/2026 - 12:17:11 AM Line coverage: 64.8% (70/108) Branch coverage: 50% (21/42) Total lines: 323 5/12/2026 - 12:15:57 AM Line coverage: 9.4% (9/95) Branch coverage: 0% (0/34) Total lines: 2835/15/2026 - 12:15:55 AM Line coverage: 9% (9/99) Branch coverage: 0% (0/36) Total lines: 3007/3/2026 - 12:15:32 AM Line coverage: 41.4% (41/99) Branch coverage: 36.1% (13/36) Total lines: 30410/29/2026 - 12:17:11 AM Line coverage: 64.8% (70/108) Branch coverage: 50% (21/42) Total lines: 323

Coverage delta

Coverage delta 37 -37

Metrics

MethodBranch coverage Crap Score Cyclomatic complexity Line coverage
.ctor(...)100%11100%
get_State()100%11100%
ApplyRequestCulture()0%620%
SendAsync()100%1166.66%
SendAsync()100%11100%
ReceiveAsync()60%462060%
IsConnectionGone(...)33.33%66100%
ProcessInternal()37.5%27833.33%
DeserializeWebSocketMessage(...)100%11100%
SendKeepAliveResponse()100%1180%
Dispose()100%11100%
Dispose(...)75%44100%
DisposeAsync()100%11100%
DisposeAsyncCore()50%2275%

File(s)

/srv/git/jellyfin/Emby.Server.Implementations/HttpServer/WebSocketConnection.cs

#LineLine coverage
 1using System;
 2using System.Buffers;
 3using System.Globalization;
 4using System.IO.Pipelines;
 5using System.Net;
 6using System.Net.WebSockets;
 7using System.Text;
 8using System.Text.Json;
 9using System.Threading;
 10using System.Threading.Tasks;
 11using Jellyfin.Extensions.Json;
 12using MediaBrowser.Controller.Net;
 13using MediaBrowser.Controller.Net.WebSocketMessages;
 14using MediaBrowser.Controller.Net.WebSocketMessages.Outbound;
 15using MediaBrowser.Model.Session;
 16using Microsoft.Extensions.Logging;
 17
 18namespace Emby.Server.Implementations.HttpServer
 19{
 20    /// <summary>
 21    /// Class WebSocketConnection.
 22    /// </summary>
 23    public class WebSocketConnection : IWebSocketConnection
 24    {
 25        /// <summary>
 26        /// The logger.
 27        /// </summary>
 28        private readonly ILogger<WebSocketConnection> _logger;
 29
 30        /// <summary>
 31        /// The json serializer options.
 32        /// </summary>
 33        private readonly JsonSerializerOptions _jsonOptions;
 34
 35        /// <summary>
 36        /// The socket.
 37        /// </summary>
 38        private readonly WebSocket _socket;
 39
 40        private bool _disposed = false;
 41
 42        /// <summary>
 43        /// Initializes a new instance of the <see cref="WebSocketConnection" /> class.
 44        /// </summary>
 45        /// <param name="logger">The logger.</param>
 46        /// <param name="socket">The socket.</param>
 47        /// <param name="authorizationInfo">The authorization information.</param>
 48        /// <param name="remoteEndPoint">The remote end point.</param>
 49        public WebSocketConnection(
 50            ILogger<WebSocketConnection> logger,
 51            WebSocket socket,
 52            AuthorizationInfo authorizationInfo,
 53            IPAddress? remoteEndPoint)
 54        {
 655            _logger = logger;
 656            _socket = socket;
 57            AuthorizationInfo = authorizationInfo;
 58            RemoteEndPoint = remoteEndPoint;
 59
 660            _jsonOptions = JsonDefaults.Options;
 661            LastActivityDate = DateTime.UtcNow;
 662        }
 63
 64        /// <inheritdoc />
 65        public event EventHandler<EventArgs>? Closed;
 66
 67        /// <inheritdoc />
 68        public AuthorizationInfo AuthorizationInfo { get; }
 69
 70        /// <inheritdoc />
 71        public IPAddress? RemoteEndPoint { get; }
 72
 73        /// <summary>
 74        /// Gets or initializes the UI culture captured from the upgrade request.
 75        /// </summary>
 76        public CultureInfo? RequestUICulture { get; init; }
 77
 78        /// <inheritdoc />
 79        public Func<WebSocketMessageInfo, Task>? OnReceive { get; set; }
 80
 81        /// <inheritdoc />
 82        public DateTime LastActivityDate { get; private set; }
 83
 84        /// <inheritdoc />
 85        public DateTime LastKeepAliveDate { get; set; }
 86
 87        /// <inheritdoc />
 288        public WebSocketState State => _socket.State;
 89
 90        /// <inheritdoc />
 91        public void ApplyRequestCulture()
 92        {
 093            if (RequestUICulture is null)
 94            {
 095                return;
 96            }
 97
 098            CultureInfo.CurrentUICulture = RequestUICulture;
 099        }
 100
 101        /// <inheritdoc />
 102        public async Task SendAsync(OutboundWebSocketMessage message, CancellationToken cancellationToken)
 103        {
 1104            var json = JsonSerializer.SerializeToUtf8Bytes(message, _jsonOptions);
 1105            await _socket.SendAsync(json, WebSocketMessageType.Text, true, cancellationToken).ConfigureAwait(false);
 0106        }
 107
 108        /// <inheritdoc />
 109        public async Task SendAsync<T>(OutboundWebSocketMessage<T> message, CancellationToken cancellationToken)
 110        {
 3111            var json = JsonSerializer.SerializeToUtf8Bytes(message, _jsonOptions);
 3112            await _socket.SendAsync(json, WebSocketMessageType.Text, true, cancellationToken).ConfigureAwait(false);
 3113        }
 114
 115        /// <inheritdoc />
 116        public async Task ReceiveAsync(CancellationToken cancellationToken = default)
 117        {
 2118            var pipe = new Pipe();
 2119            var writer = pipe.Writer;
 120
 121            ValueWebSocketReceiveResult receiveResult;
 122            do
 123            {
 124                // Allocate at least 512 bytes from the PipeWriter
 2125                Memory<byte> memory = writer.GetMemory(512);
 126                try
 127                {
 2128                    receiveResult = await _socket.ReceiveAsync(memory, cancellationToken).ConfigureAwait(false);
 1129                }
 1130                catch (Exception ex) when (IsConnectionGone(ex))
 131                {
 132                    // ObjectDisposedException/OperationCanceledException: the socket was torn
 133                    // down underneath us (e.g. by the keep-alive watchdog after the connection
 134                    // was declared lost). Fall through so Closed is still raised and the
 135                    // session can release this connection.
 1136                    _logger.LogWarning("WS {IP} error receiving data: {Message}", RemoteEndPoint, ex.Message);
 1137                    break;
 138                }
 139
 1140                int bytesRead = receiveResult.Count;
 1141                if (bytesRead == 0)
 142                {
 143                    break;
 144                }
 145
 146                // Tell the PipeWriter how much was read from the Socket
 1147                writer.Advance(bytesRead);
 148
 149                // Make the data available to the PipeReader
 1150                FlushResult flushResult = await writer.FlushAsync(cancellationToken).ConfigureAwait(false);
 1151                if (flushResult.IsCompleted)
 152                {
 153                    // The PipeReader stopped reading
 154                    break;
 155                }
 156
 1157                LastActivityDate = DateTime.UtcNow;
 158
 1159                if (receiveResult.EndOfMessage)
 160                {
 161                    try
 162                    {
 1163                        await ProcessInternal(pipe.Reader).ConfigureAwait(false);
 0164                    }
 1165                    catch (Exception ex) when (IsConnectionGone(ex))
 166                    {
 1167                        _logger.LogWarning("WS {IP} error sending data: {Message}", RemoteEndPoint, ex.Message);
 1168                        break;
 169                    }
 170                }
 171            }
 0172            while ((_socket.State == WebSocketState.Open || _socket.State == WebSocketState.Connecting)
 0173                && receiveResult.MessageType != WebSocketMessageType.Close);
 174
 2175            Closed?.Invoke(this, EventArgs.Empty);
 176
 2177            if (_socket.State == WebSocketState.Open
 2178                || _socket.State == WebSocketState.CloseReceived
 2179                || _socket.State == WebSocketState.CloseSent)
 180            {
 181                try
 182                {
 0183                    await _socket.CloseAsync(
 0184                        WebSocketCloseStatus.NormalClosure,
 0185                        string.Empty,
 0186                        cancellationToken).ConfigureAwait(false);
 0187                }
 0188                catch (Exception ex) when (IsConnectionGone(ex))
 189                {
 190                    // The peer is already gone, there is nobody left to send the close frame to.
 0191                    _logger.LogDebug("WS {IP} error closing connection: {Message}", RemoteEndPoint, ex.Message);
 0192                }
 193            }
 2194        }
 195
 196        private static bool IsConnectionGone(Exception ex)
 2197            => ex is WebSocketException or ObjectDisposedException or OperationCanceledException;
 198
 199        private async Task ProcessInternal(PipeReader reader)
 200        {
 1201            ReadResult result = await reader.ReadAsync().ConfigureAwait(false);
 1202            ReadOnlySequence<byte> buffer = result.Buffer;
 203
 1204            if (OnReceive is null)
 205            {
 206                // Tell the PipeReader how much of the buffer we have consumed
 0207                reader.AdvanceTo(buffer.End);
 0208                return;
 209            }
 210
 211            InboundWebSocketMessage<object>? stub;
 212            long bytesConsumed;
 213            try
 214            {
 1215                stub = DeserializeWebSocketMessage(buffer, out bytesConsumed);
 1216            }
 0217            catch (JsonException ex)
 218            {
 219                // Tell the PipeReader how much of the buffer we have consumed
 0220                reader.AdvanceTo(buffer.End);
 0221                _logger.LogError(ex, "Error processing web socket message: {Data}", Encoding.UTF8.GetString(buffer));
 0222                return;
 223            }
 224
 1225            if (stub is null)
 226            {
 0227                _logger.LogError("Error processing web socket message");
 0228                return;
 229            }
 230
 231            // Tell the PipeReader how much of the buffer we have consumed
 1232            reader.AdvanceTo(buffer.GetPosition(bytesConsumed));
 233
 1234            _logger.LogDebug("WS {IP} received message: {@Message}", RemoteEndPoint, stub);
 235
 1236            if (stub.MessageType == SessionMessageType.KeepAlive)
 237            {
 1238                await SendKeepAliveResponse().ConfigureAwait(false);
 239            }
 240            else
 241            {
 242                try
 243                {
 0244                    await OnReceive(
 0245                        new WebSocketMessageInfo
 0246                        {
 0247                            MessageType = stub.MessageType,
 0248                            Data = stub.Data?.ToString(), // Data can be null
 0249                            Connection = this
 0250                        }).ConfigureAwait(false);
 0251                }
 0252                catch (Exception exception)
 253                {
 0254                    _logger.LogWarning(exception, "Failed to process WebSocket message");
 0255                }
 256            }
 0257        }
 258
 259        internal InboundWebSocketMessage<object>? DeserializeWebSocketMessage(ReadOnlySequence<byte> bytes, out long byt
 260        {
 5261            var jsonReader = new Utf8JsonReader(bytes);
 5262            var ret = JsonSerializer.Deserialize<InboundWebSocketMessage<object>>(ref jsonReader, _jsonOptions);
 4263            bytesConsumed = jsonReader.BytesConsumed;
 4264            return ret;
 265        }
 266
 267        private async Task SendKeepAliveResponse()
 268        {
 1269            LastKeepAliveDate = DateTime.UtcNow;
 1270            await SendAsync(
 1271                new OutboundKeepAliveMessage(),
 1272                CancellationToken.None).ConfigureAwait(false);
 0273        }
 274
 275        /// <inheritdoc />
 276        public void Dispose()
 277        {
 1278            Dispose(true);
 1279            GC.SuppressFinalize(this);
 1280        }
 281
 282        /// <summary>
 283        /// Releases unmanaged and - optionally - managed resources.
 284        /// </summary>
 285        /// <param name="dispose"><c>true</c> to release both managed and unmanaged resources; <c>false</c> to release o
 286        protected virtual void Dispose(bool dispose)
 287        {
 2288            if (_disposed)
 289            {
 1290                return;
 291            }
 292
 1293            if (dispose)
 294            {
 1295                _socket.Dispose();
 296            }
 297
 1298            _disposed = true;
 1299        }
 300
 301        /// <inheritdoc />
 302        public async ValueTask DisposeAsync()
 303        {
 1304            await DisposeAsyncCore().ConfigureAwait(false);
 1305            Dispose(false);
 1306            GC.SuppressFinalize(this);
 1307        }
 308
 309        /// <summary>
 310        /// Used to perform asynchronous cleanup of managed resources or for cascading calls to <see cref="DisposeAsync"
 311        /// </summary>
 312        /// <returns>A ValueTask.</returns>
 313        protected virtual async ValueTask DisposeAsyncCore()
 314        {
 1315            if (_socket.State == WebSocketState.Open)
 316            {
 0317                await _socket.CloseOutputAsync(WebSocketCloseStatus.NormalClosure, "System Shutdown", CancellationToken.
 318            }
 319
 1320            _socket.Dispose();
 1321        }
 322    }
 323}
Generated by: ReportGenerator 5.5.10.0
11/1/2026 - 12:15:28 AM
GitHub | reportgenerator.io

Methods/Properties

.ctor(Microsoft.Extensions.Logging.ILogger`1<Emby.Server.Implementations.HttpServer.WebSocketConnection>,System.Net.WebSockets.WebSocket,MediaBrowser.Controller.Net.AuthorizationInfo,System.Net.IPAddress)
get_State()
ApplyRequestCulture()
SendAsync()
SendAsync()
ReceiveAsync()
IsConnectionGone(System.Exception)
ProcessInternal()
DeserializeWebSocketMessage(System.Buffers.ReadOnlySequence`1<System.Byte>,System.Int64&)
SendKeepAliveResponse()
Dispose()
Dispose(System.Boolean)
DisposeAsync()
DisposeAsyncCore()