ByteBufferStream.cs 22 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727
  1. /* Copyright 2010-2016 MongoDB Inc.
  2. *
  3. * Licensed under the Apache License, Version 2.0 (the "License");
  4. * you may not use this file except in compliance with the License.
  5. * You may obtain a copy of the License at
  6. *
  7. * http://www.apache.org/licenses/LICENSE-2.0
  8. *
  9. * Unless required by applicable law or agreed to in writing, software
  10. * distributed under the License is distributed on an "AS IS" BASIS,
  11. * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12. * See the License for the specific language governing permissions and
  13. * limitations under the License.
  14. */
  15. using System;
  16. using System.IO;
  17. using System.Text;
  18. namespace MongoDB.Bson.IO
  19. {
  20. /// <summary>
  21. /// Represents a Stream backed by an IByteBuffer. Similar to MemoryStream but backed by an IByteBuffer
  22. /// instead of a byte array and also implements the BsonStream interface for higher performance BSON I/O.
  23. /// </summary>
  24. public class ByteBufferStream : BsonStream
  25. {
  26. // private fields
  27. private IByteBuffer _buffer;
  28. private bool _disposed;
  29. private int _length;
  30. private readonly bool _ownsBuffer;
  31. private int _position;
  32. private readonly byte[] _temp = new byte[12];
  33. private readonly byte[] _tempUtf8 = new byte[128];
  34. // constructors
  35. /// <summary>
  36. /// Initializes a new instance of the <see cref="ByteBufferStream"/> class.
  37. /// </summary>
  38. /// <param name="buffer">The buffer.</param>
  39. /// <param name="ownsBuffer">Whether the stream owns the buffer and should Dispose it when done.</param>
  40. public ByteBufferStream(IByteBuffer buffer, bool ownsBuffer = false)
  41. {
  42. if (buffer == null)
  43. {
  44. throw new ArgumentNullException("buffer");
  45. }
  46. _buffer = buffer;
  47. _ownsBuffer = ownsBuffer;
  48. _length = buffer.Length;
  49. }
  50. // public properties
  51. /// <summary>
  52. /// Gets the buffer.
  53. /// </summary>
  54. /// <value>
  55. /// The buffer.
  56. /// </value>
  57. public IByteBuffer Buffer
  58. {
  59. get
  60. {
  61. ThrowIfDisposed();
  62. return _buffer;
  63. }
  64. }
  65. /// <inheritdoc/>
  66. public override bool CanRead
  67. {
  68. get { return !_disposed; }
  69. }
  70. /// <inheritdoc/>
  71. public override bool CanSeek
  72. {
  73. get { return !_disposed; }
  74. }
  75. /// <inheritdoc/>
  76. public override bool CanTimeout
  77. {
  78. get { return false; }
  79. }
  80. /// <inheritdoc/>
  81. public override bool CanWrite
  82. {
  83. get { return !_disposed && !_buffer.IsReadOnly; }
  84. }
  85. /// <inheritdoc/>
  86. public override long Length
  87. {
  88. get
  89. {
  90. ThrowIfDisposed();
  91. return _length;
  92. }
  93. }
  94. /// <inheritdoc/>
  95. public override long Position
  96. {
  97. get
  98. {
  99. ThrowIfDisposed();
  100. return _position;
  101. }
  102. set
  103. {
  104. if (value < 0 || value > int.MaxValue)
  105. {
  106. throw new ArgumentOutOfRangeException("value");
  107. }
  108. ThrowIfDisposed();
  109. _position = (int)value;
  110. }
  111. }
  112. // public methods
  113. /// <inheritdoc/>
  114. public override void Flush()
  115. {
  116. ThrowIfDisposed();
  117. // do nothing
  118. }
  119. /// <inheritdoc/>
  120. public override int Read(byte[] buffer, int offset, int count)
  121. {
  122. if (buffer == null)
  123. {
  124. throw new ArgumentNullException("buffer");
  125. }
  126. if (offset < 0 || offset > buffer.Length)
  127. {
  128. throw new ArgumentOutOfRangeException("offset");
  129. }
  130. if (count < 0 || offset + count > buffer.Length)
  131. {
  132. throw new ArgumentOutOfRangeException("count");
  133. }
  134. ThrowIfDisposed();
  135. if (_position >= _length)
  136. {
  137. return 0;
  138. }
  139. var available = _length - _position;
  140. if (count > available)
  141. {
  142. count = available;
  143. }
  144. _buffer.GetBytes(_position, buffer, offset, count);
  145. _position += count;
  146. return count;
  147. }
  148. /// <inheritdoc/>
  149. public override int ReadByte()
  150. {
  151. ThrowIfDisposed();
  152. if (_position >= _length)
  153. {
  154. return -1;
  155. }
  156. return _buffer.GetByte(_position++);
  157. }
  158. /// <inheritdoc/>
  159. public override long Seek(long offset, SeekOrigin origin)
  160. {
  161. ThrowIfDisposed();
  162. long position;
  163. switch (origin)
  164. {
  165. case SeekOrigin.Begin: position = offset; break;
  166. case SeekOrigin.Current: position = _position + offset; break;
  167. case SeekOrigin.End: position = _length + offset; break;
  168. default: throw new ArgumentException("Invalid origin.", "origin");
  169. }
  170. if (position < 0)
  171. {
  172. throw new IOException("Attempted to seek before the beginning of the stream.");
  173. }
  174. if (position > int.MaxValue)
  175. {
  176. throw new IOException("Attempted to seek beyond the maximum value that can be represented using 32 bits.");
  177. }
  178. _position = (int)position;
  179. return position;
  180. }
  181. /// <inheritdoc/>
  182. public override void SetLength(long value)
  183. {
  184. if (value < 0 || value > int.MaxValue)
  185. {
  186. throw new ArgumentOutOfRangeException("value");
  187. }
  188. ThrowIfDisposed();
  189. EnsureWriteable();
  190. _buffer.EnsureCapacity((int)value);
  191. _length = (int)value;
  192. if (_position > _length)
  193. {
  194. _position = _length;
  195. }
  196. }
  197. /// <inheritdoc/>
  198. public override void Write(byte[] buffer, int offset, int count)
  199. {
  200. if (buffer == null)
  201. {
  202. throw new ArgumentNullException("buffer");
  203. }
  204. if (offset < 0 || offset > buffer.Length)
  205. {
  206. throw new ArgumentOutOfRangeException("offset");
  207. }
  208. if (count < 0 || offset + count > buffer.Length)
  209. {
  210. throw new ArgumentOutOfRangeException("count");
  211. }
  212. ThrowIfDisposed();
  213. EnsureWriteable();
  214. PrepareToWrite(count);
  215. _buffer.SetBytes(_position, buffer, offset, count);
  216. SetPositionAfterWrite(_position + count);
  217. }
  218. /// <inheritdoc/>
  219. public override void WriteByte(byte value)
  220. {
  221. ThrowIfDisposed();
  222. PrepareToWrite(1);
  223. _buffer.SetByte(_position, value);
  224. SetPositionAfterWrite(_position + 1);
  225. }
  226. // protected methods
  227. /// <inheritdoc/>
  228. protected override void Dispose(bool disposing)
  229. {
  230. if (!_disposed)
  231. {
  232. if (_ownsBuffer)
  233. {
  234. _buffer.Dispose();
  235. }
  236. _disposed = true;
  237. }
  238. base.Dispose(disposing);
  239. }
  240. // private methods
  241. private void EnsureWriteable()
  242. {
  243. if (!CanWrite)
  244. {
  245. throw new NotSupportedException("Stream is not writeable.");
  246. }
  247. }
  248. private int FindNullByte()
  249. {
  250. var position = _position;
  251. while (position < _length)
  252. {
  253. var segment = _buffer.AccessBackingBytes(position);
  254. var endOfSegmentIndex = segment.Offset + segment.Count;
  255. for (var index = segment.Offset; index < endOfSegmentIndex; index++)
  256. {
  257. if (segment.Array[index] == 0)
  258. {
  259. return position + (index - segment.Offset);
  260. }
  261. }
  262. position += segment.Count;
  263. }
  264. throw new EndOfStreamException();
  265. }
  266. private void PrepareToWrite(int count)
  267. {
  268. var minimumCapacity = (long)_position + (long)count;
  269. if (minimumCapacity > int.MaxValue)
  270. {
  271. throw new IOException("Stream was too long.");
  272. }
  273. _buffer.EnsureCapacity((int)minimumCapacity);
  274. _buffer.Length = _buffer.Capacity;
  275. if (_length < _position)
  276. {
  277. _buffer.Clear(_length, _position - _length);
  278. }
  279. }
  280. private byte[] ReadBytes(int count)
  281. {
  282. ThrowIfEndOfStream(count);
  283. var bytes = new byte[count];
  284. _buffer.GetBytes(_position, bytes, 0, count);
  285. _position += count;
  286. return bytes;
  287. }
  288. private void SetPositionAfterWrite(int position)
  289. {
  290. _position = position;
  291. if (_length < position)
  292. {
  293. _length = position;
  294. }
  295. }
  296. private void ThrowIfDisposed()
  297. {
  298. if (_disposed)
  299. {
  300. throw new ObjectDisposedException("ByteBufferStream");
  301. }
  302. }
  303. private void ThrowIfEndOfStream(int count)
  304. {
  305. var minimumLength = (long)_position + (long)count;
  306. if (_length < minimumLength)
  307. {
  308. if (_position < _length)
  309. {
  310. _position = _length;
  311. }
  312. throw new EndOfStreamException();
  313. }
  314. }
  315. /// <inheritdoc/>
  316. public override string ReadCString(UTF8Encoding encoding)
  317. {
  318. if (encoding == null)
  319. {
  320. throw new ArgumentNullException("encoding");
  321. }
  322. ThrowIfDisposed();
  323. var bytes = ReadCStringBytes();
  324. return Utf8Helper.DecodeUtf8String(bytes.Array, bytes.Offset, bytes.Count, encoding);
  325. }
  326. /// <inheritdoc/>
  327. public override ArraySegment<byte> ReadCStringBytes()
  328. {
  329. ThrowIfDisposed();
  330. ThrowIfEndOfStream(1);
  331. var segment = _buffer.AccessBackingBytes(_position);
  332. var index = Array.IndexOf<byte>(segment.Array, 0, segment.Offset, segment.Count);
  333. if (index != -1)
  334. {
  335. var length = index - segment.Offset;
  336. _position += length + 1; // advance over the null byte
  337. return new ArraySegment<byte>(segment.Array, segment.Offset, length); // without the null byte
  338. }
  339. else
  340. {
  341. var nullPosition = FindNullByte();
  342. var length = nullPosition - _position;
  343. var cstring = ReadBytes(length + 1); // advance over the null byte
  344. return new ArraySegment<byte>(cstring, 0, length); // without the null byte
  345. }
  346. }
  347. /// <inheritdoc/>
  348. public override Decimal128 ReadDecimal128()
  349. {
  350. ThrowIfDisposed();
  351. ThrowIfEndOfStream(16);
  352. var lowBits = (ulong)ReadInt64();
  353. var highBits = (ulong)ReadInt64();
  354. return Decimal128.FromIEEEBits(highBits, lowBits);
  355. }
  356. /// <inheritdoc/>
  357. public override double ReadDouble()
  358. {
  359. ThrowIfDisposed();
  360. ThrowIfEndOfStream(8);
  361. var segment = _buffer.AccessBackingBytes(_position);
  362. if (segment.Count >= 8)
  363. {
  364. _position += 8;
  365. return BitConverter.ToDouble(segment.Array, segment.Offset);
  366. }
  367. else
  368. {
  369. this.ReadBytes(_temp, 0, 8);
  370. return BitConverter.ToDouble(_temp, 0);
  371. }
  372. }
  373. /// <inheritdoc/>
  374. public override int ReadInt32()
  375. {
  376. ThrowIfDisposed();
  377. ThrowIfEndOfStream(4);
  378. var segment = _buffer.AccessBackingBytes(_position);
  379. if (segment.Count >= 4)
  380. {
  381. _position += 4;
  382. var bytes = segment.Array;
  383. var offset = segment.Offset;
  384. return bytes[offset] | (bytes[offset + 1] << 8) | (bytes[offset + 2] << 16) | (bytes[offset + 3] << 24);
  385. }
  386. else
  387. {
  388. this.ReadBytes(_temp, 0, 4);
  389. return _temp[0] | (_temp[1] << 8) | (_temp[2] << 16) | (_temp[3] << 24);
  390. }
  391. }
  392. /// <inheritdoc/>
  393. public override long ReadInt64()
  394. {
  395. ThrowIfDisposed();
  396. ThrowIfEndOfStream(8);
  397. var segment = _buffer.AccessBackingBytes(_position);
  398. if (segment.Count >= 8)
  399. {
  400. _position += 8;
  401. return BitConverter.ToInt64(segment.Array, segment.Offset);
  402. }
  403. else
  404. {
  405. this.ReadBytes(_temp, 0, 8);
  406. return BitConverter.ToInt64(_temp, 0);
  407. }
  408. }
  409. /// <inheritdoc/>
  410. public override ObjectId ReadObjectId()
  411. {
  412. ThrowIfDisposed();
  413. ThrowIfEndOfStream(12);
  414. var segment = _buffer.AccessBackingBytes(_position);
  415. if (segment.Count >= 12)
  416. {
  417. _position += 12;
  418. return new ObjectId(segment.Array, segment.Offset);
  419. }
  420. else
  421. {
  422. this.ReadBytes(_temp, 0, 12);
  423. return new ObjectId(_temp, 0);
  424. }
  425. }
  426. /// <inheritdoc/>
  427. public override IByteBuffer ReadSlice()
  428. {
  429. ThrowIfDisposed();
  430. var position = _position;
  431. var length = ReadInt32();
  432. ThrowIfEndOfStream(length - 4);
  433. Position = position + length;
  434. return _buffer.GetSlice(position, length);
  435. }
  436. /// <inheritdoc/>
  437. public override string ReadString(UTF8Encoding encoding)
  438. {
  439. if (encoding == null)
  440. {
  441. throw new ArgumentNullException("encoding");
  442. }
  443. ThrowIfDisposed();
  444. var length = ReadInt32();
  445. if (length <= 0)
  446. {
  447. var message = string.Format("Invalid string length: {0}.", length);
  448. throw new FormatException(message);
  449. }
  450. var segment = _buffer.AccessBackingBytes(_position);
  451. if (segment.Count >= length)
  452. {
  453. ThrowIfEndOfStream(length);
  454. if (segment.Array[segment.Offset + length - 1] != 0)
  455. {
  456. throw new FormatException("String is missing terminating null byte.");
  457. }
  458. _position += length;
  459. return Utf8Helper.DecodeUtf8String(segment.Array, segment.Offset, length - 1, encoding);
  460. }
  461. else
  462. {
  463. var bytes = length <= _tempUtf8.Length ? _tempUtf8 : new byte[length];
  464. this.ReadBytes(bytes, 0, length);
  465. if (bytes[length - 1] != 0)
  466. {
  467. throw new FormatException("String is missing terminating null byte.");
  468. }
  469. return Utf8Helper.DecodeUtf8String(bytes, 0, length - 1, encoding);
  470. }
  471. }
  472. /// <inheritdoc/>
  473. public override void SkipCString()
  474. {
  475. ThrowIfDisposed();
  476. var nullPosition = FindNullByte();
  477. _position = nullPosition + 1;
  478. }
  479. /// <inheritdoc/>
  480. public override void WriteCString(string value)
  481. {
  482. if (value == null)
  483. {
  484. throw new ArgumentNullException("value");
  485. }
  486. ThrowIfDisposed();
  487. var maxLength = CStringUtf8Encoding.GetMaxByteCount(value.Length) + 1;
  488. PrepareToWrite(maxLength);
  489. int actualLength;
  490. var segment = _buffer.AccessBackingBytes(_position);
  491. if (segment.Count >= maxLength)
  492. {
  493. actualLength = CStringUtf8Encoding.GetBytes(value, segment.Array, segment.Offset, Utf8Encodings.Strict);
  494. segment.Array[segment.Offset + actualLength] = 0;
  495. }
  496. else
  497. {
  498. byte[] bytes;
  499. if (maxLength <= _tempUtf8.Length)
  500. {
  501. bytes = _tempUtf8;
  502. actualLength = CStringUtf8Encoding.GetBytes(value, bytes, 0, Utf8Encodings.Strict);
  503. }
  504. else
  505. {
  506. bytes = Utf8Encodings.Strict.GetBytes(value);
  507. if (Array.IndexOf<byte>(bytes, 0) != -1)
  508. {
  509. throw new ArgumentException("A CString cannot contain null bytes.", "value");
  510. }
  511. actualLength = bytes.Length;
  512. }
  513. _buffer.SetBytes(_position, bytes, 0, actualLength);
  514. _buffer.SetByte(_position + actualLength, 0);
  515. }
  516. SetPositionAfterWrite(_position + actualLength + 1);
  517. }
  518. /// <inheritdoc/>
  519. public override void WriteCStringBytes(byte[] value)
  520. {
  521. if (value == null)
  522. {
  523. throw new ArgumentNullException("value");
  524. }
  525. ThrowIfDisposed();
  526. var length = value.Length;
  527. PrepareToWrite(length + 1);
  528. _buffer.SetBytes(_position, value, 0, length);
  529. _buffer.SetByte(_position + length, 0);
  530. SetPositionAfterWrite(_position + length + 1);
  531. }
  532. /// <inheritdoc/>
  533. public override void WriteDecimal128(Decimal128 value)
  534. {
  535. ThrowIfDisposed();
  536. WriteInt64((long)value.GetIEEELowBits());
  537. WriteInt64((long)value.GetIEEEHighBits());
  538. }
  539. /// <inheritdoc/>
  540. public override void WriteDouble(double value)
  541. {
  542. ThrowIfDisposed();
  543. PrepareToWrite(8);
  544. var bytes = BitConverter.GetBytes(value);
  545. _buffer.SetBytes(_position, bytes, 0, 8);
  546. SetPositionAfterWrite(_position + 8);
  547. }
  548. /// <inheritdoc/>
  549. public override void WriteInt32(int value)
  550. {
  551. ThrowIfDisposed();
  552. PrepareToWrite(4);
  553. var segment = _buffer.AccessBackingBytes(_position);
  554. if (segment.Count >= 4)
  555. {
  556. segment.Array[segment.Offset] = (byte)value;
  557. segment.Array[segment.Offset + 1] = (byte)(value >> 8);
  558. segment.Array[segment.Offset + 2] = (byte)(value >> 16);
  559. segment.Array[segment.Offset + 3] = (byte)(value >> 24);
  560. }
  561. else
  562. {
  563. _temp[0] = (byte)(value);
  564. _temp[1] = (byte)(value >> 8);
  565. _temp[2] = (byte)(value >> 16);
  566. _temp[3] = (byte)(value >> 24);
  567. _buffer.SetBytes(_position, _temp, 0, 4);
  568. }
  569. SetPositionAfterWrite(_position + 4);
  570. }
  571. /// <inheritdoc/>
  572. public override void WriteInt64(long value)
  573. {
  574. ThrowIfDisposed();
  575. PrepareToWrite(8);
  576. var bytes = BitConverter.GetBytes(value);
  577. _buffer.SetBytes(_position, bytes, 0, 8);
  578. SetPositionAfterWrite(_position + 8);
  579. }
  580. /// <inheritdoc/>
  581. public override void WriteObjectId(ObjectId value)
  582. {
  583. ThrowIfDisposed();
  584. PrepareToWrite(12);
  585. var segment = _buffer.AccessBackingBytes(_position);
  586. if (segment.Count >= 12)
  587. {
  588. value.ToByteArray(segment.Array, segment.Offset);
  589. }
  590. else
  591. {
  592. var bytes = value.ToByteArray();
  593. _buffer.SetBytes(_position, bytes, 0, 12);
  594. }
  595. SetPositionAfterWrite(_position + 12);
  596. }
  597. /// <inheritdoc/>
  598. public override void WriteString(string value, UTF8Encoding encoding)
  599. {
  600. ThrowIfDisposed();
  601. var maxLength = encoding.GetMaxByteCount(value.Length) + 5;
  602. PrepareToWrite(maxLength);
  603. int actualLength;
  604. var segment = _buffer.AccessBackingBytes(_position);
  605. if (segment.Count >= maxLength)
  606. {
  607. actualLength = encoding.GetBytes(value, 0, value.Length, segment.Array, segment.Offset + 4);
  608. var lengthPlusOne = actualLength + 1;
  609. segment.Array[segment.Offset] = (byte)lengthPlusOne;
  610. segment.Array[segment.Offset + 1] = (byte)(lengthPlusOne >> 8);
  611. segment.Array[segment.Offset + 2] = (byte)(lengthPlusOne >> 16);
  612. segment.Array[segment.Offset + 3] = (byte)(lengthPlusOne >> 24);
  613. segment.Array[segment.Offset + 4 + actualLength] = 0;
  614. }
  615. else
  616. {
  617. byte[] bytes;
  618. if (maxLength <= _tempUtf8.Length)
  619. {
  620. bytes = _tempUtf8;
  621. actualLength = encoding.GetBytes(value, 0, value.Length, bytes, 0);
  622. }
  623. else
  624. {
  625. bytes = encoding.GetBytes(value);
  626. actualLength = bytes.Length;
  627. }
  628. var lengthPlusOneBytes = BitConverter.GetBytes(actualLength + 1);
  629. _buffer.SetBytes(_position, lengthPlusOneBytes, 0, 4);
  630. _buffer.SetBytes(_position + 4, bytes, 0, actualLength);
  631. _buffer.SetByte(_position + 4 + actualLength, 0);
  632. }
  633. SetPositionAfterWrite(_position + actualLength + 5);
  634. }
  635. }
  636. }