Reformat and reference update.

Author: Oren (electricessence)

Extensions.ActionBlock.cs

@@ -4,24 +4,24 @@
 
 namespace Open.Threading.Dataflow
 {
-    public static class ActionBlock
-    {
-        public static ActionBlock<T> New<T>(Action<T> action)
-            => new ActionBlock<T>(action);
+	public static class ActionBlock
+	{
+		public static ActionBlock<T> New<T>(Action<T> action)
+			=> new ActionBlock<T>(action);
 
-        public static ActionBlock<T> New<T>(Action<T> action, ExecutionDataflowBlockOptions options)
-            => new ActionBlock<T>(action, options);
+		public static ActionBlock<T> New<T>(Action<T> action, ExecutionDataflowBlockOptions options)
+			=> new ActionBlock<T>(action, options);
 
-        public static ActionBlock<T> New<T>(Action<T> consumer, int maxParallel)
-           => new ActionBlock<T>(consumer, new ExecutionDataflowBlockOptions() { MaxDegreeOfParallelism = maxParallel });
+		public static ActionBlock<T> New<T>(Action<T> consumer, int maxParallel)
+		   => new ActionBlock<T>(consumer, new ExecutionDataflowBlockOptions() { MaxDegreeOfParallelism = maxParallel });
 
-        public static ActionBlock<T> NewAsync<T>(Func<T, Task> action)
-            => new ActionBlock<T>(action);
+		public static ActionBlock<T> NewAsync<T>(Func<T, Task> action)
+			=> new ActionBlock<T>(action);
 
-        public static ActionBlock<T> NewAsync<T>(Func<T, Task> action, ExecutionDataflowBlockOptions options)
-            => new ActionBlock<T>(action, options);
+		public static ActionBlock<T> NewAsync<T>(Func<T, Task> action, ExecutionDataflowBlockOptions options)
+			=> new ActionBlock<T>(action, options);
 
-        public static ActionBlock<T> NewAsync<T>(Func<T, Task> consumer, int maxParallel)
-            => new ActionBlock<T>(consumer, new ExecutionDataflowBlockOptions() { MaxDegreeOfParallelism = maxParallel });
-    }
+		public static ActionBlock<T> NewAsync<T>(Func<T, Task> consumer, int maxParallel)
+			=> new ActionBlock<T>(consumer, new ExecutionDataflowBlockOptions() { MaxDegreeOfParallelism = maxParallel });
+	}
 }

Extensions.LinkTo.cs

@@ -6,48 +6,48 @@
 
 namespace Open.Threading.Dataflow
 {
-    public static partial class DataFlowExtensions
-    {
-        public static IDisposable LinkTo<T>(this ISourceBlock<T> producer,
-            Action<T> consumer)
-            => producer.LinkTo(new ActionBlock<T>(consumer));
+	public static partial class DataFlowExtensions
+	{
+		public static IDisposable LinkTo<T>(this ISourceBlock<T> producer,
+			Action<T> consumer)
+			=> producer.LinkTo(new ActionBlock<T>(consumer));
 
-        public static IDisposable LinkToAsync<T>(this ISourceBlock<T> producer,
-            Func<T, Task> consumer)
-            => producer.LinkTo(new ActionBlock<T>(consumer));
+		public static IDisposable LinkToAsync<T>(this ISourceBlock<T> producer,
+			Func<T, Task> consumer)
+			=> producer.LinkTo(new ActionBlock<T>(consumer));
 
-        public static IDisposable LinkToWithCompletion<T>(this ISourceBlock<T> producer,
-            ITargetBlock<T> consumer)
-            => producer.LinkTo(consumer, new DataflowLinkOptions() { PropagateCompletion = true });
+		public static IDisposable LinkToWithCompletion<T>(this ISourceBlock<T> producer,
+			ITargetBlock<T> consumer)
+			=> producer.LinkTo(consumer, new DataflowLinkOptions() { PropagateCompletion = true });
 
 
-        public static T PropagateFaultsTo<T>(this T source, params IDataflowBlock[] targets)
-            where T : IDataflowBlock
-        {
-            source.Completion.OnFaulted(ex =>
-            {
-                foreach (var target in targets.Where(t => t != null))
-                    target.Fault(ex.InnerException);
-            });
-            return source;
-        }
+		public static T PropagateFaultsTo<T>(this T source, params IDataflowBlock[] targets)
+			where T : IDataflowBlock
+		{
+			source.Completion.OnFaulted(ex =>
+			{
+				foreach (var target in targets.Where(t => t != null))
+					target.Fault(ex.InnerException);
+			});
+			return source;
+		}
 
-        public static T PropagateCompletionTo<T>(this T source, params IDataflowBlock[] targets)
-            where T : IDataflowBlock
-        {
-            source.Completion.ContinueWith(task =>
-            {
-                foreach (var target in targets.Where(t => t != null))
-                {
-                    if (task.IsFaulted)
-                        // ReSharper disable once PossibleNullReferenceException
-                        target.Fault(task.Exception.InnerException);
-                    else
-                        target.Complete();
-                }
-            });
-            return source;
-        }
+		public static T PropagateCompletionTo<T>(this T source, params IDataflowBlock[] targets)
+			where T : IDataflowBlock
+		{
+			source.Completion.ContinueWith(task =>
+			{
+				foreach (var target in targets.Where(t => t != null))
+				{
+					if (task.IsFaulted)
+						// ReSharper disable once PossibleNullReferenceException
+						target.Fault(task.Exception.InnerException);
+					else
+						target.Complete();
+				}
+			});
+			return source;
+		}
 
-    }
+	}
 }

