MessageIO.cs 6.5 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167
  1. using System;
  2. using System.Collections.Generic;
  3. using System.IO;
  4. using System.Text;
  5. namespace Taobao.Top.Link.Endpoints
  6. {
  7. //simple protocol impl, match protocolVersion=2
  8. public class MessageIO
  9. {
  10. public static Message ReadMessage(Stream input)
  11. {
  12. var buffer = new BinaryReader(input);
  13. Message msg = new Message();
  14. msg.ProtocolVersion = buffer.ReadByte();
  15. msg.MessageType = buffer.ReadByte();
  16. // read kv
  17. IDictionary<string, object> dict = new Dictionary<string, object>();
  18. short headerType = buffer.ReadInt16();
  19. while (headerType != MessageType.HeaderType.EndOfHeaders)
  20. {
  21. if (headerType == MessageType.HeaderType.Custom)
  22. dict.Add(ReadCountedString(buffer), ReadCustomValue(buffer));
  23. else if (headerType == MessageType.HeaderType.StatusCode)
  24. msg.StatusCode = buffer.ReadInt32();
  25. else if (headerType == MessageType.HeaderType.StatusPhrase)
  26. msg.StatusPhase = ReadCountedString(buffer);
  27. else if (headerType == MessageType.HeaderType.Flag)
  28. msg.Flag = buffer.ReadInt32();
  29. else if (headerType == MessageType.HeaderType.Token)
  30. msg.Token = ReadCountedString(buffer);
  31. headerType = buffer.ReadInt16();
  32. }
  33. msg.Content = dict;
  34. return msg;
  35. }
  36. public static void WriteMessage(Stream input, Message message)
  37. {
  38. var buffer = new BinaryWriter(input);
  39. buffer.Write((byte)message.ProtocolVersion);
  40. buffer.Write((byte)message.MessageType);
  41. if (message.StatusCode > 0)
  42. {
  43. buffer.Write(MessageType.HeaderType.StatusCode);
  44. buffer.Write(message.StatusCode);
  45. }
  46. if (message.StatusPhase != null && message.StatusPhase != "")
  47. {
  48. buffer.Write(MessageType.HeaderType.StatusPhrase);
  49. WriteCountedString(buffer, message.StatusPhase);
  50. }
  51. if (message.Flag > 0)
  52. {
  53. buffer.Write(MessageType.HeaderType.Flag);
  54. buffer.Write(message.Flag);
  55. }
  56. if (message.Token != null && message.Token != "")
  57. {
  58. buffer.Write(MessageType.HeaderType.Token);
  59. WriteCountedString(buffer, message.Token);
  60. }
  61. if (message.Content != null)
  62. {
  63. foreach (var i in message.Content)
  64. WriteCustomHeader(buffer, i.Key, i.Value);
  65. }
  66. buffer.Write(MessageType.HeaderType.EndOfHeaders);
  67. }
  68. // UTF-8 only
  69. private static string ReadCountedString(BinaryReader buffer)
  70. {
  71. int size = buffer.ReadInt32();
  72. return size > 0 ? Encoding.UTF8.GetString(buffer.ReadBytes(size)) : null;
  73. }
  74. private static void WriteCountedString(BinaryWriter buffer, string value)
  75. {
  76. int strLength = 0;
  77. if (value != null)
  78. strLength = value.Length;
  79. if (strLength > 0)
  80. {
  81. byte[] strBytes = Encoding.UTF8.GetBytes(value);
  82. buffer.Write(strBytes.Length);
  83. buffer.Write(strBytes);
  84. }
  85. else
  86. buffer.Write(0);
  87. }
  88. private static void WriteCustomHeader(BinaryWriter buffer, string name, object value)
  89. {
  90. buffer.Write(MessageType.HeaderType.Custom);
  91. WriteCountedString(buffer, name);
  92. WriteCustomValue(buffer, value);
  93. }
  94. private static object ReadCustomValue(BinaryReader buffer)
  95. {
  96. byte format = buffer.ReadByte();
  97. switch (format)
  98. {
  99. case MessageType.ValueFormat.Void:
  100. return null;
  101. case MessageType.ValueFormat.Byte:
  102. return buffer.ReadByte();
  103. case MessageType.ValueFormat.Int16:
  104. return buffer.ReadInt16();
  105. case MessageType.ValueFormat.Int32:
  106. return buffer.ReadInt32();
  107. case MessageType.ValueFormat.Int64:
  108. return buffer.ReadInt64();
  109. case MessageType.ValueFormat.Date:
  110. return TimeZone.CurrentTimeZone.ToLocalTime(new DateTime(1970, 1, 1).AddMilliseconds(buffer.ReadInt64()));
  111. case MessageType.ValueFormat.ByteArray:
  112. return buffer.ReadBytes(buffer.ReadInt32());
  113. default:
  114. return ReadCountedString(buffer);
  115. }
  116. }
  117. private static void WriteCustomValue(BinaryWriter buffer, object value)
  118. {
  119. if (value == null)
  120. {
  121. buffer.Write(MessageType.ValueFormat.Void);
  122. return;
  123. }
  124. Type type = value.GetType();
  125. if (typeof(byte).Equals(type) || typeof(Byte).Equals(type))
  126. {
  127. buffer.Write(MessageType.ValueFormat.Byte);
  128. buffer.Write((byte)value);
  129. }
  130. else if (typeof(short).Equals(type))
  131. {
  132. buffer.Write(MessageType.ValueFormat.Int16);
  133. buffer.Write((short)value);
  134. }
  135. else if (typeof(int).Equals(type) || typeof(Int32).Equals(type))
  136. {
  137. buffer.Write(MessageType.ValueFormat.Int32);
  138. buffer.Write((int)value);
  139. }
  140. else if (typeof(long).Equals(type) || typeof(Int64).Equals(type))
  141. {
  142. buffer.Write(MessageType.ValueFormat.Int64);
  143. buffer.Write((long)value);
  144. }
  145. else if (typeof(DateTime).Equals(type))
  146. {
  147. buffer.Write(MessageType.ValueFormat.Date);
  148. buffer.Write(((DateTime)value).Ticks);
  149. }
  150. else if (typeof(byte[]).Equals(type))
  151. {
  152. buffer.Write(MessageType.ValueFormat.ByteArray);
  153. var data = value as byte[];
  154. buffer.Write(data.Length);
  155. buffer.Write(data);
  156. }
  157. else
  158. {
  159. buffer.Write(MessageType.ValueFormat.CountedString);
  160. WriteCountedString(buffer, (string)value);
  161. }
  162. }
  163. }
  164. }