WebSocketServer.cs 30 KB

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