#if !BESTHTTP_DISABLE_SOCKETIO using System; using System.Collections.Generic; using BestHTTP.SocketIO3.Transports; using BestHTTP.Extensions; using BestHTTP.SocketIO3.Parsers; using BestHTTP.SocketIO3.Events; using BestHTTP.Logger; namespace BestHTTP.SocketIO3 { public sealed class SocketManager : IHeartbeat, IManager { /// /// Possible states of a SocketManager instance. /// public enum States { /// /// Initial state of the SocketManager /// Initial, /// /// The SocketManager is currently opening. /// Opening, /// /// The SocketManager is open, events can be sent to the server. /// Open, /// /// Paused for transport upgrade /// Paused, /// /// An error occurred, the SocketManager now trying to connect again to the server. /// Reconnecting, /// /// The SocketManager is closed, initiated by the user or by the server /// Closed } /// /// Supported Socket.IO protocol version /// public int ProtocolVersion { get { return 4; } } #region Public Properties /// /// The current state of this Socket.IO manager. /// public States State { get { return state; } private set { PreviousState = state; state = value; } } private States state; /// /// The SocketOptions instance that this manager will use. /// public SocketOptions Options { get; private set; } /// /// The Uri to the Socket.IO endpoint. /// public Uri Uri { get; private set; } /// /// The server sent and parsed Handshake data. /// public HandshakeData Handshake { get; private set; } /// /// The currently used main transport instance. /// public ITransport Transport { get; private set; } /// /// The Request counter for request-based transports. /// public ulong RequestCounter { get; internal set; } /// /// The root("/") Socket. /// public Socket Socket { get { return GetSocket(); } } /// /// Indexer to access socket associated to the given namespace. /// public Socket this[string nsp] { get { return GetSocket(nsp); } } /// /// How many reconnect attempts made. /// public int ReconnectAttempts { get; private set; } /// /// Parser to encode and decode messages and create strongly typed objects. /// public IParser Parser { get; set; } /// /// Logging context of this socket.io connection. /// public LoggingContext Context { get; private set; } #endregion #region Internal Properties /// /// Timestamp support to the request based transports. /// internal UInt64 Timestamp { get { return (UInt64)(DateTime.UtcNow.Subtract(new DateTime(1970, 1, 1))).TotalMilliseconds; } } /// /// Auto-incrementing property to return Ack ids. /// internal int NextAckId { get { return System.Threading.Interlocked.Increment(ref nextAckId); } } private int nextAckId; /// /// Internal property to store the previous state of the manager. /// internal States PreviousState { get; private set; } /// /// Transport currently upgrading. /// internal ITransport UpgradingTransport { get; set; } #endregion #region Privates /// /// Namespace name -> Socket mapping /// private Dictionary Namespaces = new Dictionary(); /// /// List of the sockets to able to iterate over them easily. /// private List Sockets = new List(); /// /// List of unsent packets. Only instantiated when we have to use it. /// private List OfflinePackets; /// /// When we sent out the last heartbeat(Ping) message. /// private DateTime LastHeartbeat = DateTime.MinValue; /// /// When we have to try to do a reconnect attempt /// private DateTime ReconnectAt; /// /// When we started to connect to the server. /// private DateTime ConnectionStarted; /// /// Private flag to avoid multiple Close call /// private bool closing; /// /// In Engine.io v4 / socket.io v3 the server sends the ping messages, not the client. /// private DateTime lastPingReceived; #endregion #region Constructors /// /// Constructor to create a SocketManager instance that will connect to the given uri. /// public SocketManager(Uri uri) :this(uri, new DefaultJsonParser(), new SocketOptions()) { } public SocketManager(Uri uri, IParser parser) : this(uri, parser, new SocketOptions()) { } public SocketManager(Uri uri, SocketOptions options) :this(uri, new DefaultJsonParser(), options) { } /// /// Constructor to create a SocketManager instance. /// public SocketManager(Uri uri, IParser parser, SocketOptions options) { this.Context = new LoggingContext(this); string path = uri.PathAndQuery; if (path.Length <= 1) { string append; if (uri.OriginalString[uri.OriginalString.Length - 1] == '/') append = "socket.io/"; else append = "/socket.io/"; uri = new Uri(uri.OriginalString + append); } this.Uri = uri; this.Options = options ?? new SocketOptions(); this.State = States.Initial; this.PreviousState = States.Initial; this.Parser = parser ?? new DefaultJsonParser(); if (uri.Scheme.StartsWith("ws")) options.ConnectWith = TransportTypes.WebSocket; } #endregion /// /// Returns with the "/" namespace, the same as the Socket property. /// public Socket GetSocket() { return GetSocket("/"); } /// /// Returns with the specified namespace /// public Socket GetSocket(string nsp) { if (string.IsNullOrEmpty(nsp)) throw new ArgumentNullException("Namespace parameter is null or empty!"); /*if (nsp[0] != '/') nsp = "/" + nsp;*/ Socket socket = null; if (!Namespaces.TryGetValue(nsp, out socket)) { // No socket found, create one socket = new Socket(nsp, this); Namespaces.Add(nsp, socket); Sockets.Add(socket); (socket as ISocket).Open(); } return socket; } /// /// Internal function to remove a Socket instance from this manager. /// /// void IManager.Remove(Socket socket) { Namespaces.Remove(socket.Namespace); Sockets.Remove(socket); if (Sockets.Count == 0) Close(); } #region Connection to the server, and upgrading /// /// This function will begin to open the Socket.IO connection by sending out the handshake request. /// If the Options' AutoConnect is true, it will be called automatically. /// public void Open() { if (State != States.Initial && State != States.Closed && State != States.Reconnecting) return; HTTPManager.Logger.Information("SocketManager", "Opening", this.Context); ReconnectAt = DateTime.MinValue; switch (Options.ConnectWith) { case TransportTypes.Polling: Transport = new PollingTransport(this); break; #if !BESTHTTP_DISABLE_WEBSOCKET case TransportTypes.WebSocket: Transport = new WebSocketTransport(this); break; #endif } Transport.Open(); (this as IManager).EmitEvent("connecting"); State = States.Opening; ConnectionStarted = DateTime.UtcNow; HTTPManager.Heartbeats.Subscribe(this); // The root namespace will be opened by default //GetSocket("/"); } /// /// Closes this Socket.IO connection. /// public void Close() { (this as IManager).Close(true); } /// /// Closes this Socket.IO connection. /// void IManager.Close(bool removeSockets) { if (State == States.Closed || closing) return; closing = true; HTTPManager.Logger.Information("SocketManager", "Closing", this.Context); HTTPManager.Heartbeats.Unsubscribe(this); // Disconnect the sockets. The Disconnect function will call the Remove function to remove it from the Sockets list. if (removeSockets) while (Sockets.Count > 0) (Sockets[Sockets.Count - 1] as ISocket).Disconnect(removeSockets); else for (int i = 0; i < Sockets.Count; ++i) (Sockets[i] as ISocket).Disconnect(removeSockets); // Set to Closed after Socket's Disconnect. This way we can send the disconnect events to the server. State = States.Closed; LastHeartbeat = DateTime.MinValue; lastPingReceived = DateTime.MinValue; if (removeSockets && OfflinePackets != null) OfflinePackets.Clear(); // Remove the references from the dictionary too. if (removeSockets) Namespaces.Clear(); Handshake = null; if (Transport != null) Transport.Close(); Transport = null; if (UpgradingTransport != null) UpgradingTransport.Close(); UpgradingTransport = null; closing = false; } /// /// Called from a ITransport implementation when an error occurs and we may have to try to reconnect. /// void IManager.TryToReconnect() { if (State == States.Reconnecting || State == States.Closed) return; if (!Options.Reconnection || HTTPManager.IsQuitting) { Close(); return; } if (++ReconnectAttempts >= Options.ReconnectionAttempts) { (this as IManager).EmitEvent("reconnect_failed"); Close(); return; } Random rand = new Random(); int delay = (int)Options.ReconnectionDelay.TotalMilliseconds * ReconnectAttempts; ReconnectAt = DateTime.UtcNow + TimeSpan.FromMilliseconds(Math.Min(rand.Next(/*rand min:*/(int)(delay - (delay * Options.RandomizationFactor)), /*rand max:*/(int)(delay + (delay * Options.RandomizationFactor))), (int)Options.ReconnectionDelayMax.TotalMilliseconds)); (this as IManager).Close(false); State = States.Reconnecting; for (int i = 0; i < Sockets.Count; ++i) (Sockets[i] as ISocket).Open(); // In the Close() function we unregistered HTTPManager.Heartbeats.Subscribe(this); HTTPManager.Logger.Information("SocketManager", "Reconnecting", this.Context); } /// /// Called by transports when they are connected to the server. /// bool IManager.OnTransportConnected(ITransport trans) { HTTPManager.Logger.Information("SocketManager", string.Format("OnTransportConnected State: {0}, PreviousState: {1}, Current Transport: {2}, Upgrading Transport: {3}", this.State, this.PreviousState, trans.Type, UpgradingTransport != null ? UpgradingTransport.Type.ToString() : "null"), this.Context); if (State != States.Opening) return false; if (PreviousState == States.Reconnecting) (this as IManager).EmitEvent("reconnect"); State = States.Open; if (PreviousState == States.Reconnecting) (this as IManager).EmitEvent("reconnect_before_offline_packets"); for (int i = 0; i < Sockets.Count; ++i) { var socket = Sockets[i]; if (socket != null) socket.OnTransportOpen(); } ReconnectAttempts = 0; // Send out packets that we collected while there were no available transport. SendOfflinePackets(); #if !BESTHTTP_DISABLE_WEBSOCKET // Can we upgrade to WebSocket transport? if (Transport.Type != TransportTypes.WebSocket && Handshake.Upgrades.Contains("websocket")) { UpgradingTransport = new WebSocketTransport(this); UpgradingTransport.Open(); } #endif return true; } void IManager.OnTransportError(ITransport trans, string err) { if (UpgradingTransport != null && trans != UpgradingTransport) return; (this as IManager).EmitError(err); trans.Close(); (this as IManager).TryToReconnect(); } void IManager.OnTransportProbed(ITransport trans) { HTTPManager.Logger.Information("SocketManager", "\"probe\" packet received", this.Context); // If we have to reconnect, we will go straight with the transport we were able to upgrade Options.ConnectWith = trans.Type; // Pause ourself to wait for any send and receive turn to finish. State = States.Paused; } #endregion #region Packet Handling /// /// Select the best transport to send out packets. /// private ITransport SelectTransport() { if (State != States.Open || Transport == null) return null; return Transport.IsRequestInProgress ? null : Transport; } /// /// Will select the best transport and sends out all packets that are in the OfflinePackets list. /// private void SendOfflinePackets() { ITransport trans = SelectTransport(); // Send out packets that we not sent while no transport was available. // This function is called before the event handlers get the 'connected' event, so // theoretically the packet orders are remains. if (OfflinePackets != null && OfflinePackets.Count > 0 && trans != null) { trans.Send(OfflinePackets); OfflinePackets.Clear(); } } /// /// Internal function that called from the Socket class. It will send out the packet instantly, or if no transport is available it will store /// the packet in the OfflinePackets list. /// void IManager.SendPacket(OutgoingPacket packet) { HTTPManager.Logger.Information("SocketManager", "SendPacket " + packet.ToString(), this.Context); ITransport trans = SelectTransport(); if (trans != null) { try { trans.Send(packet); } catch(Exception ex) { (this as IManager).EmitError(ex.Message + " " + ex.StackTrace); } } else { if (packet.IsVolatile) return; HTTPManager.Logger.Information("SocketManager", "SendPacket - Offline stashing packet", this.Context); if (OfflinePackets == null) OfflinePackets = new List(); // The same packet can be sent through multiple Sockets. OfflinePackets.Add(packet); } } /// /// Called from the currently operating Transport. Will pass forward to the Socket that has to call the callbacks. /// void IManager.OnPacket(IncomingPacket packet) { if (State == States.Closed) { HTTPManager.Logger.Information("SocketManager", "OnPacket - State == States.Closed", this.Context); return; } switch(packet.TransportEvent) { case TransportEventTypes.Open: if (Handshake == null) { Handshake = packet.DecodedArg as HandshakeData; (this as IManager).OnTransportConnected(Transport); return; } else HTTPManager.Logger.Information("SocketManager", "OnPacket - Already received handshake data!", this.Context); break; case TransportEventTypes.Ping: lastPingReceived = DateTime.UtcNow; //IncomingPacket pingPacket = new Packet(TransportEventTypes.Pong, SocketIOEventTypes.Unknown, "/", 0); (this as IManager).SendPacket(this.Parser.CreateOutgoing(TransportEventTypes.Pong, null)); break; case TransportEventTypes.Pong: break; } Socket socket = null; if (Namespaces.TryGetValue(packet.Namespace, out socket)) (socket as ISocket).OnPacket(packet); else if (packet.TransportEvent == TransportEventTypes.Message) HTTPManager.Logger.Warning("SocketManager", "Namespace \"" + packet.Namespace + "\" not found!", this.Context); } #endregion /// /// Sends an event to all available namespaces. /// public void EmitAll(string eventName, params object[] args) { for (int i = 0; i < Sockets.Count; ++i) Sockets[i].Emit(eventName, args); } /// /// Emits an internal packet-less event to the root namespace without creating it if it isn't exists yet. /// void IManager.EmitEvent(string eventName, params object[] args) { Socket socket = null; if (Namespaces.TryGetValue("/", out socket)) (socket as ISocket).EmitEvent(eventName, args); } /// /// Emits an internal packet-less event to the root namespace without creating it if it isn't exists yet. /// void IManager.EmitEvent(SocketIOEventTypes type, params object[] args) { (this as IManager).EmitEvent(EventNames.GetNameFor(type), args); } void IManager.EmitError(string msg) { var outcoming = this.Parser.CreateOutgoing(this.Sockets[0], SocketIOEventTypes.Error, -1, null, new Error(msg)); IncomingPacket inc = IncomingPacket.Empty; if (outcoming.IsBinary) inc = this.Parser.Parse(this, outcoming.PayloadData); else inc = this.Parser.Parse(this, outcoming.Payload); (this as IManager).EmitEvent(SocketIOEventTypes.Error, inc.DecodedArg ?? inc.DecodedArgs); } void IManager.EmitAll(string eventName, params object[] args) { for (int i = 0; i < Sockets.Count; ++i) (Sockets[i] as ISocket).EmitEvent(eventName, args); } #region IHeartbeat Implementation /// /// Called from the HTTPManager's OnUpdate function every frame. It's main function is to send out heartbeat messages. /// void IHeartbeat.OnHeartbeatUpdate(TimeSpan dif) { switch (State) { case States.Paused: // To ensure no messages are lost, the upgrade packet will only be sent once all the buffers of the existing transport are flushed and the transport is considered paused. if (!Transport.IsRequestInProgress && !Transport.IsPollingInProgress) { State = States.Open; // Close the current transport Transport.Close(); // and switch to the newly upgraded one Transport = UpgradingTransport; UpgradingTransport = null; // We will send an Upgrade("5") packet. Transport.Send(this.Parser.CreateOutgoing(TransportEventTypes.Upgrade, null)); goto case States.Open; } break; case States.Opening: if (DateTime.UtcNow - ConnectionStarted >= Options.Timeout) { (this as IManager).EmitError("Connection timed out!"); (this as IManager).EmitEvent("connect_error"); (this as IManager).EmitEvent("connect_timeout"); (this as IManager).TryToReconnect(); } break; case States.Reconnecting: if (ReconnectAt != DateTime.MinValue && DateTime.UtcNow >= ReconnectAt) { (this as IManager).EmitEvent("reconnect_attempt"); (this as IManager).EmitEvent("reconnecting"); Open(); } break; case States.Open: ITransport trans = null; // Select transport to use if (Transport != null && Transport.State == TransportStates.Open) trans = Transport; // not yet open? if (trans == null || trans.State != TransportStates.Open) return; // Start to poll the server for events trans.Poll(); // Start to send out unsent packets SendOfflinePackets(); // First time we reached this point. Set the LastHeartbeat to the current time, 'cause we are just opened. if (LastHeartbeat == DateTime.MinValue) { LastHeartbeat = DateTime.UtcNow; lastPingReceived = DateTime.UtcNow; return; } if (DateTime.UtcNow - lastPingReceived > TimeSpan.FromMilliseconds(Handshake.PingInterval + Handshake.PingTimeout)) (this as IManager).TryToReconnect(); break; // case States.Open: } } #endregion } } #endif