#region License /* * WebSocketSessionManager.cs * * The MIT License * * Copyright (c) 2012-2015 sta.blockhead * * Permission is hereby granted, free of charge, to any person obtaining a copy * of this software and associated documentation files (the "Software"), to deal * in the Software without restriction, including without limitation the rights * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell * copies of the Software, and to permit persons to whom the Software is * furnished to do so, subject to the following conditions: * * The above copyright notice and this permission notice shall be included in * all copies or substantial portions of the Software. * * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN * THE SOFTWARE. */ #endregion using System; using System.Collections; using System.Collections.Generic; using System.IO; using System.Text; using System.Threading; using System.Timers; namespace WebSocketSharp.Server { /// /// Manages the sessions in a Websocket service. /// public class WebSocketSessionManager { #region Private Fields private volatile bool _clean; private object _forSweep; private Logger _logger; private Dictionary _sessions; private volatile ServerState _state; private volatile bool _sweeping; private System.Timers.Timer _sweepTimer; private object _sync; private TimeSpan _waitTime; #endregion #region Internal Constructors internal WebSocketSessionManager () : this (new Logger ()) { } internal WebSocketSessionManager (Logger logger) { _logger = logger; _clean = true; _forSweep = new object (); _sessions = new Dictionary (); _state = ServerState.Ready; _sync = ((ICollection) _sessions).SyncRoot; _waitTime = TimeSpan.FromSeconds (1); setSweepTimer (60000); } #endregion #region Internal Properties internal ServerState State { get { return _state; } } #endregion #region Public Properties /// /// Gets the IDs for the active sessions in the Websocket service. /// /// /// An IEnumerable<string> instance that provides an enumerator which /// supports the iteration over the collection of the IDs for the active sessions. /// public IEnumerable ActiveIDs { get { foreach (var res in Broadping (WebSocketFrame.EmptyPingBytes, _waitTime)) if (res.Value) yield return res.Key; } } /// /// Gets the number of the sessions in the Websocket service. /// /// /// An that represents the number of the sessions. /// public int Count { get { lock (_sync) return _sessions.Count; } } /// /// Gets the IDs for the sessions in the Websocket service. /// /// /// An IEnumerable<string> instance that provides an enumerator which /// supports the iteration over the collection of the IDs for the sessions. /// public IEnumerable IDs { get { if (_state == ServerState.ShuttingDown) return new string[0]; lock (_sync) return _sessions.Keys.ToList (); } } /// /// Gets the IDs for the inactive sessions in the Websocket service. /// /// /// An IEnumerable<string> instance that provides an enumerator which /// supports the iteration over the collection of the IDs for the inactive sessions. /// public IEnumerable InactiveIDs { get { foreach (var res in Broadping (WebSocketFrame.EmptyPingBytes, _waitTime)) if (!res.Value) yield return res.Key; } } /// /// Gets the session instance with the specified . /// /// /// /// A instance or /// if not found. /// /// /// That session instance provides the function to /// access the information in the session. /// /// /// /// A that represents the ID of /// the session to find. /// /// /// is . /// /// /// is an empty string. /// public IWebSocketSession this[string id] { get { if (id == null) throw new ArgumentNullException ("id"); if (id.Length == 0) throw new ArgumentException ("An empty string.", "id"); IWebSocketSession session; tryGetSession (id, out session); return session; } } /// /// Gets a value indicating whether the manager cleans up the inactive sessions in /// the WebSocket service periodically. /// /// /// true if the manager cleans up the inactive sessions every 60 seconds; /// otherwise, false. /// public bool KeepClean { get { return _clean; } internal set { if (!(value ^ _clean)) return; _clean = value; if (_state == ServerState.Start) _sweepTimer.Enabled = value; } } /// /// Gets the sessions in the Websocket service. /// /// /// An IEnumerable<IWebSocketSession> instance that provides an enumerator /// which supports the iteration over the collection of the sessions in the service. /// public IEnumerable Sessions { get { if (_state == ServerState.ShuttingDown) return new IWebSocketSession[0]; lock (_sync) return _sessions.Values.ToList (); } } /// /// Gets the wait time for the response to the WebSocket Ping or Close. /// /// /// A that represents the wait time. /// public TimeSpan WaitTime { get { return _waitTime; } internal set { if (value == _waitTime) return; _waitTime = value; foreach (var session in Sessions) session.Context.WebSocket.WaitTime = value; } } #endregion #region Private Methods private void broadcast (Opcode opcode, byte[] data, Action completed) { var cache = new Dictionary (); try { Broadcast (opcode, data, cache); if (completed != null) completed (); } catch (Exception ex) { _logger.Fatal (ex.ToString ()); } finally { cache.Clear (); } } private void broadcast (Opcode opcode, Stream stream, Action completed) { var cache = new Dictionary (); try { Broadcast (opcode, stream, cache); if (completed != null) completed (); } catch (Exception ex) { _logger.Fatal (ex.ToString ()); } finally { foreach (var cached in cache.Values) cached.Dispose (); cache.Clear (); } } private void broadcastAsync (Opcode opcode, byte[] data, Action completed) { ThreadPool.QueueUserWorkItem (state => broadcast (opcode, data, completed)); } private void broadcastAsync (Opcode opcode, Stream stream, Action completed) { ThreadPool.QueueUserWorkItem (state => broadcast (opcode, stream, completed)); } private static string createID () { return Guid.NewGuid ().ToString ("N"); } private void setSweepTimer (double interval) { _sweepTimer = new System.Timers.Timer (interval); _sweepTimer.Elapsed += (sender, e) => Sweep (); } private void stop (PayloadData payloadData, bool send) { var bytes = send ? WebSocketFrame.CreateCloseFrame (payloadData, false).ToArray () : null; lock (_sync) { _state = ServerState.ShuttingDown; _sweepTimer.Enabled = false; foreach (var session in _sessions.Values.ToList ()) session.Context.WebSocket.Close (payloadData, bytes); _state = ServerState.Stop; } } private bool tryGetSession (string id, out IWebSocketSession session) { session = null; if (_state != ServerState.Start) return false; lock (_sync) { if (_state != ServerState.Start) return false; return _sessions.TryGetValue (id, out session); } } #endregion #region Internal Methods internal string Add (IWebSocketSession session) { lock (_sync) { if (_state != ServerState.Start) return null; var id = createID (); _sessions.Add (id, session); return id; } } internal void Broadcast ( Opcode opcode, byte[] data, Dictionary cache) { foreach (var session in Sessions) { if (_state != ServerState.Start) break; session.Context.WebSocket.Send (opcode, data, cache); } } internal void Broadcast ( Opcode opcode, Stream stream, Dictionary cache) { foreach (var session in Sessions) { if (_state != ServerState.Start) break; session.Context.WebSocket.Send (opcode, stream, cache); } } internal Dictionary Broadping (byte[] frameAsBytes, TimeSpan timeout) { var ret = new Dictionary (); foreach (var session in Sessions) { if (_state != ServerState.Start) break; ret.Add (session.ID, session.Context.WebSocket.Ping (frameAsBytes, timeout)); } return ret; } internal bool Remove (string id) { lock (_sync) return _sessions.Remove (id); } internal void Start () { lock (_sync) { _sweepTimer.Enabled = _clean; _state = ServerState.Start; } } internal void Stop (ushort code, string reason) { if (code == 1005) { // == no status stop (PayloadData.Empty, true); return; } stop (new PayloadData (code, reason), !code.IsReserved ()); } #endregion #region Public Methods /// /// Sends to every client in the WebSocket service. /// /// /// An array of that represents the binary data to send. /// /// /// The current state of the manager is not Start. /// /// /// is . /// public void Broadcast (byte[] data) { if (_state != ServerState.Start) { var msg = "The current state of the manager is not Start."; throw new InvalidOperationException (msg); } if (data == null) throw new ArgumentNullException ("data"); if (data.LongLength <= WebSocket.FragmentLength) broadcast (Opcode.Binary, data, null); else broadcast (Opcode.Binary, new MemoryStream (data), null); } /// /// Sends to every client in the WebSocket service. /// /// /// A that represents the text data to send. /// /// /// The current state of the manager is not Start. /// /// /// is . /// /// /// could not be UTF-8-encoded. /// public void Broadcast (string data) { if (_state != ServerState.Start) { var msg = "The current state of the manager is not Start."; throw new InvalidOperationException (msg); } if (data == null) throw new ArgumentNullException ("data"); byte[] bytes; if (!data.TryGetUTF8EncodedBytes (out bytes)) { var msg = "It could not be UTF-8-encoded."; throw new ArgumentException (msg, "data"); } if (bytes.LongLength <= WebSocket.FragmentLength) broadcast (Opcode.Text, bytes, null); else broadcast (Opcode.Text, new MemoryStream (bytes), null); } /// /// Sends the data from to every client in /// the WebSocket service. /// /// /// The data is sent as the binary data. /// /// /// A instance from which to read the data to send. /// /// /// An that specifies the number of bytes to send. /// /// /// The current state of the manager is not Start. /// /// /// is . /// /// /// /// cannot be read. /// /// /// -or- /// /// /// is less than 1. /// /// /// -or- /// /// /// No data could be read from . /// /// public void Broadcast (Stream stream, int length) { if (_state != ServerState.Start) { var msg = "The current state of the manager is not Start."; throw new InvalidOperationException (msg); } if (stream == null) throw new ArgumentNullException ("stream"); if (!stream.CanRead) { var msg = "It cannot be read."; throw new ArgumentException (msg, "stream"); } if (length < 1) { var msg = "Less than 1."; throw new ArgumentException (msg, "length"); } var bytes = stream.ReadBytes (length); var len = bytes.Length; if (len == 0) { var msg = "No data could be read from it."; throw new ArgumentException (msg, "stream"); } if (len < length) { _logger.Warn ( String.Format ( "Only {0} byte(s) of data could be read from the stream.", len ) ); } if (len <= WebSocket.FragmentLength) broadcast (Opcode.Binary, bytes, null); else broadcast (Opcode.Binary, new MemoryStream (bytes), null); } /// /// Sends the specified asynchronously to /// every client in the WebSocket service. /// /// /// This method does not wait for the send to be complete. /// /// /// An array of that represents /// the binary data to send. /// /// /// /// An delegate or /// if not needed. /// /// /// That delegate invokes the method called when /// the send is complete. /// /// /// /// The current state of the manager is not Start. /// /// /// is . /// public void BroadcastAsync (byte[] data, Action completed) { if (_state != ServerState.Start) { var msg = "The current state of the manager is not Start."; throw new InvalidOperationException (msg); } if (data == null) throw new ArgumentNullException ("data"); if (data.LongLength <= WebSocket.FragmentLength) broadcastAsync (Opcode.Binary, data, completed); else broadcastAsync (Opcode.Binary, new MemoryStream (data), completed); } /// /// Sends the specified asynchronously to /// every client in the WebSocket service. /// /// /// This method does not wait for the send to be complete. /// /// /// A that represents the text data to send. /// /// /// /// An delegate or /// if not needed. /// /// /// That delegate invokes the method called when the send is complete. /// /// /// /// The current state of the manager is not Start. /// /// /// is . /// /// /// could not be UTF-8-encoded. /// public void BroadcastAsync (string data, Action completed) { if (_state != ServerState.Start) { var msg = "The current state of the manager is not Start."; throw new InvalidOperationException (msg); } if (data == null) throw new ArgumentNullException ("data"); byte[] bytes; if (!data.TryGetUTF8EncodedBytes (out bytes)) { var msg = "It could not be UTF-8-encoded."; throw new ArgumentException (msg, "data"); } if (bytes.LongLength <= WebSocket.FragmentLength) broadcastAsync (Opcode.Text, bytes, completed); else broadcastAsync (Opcode.Text, new MemoryStream (bytes), completed); } /// /// Sends the specified of data from /// the specified asynchronously to /// every client in the WebSocket service. /// /// /// This method does not wait for the send to be complete. /// /// /// A from which to read the binary data to send. /// /// /// An that specifies the number of bytes to send. /// /// /// /// An delegate or /// if not needed. /// /// /// That delegate invokes the method called when the send is complete. /// /// /// /// The current state of the manager is not Start. /// /// /// is . /// /// /// /// cannot be read. /// /// /// -or- /// /// /// is less than 1. /// /// /// -or- /// /// /// No data could be read from . /// /// public void BroadcastAsync (Stream stream, int length, Action completed) { if (_state != ServerState.Start) { var msg = "The current state of the manager is not Start."; throw new InvalidOperationException (msg); } if (stream == null) throw new ArgumentNullException ("stream"); if (!stream.CanRead) throw new ArgumentException ("It cannot be read.", "stream"); if (length < 1) throw new ArgumentException ("Less than 1.", "length"); var bytes = stream.ReadBytes (length); var len = bytes.Length; if (len == 0) { var msg = "No data could be read from it."; throw new ArgumentException (msg, "stream"); } if (len < length) { _logger.Warn ( String.Format ( "Only {0} byte(s) of data could be read from the specified stream.", len ) ); } if (len <= WebSocket.FragmentLength) broadcastAsync (Opcode.Binary, bytes, completed); else broadcastAsync (Opcode.Binary, new MemoryStream (bytes), completed); } /// /// Sends a ping to every client in the WebSocket service. /// /// /// /// A Dictionary<string, bool>. /// /// /// It represents a collection of pairs of a session ID and /// a value indicating whether a pong has been received from /// its client within a time. /// /// /// /// The current state of the manager is not Start. /// [Obsolete ("This method will be removed.")] public Dictionary Broadping () { if (_state != ServerState.Start) { var msg = "The current state of the manager is not Start."; throw new InvalidOperationException (msg); } return Broadping (WebSocketFrame.EmptyPingBytes, _waitTime); } /// /// Sends a ping with to every client in /// the WebSocket service. /// /// /// /// A Dictionary<string, bool>. /// /// /// It represents a collection of pairs of a session ID and /// a value indicating whether a pong has been received from /// its client within a time. /// /// /// /// /// A that represents the message to send. /// /// /// The size must be 125 bytes or less in UTF-8. /// /// /// /// The current state of the manager is not Start. /// /// /// could not be UTF-8-encoded. /// /// /// The size of is greater than 125 bytes. /// [Obsolete ("This method will be removed.")] public Dictionary Broadping (string message) { if (_state != ServerState.Start) { var msg = "The current state of the manager is not Start."; throw new InvalidOperationException (msg); } if (message.IsNullOrEmpty ()) return Broadping (WebSocketFrame.EmptyPingBytes, _waitTime); byte[] bytes; if (!message.TryGetUTF8EncodedBytes (out bytes)) { var msg = "It could not be UTF-8-encoded."; throw new ArgumentException (msg, "message"); } if (bytes.Length > 125) { var msg = "Its size is greater than 125 bytes."; throw new ArgumentOutOfRangeException ("message", msg); } var frame = WebSocketFrame.CreatePingFrame (bytes, false); return Broadping (frame.ToArray (), _waitTime); } /// /// Closes the session with the specified . /// /// /// A that represents the ID of the session to close. /// public void CloseSession (string id) { IWebSocketSession session; if (TryGetSession (id, out session)) session.Context.WebSocket.Close (); } /// /// Closes the session with the specified , , /// and . /// /// /// A that represents the ID of the session to close. /// /// /// A that represents the status code indicating the reason for the close. /// /// /// A that represents the reason for the close. /// public void CloseSession (string id, ushort code, string reason) { IWebSocketSession session; if (TryGetSession (id, out session)) session.Context.WebSocket.Close (code, reason); } /// /// Closes the session with the specified , , /// and . /// /// /// A that represents the ID of the session to close. /// /// /// One of the enum values, represents the status code /// indicating the reason for the close. /// /// /// A that represents the reason for the close. /// public void CloseSession (string id, CloseStatusCode code, string reason) { IWebSocketSession session; if (TryGetSession (id, out session)) session.Context.WebSocket.Close (code, reason); } /// /// Sends a ping to the client using the specified session. /// /// /// true if the send has done with no error and a pong has been /// received from the client within a time; otherwise, false. /// /// /// A that represents the ID of the session. /// /// /// is . /// /// /// /// is an empty string. /// /// /// -or- /// /// /// The session could not be found. /// /// public bool PingTo (string id) { IWebSocketSession session; if (!TryGetSession (id, out session)) { var msg = "The session could not be found."; throw new ArgumentException (msg, "id"); } return session.Context.WebSocket.Ping (); } /// /// Sends a ping with to the client using /// the specified session. /// /// /// true if the send has done with no error and a pong has been /// received from the client within a time; otherwise, false. /// /// /// /// A that represents the message to send. /// /// /// The size must be 125 bytes or less in UTF-8. /// /// /// /// A that represents the ID of the session. /// /// /// is . /// /// /// /// is an empty string. /// /// /// -or- /// /// /// The session could not be found. /// /// /// -or- /// /// /// could not be UTF-8-encoded. /// /// /// /// The size of is greater than 125 bytes. /// public bool PingTo (string message, string id) { IWebSocketSession session; if (!TryGetSession (id, out session)) { var msg = "The session could not be found."; throw new ArgumentException (msg, "id"); } return session.Context.WebSocket.Ping (message); } /// /// Sends to the client using the specified session. /// /// /// An array of that represents the binary data to send. /// /// /// A that represents the ID of the session. /// /// /// /// is . /// /// /// -or- /// /// /// is . /// /// /// /// /// is an empty string. /// /// /// -or- /// /// /// The session could not be found. /// /// /// /// The current state of the WebSocket connection is not Open. /// public void SendTo (byte[] data, string id) { IWebSocketSession session; if (!TryGetSession (id, out session)) { var msg = "The session could not be found."; throw new ArgumentException (msg, "id"); } session.Context.WebSocket.Send (data); } /// /// Sends to the client using the specified session. /// /// /// A that represents the text data to send. /// /// /// A that represents the ID of the session. /// /// /// /// is . /// /// /// -or- /// /// /// is . /// /// /// /// /// is an empty string. /// /// /// -or- /// /// /// The session could not be found. /// /// /// -or- /// /// /// could not be UTF-8-encoded. /// /// /// /// The current state of the WebSocket connection is not Open. /// public void SendTo (string data, string id) { IWebSocketSession session; if (!TryGetSession (id, out session)) { var msg = "The session could not be found."; throw new ArgumentException (msg, "id"); } session.Context.WebSocket.Send (data); } /// /// Sends asynchronously to the client using /// the specified session. /// /// /// This method does not wait for the send to be complete. /// /// /// An array of that represents the binary data to send. /// /// /// A that represents the ID of the session. /// /// /// /// An Action<bool> delegate or /// if not needed. /// /// /// That delegate invokes the method called when the send is complete. /// /// /// true is passed to the method if the send has done with /// no error; otherwise, false. /// /// /// /// /// is . /// /// /// -or- /// /// /// is . /// /// /// /// /// is an empty string. /// /// /// -or- /// /// /// The session could not be found. /// /// /// /// The current state of the WebSocket connection is not Open. /// public void SendToAsync (byte[] data, string id, Action completed) { IWebSocketSession session; if (!TryGetSession (id, out session)) { var msg = "The session could not be found."; throw new ArgumentException (msg, "id"); } session.Context.WebSocket.SendAsync (data, completed); } /// /// Sends asynchronously to the client using /// the specified session. /// /// /// This method does not wait for the send to be complete. /// /// /// A that represents the text data to send. /// /// /// A that represents the ID of the session. /// /// /// /// An Action<bool> delegate or /// if not needed. /// /// /// The delegate invokes the method called when the send is complete. /// /// /// true is passed to the method if the send has done with /// no error; otherwise, false. /// /// /// /// /// is . /// /// /// -or- /// /// /// is . /// /// /// /// /// is an empty string. /// /// /// -or- /// /// /// The session could not be found. /// /// /// -or- /// /// /// could not be UTF-8-encoded. /// /// /// /// The current state of the WebSocket connection is not Open. /// public void SendToAsync (string data, string id, Action completed) { IWebSocketSession session; if (!TryGetSession (id, out session)) { var msg = "The session could not be found."; throw new ArgumentException (msg, "id"); } session.Context.WebSocket.SendAsync (data, completed); } /// /// Sends binary data from the specified asynchronously to /// the client on the session with the specified . /// /// /// This method doesn't wait for the send to be complete. /// /// /// A from which contains the binary data to send. /// /// /// An that represents the number of bytes to send. /// /// /// A that represents the ID of the session to find. /// /// /// An Action<bool> delegate that references the method(s) called when /// the send is complete. A passed to this delegate is true /// if the send is complete successfully. /// public void SendToAsync ( Stream stream, int length, string id, Action completed ) { IWebSocketSession session; if (!TryGetSession (id, out session)) { var msg = "The session could not be found."; throw new ArgumentException (msg, "id"); } session.Context.WebSocket.SendAsync (stream, length, completed); } /// /// Cleans up the inactive sessions in the WebSocket service. /// public void Sweep () { if (_state != ServerState.Start || _sweeping || Count == 0) return; lock (_forSweep) { _sweeping = true; foreach (var id in InactiveIDs) { if (_state != ServerState.Start) break; lock (_sync) { IWebSocketSession session; if (_sessions.TryGetValue (id, out session)) { var state = session.State; if (state == WebSocketState.Open) session.Context.WebSocket.Close (CloseStatusCode.ProtocolError); else if (state == WebSocketState.Closing) continue; else _sessions.Remove (id); } } } _sweeping = false; } } /// /// Tries to get the session instance with the specified /// . /// /// /// true if the session is successfully found; /// otherwise, false. /// /// /// A that represents the ID of /// the session to find. /// /// /// /// When this method returns, a /// instance or if not found. /// /// /// That session instance provides the function to /// access the information in the session. /// /// /// /// is . /// /// /// is an empty string. /// public bool TryGetSession (string id, out IWebSocketSession session) { if (id == null) throw new ArgumentNullException ("id"); if (id.Length == 0) throw new ArgumentException ("An empty string.", "id"); return tryGetSession (id, out session); } #endregion } }