OutgoingSocketExtensions.cs 32 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773
  1. using System;
  2. using System.Collections.Generic;
  3. using System.Diagnostics;
  4. using NetMQ.Utils;
  5. namespace NetMQ
  6. {
  7. /// <summary>
  8. /// This static class serves to provide extension methods for IOutgoingSocket.
  9. /// </summary>
  10. public static class OutgoingSocketExtensions
  11. {
  12. /// <summary>
  13. /// Block until the message can be sent.
  14. /// </summary>
  15. /// <remarks>
  16. /// The call blocks until the message can be sent and cannot be interrupted.
  17. /// Whether the message can be sent depends on the socket type.
  18. /// </remarks>
  19. /// <param name="socket">The socket to send the message on.</param>
  20. /// <param name="msg">An object with message's data to send.</param>
  21. /// <param name="more">Indicate if another frame is expected after this frame</param>
  22. public static void Send(this IOutgoingSocket socket, ref Msg msg, bool more)
  23. {
  24. var result = socket.TrySend(ref msg, SendReceiveConstants.InfiniteTimeout, more);
  25. Debug.Assert(result);
  26. }
  27. #region Sending Byte Array
  28. #region Blocking
  29. /// <summary>
  30. /// Transmit a byte-array of data over this socket, block until frame is sent.
  31. /// </summary>
  32. /// <param name="socket">the IOutgoingSocket to transmit on</param>
  33. /// <param name="data">the byte-array of data to send</param>
  34. /// <param name="more">set this flag to true to signal that you will be immediately sending another frame (optional: default is false)</param>
  35. public static void SendFrame(this IOutgoingSocket socket, byte[] data, bool more = false)
  36. {
  37. SendFrame(socket, data, data.Length, more);
  38. }
  39. /// <summary>
  40. /// Transmit a byte-array of data over this socket, block until frame is sent.
  41. /// </summary>
  42. /// <param name="socket">the IOutgoingSocket to transmit on</param>
  43. /// <param name="data">the byte-array of data to send</param>
  44. /// <param name="length">the number of bytes to send from <paramref name="data"/>.</param>
  45. /// <param name="more">set this flag to true to signal that you will be immediately sending another frame (optional: default is false)</param>
  46. public static void SendFrame(this IOutgoingSocket socket, byte[] data, int length, bool more = false)
  47. {
  48. var msg = new Msg();
  49. msg.InitPool(length);
  50. data.Slice(0, length).CopyTo(msg);
  51. socket.Send(ref msg, more);
  52. msg.Close();
  53. }
  54. /// <summary>
  55. /// Transmit a byte-array of data over this socket, block until frame is sent.
  56. /// Send more frame, another frame must be sent after this frame. Use to chain Send methods.
  57. /// </summary>
  58. /// <param name="socket">the IOutgoingSocket to transmit on</param>
  59. /// <param name="data">the byte-array of data to send</param>
  60. /// <returns>a reference to this IOutgoingSocket so that method-calls may be chained together</returns>
  61. public static IOutgoingSocket SendMoreFrame(this IOutgoingSocket socket, byte[] data)
  62. {
  63. SendFrame(socket, data, true);
  64. return socket;
  65. }
  66. /// <summary>
  67. /// Transmit a byte-array of data over this socket, block until frame is sent.
  68. /// Send more frame, another frame must be sent after this frame. Use to chain Send methods.
  69. /// </summary>
  70. /// <param name="socket">the IOutgoingSocket to transmit on</param>
  71. /// <param name="data">the byte-array of data to send</param>
  72. /// <param name="length">the number of bytes to send from <paramref name="data"/>.</param>
  73. /// <returns>a reference to this IOutgoingSocket so that method-calls may be chained together</returns>
  74. public static IOutgoingSocket SendMoreFrame(this IOutgoingSocket socket, byte[] data, int length)
  75. {
  76. SendFrame(socket, data, length, true);
  77. return socket;
  78. }
  79. #endregion
  80. #region Timeout
  81. /// <summary>
  82. /// Attempt to transmit a single frame on <paramref name="socket"/>.
  83. /// If message cannot be sent within <paramref name="timeout"/>, return <c>false</c>.
  84. /// </summary>
  85. /// <param name="socket">the IOutgoingSocket to transmit on</param>
  86. /// <param name="timeout">The maximum period of time to try to send a message.</param>
  87. /// <param name="data">the byte-array of data to send</param>
  88. /// <param name="length">the number of bytes to send from <paramref name="data"/>.</param>
  89. /// <param name="more">set this flag to true to signal that you will be immediately sending another frame (optional: default is false)</param>
  90. /// <returns><c>true</c> if a message was available, otherwise <c>false</c>.</returns>
  91. public static bool TrySendFrame(this IOutgoingSocket socket, TimeSpan timeout, byte[] data, int length, bool more = false)
  92. {
  93. var msg = new Msg();
  94. msg.InitPool(length);
  95. data.Slice(0, length).CopyTo(msg);
  96. if (!socket.TrySend(ref msg, timeout, more))
  97. {
  98. msg.Close();
  99. return false;
  100. }
  101. msg.Close();
  102. return true;
  103. }
  104. /// <summary>
  105. /// Attempt to transmit a single frame on <paramref name="socket"/>.
  106. /// If message cannot be sent within <paramref name="timeout"/>, return <c>false</c>.
  107. /// </summary>
  108. /// <param name="socket">the IOutgoingSocket to transmit on</param>
  109. /// <param name="timeout">The maximum period of time to try to send a message.</param>
  110. /// <param name="data">the byte-array of data to send</param>
  111. /// <param name="more">set this flag to true to signal that you will be immediately sending another frame (optional: default is false)</param>
  112. /// <returns><c>true</c> if a message was available, otherwise <c>false</c>.</returns>
  113. public static bool TrySendFrame(this IOutgoingSocket socket, TimeSpan timeout, byte[] data, bool more = false)
  114. {
  115. return TrySendFrame(socket, timeout, data, data.Length, more);
  116. }
  117. #endregion
  118. #region Immediate
  119. /// <summary>
  120. /// Attempt to transmit a single frame on <paramref name="socket"/>.
  121. /// If message cannot be sent immediately, return <c>false</c>.
  122. /// </summary>
  123. /// <param name="socket">the IOutgoingSocket to transmit on</param>
  124. /// <param name="data">the byte-array of data to send</param>
  125. /// <param name="more">set this flag to true to signal that you will be immediately sending another frame (optional: default is false)</param>
  126. /// <returns><c>true</c> if a message was available, otherwise <c>false</c>.</returns>
  127. public static bool TrySendFrame(this IOutgoingSocket socket, byte[] data,
  128. bool more = false)
  129. {
  130. return TrySendFrame(socket, TimeSpan.Zero, data, more);
  131. }
  132. /// <summary>
  133. /// Attempt to transmit a single frame on <paramref name="socket"/>.
  134. /// If message cannot be sent immediately, return <c>false</c>.
  135. /// </summary>
  136. /// <param name="socket">the IOutgoingSocket to transmit on</param>
  137. /// <param name="data">the byte-array of data to send</param>
  138. /// <param name="length">the number of bytes to send from <paramref name="data"/>.</param>
  139. /// <param name="more">set this flag to true to signal that you will be immediately sending another frame (optional: default is false)</param>
  140. /// <returns><c>true</c> if a message was available, otherwise <c>false</c>.</returns>
  141. public static bool TrySendFrame(this IOutgoingSocket socket, byte[] data, int length,
  142. bool more = false)
  143. {
  144. return TrySendFrame(socket, TimeSpan.Zero, data, length, more);
  145. }
  146. #endregion
  147. #endregion
  148. #region Sending a multipart message as byte arrays
  149. #region Blocking
  150. /// <summary>
  151. /// Send multiple frames on <paramref name="socket"/>, blocking until all frames are sent.
  152. /// </summary>
  153. /// <param name="socket">the IOutgoingSocket to transmit on</param>
  154. /// <param name="frames">frames to transmit</param>
  155. public static void SendMultipartBytes(this IOutgoingSocket socket, params byte[][] frames)
  156. {
  157. SendMultipartBytes(socket, (IEnumerable<byte[]>)frames);
  158. }
  159. /// <summary>
  160. /// Send multiple frames on <paramref name="socket"/>, blocking until all frames are sent.
  161. /// </summary>
  162. /// <param name="socket">the IOutgoingSocket to transmit on</param>
  163. /// <param name="frames">frames to transmit</param>
  164. public static void SendMultipartBytes(this IOutgoingSocket socket, IEnumerable<byte[]> frames)
  165. {
  166. var enumerator = frames.GetEnumerator();
  167. try
  168. {
  169. // move to the first element, if false frames is empty
  170. if (!enumerator.MoveNext())
  171. {
  172. throw new ArgumentException("frames is empty", nameof(frames));
  173. }
  174. var current = enumerator.Current;
  175. // we always one item back to make sure we send the last frame without the more flag
  176. while (enumerator.MoveNext())
  177. {
  178. // this is a more frame
  179. socket.SendMoreFrame(current);
  180. current = enumerator.Current;
  181. }
  182. // sending the last frame
  183. socket.SendFrame(current);
  184. }
  185. finally
  186. {
  187. enumerator.Dispose();
  188. }
  189. }
  190. #endregion
  191. #region Timeout
  192. /// <summary>
  193. /// Attempt to transmit a multiple frames on <paramref name="socket"/>.
  194. /// If frames cannot be sent within <paramref name="timeout"/>, return <c>false</c>.
  195. /// </summary>
  196. /// <param name="socket">the IOutgoingSocket to transmit on</param>
  197. /// <param name="timeout">The maximum period of time to try to send a message.</param>
  198. /// <param name="frames">frames to transmit</param>
  199. public static bool TrySendMultipartBytes(this IOutgoingSocket socket, TimeSpan timeout, params byte[][] frames)
  200. {
  201. return TrySendMultipartBytes(socket, timeout, (IEnumerable<byte[]>)frames);
  202. }
  203. /// <summary>
  204. /// Attempt to transmit a multiple frames on <paramref name="socket"/>.
  205. /// If frames cannot be sent within <paramref name="timeout"/>, return <c>false</c>.
  206. /// </summary>
  207. /// <param name="socket">the IOutgoingSocket to transmit on</param>
  208. /// <param name="timeout">The maximum period of time to try to send a message.</param>
  209. /// <param name="frames">frames to transmit</param>
  210. public static bool TrySendMultipartBytes(this IOutgoingSocket socket, TimeSpan timeout,
  211. IEnumerable<byte[]> frames)
  212. {
  213. var enumerator = frames.GetEnumerator();
  214. try
  215. {
  216. // move to the first element, if false frames is empty
  217. if (!enumerator.MoveNext())
  218. {
  219. throw new ArgumentException("frames is empty", nameof(frames));
  220. }
  221. var current = enumerator.Current;
  222. // only the first frame need to be sent with a timeout
  223. if (!enumerator.MoveNext())
  224. {
  225. return socket.TrySendFrame(timeout, current);
  226. }
  227. else
  228. {
  229. bool sentSuccessfully = socket.TrySendFrame(timeout, current, true);
  230. if (!sentSuccessfully)
  231. return false;
  232. }
  233. // fetching the second frame
  234. current = enumerator.Current;
  235. // we always one item back to make sure we send the last frame without the more flag
  236. while (enumerator.MoveNext())
  237. {
  238. // this is a more frame
  239. socket.SendMoreFrame(current);
  240. current = enumerator.Current;
  241. }
  242. // sending the last frame
  243. socket.SendFrame(current);
  244. return true;
  245. }
  246. finally
  247. {
  248. enumerator.Dispose();
  249. }
  250. }
  251. #endregion
  252. #region Immediate
  253. /// <summary>
  254. /// Attempt to transmit a multiple frames on <paramref name="socket"/>.
  255. /// If frames cannot be sent immediately, return <c>false</c>.
  256. /// </summary>
  257. /// <param name="socket">the IOutgoingSocket to transmit on</param>
  258. /// <param name="frames">frames to transmit</param>
  259. public static bool TrySendMultipartBytes(this IOutgoingSocket socket, params byte[][] frames)
  260. {
  261. return TrySendMultipartBytes(socket, TimeSpan.Zero, (IEnumerable<byte[]>)frames);
  262. }
  263. /// <summary>
  264. /// Attempt to transmit a multiple frames on <paramref name="socket"/>.
  265. /// If frames cannot be sent immediately, return <c>false</c>.
  266. /// </summary>
  267. /// <param name="socket">the IOutgoingSocket to transmit on</param>
  268. /// <param name="frames">frames to transmit</param>
  269. public static bool TrySendMultipartBytes(this IOutgoingSocket socket, IEnumerable<byte[]> frames)
  270. {
  271. return TrySendMultipartBytes(socket, TimeSpan.Zero, frames);
  272. }
  273. #endregion
  274. #endregion
  275. #region Sending Strings
  276. #region Blocking
  277. /// <summary>
  278. /// Transmit a string over this socket, block until frame is sent.
  279. /// </summary>
  280. /// <param name="socket">the IOutgoingSocket to transmit on</param>
  281. /// <param name="message">the string to send</param>
  282. /// <param name="more">set this flag to true to signal that you will be immediately sending another frame (optional: default is false)</param>
  283. public static void SendFrame(this IOutgoingSocket socket, string message, bool more = false)
  284. {
  285. var msg = new Msg();
  286. // Count the number of bytes required to encode the string.
  287. // Note that non-ASCII strings may not have an equal number of characters
  288. // and bytes. The encoding must be queried for this answer.
  289. // With this number, request a buffer from the pool.
  290. msg.InitPool(SendReceiveConstants.DefaultEncoding.GetByteCount(message));
  291. // Encode the string into the buffer
  292. SendReceiveConstants.DefaultEncoding.GetBytes(message, msg);
  293. socket.Send(ref msg, more);
  294. msg.Close();
  295. }
  296. /// <summary>
  297. /// Transmit a string over this socket, block until frame is sent.
  298. /// Send more frame, another frame must be sent after this frame. Use to chain Send methods.
  299. /// </summary>
  300. /// <param name="socket">the IOutgoingSocket to transmit on</param>
  301. /// <param name="message">the string to send</param>
  302. /// <returns>a reference to this IOutgoingSocket so that method-calls may be chained together</returns>
  303. public static IOutgoingSocket SendMoreFrame(this IOutgoingSocket socket, string message)
  304. {
  305. SendFrame(socket, message, true);
  306. return socket;
  307. }
  308. #endregion
  309. #region Timeout
  310. /// <summary>
  311. /// Attempt to transmit a single string frame on <paramref name="socket"/>.
  312. /// If message cannot be sent within <paramref name="timeout"/>, return <c>false</c>.
  313. /// </summary>
  314. /// <param name="socket">the IOutgoingSocket to transmit on</param>
  315. /// <param name="timeout">The maximum period of time to try to send a message.</param>
  316. /// <param name="message">the string to send</param>
  317. /// <param name="more">set this flag to true to signal that you will be immediately sending another frame (optional: default is false)</param>
  318. /// <returns><c>true</c> if a message was available, otherwise <c>false</c>.</returns>
  319. public static bool TrySendFrame(this IOutgoingSocket socket, TimeSpan timeout, string message, bool more = false)
  320. {
  321. var msg = new Msg();
  322. // Count the number of bytes required to encode the string.
  323. // Note that non-ASCII strings may not have an equal number of characters
  324. // and bytes. The encoding must be queried for this answer.
  325. // With this number, request a buffer from the pool.
  326. msg.InitPool(SendReceiveConstants.DefaultEncoding.GetByteCount(message));
  327. // Encode the string into the buffer
  328. SendReceiveConstants.DefaultEncoding.GetBytes(message, msg);
  329. if (!socket.TrySend(ref msg, timeout, more))
  330. {
  331. msg.Close();
  332. return false;
  333. }
  334. msg.Close();
  335. return true;
  336. }
  337. #endregion
  338. #region Immediate
  339. /// <summary>
  340. /// Attempt to transmit a single string frame on <paramref name="socket"/>.
  341. /// If message cannot be sent immediately, return <c>false</c>.
  342. /// </summary>
  343. /// <param name="socket">the IOutgoingSocket to transmit on</param>
  344. /// <param name="message">the string to send</param>
  345. /// <param name="more">set this flag to true to signal that you will be immediately sending another frame (optional: default is false)</param>
  346. /// <returns><c>true</c> if a message was available, otherwise <c>false</c>.</returns>
  347. public static bool TrySendFrame(this IOutgoingSocket socket, string message, bool more = false)
  348. {
  349. return TrySendFrame(socket, TimeSpan.Zero, message, more);
  350. }
  351. #endregion
  352. #endregion
  353. #region Sending a multipart message as NetMQMessage
  354. #region Blocking
  355. /// <summary>
  356. /// Send the multiple part message on the <paramref name="socket"/>, blocking until the entire message is sent.
  357. /// </summary>
  358. /// <param name="socket">the IOutgoingSocket to transmit on</param>
  359. /// <param name="message">message to transmit</param>
  360. public static void SendMultipartMessage(this IOutgoingSocket socket, NetMQMessage message)
  361. {
  362. if (message.FrameCount == 0)
  363. throw new ArgumentException("message is empty", nameof(message));
  364. for (int i = 0; i < message.FrameCount - 1; i++)
  365. {
  366. socket.SendMoreFrame(message[i].Buffer, message[i].MessageSize);
  367. }
  368. socket.SendFrame(message.Last.Buffer, message.Last.MessageSize);
  369. }
  370. #endregion
  371. #region Timeout
  372. /// <summary>
  373. /// Attempt to transmit a multiple message on <paramref name="socket"/>.
  374. /// If message cannot be sent within <paramref name="timeout"/>, return <c>false</c>.
  375. /// </summary>
  376. /// <param name="socket">the IOutgoingSocket to transmit on</param>
  377. /// <param name="timeout">The maximum period of time to try to send a message.</param>
  378. /// <param name="message">message to transmit</param>
  379. public static bool TrySendMultipartMessage(this IOutgoingSocket socket, TimeSpan timeout, NetMQMessage message)
  380. {
  381. if (message.FrameCount == 0)
  382. throw new ArgumentException("message is empty", nameof(message));
  383. else if (message.FrameCount == 1)
  384. {
  385. return TrySendFrame(socket, timeout, message[0].Buffer, message[0].MessageSize);
  386. }
  387. else
  388. {
  389. bool sentSuccessfully = TrySendFrame(socket, timeout, message[0].Buffer, message[0].MessageSize, true);
  390. if (!sentSuccessfully)
  391. return false;
  392. }
  393. for (int i = 1; i < message.FrameCount - 1; i++)
  394. {
  395. socket.SendMoreFrame(message[i].Buffer, message[i].MessageSize);
  396. }
  397. socket.SendFrame(message.Last.Buffer, message.Last.MessageSize);
  398. return true;
  399. }
  400. #endregion
  401. #region Immediate
  402. /// <summary>
  403. /// Attempt to transmit a multiple message on <paramref name="socket"/>.
  404. /// If frames cannot be sent immediately, return <c>false</c>.
  405. /// </summary>
  406. /// <param name="socket">the IOutgoingSocket to transmit on</param>
  407. /// <param name="message">message to transmit</param>
  408. public static bool TrySendMultipartMessage(this IOutgoingSocket socket, NetMQMessage message)
  409. {
  410. return TrySendMultipartMessage(socket, TimeSpan.Zero, message);
  411. }
  412. #endregion
  413. #endregion
  414. #region Sending an empty frame
  415. #region Blocking
  416. /// <summary>
  417. /// Transmit an empty frame over this socket, block until frame is sent.
  418. /// </summary>
  419. /// <param name="socket">the IOutgoingSocket to transmit on</param>
  420. /// <param name="more">set this flag to true to signal that you will be immediately sending another frame (optional: default is false)</param>
  421. public static void SendFrameEmpty(this IOutgoingSocket socket, bool more = false)
  422. {
  423. SendFrame(socket, EmptyArray<byte>.Instance, more);
  424. }
  425. /// <summary>
  426. /// Transmit an empty frame over this socket, block until frame is sent.
  427. /// Send more frame, another frame must be sent after this frame. Use to chain Send methods.
  428. /// </summary>
  429. /// <param name="socket">the IOutgoingSocket to transmit on</param>
  430. /// <returns>a reference to this IOutgoingSocket so that method-calls may be chained together</returns>
  431. public static IOutgoingSocket SendMoreFrameEmpty(this IOutgoingSocket socket)
  432. {
  433. SendFrame(socket, EmptyArray<byte>.Instance, true);
  434. return socket;
  435. }
  436. #endregion
  437. #region Timeout
  438. /// <summary>
  439. /// Attempt to transmit an empty frame on <paramref name="socket"/>.
  440. /// If message cannot be sent within <paramref name="timeout"/>, return <c>false</c>.
  441. /// </summary>
  442. /// <param name="socket">the IOutgoingSocket to transmit on</param>
  443. /// <param name="timeout">The maximum period of time to try to send a message.</param>
  444. /// <param name="more">set this flag to true to signal that you will be immediately sending another frame (optional: default is false)</param>
  445. /// <returns><c>true</c> if a message was available, otherwise <c>false</c>.</returns>
  446. public static bool TrySendFrameEmpty(this IOutgoingSocket socket, TimeSpan timeout, bool more = false)
  447. {
  448. return TrySendFrame(socket, timeout, EmptyArray<byte>.Instance, more);
  449. }
  450. #endregion
  451. #region Immediate
  452. /// <summary>
  453. /// Attempt to transmit an empty frame on <paramref name="socket"/>.
  454. /// If message cannot be sent immediately, return <c>false</c>.
  455. /// </summary>
  456. /// <param name="socket">the IOutgoingSocket to transmit on</param>
  457. /// <param name="more">set this flag to true to signal that you will be immediately sending another frame (optional: default is false)</param>
  458. /// <returns><c>true</c> if a message was available, otherwise <c>false</c>.</returns>
  459. public static bool TrySendFrameEmpty(this IOutgoingSocket socket, bool more = false)
  460. {
  461. return TrySendFrame(socket, EmptyArray<byte>.Instance, more);
  462. }
  463. #endregion
  464. #endregion
  465. #region Sending Signals
  466. /// <summary>
  467. /// Transmit a status-signal over this socket.
  468. /// </summary>
  469. /// <param name="socket">the IOutgoingSocket to transmit on</param>
  470. /// <param name="status">a byte that contains the status signal to send</param>
  471. private static void Signal(this IOutgoingSocket socket, byte status)
  472. {
  473. long signalValue = 0x7766554433221100L + status;
  474. Msg msg = new Msg();
  475. msg.InitPool(8);
  476. NetworkOrderBitsConverter.PutInt64(signalValue, msg);
  477. socket.Send(ref msg, false);
  478. msg.Close();
  479. }
  480. /// <summary>
  481. /// Attempt to transmit a status-signal over this socket.
  482. /// If signal cannot be sent immediately, return <c>false</c>.
  483. /// </summary>
  484. /// <param name="socket">the IOutgoingSocket to transmit on</param>
  485. /// <param name="status">a byte that contains the status signal to send</param>
  486. private static bool TrySignal(this IOutgoingSocket socket, byte status)
  487. {
  488. long signalValue = 0x7766554433221100L + status;
  489. Msg msg = new Msg();
  490. msg.InitPool(8);
  491. NetworkOrderBitsConverter.PutInt64(signalValue, msg);
  492. if (!socket.TrySend(ref msg, TimeSpan.Zero, false))
  493. {
  494. msg.Close();
  495. return false;
  496. }
  497. msg.Close();
  498. return true;
  499. }
  500. /// <summary>
  501. /// Transmit a specific status-signal over this socket that indicates OK.
  502. /// </summary>
  503. /// <param name="socket">the IOutgoingSocket to transmit on</param>
  504. public static void SignalOK(this IOutgoingSocket socket)
  505. {
  506. socket.Signal(0);
  507. }
  508. /// <summary>
  509. /// Attempt to transmit a specific status-signal over this socket that indicates OK.
  510. /// If signal cannot be sent immediately, return <c>false</c>.
  511. /// </summary>
  512. /// <param name="socket">the IOutgoingSocket to transmit on</param>
  513. public static bool TrySignalOK(this IOutgoingSocket socket)
  514. {
  515. return TrySignal(socket, 0);
  516. }
  517. /// <summary>
  518. /// Transmit a specific status-signal over this socket that indicates there is an error.
  519. /// </summary>
  520. /// <param name="socket">the IOutgoingSocket to transmit on</param>
  521. public static void SignalError(this IOutgoingSocket socket)
  522. {
  523. socket.Signal(1);
  524. }
  525. /// <summary>
  526. /// Attempt to transmit a specific status-signal over this socket that indicates there is an error.
  527. /// If signal cannot be sent immediately, return <c>false</c>.
  528. /// </summary>
  529. /// <param name="socket">the IOutgoingSocket to transmit on</param>
  530. public static bool TrySignalError(this IOutgoingSocket socket)
  531. {
  532. return socket.TrySignal(1);
  533. }
  534. #endregion
  535. #region Sending Routing Key
  536. /// <summary>
  537. /// Send routing key over <paramref name="socket"/>.
  538. /// </summary>
  539. /// <param name="socket">the IOutgoingSocket to transmit on</param>
  540. /// <param name="routingKey">the routing key to send</param>
  541. public static void SendMoreFrame(this IOutgoingSocket socket, RoutingKey routingKey)
  542. {
  543. socket.SendMoreFrame(routingKey.Bytes);
  544. }
  545. /// <summary>
  546. /// Attempt to transmit routing key over <paramref name="socket"/>.
  547. /// If message cannot be sent immediately, return <c>false</c>.
  548. /// Routing is always sent as more frame.
  549. /// </summary>
  550. /// <param name="socket">the IOutgoingSocket to transmit on</param>
  551. /// <param name="routingKey">the routing key to send</param>
  552. /// <returns><c>true</c> if a message was available, otherwise <c>false</c>.</returns>
  553. public static bool TrySendFrame(this IOutgoingSocket socket, RoutingKey routingKey)
  554. {
  555. return socket.TrySendFrame(routingKey.Bytes, true);
  556. }
  557. /// <summary>
  558. /// Attempt to transmit routing key over <paramref name="socket"/>.
  559. /// If message cannot be sent within <paramref name="timeout"/>, return <c>false</c>.
  560. /// Routing is always sent as more frame.
  561. /// </summary>
  562. /// <param name="socket">the IOutgoingSocket to transmit on</param>
  563. /// <param name="timeout">The maximum period of time to try to send a message.</param>
  564. /// <param name="routingKey">the routing key to send</param>
  565. /// <returns><c>true</c> if a message was available, otherwise <c>false</c>.</returns>
  566. public static bool TrySendFrame(this IOutgoingSocket socket, TimeSpan timeout, RoutingKey routingKey)
  567. {
  568. return socket.TrySendFrame(timeout, routingKey.Bytes, true);
  569. }
  570. #endregion
  571. #region Sending Routing Keys
  572. /// <summary>
  573. /// Send empty list of routing keys over <paramref name="socket"/>, append an empty message at the end of the keys.
  574. /// </summary>
  575. /// <param name="socket">the IOutgoingSocket to transmit on</param>
  576. public static IOutgoingSocket SendEmptyRoutingKeys(this IOutgoingSocket socket)
  577. {
  578. return socket.SendMoreFrameEmpty();
  579. }
  580. /// <summary>
  581. /// Send a single routing key over <paramref name="socket"/>, append an empty message afterwards.
  582. /// </summary>
  583. /// <param name="socket">the IOutgoingSocket to transmit on</param>
  584. /// <param name="routingKeys">the routing keys to send</param>
  585. public static IOutgoingSocket SendRoutingKeys(this IOutgoingSocket socket, params RoutingKey[] routingKeys)
  586. {
  587. foreach(var routingKey in routingKeys)
  588. socket.SendMoreFrame(routingKey);
  589. socket.SendMoreFrameEmpty();
  590. return socket;
  591. }
  592. /// <summary>
  593. /// Send routing keys over <paramref name="socket"/>, append an empty message at the end of the keys.
  594. /// </summary>
  595. /// <param name="socket">the IOutgoingSocket to transmit on</param>
  596. /// <param name="routingKeys">the routing keys to send</param>
  597. public static IOutgoingSocket SendRoutingKeys(this IOutgoingSocket socket, IEnumerable<RoutingKey> routingKeys)
  598. {
  599. foreach(var routingKey in routingKeys)
  600. socket.SendMoreFrame(routingKey);
  601. socket.SendMoreFrameEmpty();
  602. return socket;
  603. }
  604. /// <summary>
  605. /// Attempt to transmit routing keys over <paramref name="socket"/>.
  606. /// If message cannot be sent immediately, return <c>false</c>.
  607. /// Routing is always sent as more frame.
  608. /// </summary>
  609. /// <param name="socket">the IOutgoingSocket to transmit on</param>
  610. /// <param name="routingKeys">the routing keys to send</param>
  611. /// <returns><c>true</c> if a message was available, otherwise <c>false</c>.</returns>
  612. public static bool TrySendRoutingKeys(this IOutgoingSocket socket, IEnumerable<RoutingKey> routingKeys)
  613. {
  614. return socket.TrySendRoutingKeys(TimeSpan.Zero, routingKeys);
  615. }
  616. /// <summary>
  617. /// Attempt to transmit routing key over <paramref name="socket"/>.
  618. /// If message cannot be sent within <paramref name="timeout"/>, return <c>false</c>.
  619. /// Routing is always sent as more frame.
  620. /// </summary>
  621. /// <param name="socket">the IOutgoingSocket to transmit on</param>
  622. /// <param name="timeout">The maximum period of time to try to send a message.</param>
  623. /// <param name="routingKeys">the routing keys to send</param>
  624. /// <returns><c>true</c> if a message was available, otherwise <c>false</c>.</returns>
  625. public static bool TrySendRoutingKeys(this IOutgoingSocket socket, TimeSpan timeout, IEnumerable<RoutingKey> routingKeys)
  626. {
  627. var enumerator = routingKeys.GetEnumerator();
  628. // Empty collection, just trying to send the empty message
  629. if (!enumerator.MoveNext())
  630. return socket.TrySendFrameEmpty(timeout, true);
  631. if (!socket.TrySendFrame(enumerator.Current))
  632. return false;
  633. while (enumerator.MoveNext())
  634. socket.SendMoreFrame(enumerator.Current);
  635. socket.SendMoreFrameEmpty();
  636. return true;
  637. }
  638. #endregion
  639. }
  640. }