Publish.cs 7.1 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197
  1. // Licensed to the .NET Foundation under one or more agreements.
  2. // The .NET Foundation licenses this file to you under the Apache 2.0 License.
  3. // See the LICENSE file in the project root for more information.
  4. using System.Collections;
  5. using System.Collections.Generic;
  6. namespace System.Linq
  7. {
  8. public static partial class EnumerableEx
  9. {
  10. /// <summary>
  11. /// Creates a buffer with a view over the source sequence, causing each enumerator to obtain access to the remainder of
  12. /// the sequence from the current index in the buffer.
  13. /// </summary>
  14. /// <typeparam name="TSource">Source sequence element type.</typeparam>
  15. /// <param name="source">Source sequence.</param>
  16. /// <returns>
  17. /// Buffer enabling each enumerator to retrieve elements from the shared source sequence, starting from the index
  18. /// at the point of obtaining the enumerator.
  19. /// </returns>
  20. /// <example>
  21. /// var rng = Enumerable.Range(0, 10).Publish();
  22. /// var e1 = rng.GetEnumerator(); // e1 has a view on the source starting from element 0
  23. /// Assert.IsTrue(e1.MoveNext());
  24. /// Assert.AreEqual(0, e1.Current);
  25. /// Assert.IsTrue(e1.MoveNext());
  26. /// Assert.AreEqual(1, e1.Current);
  27. /// var e2 = rng.GetEnumerator();
  28. /// Assert.IsTrue(e2.MoveNext()); // e2 has a view on the source starting from element 2
  29. /// Assert.AreEqual(2, e2.Current);
  30. /// Assert.IsTrue(e1.MoveNext()); // e1 continues to enumerate over its view
  31. /// Assert.AreEqual(2, e1.Current);
  32. /// </example>
  33. public static IBuffer<TSource> Publish<TSource>(this IEnumerable<TSource> source)
  34. {
  35. if (source == null)
  36. throw new ArgumentNullException(nameof(source));
  37. return new PublishedBuffer<TSource>(source.GetEnumerator());
  38. }
  39. /// <summary>
  40. /// Publishes the source sequence within a selector function where each enumerator can obtain a view over a tail of the
  41. /// source sequence.
  42. /// </summary>
  43. /// <typeparam name="TSource">Source sequence element type.</typeparam>
  44. /// <typeparam name="TResult">Result sequence element type.</typeparam>
  45. /// <param name="source">Source sequence.</param>
  46. /// <param name="selector">Selector function with published access to the source sequence for each enumerator.</param>
  47. /// <returns>Sequence resulting from applying the selector function to the published view over the source sequence.</returns>
  48. public static IEnumerable<TResult> Publish<TSource, TResult>(this IEnumerable<TSource> source, Func<IEnumerable<TSource>, IEnumerable<TResult>> selector)
  49. {
  50. if (source == null)
  51. throw new ArgumentNullException(nameof(source));
  52. if (selector == null)
  53. throw new ArgumentNullException(nameof(selector));
  54. return Create(() => selector(source.Publish()).GetEnumerator());
  55. }
  56. private sealed class PublishedBuffer<T> : IBuffer<T>
  57. {
  58. private RefCountList<T> _buffer;
  59. private bool _disposed;
  60. private Exception _error;
  61. private IEnumerator<T> _source;
  62. private bool _stopped;
  63. public PublishedBuffer(IEnumerator<T> source)
  64. {
  65. _buffer = new RefCountList<T>(0);
  66. _source = source;
  67. }
  68. public IEnumerator<T> GetEnumerator()
  69. {
  70. if (_disposed)
  71. throw new ObjectDisposedException("");
  72. var i = default(int);
  73. lock (_source)
  74. {
  75. i = _buffer.Count;
  76. _buffer.ReaderCount++;
  77. }
  78. return GetEnumeratorCore(i);
  79. }
  80. IEnumerator IEnumerable.GetEnumerator()
  81. {
  82. if (_disposed)
  83. throw new ObjectDisposedException("");
  84. return GetEnumerator();
  85. }
  86. public void Dispose()
  87. {
  88. lock (_source)
  89. {
  90. if (!_disposed)
  91. {
  92. _source.Dispose();
  93. _source = null;
  94. _buffer.Clear();
  95. _buffer = null;
  96. }
  97. _disposed = true;
  98. }
  99. }
  100. private IEnumerator<T> GetEnumeratorCore(int i)
  101. {
  102. try
  103. {
  104. while (true)
  105. {
  106. if (_disposed)
  107. throw new ObjectDisposedException("");
  108. var hasValue = default(bool);
  109. var current = default(T);
  110. lock (_source)
  111. {
  112. if (i >= _buffer.Count)
  113. {
  114. if (!_stopped)
  115. {
  116. try
  117. {
  118. hasValue = _source.MoveNext();
  119. if (hasValue)
  120. {
  121. current = _source.Current;
  122. }
  123. }
  124. catch (Exception ex)
  125. {
  126. _stopped = true;
  127. _error = ex;
  128. _source.Dispose();
  129. }
  130. }
  131. if (_stopped)
  132. {
  133. if (_error != null)
  134. {
  135. throw _error;
  136. }
  137. else
  138. {
  139. break;
  140. }
  141. }
  142. if (hasValue)
  143. {
  144. _buffer.Add(current);
  145. }
  146. }
  147. else
  148. {
  149. hasValue = true;
  150. }
  151. }
  152. if (hasValue)
  153. {
  154. yield return _buffer[i];
  155. }
  156. else
  157. {
  158. break;
  159. }
  160. i++;
  161. }
  162. }
  163. finally
  164. {
  165. if (_buffer != null)
  166. {
  167. _buffer.Done(i + 1);
  168. }
  169. }
  170. }
  171. }
  172. }
  173. }