< 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
41%
Covered lines: 41
Uncovered lines: 58
Coverable lines: 99
Total lines: 304
Line coverage: 41.4%
Branch coverage
36%
Covered branches: 13
Total branches: 36
Branch coverage: 36.1%
Method coverage

Feature is only available for sponsors

Upgrade to PRO version

Coverage history

Coverage history 0 25 50 75 100 3/26/2026 - 12:14:14 AM Line coverage: 47.3% (9/19) Branch coverage: 0% (0/4) Total lines: 2834/19/2026 - 12:14:27 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: 304 3/26/2026 - 12:14:14 AM Line coverage: 47.3% (9/19) Branch coverage: 0% (0/4) Total lines: 2834/19/2026 - 12:14:27 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: 304

Coverage delta

Coverage delta 38 -38

Metrics

MethodBranch coverage Crap Score Cyclomatic complexity Line coverage
.ctor(...)100%11100%
get_State()100%11100%
ApplyRequestCulture()0%620%
SendAsync()100%210%
SendAsync()100%11100%
ReceiveAsync()45%892044.44%
ProcessInternal()0%7280%
DeserializeWebSocketMessage(...)100%11100%
SendKeepAliveResponse()100%210%
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        {
 555            _logger = logger;
 556            _socket = socket;
 57            AuthorizationInfo = authorizationInfo;
 58            RemoteEndPoint = remoteEndPoint;
 59
 560            _jsonOptions = JsonDefaults.Options;
 561            LastActivityDate = DateTime.UtcNow;
 562        }
 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        {
 0104            var json = JsonSerializer.SerializeToUtf8Bytes(message, _jsonOptions);
 0105            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        {
 1118            var pipe = new Pipe();
 1119            var writer = pipe.Writer;
 120
 121            ValueWebSocketReceiveResult receiveResult;
 122            do
 123            {
 124                // Allocate at least 512 bytes from the PipeWriter
 1125                Memory<byte> memory = writer.GetMemory(512);
 126                try
 127                {
 1128                    receiveResult = await _socket.ReceiveAsync(memory, cancellationToken).ConfigureAwait(false);
 0129                }
 1130                catch (Exception ex) when (ex is WebSocketException or ObjectDisposedException or OperationCanceledExcep
 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
 0140                int bytesRead = receiveResult.Count;
 0141                if (bytesRead == 0)
 142                {
 143                    break;
 144                }
 145
 146                // Tell the PipeWriter how much was read from the Socket
 0147                writer.Advance(bytesRead);
 148
 149                // Make the data available to the PipeReader
 0150                FlushResult flushResult = await writer.FlushAsync(cancellationToken).ConfigureAwait(false);
 0151                if (flushResult.IsCompleted)
 152                {
 153                    // The PipeReader stopped reading
 154                    break;
 155                }
 156
 0157                LastActivityDate = DateTime.UtcNow;
 158
 0159                if (receiveResult.EndOfMessage)
 160                {
 0161                    await ProcessInternal(pipe.Reader).ConfigureAwait(false);
 162                }
 163            }
 0164            while ((_socket.State == WebSocketState.Open || _socket.State == WebSocketState.Connecting)
 0165                && receiveResult.MessageType != WebSocketMessageType.Close);
 166
 1167            Closed?.Invoke(this, EventArgs.Empty);
 168
 1169            if (_socket.State == WebSocketState.Open
 1170                || _socket.State == WebSocketState.CloseReceived
 1171                || _socket.State == WebSocketState.CloseSent)
 172            {
 0173                await _socket.CloseAsync(
 0174                    WebSocketCloseStatus.NormalClosure,
 0175                    string.Empty,
 0176                    cancellationToken).ConfigureAwait(false);
 177            }
 1178        }
 179
 180        private async Task ProcessInternal(PipeReader reader)
 181        {
 0182            ReadResult result = await reader.ReadAsync().ConfigureAwait(false);
 0183            ReadOnlySequence<byte> buffer = result.Buffer;
 184
 0185            if (OnReceive is null)
 186            {
 187                // Tell the PipeReader how much of the buffer we have consumed
 0188                reader.AdvanceTo(buffer.End);
 0189                return;
 190            }
 191
 192            InboundWebSocketMessage<object>? stub;
 193            long bytesConsumed;
 194            try
 195            {
 0196                stub = DeserializeWebSocketMessage(buffer, out bytesConsumed);
 0197            }
 0198            catch (JsonException ex)
 199            {
 200                // Tell the PipeReader how much of the buffer we have consumed
 0201                reader.AdvanceTo(buffer.End);
 0202                _logger.LogError(ex, "Error processing web socket message: {Data}", Encoding.UTF8.GetString(buffer));
 0203                return;
 204            }
 205
 0206            if (stub is null)
 207            {
 0208                _logger.LogError("Error processing web socket message");
 0209                return;
 210            }
 211
 212            // Tell the PipeReader how much of the buffer we have consumed
 0213            reader.AdvanceTo(buffer.GetPosition(bytesConsumed));
 214
 0215            _logger.LogDebug("WS {IP} received message: {@Message}", RemoteEndPoint, stub);
 216
 0217            if (stub.MessageType == SessionMessageType.KeepAlive)
 218            {
 0219                await SendKeepAliveResponse().ConfigureAwait(false);
 220            }
 221            else
 222            {
 223                try
 224                {
 0225                    await OnReceive(
 0226                        new WebSocketMessageInfo
 0227                        {
 0228                            MessageType = stub.MessageType,
 0229                            Data = stub.Data?.ToString(), // Data can be null
 0230                            Connection = this
 0231                        }).ConfigureAwait(false);
 0232                }
 0233                catch (Exception exception)
 234                {
 0235                    _logger.LogWarning(exception, "Failed to process WebSocket message");
 0236                }
 237            }
 0238        }
 239
 240        internal InboundWebSocketMessage<object>? DeserializeWebSocketMessage(ReadOnlySequence<byte> bytes, out long byt
 241        {
 4242            var jsonReader = new Utf8JsonReader(bytes);
 4243            var ret = JsonSerializer.Deserialize<InboundWebSocketMessage<object>>(ref jsonReader, _jsonOptions);
 3244            bytesConsumed = jsonReader.BytesConsumed;
 3245            return ret;
 246        }
 247
 248        private async Task SendKeepAliveResponse()
 249        {
 0250            LastKeepAliveDate = DateTime.UtcNow;
 0251            await SendAsync(
 0252                new OutboundKeepAliveMessage(),
 0253                CancellationToken.None).ConfigureAwait(false);
 0254        }
 255
 256        /// <inheritdoc />
 257        public void Dispose()
 258        {
 1259            Dispose(true);
 1260            GC.SuppressFinalize(this);
 1261        }
 262
 263        /// <summary>
 264        /// Releases unmanaged and - optionally - managed resources.
 265        /// </summary>
 266        /// <param name="dispose"><c>true</c> to release both managed and unmanaged resources; <c>false</c> to release o
 267        protected virtual void Dispose(bool dispose)
 268        {
 2269            if (_disposed)
 270            {
 1271                return;
 272            }
 273
 1274            if (dispose)
 275            {
 1276                _socket.Dispose();
 277            }
 278
 1279            _disposed = true;
 1280        }
 281
 282        /// <inheritdoc />
 283        public async ValueTask DisposeAsync()
 284        {
 1285            await DisposeAsyncCore().ConfigureAwait(false);
 1286            Dispose(false);
 1287            GC.SuppressFinalize(this);
 1288        }
 289
 290        /// <summary>
 291        /// Used to perform asynchronous cleanup of managed resources or for cascading calls to <see cref="DisposeAsync"
 292        /// </summary>
 293        /// <returns>A ValueTask.</returns>
 294        protected virtual async ValueTask DisposeAsyncCore()
 295        {
 1296            if (_socket.State == WebSocketState.Open)
 297            {
 0298                await _socket.CloseOutputAsync(WebSocketCloseStatus.NormalClosure, "System Shutdown", CancellationToken.
 299            }
 300
 1301            _socket.Dispose();
 1302        }
 303    }
 304}
Generated by: ReportGenerator 5.5.10.0
7/6/2026 - 12:16: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()
ProcessInternal()
DeserializeWebSocketMessage(System.Buffers.ReadOnlySequence`1<System.Byte>,System.Int64&)
SendKeepAliveResponse()
Dispose()
Dispose(System.Boolean)
DisposeAsync()
DisposeAsyncCore()