<commit_before>using System;
using System.Collections.Generic;
using System.Collections.Concurrent;
using System.Linq;
using System.Threading.Tasks;
using System.Reactive.Concurrency;
using System.Reactive.Disposables;

namespace System.Reactive.Subjects
{
	// see http://leecampbell.blogspot.com/2010/05/intro-to-rx.html
	public sealed class Subject<T>
		: ISubject<T>, ISubject<T, T>, IObserver<T>, IObservable<T>, IDisposable
	{
		bool done;
		bool disposed;

		public void Dispose ()
		{
			disposed = true;
		}
		
		void CheckDisposed ()
		{
			if (disposed)
				throw new ObjectDisposedException ("subject");
		}
		
		ConcurrentQueue<Notification<T>> notifications = new ConcurrentQueue<Notification<T>> ();
		
		public void OnCompleted ()
		{
			CheckDisposed ();
			if (subscribed.Count > 0) {
				if (!done)
					// ToArray is to avoid InvalidOperationException when OnCompleted() unsubscribes item itself from the list.
					foreach (var s in subscribed.ToArray ())
						s.OnCompleted ();
				done = true;
			} else {
				var n = Notification.CreateOnCompleted<T> ();
				if (!notifications.Contains (n))
					notifications.Enqueue (n);
			}
		}
		
		public void OnError (Exception error)
		{
			CheckDisposed ();
			if (subscribed.Count > 0) {
				if (!done)
					// ToArray is to avoid InvalidOperationException when OnError() unsubscribes item itself from the list.
					foreach (var s in subscribed.ToArray ())
						s.OnError (error);
				done = true;
			} else {
				var n = Notification.CreateOnError<T> (error);
				if (!notifications.Contains (n))
					notifications.Enqueue (n);
			}
		}
		
		public void OnNext (T value)
		{
			CheckDisposed ();
			if (subscribed.Count > 0) {
				if (!done)
					// ToArray is to avoid InvalidOperationException when OnNext() unsubscribes item itself from the list.
					foreach (var s in subscribed.ToArray ())
						s.OnNext (value);
			} else {
				var n = Notification.CreateOnNext<T> (value);
				if (!notifications.Contains (n))
					notifications.Enqueue (n);
			}
		}
		
		List<IObserver<T>> subscribed = new List<IObserver<T>> ();
		
		// The returned IDisposable unregisters the observer when Dispose() is invoked.
		public IDisposable Subscribe (IObserver<T> observer)
		{
			CheckDisposed ();

			// If there was registered events (OnCompleted/OnError/OnNext), they are dequeued and handled here.
			while (notifications.Count > 0) {
				Notification<T> n;
				if (notifications.TryDequeue (out n))
					n.Accept (observer);
			}

			subscribed.Add (observer);
			return Disposable.Create (() => subscribed.Remove (observer));
		}
	}
}
<commit_msg>fix pending event queueing in Subject<T>.<commit_after>using System;
using System.Collections.Generic;
using System.Collections.Concurrent;
using System.Linq;
using System.Threading.Tasks;
using System.Reactive.Concurrency;
using System.Reactive.Disposables;

namespace System.Reactive.Subjects
{
	// see http://leecampbell.blogspot.com/2010/05/intro-to-rx.html
	public sealed class Subject<T>
		: ISubject<T>, ISubject<T, T>, IObserver<T>, IObservable<T>, IDisposable
	{
		bool done;
		bool disposed;

		public void Dispose ()
		{
			disposed = true;
		}
		
		void CheckDisposed ()
		{
			if (disposed)
				throw new ObjectDisposedException ("subject");
		}
		
		Queue<Notification<T>> notifications = new Queue<Notification<T>> ();
		
		public void OnCompleted ()
		{
			CheckDisposed ();
			if (subscribed.Count > 0) {
				if (!done)
					// ToArray is to avoid InvalidOperationException when OnCompleted() unsubscribes item itself from the list.
					foreach (var s in subscribed.ToArray ())
						s.OnCompleted ();
				done = true;
			} else {
				if (!notifications.Any (n => n.Kind == NotificationKind.OnCompleted))
					notifications.Enqueue (Notification.CreateOnCompleted<T> ());
			}
		}
		
		public void OnError (Exception error)
		{
			CheckDisposed ();
			if (subscribed.Count > 0) {
				if (!done)
					// ToArray is to avoid InvalidOperationException when OnError() unsubscribes item itself from the list.
					foreach (var s in subscribed.ToArray ())
						s.OnError (error);
				done = true;
			} else {
				if (!notifications.Any (n => n.Kind == NotificationKind.OnError))
					notifications.Enqueue (Notification.CreateOnError<T> (error));
			}
		}
		
		public void OnNext (T value)
		{
			CheckDisposed ();
			if (subscribed.Count > 0) {
				if (!done)
					// ToArray is to avoid InvalidOperationException when OnNext() unsubscribes item itself from the list.
					foreach (var s in subscribed.ToArray ())
						s.OnNext (value);
			} else {
				var n = Notification.CreateOnNext<T> (value);
				notifications.Enqueue (n);
			}
		}
		
		List<IObserver<T>> subscribed = new List<IObserver<T>> ();
		
		// The returned IDisposable unregisters the observer when Dispose() is invoked.
		public IDisposable Subscribe (IObserver<T> observer)
		{
			CheckDisposed ();

			// If there were registered events (OnCompleted/OnError/OnNext), they are dequeued and handled here.
			if (notifications.Count > 0)
				lock (notifications)
					while (notifications.Count > 0)
						notifications.Dequeue ().Accept (observer);
			subscribed.Add (observer);
			return Disposable.Create (() => subscribed.Remove (observer));
		}
	}
}

{'fraction_non_alphanumeric': 0.11286764705882353, 'fraction_numerical': 0.003860294117647059, 'mean_word_length': 2.872597864768683, 'pattern_counts': {'":': 0, '<': 39, '<?xml version=': 0, '>': 52, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7318660', 'n_tokens_mistral': 2064, 'n_tokens_neox': 1644, 'n_words': 442}