WebSocketServer.cs 29 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852
  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. try
  439. {
  440. if (m_NewMessageReceived == null)
  441. {
  442. ExecuteMessage(session, message);
  443. }
  444. else
  445. {
  446. m_NewMessageReceived(session, message);
  447. }
  448. }
  449. catch (Exception ex)
  450. {
  451. if (Logger.IsErrorEnabled)
  452. Logger.Error(message + " " + ex.ToString());
  453. }
  454. }
  455. private SessionHandler<TWebSocketSession, byte[]> m_NewDataReceived;
  456. /// <summary>
  457. /// Occurs when [new data received].
  458. /// </summary>
  459. public event SessionHandler<TWebSocketSession, byte[]> NewDataReceived
  460. {
  461. add
  462. {
  463. m_NewDataReceived += value;
  464. }
  465. remove
  466. {
  467. m_NewDataReceived -= value;
  468. }
  469. }
  470. internal void OnNewDataReceived(TWebSocketSession session, byte[] data)
  471. {
  472. if (m_NewDataReceived == null)
  473. {
  474. var converter = m_BinaryDataConverter;
  475. if (converter != null)
  476. {
  477. ExecuteMessage(session, converter.ToString(data, 0, data.Length));
  478. }
  479. return;
  480. }
  481. m_NewDataReceived(session, data);
  482. }
  483. private const string m_Tab = "\t";
  484. private const char m_Colon = ':';
  485. private const string m_Space = " ";
  486. private const char m_SpaceChar = ' ';
  487. private const string m_ValueSeparator = ", ";
  488. internal static void ParseHandshake(IWebSocketSession session, TextReader reader)
  489. {
  490. string line;
  491. string firstLine = string.Empty;
  492. string prevKey = string.Empty;
  493. while (!string.IsNullOrEmpty(line = reader.ReadLine()))
  494. {
  495. Console.WriteLine(session.SessionID + " " + line);
  496. if (string.IsNullOrEmpty(firstLine))
  497. {
  498. firstLine = line;
  499. continue;
  500. }
  501. if (line.StartsWith(m_Tab) && !string.IsNullOrEmpty(prevKey))
  502. {
  503. string currentValue = session.Items.GetValue<string>(prevKey, string.Empty);
  504. session.Items[prevKey] = currentValue + line.Trim();
  505. continue;
  506. }
  507. int pos = line.IndexOf(m_Colon);
  508. if (pos <= 0)
  509. continue;
  510. string key = line.Substring(0, pos);
  511. if (!string.IsNullOrEmpty(key))
  512. key = key.Trim();
  513. var valueOffset = pos + 1;
  514. if (line.Length <= valueOffset) //No value in this line
  515. continue;
  516. string value = line.Substring(valueOffset);
  517. if (!string.IsNullOrEmpty(value) && value.StartsWith(m_Space) && value.Length > 1)
  518. value = value.Substring(1);
  519. if (string.IsNullOrEmpty(key))
  520. continue;
  521. object oldValue;
  522. if (!session.Items.TryGetValue(key, out oldValue))
  523. {
  524. session.Items.Add(key, value);
  525. }
  526. else
  527. {
  528. session.Items[key] = oldValue + m_ValueSeparator + value;
  529. }
  530. prevKey = key;
  531. }
  532. var metaInfo = firstLine.Split(m_SpaceChar);
  533. session.Method = metaInfo[0];
  534. session.Path = metaInfo[1];
  535. session.HttpVersion = metaInfo[2];
  536. }
  537. /// <summary>
  538. /// Setups the commands.
  539. /// </summary>
  540. /// <param name="discoveredCommands">The discovered commands.</param>
  541. /// <returns></returns>
  542. protected override bool SetupCommands(Dictionary<string, ICommand<TWebSocketSession, IWebSocketFragment>> discoveredCommands)
  543. {
  544. var commands = new List<ICommand<TWebSocketSession, IWebSocketFragment>>
  545. {
  546. new HandShake<TWebSocketSession>(),
  547. new Text<TWebSocketSession>(),
  548. new Binary<TWebSocketSession>(),
  549. new Close<TWebSocketSession>(),
  550. new Ping<TWebSocketSession>(),
  551. new Pong<TWebSocketSession>(),
  552. new Continuation<TWebSocketSession>(),
  553. new Plain<TWebSocketSession>()
  554. };
  555. commands.ForEach(c => discoveredCommands.Add(c.Name, c));
  556. if (!SetupSubProtocols(Config))
  557. return false;
  558. return true;
  559. }
  560. /// <summary>
  561. /// Executes the command.
  562. /// </summary>
  563. /// <param name="session">The session.</param>
  564. /// <param name="requestInfo">The request info.</param>
  565. protected override void ExecuteCommand(TWebSocketSession session, IWebSocketFragment requestInfo)
  566. {
  567. if (session.InClosing)
  568. {
  569. //Only handle closing handshake if the session is in closing
  570. if (requestInfo.Key != OpCode.CloseTag)
  571. return;
  572. }
  573. base.ExecuteCommand(session, requestInfo);
  574. }
  575. private void ExecuteSubCommand(TWebSocketSession session, SubRequestInfo requestInfo)
  576. {
  577. ISubCommand<TWebSocketSession> subCommand;
  578. if (session.SubProtocol.TryGetCommand(requestInfo.Key, out subCommand))
  579. {
  580. session.CurrentCommand = requestInfo.Key;
  581. subCommand.ExecuteCommand(session, requestInfo);
  582. session.PrevCommand = requestInfo.Key;
  583. if (Config.LogCommand && Logger.IsInfoEnabled)
  584. Logger.Info(session, string.Format("Command - {0} - {1}", session.SessionID, requestInfo.Key));
  585. }
  586. else
  587. {
  588. //session.SubProtocol.TryGetCommand("OCPP", out subCommand);
  589. //session.CurrentCommand = "OCPP";
  590. //subCommand.ExecuteCommand(session, requestInfo);
  591. //session.PrevCommand = "OCPP";
  592. session.HandleUnknownCommand(requestInfo);
  593. }
  594. session.LastActiveTime = DateTime.Now;
  595. }
  596. /// <summary>
  597. /// Broadcasts data to the specified sessions.
  598. /// </summary>
  599. /// <param name="sessions">The sessions.</param>
  600. /// <param name="data">The data.</param>
  601. /// <param name="offset">The offset.</param>
  602. /// <param name="length">The length.</param>
  603. /// <param name="sendFeedback">The send feedback.</param>
  604. public void Broadcast(IEnumerable<TWebSocketSession> sessions, byte[] data, int offset, int length, Action<TWebSocketSession, bool> sendFeedback)
  605. {
  606. IList<ArraySegment<byte>> encodedPackage = null;
  607. IProtocolProcessor encodingProcessor = null;
  608. foreach (var s in sessions)
  609. {
  610. if (!s.Connected)
  611. continue;
  612. var currentProtocolProcessor = s.ProtocolProcessor;
  613. if (currentProtocolProcessor == null || !currentProtocolProcessor.CanSendBinaryData)
  614. continue;
  615. if (encodedPackage == null || currentProtocolProcessor != encodingProcessor)
  616. {
  617. encodedPackage = currentProtocolProcessor.GetEncodedPackage(OpCode.Binary, data, offset, length);
  618. encodingProcessor = currentProtocolProcessor;
  619. }
  620. Task.Factory.StartNew(SendRawDataToSession, new BroadcastState(s, encodedPackage, sendFeedback));
  621. }
  622. }
  623. /// <summary>
  624. /// Broadcasts message to the specified sessions.
  625. /// </summary>
  626. /// <param name="sessions">The sessions.</param>
  627. /// <param name="message">The message.</param>
  628. /// <param name="sendFeedback">The send feedback.</param>
  629. public void Broadcast(IEnumerable<TWebSocketSession> sessions, string message, Action<TWebSocketSession, bool> sendFeedback)
  630. {
  631. IList<ArraySegment<byte>> encodedPackage = null;
  632. IProtocolProcessor encodingProcessor = null;
  633. foreach (var s in sessions)
  634. {
  635. if (!s.Connected)
  636. continue;
  637. var currentProtocolProcessor = s.ProtocolProcessor;
  638. if (currentProtocolProcessor == null)
  639. continue;
  640. if (encodedPackage == null || encodingProcessor != currentProtocolProcessor)
  641. {
  642. encodedPackage = currentProtocolProcessor.GetEncodedPackage(OpCode.Text, message);
  643. encodingProcessor = currentProtocolProcessor;
  644. }
  645. Task.Factory.StartNew(SendRawDataToSession, new BroadcastState(s, encodedPackage, sendFeedback));
  646. }
  647. }
  648. private void SendRawDataToSession(object state)
  649. {
  650. var param = state as BroadcastState;
  651. var session = param.Session;
  652. var sendFeedback = param.FeedbackFunc;
  653. var sendOk = false;
  654. try
  655. {
  656. sendOk = session.TrySendRawData(param.Data);
  657. }
  658. catch (Exception e)
  659. {
  660. session.Logger.Error(e);
  661. }
  662. sendFeedback(session, sendOk);
  663. }
  664. #region JSON serialize/deserialize
  665. /// <summary>
  666. /// Serialize the target object by JSON
  667. /// </summary>
  668. /// <param name="target">The target.</param>
  669. /// <returns></returns>
  670. public virtual string JsonSerialize(object target)
  671. {
  672. return JsonConvert.SerializeObject(target);
  673. }
  674. /// <summary>
  675. /// Deserialize the JSON string to target type object.
  676. /// </summary>
  677. /// <param name="json">The json.</param>
  678. /// <param name="type">The type.</param>
  679. /// <returns></returns>
  680. public virtual object JsonDeserialize(string json, Type type)
  681. {
  682. return JsonConvert.DeserializeObject(json, type);
  683. }
  684. #endregion
  685. class BroadcastState
  686. {
  687. public TWebSocketSession Session { get; private set; }
  688. public IList<ArraySegment<byte>> Data { get; private set; }
  689. public Action<TWebSocketSession, bool> FeedbackFunc { get; private set; }
  690. public BroadcastState(TWebSocketSession session, IList<ArraySegment<byte>> data, Action<TWebSocketSession, bool> feedbackFunc)
  691. {
  692. Session = session;
  693. Data = data;
  694. FeedbackFunc = feedbackFunc;
  695. }
  696. }
  697. }
  698. }