BsonStreamAdapter.cs 15 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549
  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. using System.Threading;
  19. using System.Threading.Tasks;
  20. namespace MongoDB.Bson.IO
  21. {
  22. /// <summary>
  23. /// A Stream that wraps another Stream while implementing the BsonStream abstract methods.
  24. /// </summary>
  25. public sealed class BsonStreamAdapter : BsonStream
  26. {
  27. // fields
  28. private bool _disposed;
  29. private bool _ownsStream;
  30. private readonly Stream _stream;
  31. private readonly byte[] _temp = new byte[12];
  32. // constructors
  33. /// <summary>
  34. /// Initializes a new instance of the <see cref="BsonStreamAdapter"/> class.
  35. /// </summary>
  36. /// <param name="stream">The stream.</param>
  37. /// <param name="ownsStream">if set to <c>true</c> [owns stream].</param>
  38. /// <exception cref="System.ArgumentNullException">stream</exception>
  39. public BsonStreamAdapter(Stream stream, bool ownsStream = false)
  40. {
  41. if (stream == null)
  42. {
  43. throw new ArgumentNullException("stream");
  44. }
  45. _stream = stream;
  46. _ownsStream = ownsStream;
  47. }
  48. // properties
  49. /// <summary>
  50. /// Gets the base stream.
  51. /// </summary>
  52. /// <value>
  53. /// The base stream.
  54. /// </value>
  55. public Stream BaseStream
  56. {
  57. get
  58. {
  59. ThrowIfDisposed();
  60. return _stream;
  61. }
  62. }
  63. /// <inheritdoc/>
  64. public override bool CanRead
  65. {
  66. get
  67. {
  68. ThrowIfDisposed();
  69. return _stream.CanRead;
  70. }
  71. }
  72. /// <inheritdoc/>
  73. public override bool CanSeek
  74. {
  75. get
  76. {
  77. ThrowIfDisposed();
  78. return _stream.CanSeek;
  79. }
  80. }
  81. /// <inheritdoc/>
  82. public override bool CanTimeout
  83. {
  84. get
  85. {
  86. ThrowIfDisposed();
  87. return _stream.CanTimeout;
  88. }
  89. }
  90. /// <inheritdoc/>
  91. public override bool CanWrite
  92. {
  93. get
  94. {
  95. ThrowIfDisposed();
  96. return _stream.CanWrite;
  97. }
  98. }
  99. /// <inheritdoc/>
  100. public override long Length
  101. {
  102. get
  103. {
  104. ThrowIfDisposed();
  105. return _stream.Length;
  106. }
  107. }
  108. /// <inheritdoc/>
  109. public override long Position
  110. {
  111. get
  112. {
  113. ThrowIfDisposed();
  114. return _stream.Position;
  115. }
  116. set
  117. {
  118. ThrowIfDisposed();
  119. _stream.Position = value;
  120. }
  121. }
  122. /// <inheritdoc/>
  123. public override int ReadTimeout
  124. {
  125. get
  126. {
  127. ThrowIfDisposed();
  128. return _stream.ReadTimeout;
  129. }
  130. set
  131. {
  132. ThrowIfDisposed();
  133. _stream.ReadTimeout = value;
  134. }
  135. }
  136. /// <inheritdoc/>
  137. public override int WriteTimeout
  138. {
  139. get
  140. {
  141. ThrowIfDisposed();
  142. return _stream.WriteTimeout;
  143. }
  144. set
  145. {
  146. ThrowIfDisposed();
  147. _stream.WriteTimeout = value;
  148. }
  149. }
  150. // methods
  151. /// <inheritdoc/>
  152. public override IAsyncResult BeginRead(byte[] buffer, int offset, int count, AsyncCallback callback, object state)
  153. {
  154. ThrowIfDisposed();
  155. return _stream.BeginRead(buffer, offset, count, callback, state);
  156. }
  157. /// <inheritdoc/>
  158. public override IAsyncResult BeginWrite(byte[] buffer, int offset, int count, AsyncCallback callback, object state)
  159. {
  160. ThrowIfDisposed();
  161. return _stream.BeginWrite(buffer, offset, count, callback, state);
  162. }
  163. /// <inheritdoc/>
  164. public override void Close()
  165. {
  166. base.Close(); // base class will call Dispose
  167. }
  168. /// <inheritdoc/>
  169. public override Task CopyToAsync(Stream destination, int bufferSize, CancellationToken cancellationToken)
  170. {
  171. ThrowIfDisposed();
  172. return _stream.CopyToAsync(destination, bufferSize, cancellationToken);
  173. }
  174. /// <inheritdoc/>
  175. protected override void Dispose(bool disposing)
  176. {
  177. if (!_disposed)
  178. {
  179. if (disposing)
  180. {
  181. if (_ownsStream)
  182. {
  183. _stream.Dispose();
  184. }
  185. }
  186. _disposed = true;
  187. }
  188. base.Dispose(disposing);
  189. }
  190. /// <inheritdoc/>
  191. public override int EndRead(IAsyncResult asyncResult)
  192. {
  193. ThrowIfDisposed();
  194. return _stream.EndRead(asyncResult);
  195. }
  196. /// <inheritdoc/>
  197. public override void EndWrite(IAsyncResult asyncResult)
  198. {
  199. ThrowIfDisposed();
  200. _stream.EndWrite(asyncResult);
  201. }
  202. /// <inheritdoc/>
  203. public override void Flush()
  204. {
  205. ThrowIfDisposed();
  206. _stream.Flush();
  207. }
  208. /// <inheritdoc/>
  209. public override Task FlushAsync(CancellationToken cancellationToken)
  210. {
  211. ThrowIfDisposed();
  212. return _stream.FlushAsync(cancellationToken);
  213. }
  214. /// <inheritdoc/>
  215. public override int Read(byte[] buffer, int offset, int count)
  216. {
  217. ThrowIfDisposed();
  218. return _stream.Read(buffer, offset, count);
  219. }
  220. /// <inheritdoc/>
  221. public override Task<int> ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
  222. {
  223. ThrowIfDisposed();
  224. return _stream.ReadAsync(buffer, offset, count, cancellationToken);
  225. }
  226. /// <inheritdoc/>
  227. public override int ReadByte()
  228. {
  229. ThrowIfDisposed();
  230. return _stream.ReadByte();
  231. }
  232. /// <inheritdoc/>
  233. public override string ReadCString(UTF8Encoding encoding)
  234. {
  235. if (encoding == null)
  236. {
  237. throw new ArgumentNullException("encoding");
  238. }
  239. ThrowIfDisposed();
  240. var bytes = ReadCStringBytes();
  241. return Utf8Helper.DecodeUtf8String(bytes.Array, 0, bytes.Count, encoding);
  242. }
  243. /// <inheritdoc/>
  244. public override ArraySegment<byte> ReadCStringBytes()
  245. {
  246. ThrowIfDisposed();
  247. var memoryStream = new MemoryStream(32);
  248. while (true)
  249. {
  250. var b = _stream.ReadByte();
  251. if (b == -1)
  252. {
  253. throw new EndOfStreamException();
  254. }
  255. if (b == 0)
  256. {
  257. byte[] memoryStreamBuffer;
  258. memoryStreamBuffer = memoryStream.GetBuffer();
  259. return new ArraySegment<byte>(memoryStreamBuffer, 0, (int)memoryStream.Length);
  260. }
  261. memoryStream.WriteByte((byte)b);
  262. }
  263. }
  264. /// <inheritdoc/>
  265. public override Decimal128 ReadDecimal128()
  266. {
  267. ThrowIfDisposed();
  268. var lowBits = (ulong)ReadInt64();
  269. var highBits = (ulong)ReadInt64();
  270. return Decimal128.FromIEEEBits(highBits, lowBits);
  271. }
  272. /// <inheritdoc/>
  273. public override double ReadDouble()
  274. {
  275. ThrowIfDisposed();
  276. this.ReadBytes(_temp, 0, 8);
  277. return BitConverter.ToDouble(_temp, 0);
  278. }
  279. /// <inheritdoc/>
  280. public override int ReadInt32()
  281. {
  282. ThrowIfDisposed();
  283. this.ReadBytes(_temp, 0, 4);
  284. return _temp[0] | (_temp[1] << 8) | (_temp[2] << 16) | (_temp[3] << 24);
  285. }
  286. /// <inheritdoc/>
  287. public override long ReadInt64()
  288. {
  289. ThrowIfDisposed();
  290. this.ReadBytes(_temp, 0, 8);
  291. return BitConverter.ToInt64(_temp, 0);
  292. }
  293. /// <inheritdoc/>
  294. public override ObjectId ReadObjectId()
  295. {
  296. ThrowIfDisposed();
  297. this.ReadBytes(_temp, 0, 12);
  298. return new ObjectId(_temp, 0);
  299. }
  300. /// <inheritdoc/>
  301. public override IByteBuffer ReadSlice()
  302. {
  303. ThrowIfDisposed();
  304. var position = _stream.Position;
  305. var length = ReadInt32();
  306. var bytes = new byte[length];
  307. _stream.Position = position;
  308. this.ReadBytes(bytes, 0, length);
  309. return new ByteArrayBuffer(bytes, isReadOnly: true);
  310. }
  311. /// <inheritdoc/>
  312. public override string ReadString(UTF8Encoding encoding)
  313. {
  314. if (encoding == null)
  315. {
  316. throw new ArgumentNullException("encoding");
  317. }
  318. ThrowIfDisposed();
  319. var length = ReadInt32();
  320. using var rentedBuffer = ThreadStaticBuffer.RentBuffer(length);
  321. var bytes = rentedBuffer.Bytes;
  322. this.ReadBytes(bytes, 0, length);
  323. if (bytes[length - 1] != 0)
  324. {
  325. throw new FormatException("String is missing terminating null byte.");
  326. }
  327. return encoding.GetString(bytes, 0, length - 1);
  328. }
  329. /// <inheritdoc/>
  330. public override long Seek(long offset, SeekOrigin origin)
  331. {
  332. ThrowIfDisposed();
  333. return _stream.Seek(offset, origin);
  334. }
  335. /// <inheritdoc/>
  336. public override void SetLength(long value)
  337. {
  338. ThrowIfDisposed();
  339. _stream.SetLength(value);
  340. }
  341. /// <inheritdoc/>
  342. public override void SkipCString()
  343. {
  344. ThrowIfDisposed();
  345. while (true)
  346. {
  347. var b = _stream.ReadByte();
  348. if (b == -1)
  349. {
  350. throw new EndOfStreamException();
  351. }
  352. if (b == 0)
  353. {
  354. return;
  355. }
  356. }
  357. }
  358. /// <inheritdoc/>
  359. private void ThrowIfDisposed()
  360. {
  361. if (_disposed)
  362. {
  363. throw new ObjectDisposedException(GetType().Name);
  364. }
  365. }
  366. /// <inheritdoc/>
  367. public override void Write(byte[] buffer, int offset, int count)
  368. {
  369. ThrowIfDisposed();
  370. _stream.Write(buffer, offset, count);
  371. }
  372. /// <inheritdoc/>
  373. public override Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
  374. {
  375. ThrowIfDisposed();
  376. return _stream.WriteAsync(buffer, offset, count, cancellationToken);
  377. }
  378. /// <inheritdoc/>
  379. public override void WriteByte(byte value)
  380. {
  381. ThrowIfDisposed();
  382. _stream.WriteByte(value);
  383. }
  384. /// <inheritdoc/>
  385. public override void WriteCString(string value)
  386. {
  387. if (value == null)
  388. {
  389. throw new ArgumentNullException("value");
  390. }
  391. ThrowIfDisposed();
  392. // Compare to 128 to preserve original behavior
  393. const int maxLengthToUseCStringUtf8EncodingWith = 128;
  394. if (CStringUtf8Encoding.GetMaxByteCount(value.Length) <= maxLengthToUseCStringUtf8EncodingWith)
  395. {
  396. using var rentedBuffer = ThreadStaticBuffer.RentBuffer(maxLengthToUseCStringUtf8EncodingWith);
  397. var length = CStringUtf8Encoding.GetBytes(value, rentedBuffer.Bytes, 0, Utf8Encodings.Strict);
  398. WriteBytes(rentedBuffer.Bytes, length);
  399. }
  400. else
  401. {
  402. using var rentedSegment = Utf8Encodings.Strict.GetBytesUsingThreadStaticBuffer(value);
  403. var segment = rentedSegment.Segment;
  404. if (Array.IndexOf<byte>(segment.Array, 0, 0, segment.Count) != -1)
  405. {
  406. throw new ArgumentException("A CString cannot contain null bytes.", "value");
  407. }
  408. WriteBytes(segment.Array, segment.Count);
  409. }
  410. void WriteBytes(byte[] bytes, int length)
  411. {
  412. _stream.Write(bytes, 0, length);
  413. _stream.WriteByte(0);
  414. }
  415. }
  416. /// <inheritdoc/>
  417. public override void WriteCStringBytes(byte[] value)
  418. {
  419. if (value == null)
  420. {
  421. throw new ArgumentNullException("value");
  422. }
  423. ThrowIfDisposed();
  424. this.WriteBytes(value, 0, value.Length);
  425. WriteByte(0);
  426. }
  427. /// <inheritdoc/>
  428. public override void WriteDecimal128(Decimal128 value)
  429. {
  430. ThrowIfDisposed();
  431. WriteInt64((long)value.GetIEEELowBits());
  432. WriteInt64((long)value.GetIEEEHighBits());
  433. }
  434. /// <inheritdoc/>
  435. public override void WriteDouble(double value)
  436. {
  437. ThrowIfDisposed();
  438. var bytes = BitConverter.GetBytes(value);
  439. _stream.Write(bytes, 0, 8);
  440. }
  441. /// <inheritdoc/>
  442. public override void WriteInt32(int value)
  443. {
  444. ThrowIfDisposed();
  445. _temp[0] = (byte)(value);
  446. _temp[1] = (byte)(value >> 8);
  447. _temp[2] = (byte)(value >> 16);
  448. _temp[3] = (byte)(value >> 24);
  449. _stream.Write(_temp, 0, 4);
  450. }
  451. /// <inheritdoc/>
  452. public override void WriteInt64(long value)
  453. {
  454. ThrowIfDisposed();
  455. var bytes = BitConverter.GetBytes(value);
  456. _stream.Write(bytes, 0, 8);
  457. }
  458. /// <inheritdoc/>
  459. public override void WriteObjectId(ObjectId value)
  460. {
  461. ThrowIfDisposed();
  462. value.ToByteArray(_temp, 0);
  463. _stream.Write(_temp, 0, 12);
  464. }
  465. /// <inheritdoc/>
  466. public override void WriteString(string value, UTF8Encoding encoding)
  467. {
  468. if (value == null)
  469. {
  470. throw new ArgumentNullException("value");
  471. }
  472. if (encoding == null)
  473. {
  474. throw new ArgumentNullException("encoding");
  475. }
  476. ThrowIfDisposed();
  477. using var rentedSegment = encoding.GetBytesUsingThreadStaticBuffer(value);
  478. var segment = rentedSegment.Segment;
  479. WriteInt32(segment.Count + 1);
  480. _stream.Write(segment.Array, 0, segment.Count);
  481. _stream.WriteByte(0);
  482. }
  483. }
  484. }