WebSocketServer.cs 29 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841
  1. using System;
  2. using System.Collections.Concurrent;
  3. using System.Collections.Generic;
  4. using System.Collections.Specialized;
  5. using System.ComponentModel;
  6. using System.IO;
  7. using System.Linq;
  8. using System.Net;
  9. using System.Security.Cryptography;
  10. using System.Text;
  11. using System.Text.RegularExpressions;
  12. using System.Threading;
  13. using System.Threading.Tasks;
  14. using Newtonsoft.Json;
  15. using SuperSocket.Common;
  16. using SuperSocket.SocketBase;
  17. using SuperSocket.SocketBase.Command;
  18. using SuperSocket.SocketBase.Config;
  19. using SuperSocket.SocketBase.Protocol;
  20. using SuperWebSocket.Command;
  21. using SuperWebSocket.Config;
  22. using SuperWebSocket.Protocol;
  23. using SuperWebSocket.SubProtocol;
  24. namespace SuperWebSocket
  25. {
  26. /// <summary>
  27. /// WebSocket server interface
  28. /// </summary>
  29. public interface IWebSocketServer : IAppServer
  30. {
  31. /// <summary>
  32. /// Gets the web socket protocol processor.
  33. /// </summary>
  34. IProtocolProcessor WebSocketProtocolProcessor { get; }
  35. /// <summary>
  36. /// Validates the handshake request.
  37. /// </summary>
  38. /// <param name="session">The session.</param>
  39. /// <param name="origin">The origin.</param>
  40. /// <returns>the validation result</returns>
  41. bool ValidateHandshake(IWebSocketSession session, string origin);
  42. }
  43. /// <summary>
  44. /// WebSocket AppServer
  45. /// </summary>
  46. public class WebSocketServer : WebSocketServer<WebSocketSession>
  47. {
  48. /// <summary>
  49. /// Initializes a new instance of the <see cref="WebSocketServer"/> class.
  50. /// </summary>
  51. /// <param name="subProtocols">The sub protocols.</param>
  52. public WebSocketServer(IEnumerable<ISubProtocol<WebSocketSession>> subProtocols)
  53. : base(subProtocols)
  54. {
  55. }
  56. /// <summary>
  57. /// Initializes a new instance of the <see cref="WebSocketServer"/> class.
  58. /// </summary>
  59. /// <param name="subProtocol">The sub protocol.</param>
  60. public WebSocketServer(ISubProtocol<WebSocketSession> subProtocol)
  61. : base(subProtocol)
  62. {
  63. }
  64. /// <summary>
  65. /// Initializes a new instance of the <see cref="WebSocketServer"/> class.
  66. /// </summary>
  67. public WebSocketServer()
  68. : base(new List<ISubProtocol<WebSocketSession>>())
  69. {
  70. }
  71. }
  72. /// <summary>
  73. /// WebSocket AppServer
  74. /// </summary>
  75. /// <typeparam name="TWebSocketSession">The type of the web socket session.</typeparam>
  76. public abstract class WebSocketServer<TWebSocketSession> : AppServer<TWebSocketSession, IWebSocketFragment>, IWebSocketServer
  77. where TWebSocketSession : WebSocketSession<TWebSocketSession>, new()
  78. {
  79. private IBinaryDataConverter m_BinaryDataConverter;
  80. /// <summary>
  81. /// Gets or sets the binary data converter.
  82. /// </summary>
  83. /// <value>
  84. /// The binary data converter.
  85. /// </value>
  86. protected IBinaryDataConverter BinaryDataConverter
  87. {
  88. get { return m_BinaryDataConverter; }
  89. set { m_BinaryDataConverter = value; }
  90. }
  91. /// <summary>
  92. /// Initializes a new instance of the <see cref="WebSocketServer&lt;TWebSocketSession&gt;"/> class.
  93. /// </summary>
  94. /// <param name="subProtocols">The sub protocols.</param>
  95. public WebSocketServer(IEnumerable<ISubProtocol<TWebSocketSession>> subProtocols)
  96. : this()
  97. {
  98. if (!subProtocols.Any())
  99. return;
  100. foreach (var protocol in subProtocols)
  101. {
  102. if (!RegisterSubProtocol(protocol))
  103. throw new Exception("Failed to register sub protocol!");
  104. }
  105. m_SubProtocolConfigured = true;
  106. }
  107. /// <summary>
  108. /// Initializes a new instance of the <see cref="WebSocketServer&lt;TWebSocketSession&gt;"/> class.
  109. /// </summary>
  110. /// <param name="subProtocol">The sub protocol.</param>
  111. public WebSocketServer(ISubProtocol<TWebSocketSession> subProtocol)
  112. : this(new List<ISubProtocol<TWebSocketSession>> { subProtocol })
  113. {
  114. }
  115. /// <summary>
  116. /// Initializes a new instance of the <see cref="WebSocketServer&lt;TWebSocketSession&gt;"/> class.
  117. /// </summary>
  118. public WebSocketServer()
  119. : base(new WebSocketProtocol())
  120. {
  121. }
  122. private Dictionary<string, ISubProtocol<TWebSocketSession>> m_SubProtocols = new Dictionary<string, ISubProtocol<TWebSocketSession>>(StringComparer.OrdinalIgnoreCase);
  123. internal ISubProtocol<TWebSocketSession> DefaultSubProtocol { get; private set; }
  124. private bool m_SubProtocolConfigured = false;
  125. private ConcurrentQueue<TWebSocketSession> m_OpenHandshakePendingQueue = new ConcurrentQueue<TWebSocketSession>();
  126. private ConcurrentQueue<TWebSocketSession> m_CloseHandshakePendingQueue = new ConcurrentQueue<TWebSocketSession>();
  127. /// <summary>
  128. /// The openning handshake timeout, in seconds
  129. /// </summary>
  130. private int m_OpenHandshakeTimeOut;
  131. /// <summary>
  132. /// The closing handshake timeout, in seconds
  133. /// </summary>
  134. private int m_CloseHandshakeTimeOut;
  135. /// <summary>
  136. /// The interval of checking handshake pending queue, in seconds
  137. /// </summary>
  138. private int m_HandshakePendingQueueCheckingInterval;
  139. private Timer m_HandshakePendingQueueCheckingTimer;
  140. /// <summary>
  141. /// Gets the sub protocol by sub protocol name.
  142. /// </summary>
  143. /// <param name="name">The name.</param>
  144. /// <returns></returns>
  145. internal ISubProtocol<TWebSocketSession> GetSubProtocol(string name)
  146. {
  147. ISubProtocol<TWebSocketSession> subProtocol;
  148. if (m_SubProtocols.TryGetValue(name, out subProtocol))
  149. return subProtocol;
  150. else
  151. return null;
  152. }
  153. private IProtocolProcessor m_WebSocketProtocolProcessor;
  154. IProtocolProcessor IWebSocketServer.WebSocketProtocolProcessor
  155. {
  156. get { return m_WebSocketProtocolProcessor; }
  157. }
  158. /// <summary>
  159. /// Gets the request filter factory.
  160. /// </summary>
  161. public new WebSocketProtocol ReceiveFilterFactory
  162. {
  163. get
  164. {
  165. return (WebSocketProtocol)base.ReceiveFilterFactory;
  166. }
  167. }
  168. bool IWebSocketServer.ValidateHandshake(IWebSocketSession session, string origin)
  169. {
  170. var s = (TWebSocketSession)session;
  171. s.Origin = origin;
  172. return ValidateHandshake(s, origin);
  173. }
  174. /// <summary>
  175. /// Validates the handshake request.
  176. /// </summary>
  177. /// <param name="session">The session.</param>
  178. /// <param name="origin">The origin in the handshake request.</param>
  179. /// <returns></returns>
  180. protected virtual bool ValidateHandshake(TWebSocketSession session, string origin)
  181. {
  182. return true;
  183. }
  184. bool RegisterSubProtocol(ISubProtocol<TWebSocketSession> subProtocol)
  185. {
  186. if (m_SubProtocols.ContainsKey(subProtocol.Name))
  187. {
  188. if(Logger.IsErrorEnabled)
  189. Logger.ErrorFormat("Cannot register duplicate name sub protocol! Duplicate name: {0}.", subProtocol.Name);
  190. return false;
  191. }
  192. m_SubProtocols.Add(subProtocol.Name, subProtocol);
  193. return true;
  194. }
  195. private bool SetupSubProtocols(IServerConfig config)
  196. {
  197. //Preparing sub protocols' configuration
  198. var subProtocolConfigSection = config.GetChildConfig<SubProtocolConfigCollection>("subProtocols");
  199. var subProtocolConfigDict = new Dictionary<string, SubProtocolConfig>(subProtocolConfigSection == null ? 0 : subProtocolConfigSection.Count, StringComparer.OrdinalIgnoreCase);
  200. if (subProtocolConfigSection != null)
  201. {
  202. foreach (var protocolConfig in subProtocolConfigSection)
  203. {
  204. string originalProtocolName = protocolConfig.Name;
  205. string protocolName;
  206. ISubProtocol<TWebSocketSession> subProtocolInstance;
  207. if (!string.IsNullOrEmpty(originalProtocolName))
  208. {
  209. protocolName = originalProtocolName;
  210. if (!string.IsNullOrEmpty(protocolConfig.Type))
  211. {
  212. try
  213. {
  214. subProtocolInstance = AssemblyUtil.CreateInstance<ISubProtocol<TWebSocketSession>>(protocolConfig.Type, new object[] { originalProtocolName });
  215. }
  216. catch (Exception e)
  217. {
  218. Logger.Error(e);
  219. return false;
  220. }
  221. if (!RegisterSubProtocol(subProtocolInstance))
  222. return false;
  223. }
  224. else
  225. {
  226. if (!m_SubProtocols.ContainsKey(protocolName))
  227. {
  228. subProtocolInstance = new BasicSubProtocol<TWebSocketSession>(protocolName);
  229. if (!RegisterSubProtocol(subProtocolInstance))
  230. return false;
  231. }
  232. }
  233. }
  234. else
  235. {
  236. protocolName = BasicSubProtocol<TWebSocketSession>.DefaultName;
  237. if (!string.IsNullOrEmpty(protocolConfig.Type))
  238. {
  239. if(Logger.IsErrorEnabled)
  240. Logger.Error("You needn't set Type attribute for SubProtocol, if you don't set Name attribute!");
  241. return false;
  242. }
  243. }
  244. subProtocolConfigDict[protocolName] = protocolConfig;
  245. }
  246. if(subProtocolConfigDict.Values.Any())
  247. m_SubProtocolConfigured = true;
  248. }
  249. if (m_SubProtocols.Count <= 0 || (subProtocolConfigDict.ContainsKey(BasicSubProtocol<TWebSocketSession>.DefaultName) && !m_SubProtocols.ContainsKey(BasicSubProtocol<TWebSocketSession>.DefaultName)))
  250. {
  251. if (!RegisterSubProtocol(BasicSubProtocol<TWebSocketSession>.CreateDefaultSubProtocol()))
  252. return false;
  253. }
  254. //Initialize sub protocols
  255. foreach (var subProtocol in m_SubProtocols.Values)
  256. {
  257. SubProtocolConfig protocolConfig = null;
  258. subProtocolConfigDict.TryGetValue(subProtocol.Name, out protocolConfig);
  259. bool initialized = false;
  260. try
  261. {
  262. initialized = subProtocol.Initialize(this, protocolConfig, Logger);
  263. }
  264. catch (Exception e)
  265. {
  266. initialized = false;
  267. Logger.Error(e);
  268. }
  269. if (!initialized)
  270. {
  271. if (Logger.IsErrorEnabled)
  272. Logger.ErrorFormat("Failed to initialize the sub protocol '{0}'!", subProtocol.Name);
  273. return false;
  274. }
  275. }
  276. return true;
  277. }
  278. /// <summary>
  279. /// Setups with the specified root config.
  280. /// </summary>
  281. /// <param name="rootConfig">The root config.</param>
  282. /// <param name="config">The config.</param>
  283. /// <returns></returns>
  284. protected override bool Setup(IRootConfig rootConfig, IServerConfig config)
  285. {
  286. if (m_SubProtocols != null && m_SubProtocols.Count > 0)
  287. DefaultSubProtocol = m_SubProtocols.Values.FirstOrDefault();
  288. m_WebSocketProtocolProcessor = new DraftHybi10Processor
  289. {
  290. NextProcessor = new Rfc6455Processor
  291. {
  292. NextProcessor = new DraftHybi00Processor()
  293. }
  294. };
  295. SetupMultipleProtocolSwitch(m_WebSocketProtocolProcessor);
  296. if (!int.TryParse(config.Options.GetValue("handshakePendingQueueCheckingInterval"), out m_HandshakePendingQueueCheckingInterval))
  297. m_HandshakePendingQueueCheckingInterval = 60;// 1 minute default
  298. if (!int.TryParse(config.Options.GetValue("openHandshakeTimeOut"), out m_OpenHandshakeTimeOut))
  299. m_OpenHandshakeTimeOut = 120;// 2 minute default
  300. if (!int.TryParse(config.Options.GetValue("closeHandshakeTimeOut"), out m_CloseHandshakeTimeOut))
  301. m_CloseHandshakeTimeOut = 120;// 2 minute default
  302. if (m_BinaryDataConverter == null)
  303. {
  304. m_BinaryDataConverter = new TextEncodingBinaryDataConverter(Encoding.UTF8);
  305. }
  306. return true;
  307. }
  308. private void SetupMultipleProtocolSwitch(IProtocolProcessor rootProcessor)
  309. {
  310. var thisProcessor = rootProcessor;
  311. List<int> availableVersions = new List<int>();
  312. while (true)
  313. {
  314. if (thisProcessor.Version > 0)
  315. availableVersions.Add(thisProcessor.Version);
  316. if (thisProcessor.NextProcessor == null)
  317. break;
  318. thisProcessor = thisProcessor.NextProcessor;
  319. }
  320. thisProcessor.NextProcessor = new MultipleProtocolSwitchProcessor(availableVersions.ToArray());
  321. }
  322. /// <summary>
  323. /// Called when [started].
  324. /// </summary>
  325. protected override void OnStarted()
  326. {
  327. m_HandshakePendingQueueCheckingTimer = new Timer(HandshakePendingQueueCheckingCallback, null, m_HandshakePendingQueueCheckingInterval * 1000, m_HandshakePendingQueueCheckingInterval * 1000);
  328. base.OnStarted();
  329. }
  330. private void HandshakePendingQueueCheckingCallback(object state)
  331. {
  332. try
  333. {
  334. m_HandshakePendingQueueCheckingTimer.Change(Timeout.Infinite, Timeout.Infinite);
  335. while (true)
  336. {
  337. TWebSocketSession session;
  338. if (!m_OpenHandshakePendingQueue.TryPeek(out session))
  339. break;
  340. if (session.Handshaked || !session.Connected)
  341. {
  342. //Handshaked or not connected
  343. m_OpenHandshakePendingQueue.TryDequeue(out session);
  344. continue;
  345. }
  346. if (DateTime.Now < session.StartTime.AddSeconds(m_OpenHandshakeTimeOut))
  347. break;
  348. //Timeout, dequeue and then close
  349. m_OpenHandshakePendingQueue.TryDequeue(out session);
  350. session.Close(CloseReason.TimeOut);
  351. }
  352. while (true)
  353. {
  354. TWebSocketSession session;
  355. if (!m_CloseHandshakePendingQueue.TryPeek(out session))
  356. break;
  357. if (!session.Connected)
  358. {
  359. //the session has been closed
  360. m_CloseHandshakePendingQueue.TryDequeue(out session);
  361. continue;
  362. }
  363. if (DateTime.Now < session.StartClosingHandshakeTime.AddSeconds(m_CloseHandshakeTimeOut))
  364. break;
  365. //Timeout, dequeue and then close
  366. m_CloseHandshakePendingQueue.TryDequeue(out session);
  367. //Needn't send closing handshake again
  368. session.Close(CloseReason.ServerClosing);
  369. }
  370. }
  371. catch (Exception e)
  372. {
  373. if(Logger.IsErrorEnabled)
  374. Logger.Error(e);
  375. }
  376. finally
  377. {
  378. m_HandshakePendingQueueCheckingTimer.Change(m_HandshakePendingQueueCheckingInterval * 1000, m_HandshakePendingQueueCheckingInterval * 1000);
  379. }
  380. }
  381. internal void PushToCloseHandshakeQueue(IAppSession appSession)
  382. {
  383. m_CloseHandshakePendingQueue.Enqueue((TWebSocketSession)appSession);
  384. }
  385. /// <summary>
  386. /// Called when [new session connected].
  387. /// </summary>
  388. /// <param name="session">The session.</param>
  389. protected override void OnNewSessionConnected(TWebSocketSession session)
  390. {
  391. m_OpenHandshakePendingQueue.Enqueue(session);
  392. }
  393. internal void FireOnNewSessionConnected(IAppSession appSession)
  394. {
  395. base.OnNewSessionConnected((TWebSocketSession)appSession);
  396. }
  397. /// <summary>
  398. /// Occurs when [new request received].
  399. /// </summary>
  400. /// <exception cref="System.NotSupportedException"></exception>
  401. [Browsable(false)]
  402. [EditorBrowsable(EditorBrowsableState.Never)]
  403. public override event RequestHandler<TWebSocketSession,IWebSocketFragment> NewRequestReceived
  404. {
  405. add { throw new NotSupportedException("Please use NewMessageReceived instead!"); }
  406. remove { throw new NotSupportedException("Please use NewMessageReceived instead!"); }
  407. }
  408. private SessionHandler<TWebSocketSession, string> m_NewMessageReceived;
  409. /// <summary>
  410. /// Occurs when [new message received].
  411. /// </summary>
  412. public event SessionHandler<TWebSocketSession, string> NewMessageReceived
  413. {
  414. add
  415. {
  416. if (m_SubProtocolConfigured)
  417. throw new Exception("If you have defined any sub protocol, you cannot subscribe NewMessageReceived event!");
  418. m_NewMessageReceived += value;
  419. }
  420. remove
  421. {
  422. m_NewMessageReceived -= value;
  423. }
  424. }
  425. void ExecuteMessage(TWebSocketSession session, string message)
  426. {
  427. if (session.SubProtocol == null)
  428. {
  429. if (Logger.IsErrorEnabled)
  430. Logger.Error("No SubProtocol selected! This session cannot process any message!");
  431. session.CloseWithHandshake(session.ProtocolProcessor.CloseStatusClode.ProtocolError, "No SubProtocol selected");
  432. return;
  433. }
  434. ExecuteSubCommand(session, session.SubProtocol.SubRequestParser.ParseRequestInfo(message));
  435. }
  436. internal void OnNewMessageReceived(TWebSocketSession session, string message)
  437. {
  438. if (m_NewMessageReceived == null)
  439. {
  440. ExecuteMessage(session, message);
  441. }
  442. else
  443. {
  444. m_NewMessageReceived(session, message);
  445. }
  446. }
  447. private SessionHandler<TWebSocketSession, byte[]> m_NewDataReceived;
  448. /// <summary>
  449. /// Occurs when [new data received].
  450. /// </summary>
  451. public event SessionHandler<TWebSocketSession, byte[]> NewDataReceived
  452. {
  453. add
  454. {
  455. m_NewDataReceived += value;
  456. }
  457. remove
  458. {
  459. m_NewDataReceived -= value;
  460. }
  461. }
  462. internal void OnNewDataReceived(TWebSocketSession session, byte[] data)
  463. {
  464. if (m_NewDataReceived == null)
  465. {
  466. var converter = m_BinaryDataConverter;
  467. if (converter != null)
  468. {
  469. ExecuteMessage(session, converter.ToString(data, 0, data.Length));
  470. }
  471. return;
  472. }
  473. m_NewDataReceived(session, data);
  474. }
  475. private const string m_Tab = "\t";
  476. private const char m_Colon = ':';
  477. private const string m_Space = " ";
  478. private const char m_SpaceChar = ' ';
  479. private const string m_ValueSeparator = ", ";
  480. internal static void ParseHandshake(IWebSocketSession session, TextReader reader)
  481. {
  482. string line;
  483. string firstLine = string.Empty;
  484. string prevKey = string.Empty;
  485. while (!string.IsNullOrEmpty(line = reader.ReadLine()))
  486. {
  487. if (string.IsNullOrEmpty(firstLine))
  488. {
  489. firstLine = line;
  490. continue;
  491. }
  492. if (line.StartsWith(m_Tab) && !string.IsNullOrEmpty(prevKey))
  493. {
  494. string currentValue = session.Items.GetValue<string>(prevKey, string.Empty);
  495. session.Items[prevKey] = currentValue + line.Trim();
  496. continue;
  497. }
  498. int pos = line.IndexOf(m_Colon);
  499. if (pos <= 0)
  500. continue;
  501. string key = line.Substring(0, pos);
  502. if (!string.IsNullOrEmpty(key))
  503. key = key.Trim();
  504. var valueOffset = pos + 1;
  505. if (line.Length <= valueOffset) //No value in this line
  506. continue;
  507. string value = line.Substring(valueOffset);
  508. if (!string.IsNullOrEmpty(value) && value.StartsWith(m_Space) && value.Length > 1)
  509. value = value.Substring(1);
  510. if (string.IsNullOrEmpty(key))
  511. continue;
  512. object oldValue;
  513. if (!session.Items.TryGetValue(key, out oldValue))
  514. {
  515. session.Items.Add(key, value);
  516. }
  517. else
  518. {
  519. session.Items[key] = oldValue + m_ValueSeparator + value;
  520. }
  521. prevKey = key;
  522. }
  523. var metaInfo = firstLine.Split(m_SpaceChar);
  524. session.Method = metaInfo[0];
  525. session.Path = metaInfo[1];
  526. session.HttpVersion = metaInfo[2];
  527. }
  528. /// <summary>
  529. /// Setups the commands.
  530. /// </summary>
  531. /// <param name="discoveredCommands">The discovered commands.</param>
  532. /// <returns></returns>
  533. protected override bool SetupCommands(Dictionary<string, ICommand<TWebSocketSession, IWebSocketFragment>> discoveredCommands)
  534. {
  535. var commands = new List<ICommand<TWebSocketSession, IWebSocketFragment>>
  536. {
  537. new HandShake<TWebSocketSession>(),
  538. new Text<TWebSocketSession>(),
  539. new Binary<TWebSocketSession>(),
  540. new Close<TWebSocketSession>(),
  541. new Ping<TWebSocketSession>(),
  542. new Pong<TWebSocketSession>(),
  543. new Continuation<TWebSocketSession>(),
  544. new Plain<TWebSocketSession>()
  545. };
  546. commands.ForEach(c => discoveredCommands.Add(c.Name, c));
  547. if (!SetupSubProtocols(Config))
  548. return false;
  549. return true;
  550. }
  551. /// <summary>
  552. /// Executes the command.
  553. /// </summary>
  554. /// <param name="session">The session.</param>
  555. /// <param name="requestInfo">The request info.</param>
  556. protected override void ExecuteCommand(TWebSocketSession session, IWebSocketFragment requestInfo)
  557. {
  558. if (session.InClosing)
  559. {
  560. //Only handle closing handshake if the session is in closing
  561. if (requestInfo.Key != OpCode.CloseTag)
  562. return;
  563. }
  564. base.ExecuteCommand(session, requestInfo);
  565. }
  566. private void ExecuteSubCommand(TWebSocketSession session, SubRequestInfo requestInfo)
  567. {
  568. ISubCommand<TWebSocketSession> subCommand;
  569. if (session.SubProtocol.TryGetCommand(requestInfo.Key, out subCommand))
  570. {
  571. session.CurrentCommand = requestInfo.Key;
  572. subCommand.ExecuteCommand(session, requestInfo);
  573. session.PrevCommand = requestInfo.Key;
  574. if (Config.LogCommand && Logger.IsInfoEnabled)
  575. Logger.Info(session, string.Format("Command - {0} - {1}", session.SessionID, requestInfo.Key));
  576. }
  577. else
  578. {
  579. //session.SubProtocol.TryGetCommand("OCPP", out subCommand);
  580. //session.CurrentCommand = "OCPP";
  581. //subCommand.ExecuteCommand(session, requestInfo);
  582. //session.PrevCommand = "OCPP";
  583. session.HandleUnknownCommand(requestInfo);
  584. }
  585. session.LastActiveTime = DateTime.Now;
  586. }
  587. /// <summary>
  588. /// Broadcasts data to the specified sessions.
  589. /// </summary>
  590. /// <param name="sessions">The sessions.</param>
  591. /// <param name="data">The data.</param>
  592. /// <param name="offset">The offset.</param>
  593. /// <param name="length">The length.</param>
  594. /// <param name="sendFeedback">The send feedback.</param>
  595. public void Broadcast(IEnumerable<TWebSocketSession> sessions, byte[] data, int offset, int length, Action<TWebSocketSession, bool> sendFeedback)
  596. {
  597. IList<ArraySegment<byte>> encodedPackage = null;
  598. IProtocolProcessor encodingProcessor = null;
  599. foreach (var s in sessions)
  600. {
  601. if (!s.Connected)
  602. continue;
  603. var currentProtocolProcessor = s.ProtocolProcessor;
  604. if (currentProtocolProcessor == null || !currentProtocolProcessor.CanSendBinaryData)
  605. continue;
  606. if (encodedPackage == null || currentProtocolProcessor != encodingProcessor)
  607. {
  608. encodedPackage = currentProtocolProcessor.GetEncodedPackage(OpCode.Binary, data, offset, length);
  609. encodingProcessor = currentProtocolProcessor;
  610. }
  611. Task.Factory.StartNew(SendRawDataToSession, new BroadcastState(s, encodedPackage, sendFeedback));
  612. }
  613. }
  614. /// <summary>
  615. /// Broadcasts message to the specified sessions.
  616. /// </summary>
  617. /// <param name="sessions">The sessions.</param>
  618. /// <param name="message">The message.</param>
  619. /// <param name="sendFeedback">The send feedback.</param>
  620. public void Broadcast(IEnumerable<TWebSocketSession> sessions, string message, Action<TWebSocketSession, bool> sendFeedback)
  621. {
  622. IList<ArraySegment<byte>> encodedPackage = null;
  623. IProtocolProcessor encodingProcessor = null;
  624. foreach (var s in sessions)
  625. {
  626. if (!s.Connected)
  627. continue;
  628. var currentProtocolProcessor = s.ProtocolProcessor;
  629. if (currentProtocolProcessor == null)
  630. continue;
  631. if (encodedPackage == null || encodingProcessor != currentProtocolProcessor)
  632. {
  633. encodedPackage = currentProtocolProcessor.GetEncodedPackage(OpCode.Text, message);
  634. encodingProcessor = currentProtocolProcessor;
  635. }
  636. Task.Factory.StartNew(SendRawDataToSession, new BroadcastState(s, encodedPackage, sendFeedback));
  637. }
  638. }
  639. private void SendRawDataToSession(object state)
  640. {
  641. var param = state as BroadcastState;
  642. var session = param.Session;
  643. var sendFeedback = param.FeedbackFunc;
  644. var sendOk = false;
  645. try
  646. {
  647. sendOk = session.TrySendRawData(param.Data);
  648. }
  649. catch (Exception e)
  650. {
  651. session.Logger.Error(e);
  652. }
  653. sendFeedback(session, sendOk);
  654. }
  655. #region JSON serialize/deserialize
  656. /// <summary>
  657. /// Serialize the target object by JSON
  658. /// </summary>
  659. /// <param name="target">The target.</param>
  660. /// <returns></returns>
  661. public virtual string JsonSerialize(object target)
  662. {
  663. return JsonConvert.SerializeObject(target);
  664. }
  665. /// <summary>
  666. /// Deserialize the JSON string to target type object.
  667. /// </summary>
  668. /// <param name="json">The json.</param>
  669. /// <param name="type">The type.</param>
  670. /// <returns></returns>
  671. public virtual object JsonDeserialize(string json, Type type)
  672. {
  673. return JsonConvert.DeserializeObject(json, type);
  674. }
  675. #endregion
  676. class BroadcastState
  677. {
  678. public TWebSocketSession Session { get; private set; }
  679. public IList<ArraySegment<byte>> Data { get; private set; }
  680. public Action<TWebSocketSession, bool> FeedbackFunc { get; private set; }
  681. public BroadcastState(TWebSocketSession session, IList<ArraySegment<byte>> data, Action<TWebSocketSession, bool> feedbackFunc)
  682. {
  683. Session = session;
  684. Data = data;
  685. FeedbackFunc = feedbackFunc;
  686. }
  687. }
  688. }
  689. }