BsonStreamAdapter.cs 15 KB

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