BasePeriodicWebSocketListener.cs 10 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341
  1. using MediaBrowser.Model.Logging;
  2. using MediaBrowser.Model.Net;
  3. using System;
  4. using System.Collections.Generic;
  5. using System.Globalization;
  6. using System.Linq;
  7. using System.Threading;
  8. using System.Threading.Tasks;
  9. namespace MediaBrowser.Controller.Net
  10. {
  11. /// <summary>
  12. /// Starts sending data over a web socket periodically when a message is received, and then stops when a corresponding stop message is received
  13. /// </summary>
  14. /// <typeparam name="TReturnDataType">The type of the T return data type.</typeparam>
  15. /// <typeparam name="TStateType">The type of the T state type.</typeparam>
  16. public abstract class BasePeriodicWebSocketListener<TReturnDataType, TStateType> : IWebSocketListener, IDisposable
  17. where TStateType : WebSocketListenerState, new()
  18. where TReturnDataType : class
  19. {
  20. /// <summary>
  21. /// The _active connections
  22. /// </summary>
  23. protected readonly List<Tuple<IWebSocketConnection, CancellationTokenSource, Timer, TStateType, SemaphoreSlim>> ActiveConnections =
  24. new List<Tuple<IWebSocketConnection, CancellationTokenSource, Timer, TStateType, SemaphoreSlim>>();
  25. /// <summary>
  26. /// Gets the name.
  27. /// </summary>
  28. /// <value>The name.</value>
  29. protected abstract string Name { get; }
  30. /// <summary>
  31. /// Gets the data to send.
  32. /// </summary>
  33. /// <param name="state">The state.</param>
  34. /// <returns>Task{`1}.</returns>
  35. protected abstract Task<TReturnDataType> GetDataToSend(TStateType state);
  36. /// <summary>
  37. /// The logger
  38. /// </summary>
  39. protected ILogger Logger;
  40. /// <summary>
  41. /// Initializes a new instance of the <see cref="BasePeriodicWebSocketListener{TStateType}" /> class.
  42. /// </summary>
  43. /// <param name="logger">The logger.</param>
  44. /// <exception cref="System.ArgumentNullException">logger</exception>
  45. protected BasePeriodicWebSocketListener(ILogger logger)
  46. {
  47. if (logger == null)
  48. {
  49. throw new ArgumentNullException("logger");
  50. }
  51. Logger = logger;
  52. }
  53. /// <summary>
  54. /// The null task result
  55. /// </summary>
  56. protected Task NullTaskResult = Task.FromResult(true);
  57. /// <summary>
  58. /// Processes the message.
  59. /// </summary>
  60. /// <param name="message">The message.</param>
  61. /// <returns>Task.</returns>
  62. public Task ProcessMessage(WebSocketMessageInfo message)
  63. {
  64. if (message == null)
  65. {
  66. throw new ArgumentNullException("message");
  67. }
  68. if (string.Equals(message.MessageType, Name + "Start", StringComparison.OrdinalIgnoreCase))
  69. {
  70. Start(message);
  71. }
  72. if (string.Equals(message.MessageType, Name + "Stop", StringComparison.OrdinalIgnoreCase))
  73. {
  74. Stop(message);
  75. }
  76. return NullTaskResult;
  77. }
  78. protected readonly CultureInfo UsCulture = new CultureInfo("en-US");
  79. protected virtual bool SendOnTimer
  80. {
  81. get
  82. {
  83. return true;
  84. }
  85. }
  86. protected virtual void ParseMessageParams(string[] values)
  87. {
  88. }
  89. /// <summary>
  90. /// Starts sending messages over a web socket
  91. /// </summary>
  92. /// <param name="message">The message.</param>
  93. private void Start(WebSocketMessageInfo message)
  94. {
  95. var vals = message.Data.Split(',');
  96. var dueTimeMs = long.Parse(vals[0], UsCulture);
  97. var periodMs = long.Parse(vals[1], UsCulture);
  98. if (vals.Length > 2)
  99. {
  100. ParseMessageParams(vals.Skip(2).ToArray());
  101. }
  102. var cancellationTokenSource = new CancellationTokenSource();
  103. Logger.Debug("{1} Begin transmitting over websocket to {0}", message.Connection.RemoteEndPoint, GetType().Name);
  104. var timer = SendOnTimer ?
  105. new Timer(TimerCallback, message.Connection, Timeout.Infinite, Timeout.Infinite) :
  106. null;
  107. var state = new TStateType
  108. {
  109. IntervalMs = periodMs,
  110. InitialDelayMs = dueTimeMs
  111. };
  112. var semaphore = new SemaphoreSlim(1, 1);
  113. lock (ActiveConnections)
  114. {
  115. ActiveConnections.Add(new Tuple<IWebSocketConnection, CancellationTokenSource, Timer, TStateType, SemaphoreSlim>(message.Connection, cancellationTokenSource, timer, state, semaphore));
  116. }
  117. if (timer != null)
  118. {
  119. timer.Change(TimeSpan.FromMilliseconds(dueTimeMs), TimeSpan.FromMilliseconds(periodMs));
  120. }
  121. }
  122. /// <summary>
  123. /// Timers the callback.
  124. /// </summary>
  125. /// <param name="state">The state.</param>
  126. private void TimerCallback(object state)
  127. {
  128. var connection = (IWebSocketConnection)state;
  129. Tuple<IWebSocketConnection, CancellationTokenSource, Timer, TStateType, SemaphoreSlim> tuple;
  130. lock (ActiveConnections)
  131. {
  132. tuple = ActiveConnections.FirstOrDefault(c => c.Item1 == connection);
  133. }
  134. if (tuple == null)
  135. {
  136. return;
  137. }
  138. if (connection.State != WebSocketState.Open || tuple.Item2.IsCancellationRequested)
  139. {
  140. DisposeConnection(tuple);
  141. return;
  142. }
  143. SendData(tuple);
  144. }
  145. protected void SendData(bool force)
  146. {
  147. List<Tuple<IWebSocketConnection, CancellationTokenSource, Timer, TStateType, SemaphoreSlim>> tuples;
  148. lock (ActiveConnections)
  149. {
  150. tuples = ActiveConnections
  151. .Where(c =>
  152. {
  153. if (c.Item1.State == WebSocketState.Open && !c.Item2.IsCancellationRequested)
  154. {
  155. var state = c.Item4;
  156. if (force || (DateTime.UtcNow - state.DateLastSendUtc).TotalMilliseconds >= state.IntervalMs)
  157. {
  158. return true;
  159. }
  160. }
  161. return false;
  162. })
  163. .ToList();
  164. }
  165. foreach (var tuple in tuples)
  166. {
  167. SendData(tuple);
  168. }
  169. }
  170. private async void SendData(Tuple<IWebSocketConnection, CancellationTokenSource, Timer, TStateType, SemaphoreSlim> tuple)
  171. {
  172. var connection = tuple.Item1;
  173. try
  174. {
  175. await tuple.Item5.WaitAsync(tuple.Item2.Token).ConfigureAwait(false);
  176. var state = tuple.Item4;
  177. var data = await GetDataToSend(state).ConfigureAwait(false);
  178. if (data != null)
  179. {
  180. await connection.SendAsync(new WebSocketMessage<TReturnDataType>
  181. {
  182. MessageType = Name,
  183. Data = data
  184. }, tuple.Item2.Token).ConfigureAwait(false);
  185. state.DateLastSendUtc = DateTime.UtcNow;
  186. }
  187. tuple.Item5.Release();
  188. }
  189. catch (OperationCanceledException)
  190. {
  191. if (tuple.Item2.IsCancellationRequested)
  192. {
  193. DisposeConnection(tuple);
  194. }
  195. }
  196. catch (Exception ex)
  197. {
  198. Logger.ErrorException("Error sending web socket message {0}", ex, Name);
  199. DisposeConnection(tuple);
  200. }
  201. }
  202. /// <summary>
  203. /// Stops sending messages over a web socket
  204. /// </summary>
  205. /// <param name="message">The message.</param>
  206. private void Stop(WebSocketMessageInfo message)
  207. {
  208. lock (ActiveConnections)
  209. {
  210. var connection = ActiveConnections.FirstOrDefault(c => c.Item1 == message.Connection);
  211. if (connection != null)
  212. {
  213. DisposeConnection(connection);
  214. }
  215. }
  216. }
  217. /// <summary>
  218. /// Disposes the connection.
  219. /// </summary>
  220. /// <param name="connection">The connection.</param>
  221. private void DisposeConnection(Tuple<IWebSocketConnection, CancellationTokenSource, Timer, TStateType, SemaphoreSlim> connection)
  222. {
  223. Logger.Debug("{1} stop transmitting over websocket to {0}", connection.Item1.RemoteEndPoint, GetType().Name);
  224. var timer = connection.Item3;
  225. if (timer != null)
  226. {
  227. try
  228. {
  229. timer.Dispose();
  230. }
  231. catch (ObjectDisposedException)
  232. {
  233. }
  234. }
  235. try
  236. {
  237. connection.Item2.Cancel();
  238. connection.Item2.Dispose();
  239. }
  240. catch (ObjectDisposedException)
  241. {
  242. }
  243. try
  244. {
  245. connection.Item5.Dispose();
  246. }
  247. catch (ObjectDisposedException)
  248. {
  249. }
  250. ActiveConnections.Remove(connection);
  251. }
  252. /// <summary>
  253. /// Releases unmanaged and - optionally - managed resources.
  254. /// </summary>
  255. /// <param name="dispose"><c>true</c> to release both managed and unmanaged resources; <c>false</c> to release only unmanaged resources.</param>
  256. protected virtual void Dispose(bool dispose)
  257. {
  258. if (dispose)
  259. {
  260. lock (ActiveConnections)
  261. {
  262. foreach (var connection in ActiveConnections.ToList())
  263. {
  264. DisposeConnection(connection);
  265. }
  266. }
  267. }
  268. }
  269. /// <summary>
  270. /// Performs application-defined tasks associated with freeing, releasing, or resetting unmanaged resources.
  271. /// </summary>
  272. public void Dispose()
  273. {
  274. Dispose(true);
  275. }
  276. }
  277. public class WebSocketListenerState
  278. {
  279. public DateTime DateLastSendUtc { get; set; }
  280. public long InitialDelayMs { get; set; }
  281. public long IntervalMs { get; set; }
  282. }
  283. }