Circularbuffer.cs 7.6 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306
  1. using System;
  2. using System.Collections.Generic;
  3. using System.IO;
  4. namespace ET
  5. {
  6. public class CircularBuffer: Stream
  7. {
  8. public int ChunkSize = 8192;
  9. private readonly Queue<byte[]> bufferQueue = new Queue<byte[]>();
  10. private readonly Queue<byte[]> bufferCache = new Queue<byte[]>();
  11. public int LastIndex { get; set; }
  12. public int FirstIndex { get; set; }
  13. private byte[] lastBuffer;
  14. public CircularBuffer()
  15. {
  16. this.AddLast();
  17. }
  18. public override long Length
  19. {
  20. get
  21. {
  22. int c = 0;
  23. if (this.bufferQueue.Count == 0)
  24. {
  25. c = 0;
  26. }
  27. else
  28. {
  29. c = (this.bufferQueue.Count - 1) * ChunkSize + this.LastIndex - this.FirstIndex;
  30. }
  31. if (c < 0)
  32. {
  33. Log.Error("CircularBuffer count < 0: {0}, {1}, {2}".Fmt(this.bufferQueue.Count, this.LastIndex, this.FirstIndex));
  34. }
  35. return c;
  36. }
  37. }
  38. public void AddLast()
  39. {
  40. byte[] buffer;
  41. if (this.bufferCache.Count > 0)
  42. {
  43. buffer = this.bufferCache.Dequeue();
  44. }
  45. else
  46. {
  47. buffer = new byte[ChunkSize];
  48. }
  49. this.bufferQueue.Enqueue(buffer);
  50. this.lastBuffer = buffer;
  51. }
  52. public void RemoveFirst()
  53. {
  54. this.bufferCache.Enqueue(bufferQueue.Dequeue());
  55. }
  56. public byte[] First
  57. {
  58. get
  59. {
  60. if (this.bufferQueue.Count == 0)
  61. {
  62. this.AddLast();
  63. }
  64. return this.bufferQueue.Peek();
  65. }
  66. }
  67. public byte[] Last
  68. {
  69. get
  70. {
  71. if (this.bufferQueue.Count == 0)
  72. {
  73. this.AddLast();
  74. }
  75. return this.lastBuffer;
  76. }
  77. }
  78. /// <summary>
  79. /// 从CircularBuffer读到stream中
  80. /// </summary>
  81. /// <param name="stream"></param>
  82. /// <returns></returns>
  83. //public async ETTask ReadAsync(Stream stream)
  84. //{
  85. // long buffLength = this.Length;
  86. // int sendSize = this.ChunkSize - this.FirstIndex;
  87. // if (sendSize > buffLength)
  88. // {
  89. // sendSize = (int)buffLength;
  90. // }
  91. //
  92. // await stream.WriteAsync(this.First, this.FirstIndex, sendSize);
  93. //
  94. // this.FirstIndex += sendSize;
  95. // if (this.FirstIndex == this.ChunkSize)
  96. // {
  97. // this.FirstIndex = 0;
  98. // this.RemoveFirst();
  99. // }
  100. //}
  101. // 从CircularBuffer读到stream
  102. public void Read(Stream stream, int count)
  103. {
  104. if (count > this.Length)
  105. {
  106. throw new Exception($"bufferList length < count, {Length} {count}");
  107. }
  108. int alreadyCopyCount = 0;
  109. while (alreadyCopyCount < count)
  110. {
  111. int n = count - alreadyCopyCount;
  112. if (ChunkSize - this.FirstIndex > n)
  113. {
  114. stream.Write(this.First, this.FirstIndex, n);
  115. this.FirstIndex += n;
  116. alreadyCopyCount += n;
  117. }
  118. else
  119. {
  120. stream.Write(this.First, this.FirstIndex, ChunkSize - this.FirstIndex);
  121. alreadyCopyCount += ChunkSize - this.FirstIndex;
  122. this.FirstIndex = 0;
  123. this.RemoveFirst();
  124. }
  125. }
  126. }
  127. // 从stream写入CircularBuffer
  128. public void Write(Stream stream)
  129. {
  130. int count = (int)(stream.Length - stream.Position);
  131. int alreadyCopyCount = 0;
  132. while (alreadyCopyCount < count)
  133. {
  134. if (this.LastIndex == ChunkSize)
  135. {
  136. this.AddLast();
  137. this.LastIndex = 0;
  138. }
  139. int n = count - alreadyCopyCount;
  140. if (ChunkSize - this.LastIndex > n)
  141. {
  142. stream.Read(this.lastBuffer, this.LastIndex, n);
  143. this.LastIndex += count - alreadyCopyCount;
  144. alreadyCopyCount += n;
  145. }
  146. else
  147. {
  148. stream.Read(this.lastBuffer, this.LastIndex, ChunkSize - this.LastIndex);
  149. alreadyCopyCount += ChunkSize - this.LastIndex;
  150. this.LastIndex = ChunkSize;
  151. }
  152. }
  153. }
  154. /// <summary>
  155. /// 从stream写入CircularBuffer
  156. /// </summary>
  157. /// <param name="stream"></param>
  158. /// <returns></returns>
  159. //public async ETTask<int> WriteAsync(Stream stream)
  160. //{
  161. // int size = this.ChunkSize - this.LastIndex;
  162. //
  163. // int n = await stream.ReadAsync(this.Last, this.LastIndex, size);
  164. //
  165. // if (n == 0)
  166. // {
  167. // return 0;
  168. // }
  169. //
  170. // this.LastIndex += n;
  171. //
  172. // if (this.LastIndex == this.ChunkSize)
  173. // {
  174. // this.AddLast();
  175. // this.LastIndex = 0;
  176. // }
  177. //
  178. // return n;
  179. //}
  180. // 把CircularBuffer中数据写入buffer
  181. public override int Read(byte[] buffer, int offset, int count)
  182. {
  183. if (buffer.Length < offset + count)
  184. {
  185. throw new Exception($"bufferList length < coutn, buffer length: {buffer.Length} {offset} {count}");
  186. }
  187. long length = this.Length;
  188. if (length < count)
  189. {
  190. count = (int)length;
  191. }
  192. int alreadyCopyCount = 0;
  193. while (alreadyCopyCount < count)
  194. {
  195. int n = count - alreadyCopyCount;
  196. if (ChunkSize - this.FirstIndex > n)
  197. {
  198. Array.Copy(this.First, this.FirstIndex, buffer, alreadyCopyCount + offset, n);
  199. this.FirstIndex += n;
  200. alreadyCopyCount += n;
  201. }
  202. else
  203. {
  204. Array.Copy(this.First, this.FirstIndex, buffer, alreadyCopyCount + offset, ChunkSize - this.FirstIndex);
  205. alreadyCopyCount += ChunkSize - this.FirstIndex;
  206. this.FirstIndex = 0;
  207. this.RemoveFirst();
  208. }
  209. }
  210. return count;
  211. }
  212. // 把buffer写入CircularBuffer中
  213. public override void Write(byte[] buffer, int offset, int count)
  214. {
  215. int alreadyCopyCount = 0;
  216. while (alreadyCopyCount < count)
  217. {
  218. if (this.LastIndex == ChunkSize)
  219. {
  220. this.AddLast();
  221. this.LastIndex = 0;
  222. }
  223. int n = count - alreadyCopyCount;
  224. if (ChunkSize - this.LastIndex > n)
  225. {
  226. Array.Copy(buffer, alreadyCopyCount + offset, this.lastBuffer, this.LastIndex, n);
  227. this.LastIndex += count - alreadyCopyCount;
  228. alreadyCopyCount += n;
  229. }
  230. else
  231. {
  232. Array.Copy(buffer, alreadyCopyCount + offset, this.lastBuffer, this.LastIndex, ChunkSize - this.LastIndex);
  233. alreadyCopyCount += ChunkSize - this.LastIndex;
  234. this.LastIndex = ChunkSize;
  235. }
  236. }
  237. }
  238. public override void Flush()
  239. {
  240. throw new NotImplementedException();
  241. }
  242. public override long Seek(long offset, SeekOrigin origin)
  243. {
  244. throw new NotImplementedException();
  245. }
  246. public override void SetLength(long value)
  247. {
  248. throw new NotImplementedException();
  249. }
  250. public override bool CanRead
  251. {
  252. get
  253. {
  254. return true;
  255. }
  256. }
  257. public override bool CanSeek
  258. {
  259. get
  260. {
  261. return false;
  262. }
  263. }
  264. public override bool CanWrite
  265. {
  266. get
  267. {
  268. return true;
  269. }
  270. }
  271. public override long Position { get; set; }
  272. }
  273. }