Extensions.Observable.cs

@@ -2,60 +2,60 @@
 
 namespace Open.Threading.Dataflow
 {
-    public static partial class DataFlowExtensions
-    {
-        class Observer<T> : IObserver<T>, IDisposable
-        {
-            public static IObserver<T> New(
-                Action<T> onNext,
-                Action<Exception> onError,
-                Action onCompleted)
-                => new Observer<T>()
-                {
-                    _onNext = onNext,
-                    _onError = onError,
-                    _onCompleted = onCompleted
-                };
-
-            Action _onCompleted;
-            Action<Exception> _onError;
-            Action<T> _onNext;
-
-
-            public void OnNext(T value)
-            {
-                _onNext?.Invoke(value);
-            }
-
-            public void OnError(Exception error)
-            {
-                _onError?.Invoke(error);
-            }
-
-            public void OnCompleted()
-            {
-                _onCompleted?.Invoke();
-            }
-
-
-            public void Dispose()
-            {
-                _onNext = null;
-                _onError = null;
-                _onCompleted = null;
-            }
-        }
-
-        public static IDisposable Subscribe<T>(this IObservable<T> observable,
-            Action<T> onNext,
-            Action<Exception> onError,
-            Action onCompleted = null)
-            => observable.Subscribe(Observer<T>.New(onNext, onError, onCompleted));
-
-        public static IDisposable Subscribe<T>(this IObservable<T> observable,
-            Action<T> onNext,
-            Action onCompleted = null)
-            => observable.Subscribe(Observer<T>.New(onNext, null, onCompleted));
-
-    }
+	public static partial class DataFlowExtensions
+	{
+		class Observer<T> : IObserver<T>, IDisposable
+		{
+			public static IObserver<T> New(
+				Action<T> onNext,
+				Action<Exception> onError,
+				Action onCompleted)
+				=> new Observer<T>()
+				{
+					_onNext = onNext,
+					_onError = onError,
+					_onCompleted = onCompleted
+				};
+
+			Action _onCompleted;
+			Action<Exception> _onError;
+			Action<T> _onNext;
+
+
+			public void OnNext(T value)
+			{
+				_onNext?.Invoke(value);
+			}
+
+			public void OnError(Exception error)
+			{
+				_onError?.Invoke(error);
+			}
+
+			public void OnCompleted()
+			{
+				_onCompleted?.Invoke();
+			}
+
+
+			public void Dispose()
+			{
+				_onNext = null;
+				_onError = null;
+				_onCompleted = null;
+			}
+		}
+
+		public static IDisposable Subscribe<T>(this IObservable<T> observable,
+			Action<T> onNext,
+			Action<Exception> onError,
+			Action onCompleted = null)
+			=> observable.Subscribe(Observer<T>.New(onNext, onError, onCompleted));
+
+		public static IDisposable Subscribe<T>(this IObservable<T> observable,
+			Action<T> onNext,
+			Action onCompleted = null)
+			=> observable.Subscribe(Observer<T>.New(onNext, null, onCompleted));
+
+	}
 }