WebSocketServer.cs 29 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849
  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 Microsoft.Extensions.Logging;
  15. using Newtonsoft.Json;
  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;
  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. Logger.LogError("Cannot register duplicate name sub protocol! Duplicate name: {0}.", subProtocol.Name);
  191. return false;
  192. }
  193. m_SubProtocols.Add(subProtocol.Name, subProtocol);
  194. return true;
  195. }
  196. private bool SetupSubProtocols(IServerConfig config)
  197. {
  198. //Preparing sub protocols' configuration
  199. var subProtocolConfigSection = config.GetChildConfig<SubProtocolConfigCollection>("subProtocols");
  200. var subProtocolConfigDict = new Dictionary<string, SubProtocolConfig>(subProtocolConfigSection == null ? 0 : subProtocolConfigSection.Count, StringComparer.OrdinalIgnoreCase);
  201. if (subProtocolConfigSection != null)
  202. {
  203. foreach (var protocolConfig in subProtocolConfigSection)
  204. {
  205. string originalProtocolName = protocolConfig.Name;
  206. string protocolName;
  207. ISubProtocol<TWebSocketSession> subProtocolInstance;
  208. if (!string.IsNullOrEmpty(originalProtocolName))
  209. {
  210. protocolName = originalProtocolName;
  211. if (!string.IsNullOrEmpty(protocolConfig.Type))
  212. {
  213. try
  214. {
  215. subProtocolInstance = AssemblyUtil.CreateInstance<ISubProtocol<TWebSocketSession>>(protocolConfig.Type, new object[] { originalProtocolName });
  216. }
  217. catch (Exception e)
  218. {
  219. Logger.LogError(e.Message);
  220. return false;
  221. }
  222. if (!RegisterSubProtocol(subProtocolInstance))
  223. return false;
  224. }
  225. else
  226. {
  227. if (!m_SubProtocols.ContainsKey(protocolName))
  228. {
  229. subProtocolInstance = new BasicSubProtocol<TWebSocketSession>(protocolName);
  230. if (!RegisterSubProtocol(subProtocolInstance))
  231. return false;
  232. }
  233. }
  234. }
  235. else
  236. {
  237. protocolName = BasicSubProtocol<TWebSocketSession>.DefaultName;
  238. if (!string.IsNullOrEmpty(protocolConfig.Type))
  239. {
  240. Logger.LogError("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.LogError(e,e.Message);
  268. }
  269. if (!initialized)
  270. {
  271. Logger.LogError("Failed to initialize the sub protocol '{0}'!", subProtocol.Name);
  272. return false;
  273. }
  274. }
  275. return true;
  276. }
  277. /// <summary>
  278. /// Setups with the specified root config.
  279. /// </summary>
  280. /// <param name="rootConfig">The root config.</param>
  281. /// <param name="config">The config.</param>
  282. /// <returns></returns>
  283. protected override bool Setup(IRootConfig rootConfig, IServerConfig config)
  284. {
  285. if (m_SubProtocols != null && m_SubProtocols.Count > 0)
  286. DefaultSubProtocol = m_SubProtocols.Values.FirstOrDefault();
  287. m_WebSocketProtocolProcessor = new DraftHybi10Processor
  288. {
  289. NextProcessor = new Rfc6455Processor
  290. {
  291. NextProcessor = new DraftHybi00Processor()
  292. }
  293. };
  294. SetupMultipleProtocolSwitch(m_WebSocketProtocolProcessor);
  295. if (!int.TryParse(config.Options.GetValue("handshakePendingQueueCheckingInterval"), out m_HandshakePendingQueueCheckingInterval))
  296. m_HandshakePendingQueueCheckingInterval = 60;// 1 minute default
  297. if (!int.TryParse(config.Options.GetValue("openHandshakeTimeOut"), out m_OpenHandshakeTimeOut))
  298. m_OpenHandshakeTimeOut = 120;// 2 minute default
  299. if (!int.TryParse(config.Options.GetValue("closeHandshakeTimeOut"), out m_CloseHandshakeTimeOut))
  300. m_CloseHandshakeTimeOut = 120;// 2 minute default
  301. if (m_BinaryDataConverter == null)
  302. {
  303. m_BinaryDataConverter = new TextEncodingBinaryDataConverter(Encoding.UTF8);
  304. }
  305. return true;
  306. }
  307. private void SetupMultipleProtocolSwitch(IProtocolProcessor rootProcessor)
  308. {
  309. var thisProcessor = rootProcessor;
  310. List<int> availableVersions = new List<int>();
  311. while (true)
  312. {
  313. if (thisProcessor.Version > 0)
  314. availableVersions.Add(thisProcessor.Version);
  315. if (thisProcessor.NextProcessor == null)
  316. break;
  317. thisProcessor = thisProcessor.NextProcessor;
  318. }
  319. thisProcessor.NextProcessor = new MultipleProtocolSwitchProcessor(availableVersions.ToArray());
  320. }
  321. /// <summary>
  322. /// Called when [started].
  323. /// </summary>
  324. protected override void OnStarted()
  325. {
  326. m_HandshakePendingQueueCheckingTimer = new Timer(HandshakePendingQueueCheckingCallback, null, m_HandshakePendingQueueCheckingInterval * 1000, m_HandshakePendingQueueCheckingInterval * 1000);
  327. base.OnStarted();
  328. }
  329. private void HandshakePendingQueueCheckingCallback(object state)
  330. {
  331. try
  332. {
  333. m_HandshakePendingQueueCheckingTimer.Change(Timeout.Infinite, Timeout.Infinite);
  334. while (true)
  335. {
  336. TWebSocketSession session;
  337. if (!m_OpenHandshakePendingQueue.TryPeek(out session))
  338. break;
  339. if (session.Handshaked || !session.Connected)
  340. {
  341. //Handshaked or not connected
  342. m_OpenHandshakePendingQueue.TryDequeue(out session);
  343. continue;
  344. }
  345. if (DateTime.UtcNow < session.StartTime.AddSeconds(m_OpenHandshakeTimeOut))
  346. break;
  347. //Timeout, dequeue and then close
  348. m_OpenHandshakePendingQueue.TryDequeue(out session);
  349. session.Close(CloseReason.TimeOut);
  350. }
  351. while (true)
  352. {
  353. TWebSocketSession session;
  354. if (!m_CloseHandshakePendingQueue.TryPeek(out session))
  355. break;
  356. if (!session.Connected)
  357. {
  358. //the session has been closed
  359. m_CloseHandshakePendingQueue.TryDequeue(out session);
  360. continue;
  361. }
  362. if (DateTime.UtcNow < session.StartClosingHandshakeTime.AddSeconds(m_CloseHandshakeTimeOut))
  363. break;
  364. //Timeout, dequeue and then close
  365. m_CloseHandshakePendingQueue.TryDequeue(out session);
  366. //Needn't send closing handshake again
  367. session.Close(CloseReason.ServerClosing);
  368. }
  369. }
  370. catch (Exception e)
  371. {
  372. Logger.LogError(e,e.Message);
  373. }
  374. finally
  375. {
  376. m_HandshakePendingQueueCheckingTimer.Change(m_HandshakePendingQueueCheckingInterval * 1000, m_HandshakePendingQueueCheckingInterval * 1000);
  377. }
  378. }
  379. internal void PushToCloseHandshakeQueue(IAppSession appSession)
  380. {
  381. m_CloseHandshakePendingQueue.Enqueue((TWebSocketSession)appSession);
  382. }
  383. /// <summary>
  384. /// Called when [new session connected].
  385. /// </summary>
  386. /// <param name="session">The session.</param>
  387. protected override void OnNewSessionConnected(TWebSocketSession session)
  388. {
  389. m_OpenHandshakePendingQueue.Enqueue(session);
  390. }
  391. internal void FireOnNewSessionConnected(IAppSession appSession)
  392. {
  393. base.OnNewSessionConnected((TWebSocketSession)appSession);
  394. }
  395. /// <summary>
  396. /// Occurs when [new request received].
  397. /// </summary>
  398. /// <exception cref="System.NotSupportedException"></exception>
  399. [Browsable(false)]
  400. [EditorBrowsable(EditorBrowsableState.Never)]
  401. public override event RequestHandler<TWebSocketSession, IWebSocketFragment> NewRequestReceived
  402. {
  403. add { throw new NotSupportedException("Please use NewMessageReceived instead!"); }
  404. remove { throw new NotSupportedException("Please use NewMessageReceived instead!"); }
  405. }
  406. private SessionHandler<TWebSocketSession, string> m_NewMessageReceived;
  407. /// <summary>
  408. /// Occurs when [new message received].
  409. /// </summary>
  410. public event SessionHandler<TWebSocketSession, string> NewMessageReceived
  411. {
  412. add
  413. {
  414. if (m_SubProtocolConfigured)
  415. throw new Exception("If you have defined any sub protocol, you cannot subscribe NewMessageReceived event!");
  416. m_NewMessageReceived += value;
  417. }
  418. remove
  419. {
  420. m_NewMessageReceived -= value;
  421. }
  422. }
  423. void ExecuteMessage(TWebSocketSession session, string message)
  424. {
  425. if (session.SubProtocol == null)
  426. {
  427. Logger.LogError("No SubProtocol selected! This session cannot process any message!");
  428. session.CloseWithHandshake(session.ProtocolProcessor.CloseStatusClode.ProtocolError, "No SubProtocol selected");
  429. return;
  430. }
  431. ExecuteSubCommand(session, session.SubProtocol.SubRequestParser.ParseRequestInfo(message));
  432. }
  433. internal void OnNewMessageReceived(TWebSocketSession session, string message)
  434. {
  435. try
  436. {
  437. if (m_NewMessageReceived == null)
  438. {
  439. ExecuteMessage(session, message);
  440. }
  441. else
  442. {
  443. m_NewMessageReceived(session, message);
  444. }
  445. }
  446. catch (Exception ex)
  447. {
  448. Logger.LogError(message + " " + ex.ToString());
  449. }
  450. }
  451. private SessionHandler<TWebSocketSession, byte[]> m_NewDataReceived;
  452. /// <summary>
  453. /// Occurs when [new data received].
  454. /// </summary>
  455. public event SessionHandler<TWebSocketSession, byte[]> NewDataReceived
  456. {
  457. add
  458. {
  459. m_NewDataReceived += value;
  460. }
  461. remove
  462. {
  463. m_NewDataReceived -= value;
  464. }
  465. }
  466. internal void OnNewDataReceived(TWebSocketSession session, byte[] data)
  467. {
  468. if (m_NewDataReceived == null)
  469. {
  470. var converter = m_BinaryDataConverter;
  471. if (converter != null)
  472. {
  473. ExecuteMessage(session, converter.ToString(data, 0, data.Length));
  474. }
  475. return;
  476. }
  477. m_NewDataReceived(session, data);
  478. }
  479. private const string m_Tab = "\t";
  480. private const char m_Colon = ':';
  481. private const string m_Space = " ";
  482. private const char m_SpaceChar = ' ';
  483. private const string m_ValueSeparator = ", ";
  484. internal static void ParseHandshake(IWebSocketSession session, TextReader reader)
  485. {
  486. string line;
  487. string firstLine = string.Empty;
  488. string prevKey = string.Empty;
  489. //logger.LogInformation("============================================================");
  490. while (!string.IsNullOrEmpty(line = reader.ReadLine()))
  491. {
  492. //logger.LogInformation(session.SessionID + " " + line);
  493. if (string.IsNullOrEmpty(firstLine))
  494. {
  495. firstLine = line;
  496. continue;
  497. }
  498. if (line.StartsWith(m_Tab) && !string.IsNullOrEmpty(prevKey))
  499. {
  500. string currentValue = session.Items.GetValue<string>(prevKey, string.Empty);
  501. session.Items[prevKey] = currentValue + line.Trim();
  502. continue;
  503. }
  504. int pos = line.IndexOf(m_Colon);
  505. if (pos <= 0)
  506. continue;
  507. string key = line.Substring(0, pos);
  508. if (!string.IsNullOrEmpty(key))
  509. key = key.Trim();
  510. var valueOffset = pos + 1;
  511. if (line.Length <= valueOffset) //No value in this line
  512. continue;
  513. string value = line.Substring(valueOffset);
  514. if (!string.IsNullOrEmpty(value) && value.StartsWith(m_Space) && value.Length > 1)
  515. value = value.Substring(1);
  516. if (string.IsNullOrEmpty(key))
  517. continue;
  518. object oldValue;
  519. if (!session.Items.TryGetValue(key, out oldValue))
  520. {
  521. session.Items.Add(key, value);
  522. }
  523. else
  524. {
  525. session.Items[key] = oldValue + m_ValueSeparator + value;
  526. }
  527. prevKey = key;
  528. }
  529. //logger.LogInformation("============================================================");
  530. var metaInfo = firstLine.Split(m_SpaceChar);
  531. session.Method = metaInfo[0];
  532. session.Path = metaInfo[1];
  533. session.HttpVersion = metaInfo[2];
  534. }
  535. /// <summary>
  536. /// Setups the commands.
  537. /// </summary>
  538. /// <param name="discoveredCommands">The discovered commands.</param>
  539. /// <returns></returns>
  540. protected override bool SetupCommands(Dictionary<string, ICommand<TWebSocketSession, IWebSocketFragment>> discoveredCommands)
  541. {
  542. var commands = new List<ICommand<TWebSocketSession, IWebSocketFragment>>
  543. {
  544. new HandShake<TWebSocketSession>(),
  545. new Text<TWebSocketSession>(),
  546. new Binary<TWebSocketSession>(),
  547. new Close<TWebSocketSession>(),
  548. new Ping<TWebSocketSession>(),
  549. new Pong<TWebSocketSession>(),
  550. new Continuation<TWebSocketSession>(),
  551. new Plain<TWebSocketSession>()
  552. };
  553. commands.ForEach(c => discoveredCommands.Add(c.Name, c));
  554. if (!SetupSubProtocols(Config))
  555. return false;
  556. return true;
  557. }
  558. /// <summary>
  559. /// Executes the command.
  560. /// </summary>
  561. /// <param name="session">The session.</param>
  562. /// <param name="requestInfo">The request info.</param>
  563. protected override void ExecuteCommand(TWebSocketSession session, IWebSocketFragment requestInfo)
  564. {
  565. if (session.InClosing)
  566. {
  567. //Only handle closing handshake if the session is in closing
  568. if (requestInfo.Key != OpCode.CloseTag)
  569. return;
  570. }
  571. base.ExecuteCommand(session, requestInfo);
  572. }
  573. private void ExecuteSubCommand(TWebSocketSession session, SubRequestInfo requestInfo)
  574. {
  575. ISubCommand<TWebSocketSession> subCommand;
  576. if (session.SubProtocol.TryGetCommand(requestInfo.Key, out subCommand))
  577. {
  578. session.CurrentCommand = requestInfo.Key;
  579. subCommand.ExecuteCommand(session, requestInfo);
  580. session.PrevCommand = requestInfo.Key;
  581. Logger.LogInfo(session, string.Format("Command - {0} - {1}", session.SessionID, requestInfo.Key));
  582. }
  583. else
  584. {
  585. //session.SubProtocol.TryGetCommand("OCPP", out subCommand);
  586. //session.CurrentCommand = "OCPP";
  587. //subCommand.ExecuteCommand(session, requestInfo);
  588. //session.PrevCommand = "OCPP";
  589. session.HandleUnknownCommand(requestInfo);
  590. }
  591. session.LastActiveTime = DateTime.UtcNow;
  592. }
  593. /// <summary>
  594. /// Broadcasts data to the specified sessions.
  595. /// </summary>
  596. /// <param name="sessions">The sessions.</param>
  597. /// <param name="data">The data.</param>
  598. /// <param name="offset">The offset.</param>
  599. /// <param name="length">The length.</param>
  600. /// <param name="sendFeedback">The send feedback.</param>
  601. public void Broadcast(IEnumerable<TWebSocketSession> sessions, byte[] data, int offset, int length, Action<TWebSocketSession, bool> sendFeedback)
  602. {
  603. IList<ArraySegment<byte>> encodedPackage = null;
  604. IProtocolProcessor encodingProcessor = null;
  605. foreach (var s in sessions)
  606. {
  607. if (!s.Connected)
  608. continue;
  609. var currentProtocolProcessor = s.ProtocolProcessor;
  610. if (currentProtocolProcessor == null || !currentProtocolProcessor.CanSendBinaryData)
  611. continue;
  612. if (encodedPackage == null || currentProtocolProcessor != encodingProcessor)
  613. {
  614. encodedPackage = currentProtocolProcessor.GetEncodedPackage(OpCode.Binary, data, offset, length);
  615. encodingProcessor = currentProtocolProcessor;
  616. }
  617. Task.Factory.StartNew(SendRawDataToSession, new BroadcastState(s, encodedPackage, sendFeedback));
  618. }
  619. }
  620. /// <summary>
  621. /// Broadcasts message to the specified sessions.
  622. /// </summary>
  623. /// <param name="sessions">The sessions.</param>
  624. /// <param name="message">The message.</param>
  625. /// <param name="sendFeedback">The send feedback.</param>
  626. public void Broadcast(IEnumerable<TWebSocketSession> sessions, string message, Action<TWebSocketSession, bool> sendFeedback)
  627. {
  628. IList<ArraySegment<byte>> encodedPackage = null;
  629. IProtocolProcessor encodingProcessor = null;
  630. foreach (var s in sessions)
  631. {
  632. if (!s.Connected)
  633. continue;
  634. var currentProtocolProcessor = s.ProtocolProcessor;
  635. if (currentProtocolProcessor == null)
  636. continue;
  637. if (encodedPackage == null || encodingProcessor != currentProtocolProcessor)
  638. {
  639. encodedPackage = currentProtocolProcessor.GetEncodedPackage(OpCode.Text, message);
  640. encodingProcessor = currentProtocolProcessor;
  641. }
  642. Task.Factory.StartNew(SendRawDataToSession, new BroadcastState(s, encodedPackage, sendFeedback));
  643. }
  644. }
  645. private void SendRawDataToSession(object state)
  646. {
  647. var param = state as BroadcastState;
  648. var session = param.Session;
  649. var sendFeedback = param.FeedbackFunc;
  650. var sendOk = false;
  651. try
  652. {
  653. sendOk = session.TrySendRawData(param.Data);
  654. }
  655. catch (Exception e)
  656. {
  657. session.Logger.LogError(e,e.Message);
  658. }
  659. sendFeedback(session, sendOk);
  660. }
  661. #region JSON serialize/deserialize
  662. /// <summary>
  663. /// Serialize the target object by JSON
  664. /// </summary>
  665. /// <param name="target">The target.</param>
  666. /// <returns></returns>
  667. public virtual string JsonSerialize(object target)
  668. {
  669. return JsonConvert.SerializeObject(target);
  670. }
  671. /// <summary>
  672. /// Deserialize the JSON string to target type object.
  673. /// </summary>
  674. /// <param name="json">The json.</param>
  675. /// <param name="type">The type.</param>
  676. /// <returns></returns>
  677. public virtual object JsonDeserialize(string json, Type type)
  678. {
  679. return JsonConvert.DeserializeObject(json, type);
  680. }
  681. #endregion
  682. class BroadcastState
  683. {
  684. public TWebSocketSession Session { get; private set; }
  685. public IList<ArraySegment<byte>> Data { get; private set; }
  686. public Action<TWebSocketSession, bool> FeedbackFunc { get; private set; }
  687. public BroadcastState(TWebSocketSession session, IList<ArraySegment<byte>> data, Action<TWebSocketSession, bool> feedbackFunc)
  688. {
  689. Session = session;
  690. Data = data;
  691. FeedbackFunc = feedbackFunc;
  692. }
  693. }
  694. }
  695. }