ByteBufferStream.cs 23 KB

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