SessionWebSocketListener.cs 12 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344
  1. using System;
  2. using System.Collections.Generic;
  3. using System.Linq;
  4. using System.Net.WebSockets;
  5. using System.Threading;
  6. using System.Threading.Tasks;
  7. using MediaBrowser.Controller.Net;
  8. using MediaBrowser.Controller.Session;
  9. using MediaBrowser.Model.Net;
  10. using MediaBrowser.Model.Session;
  11. using Microsoft.AspNetCore.Http;
  12. using Microsoft.Extensions.Logging;
  13. namespace Emby.Server.Implementations.Session
  14. {
  15. /// <summary>
  16. /// Class SessionWebSocketListener.
  17. /// </summary>
  18. public sealed class SessionWebSocketListener : IWebSocketListener, IDisposable
  19. {
  20. /// <summary>
  21. /// The timeout in seconds after which a WebSocket is considered to be lost.
  22. /// </summary>
  23. private const int WebSocketLostTimeout = 60;
  24. /// <summary>
  25. /// The keep-alive interval factor; controls how often the watcher will check on the status of the WebSockets.
  26. /// </summary>
  27. private const float IntervalFactor = 0.2f;
  28. /// <summary>
  29. /// The ForceKeepAlive factor; controls when a ForceKeepAlive is sent.
  30. /// </summary>
  31. private const float ForceKeepAliveFactor = 0.75f;
  32. /// <summary>
  33. /// Lock used for accesing the KeepAlive cancellation token.
  34. /// </summary>
  35. private readonly object _keepAliveLock = new object();
  36. /// <summary>
  37. /// The WebSocket watchlist.
  38. /// </summary>
  39. private readonly HashSet<IWebSocketConnection> _webSockets = new HashSet<IWebSocketConnection>();
  40. /// <summary>
  41. /// Lock used for accessing the WebSockets watchlist.
  42. /// </summary>
  43. private readonly object _webSocketsLock = new object();
  44. /// <summary>
  45. /// The _session manager.
  46. /// </summary>
  47. private readonly ISessionManager _sessionManager;
  48. /// <summary>
  49. /// The _logger.
  50. /// </summary>
  51. private readonly ILogger<SessionWebSocketListener> _logger;
  52. private readonly ILoggerFactory _loggerFactory;
  53. /// <summary>
  54. /// The KeepAlive cancellation token.
  55. /// </summary>
  56. private CancellationTokenSource _keepAliveCancellationToken;
  57. /// <summary>
  58. /// Initializes a new instance of the <see cref="SessionWebSocketListener" /> class.
  59. /// </summary>
  60. /// <param name="logger">The logger.</param>
  61. /// <param name="sessionManager">The session manager.</param>
  62. /// <param name="loggerFactory">The logger factory.</param>
  63. public SessionWebSocketListener(
  64. ILogger<SessionWebSocketListener> logger,
  65. ISessionManager sessionManager,
  66. ILoggerFactory loggerFactory)
  67. {
  68. _logger = logger;
  69. _sessionManager = sessionManager;
  70. _loggerFactory = loggerFactory;
  71. }
  72. /// <inheritdoc />
  73. public void Dispose()
  74. {
  75. StopKeepAlive();
  76. }
  77. /// <summary>
  78. /// Processes the message.
  79. /// </summary>
  80. /// <param name="message">The message.</param>
  81. /// <returns>Task.</returns>
  82. public Task ProcessMessageAsync(WebSocketMessageInfo message)
  83. => Task.CompletedTask;
  84. /// <inheritdoc />
  85. public async Task ProcessWebSocketConnectedAsync(IWebSocketConnection connection)
  86. {
  87. var session = GetSession(connection.QueryString, connection.RemoteEndPoint.ToString());
  88. if (session != null)
  89. {
  90. EnsureController(session, connection);
  91. await KeepAliveWebSocket(connection).ConfigureAwait(false);
  92. }
  93. else
  94. {
  95. _logger.LogWarning("Unable to determine session based on query string: {0}", connection.QueryString);
  96. }
  97. }
  98. private SessionInfo GetSession(IQueryCollection queryString, string remoteEndpoint)
  99. {
  100. if (queryString == null)
  101. {
  102. return null;
  103. }
  104. var token = queryString["api_key"];
  105. if (string.IsNullOrWhiteSpace(token))
  106. {
  107. return null;
  108. }
  109. var deviceId = queryString["deviceId"];
  110. return _sessionManager.GetSessionByAuthenticationToken(token, deviceId, remoteEndpoint);
  111. }
  112. private void EnsureController(SessionInfo session, IWebSocketConnection connection)
  113. {
  114. var controllerInfo = session.EnsureController<WebSocketController>(
  115. s => new WebSocketController(_loggerFactory.CreateLogger<WebSocketController>(), s, _sessionManager));
  116. var controller = (WebSocketController)controllerInfo.Item1;
  117. controller.AddWebSocket(connection);
  118. _sessionManager.OnSessionControllerConnected(session);
  119. }
  120. /// <summary>
  121. /// Called when a WebSocket is closed.
  122. /// </summary>
  123. /// <param name="sender">The WebSocket.</param>
  124. /// <param name="e">The event arguments.</param>
  125. private void OnWebSocketClosed(object sender, EventArgs e)
  126. {
  127. var webSocket = (IWebSocketConnection)sender;
  128. _logger.LogDebug("WebSocket {0} is closed.", webSocket);
  129. RemoveWebSocket(webSocket);
  130. }
  131. /// <summary>
  132. /// Adds a WebSocket to the KeepAlive watchlist.
  133. /// </summary>
  134. /// <param name="webSocket">The WebSocket to monitor.</param>
  135. private async Task KeepAliveWebSocket(IWebSocketConnection webSocket)
  136. {
  137. lock (_webSocketsLock)
  138. {
  139. if (!_webSockets.Add(webSocket))
  140. {
  141. _logger.LogWarning("Multiple attempts to keep alive single WebSocket {0}", webSocket);
  142. return;
  143. }
  144. webSocket.Closed += OnWebSocketClosed;
  145. webSocket.LastKeepAliveDate = DateTime.UtcNow;
  146. StartKeepAlive();
  147. }
  148. // Notify WebSocket about timeout
  149. try
  150. {
  151. await SendForceKeepAlive(webSocket).ConfigureAwait(false);
  152. }
  153. catch (WebSocketException exception)
  154. {
  155. _logger.LogWarning(exception, "Cannot send ForceKeepAlive message to WebSocket {0}.", webSocket);
  156. }
  157. }
  158. /// <summary>
  159. /// Removes a WebSocket from the KeepAlive watchlist.
  160. /// </summary>
  161. /// <param name="webSocket">The WebSocket to remove.</param>
  162. private void RemoveWebSocket(IWebSocketConnection webSocket)
  163. {
  164. lock (_webSocketsLock)
  165. {
  166. if (!_webSockets.Remove(webSocket))
  167. {
  168. _logger.LogWarning("WebSocket {0} not on watchlist.", webSocket);
  169. }
  170. else
  171. {
  172. webSocket.Closed -= OnWebSocketClosed;
  173. }
  174. }
  175. }
  176. /// <summary>
  177. /// Starts the KeepAlive watcher.
  178. /// </summary>
  179. private void StartKeepAlive()
  180. {
  181. lock (_keepAliveLock)
  182. {
  183. if (_keepAliveCancellationToken == null)
  184. {
  185. _keepAliveCancellationToken = new CancellationTokenSource();
  186. // Start KeepAlive watcher
  187. _ = RepeatAsyncCallbackEvery(
  188. KeepAliveSockets,
  189. TimeSpan.FromSeconds(WebSocketLostTimeout * IntervalFactor),
  190. _keepAliveCancellationToken.Token);
  191. }
  192. }
  193. }
  194. /// <summary>
  195. /// Stops the KeepAlive watcher.
  196. /// </summary>
  197. private void StopKeepAlive()
  198. {
  199. lock (_keepAliveLock)
  200. {
  201. if (_keepAliveCancellationToken != null)
  202. {
  203. _keepAliveCancellationToken.Cancel();
  204. _keepAliveCancellationToken.Dispose();
  205. _keepAliveCancellationToken = null;
  206. }
  207. }
  208. lock (_webSocketsLock)
  209. {
  210. foreach (var webSocket in _webSockets)
  211. {
  212. webSocket.Closed -= OnWebSocketClosed;
  213. }
  214. _webSockets.Clear();
  215. }
  216. }
  217. /// <summary>
  218. /// Checks status of KeepAlive of WebSockets.
  219. /// </summary>
  220. private async Task KeepAliveSockets()
  221. {
  222. List<IWebSocketConnection> inactive;
  223. List<IWebSocketConnection> lost;
  224. lock (_webSocketsLock)
  225. {
  226. _logger.LogDebug("Watching {0} WebSockets.", _webSockets.Count);
  227. inactive = _webSockets.Where(i =>
  228. {
  229. var elapsed = (DateTime.UtcNow - i.LastKeepAliveDate).TotalSeconds;
  230. return (elapsed > WebSocketLostTimeout * ForceKeepAliveFactor) && (elapsed < WebSocketLostTimeout);
  231. }).ToList();
  232. lost = _webSockets.Where(i => (DateTime.UtcNow - i.LastKeepAliveDate).TotalSeconds >= WebSocketLostTimeout).ToList();
  233. }
  234. if (inactive.Count > 0)
  235. {
  236. _logger.LogInformation("Sending ForceKeepAlive message to {0} inactive WebSockets.", inactive.Count);
  237. }
  238. foreach (var webSocket in inactive)
  239. {
  240. try
  241. {
  242. await SendForceKeepAlive(webSocket).ConfigureAwait(false);
  243. }
  244. catch (WebSocketException exception)
  245. {
  246. _logger.LogInformation(exception, "Error sending ForceKeepAlive message to WebSocket.");
  247. lost.Add(webSocket);
  248. }
  249. }
  250. lock (_webSocketsLock)
  251. {
  252. if (lost.Count > 0)
  253. {
  254. _logger.LogInformation("Lost {0} WebSockets.", lost.Count);
  255. foreach (var webSocket in lost)
  256. {
  257. // TODO: handle session relative to the lost webSocket
  258. RemoveWebSocket(webSocket);
  259. }
  260. }
  261. if (_webSockets.Count == 0)
  262. {
  263. StopKeepAlive();
  264. }
  265. }
  266. }
  267. /// <summary>
  268. /// Sends a ForceKeepAlive message to a WebSocket.
  269. /// </summary>
  270. /// <param name="webSocket">The WebSocket.</param>
  271. /// <returns>Task.</returns>
  272. private Task SendForceKeepAlive(IWebSocketConnection webSocket)
  273. {
  274. return webSocket.SendAsync(
  275. new WebSocketMessage<int>
  276. {
  277. MessageType = SessionMessageType.ForceKeepAlive,
  278. Data = WebSocketLostTimeout
  279. },
  280. CancellationToken.None);
  281. }
  282. /// <summary>
  283. /// Runs a given async callback once every specified interval time, until cancelled.
  284. /// </summary>
  285. /// <param name="callback">The async callback.</param>
  286. /// <param name="interval">The interval time.</param>
  287. /// <param name="cancellationToken">The cancellation token.</param>
  288. /// <returns>Task.</returns>
  289. private async Task RepeatAsyncCallbackEvery(Func<Task> callback, TimeSpan interval, CancellationToken cancellationToken)
  290. {
  291. while (!cancellationToken.IsCancellationRequested)
  292. {
  293. await callback().ConfigureAwait(false);
  294. try
  295. {
  296. await Task.Delay(interval, cancellationToken).ConfigureAwait(false);
  297. }
  298. catch (TaskCanceledException)
  299. {
  300. return;
  301. }
  302. }
  303. }
  304. }
  305. }