Skip to content

Commit

Permalink
Explicitly type variables according to guidelines
Browse files Browse the repository at this point in the history 
Changed usage of 'var' in TPL Dataflow to match with .NET Core guidelines.  'var' is now only used when the right-hand side of a variable declaration is a constructor, a cast, or factory method where the resulting type is obvious.
  • Loading branch information
@stephentoub
stephentoub committed Jan 11, 2015
1 parent ad16f98 commit dca1705
Show file tree
Hide file tree
Showing 18 changed files with 158 additions and 157 deletions.
54 changes: 27 additions & 27 deletions src/System.Threading.Tasks.Dataflow/src/Base/DataflowBlock.cs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -576,7 +576,7 @@ private void OfferToTargetAsync()
/// </param>
private static void CancellationHandler(object state)
{
var source = Common.UnwrapWeakReference<SendAsyncSource<TOutput>>(state);
SendAsyncSource<TOutput> source = Common.UnwrapWeakReference<SendAsyncSource<TOutput>>(state);
if (source != null)
{
Contract.Assert(source._cancellationState != CANCELLATION_STATE_NONE,
Expand Down Expand Up @@ -676,7 +676,7 @@ TOutput ISourceBlock<TOutput>.ConsumeMessage(DataflowMessageHeader messageHeader

if (validMessage)
{
var curState = _cancellationState;
int curState = _cancellationState;
Contract.Assert(
curState == CANCELLATION_STATE_NONE || curState == CANCELLATION_STATE_REGISTERED ||
curState == CANCELLATION_STATE_RESERVED || curState == CANCELLATION_STATE_COMPLETING,
Expand Down Expand Up @@ -990,7 +990,7 @@ public static TOutput Receive<TOutput>(
// Get a TCS to represent the receive operation and wait for it to complete.
// If it completes successfully, return the result. Otherwise, throw the
// original inner exception representing the cause. This could be an OCE.
var task = ReceiveCore(source, false, timeout, cancellationToken);
Task<TOutput> task = ReceiveCore(source, false, timeout, cancellationToken);
try
{
return task.GetAwaiter().GetResult(); // block until the result is available
Expand Down Expand Up @@ -1125,7 +1125,7 @@ private static Task<TOutput> ReceiveCoreByLinking<TOutput>(ISourceBlock<TOutput>
}

// Link the target to the source
var unlink = source.LinkTo(target, DataflowLinkOptions.UnlinkAfterOneAndPropagateCompletion);
IDisposable unlink = source.LinkTo(target, DataflowLinkOptions.UnlinkAfterOneAndPropagateCompletion);
target._unlink = unlink;

// If completion has started, there is a chance it started after we linked.
Expand All @@ -1134,7 +1134,7 @@ private static Task<TOutput> ReceiveCoreByLinking<TOutput>(ISourceBlock<TOutput>
// So we are racing to dispose of the unlinker.
if (Volatile.Read(ref target._cleanupReserved))
{
var disposableUnlink = Interlocked.CompareExchange(ref target._unlink, null, unlink);
IDisposable disposableUnlink = Interlocked.CompareExchange(ref target._unlink, null, unlink);
if (disposableUnlink != null) disposableUnlink.Dispose();
}
}
Expand Down Expand Up @@ -1203,7 +1203,7 @@ DataflowMessageStatus ITargetBlock<T>.OfferMessage(DataflowMessageHeader message
if (source == null && consumeToAccept) throw new ArgumentException(Strings.Argument_CantConsumeFromANullSource, "consumeToAccept");
Contract.EndContractBlock();

var status = DataflowMessageStatus.NotAvailable;
DataflowMessageStatus status = DataflowMessageStatus.NotAvailable;

// If we're already one our way to being done, don't accept anything.
// This is a fast-path check prior to taking the incoming lock;
Expand All @@ -1219,7 +1219,7 @@ DataflowMessageStatus ITargetBlock<T>.OfferMessage(DataflowMessageHeader message
{
// Accept the message if possible and complete this task with the message's value.
bool consumed = true;
var acceptedValue = consumeToAccept ? source.ConsumeMessage(messageHeader, this, out consumed) : messageValue;
T acceptedValue = consumeToAccept ? source.ConsumeMessage(messageHeader, this, out consumed) : messageValue;
if (consumed)
{
status = DataflowMessageStatus.Accepted;
Expand Down Expand Up @@ -1287,10 +1287,10 @@ private void CleanupAndComplete(ReceiveCoreByLinkingCleanupReason reason)
// completed, this is unnecessary, as the source should have already
// emptied out its target registry, or at least be in the process of doing so.
// We are racing with the linking code - only one can dispose of the unlinker.
var unlink = _unlink;
IDisposable unlink = _unlink;
if (reason != ReceiveCoreByLinkingCleanupReason.SourceCompletion && unlink != null)
{
var disposableUnlink = Interlocked.CompareExchange(ref _unlink, null, unlink);
IDisposable disposableUnlink = Interlocked.CompareExchange(ref _unlink, null, unlink);
if (disposableUnlink != null)
{
// If an error occurs, fault the target and override the reason to
Expand Down Expand Up @@ -1574,7 +1574,7 @@ private static void CancelAndUnlink(object state)
internal void AttemptThreadSafeUnlink()
{
// A race is possible. Therefore use an interlocked operation.
var cachedUnlinker = _unlinker;
IDisposable cachedUnlinker = _unlinker;
if (cachedUnlinker != null && Interlocked.CompareExchange(ref _unlinker, null, cachedUnlinker) == cachedUnlinker)
{
cachedUnlinker.Dispose();
Expand Down Expand Up @@ -1987,7 +1987,7 @@ private static Task<Int32> ChooseCore<T1, T2, T3>(
Task<int> resultTask;
try
{
var scheduler = dataflowBlockOptions.TaskScheduler;
TaskScheduler scheduler = dataflowBlockOptions.TaskScheduler;
if (TryChooseFromSource(source1, action1, 0, scheduler, out resultTask) ||
TryChooseFromSource(source2, action2, 1, scheduler, out resultTask) ||
(hasThirdSource && TryChooseFromSource(source3, action3, 2, scheduler, out resultTask)))
Expand Down Expand Up @@ -2073,10 +2073,10 @@ private static Task<Int32> ChooseCoreByLinking<T1, T2, T3>(
// Set up teardown cancellation. We will request cancellation when a) the supplied options token
// has cancellation requested or b) when we actually complete somewhere in order to tear down
// the rest of our configured set up.
var cts = CancellationTokenSource.CreateLinkedTokenSource(dataflowBlockOptions.CancellationToken, CancellationToken.None);
CancellationTokenSource cts = CancellationTokenSource.CreateLinkedTokenSource(dataflowBlockOptions.CancellationToken, CancellationToken.None);

// Set up the branches.
var scheduler = dataflowBlockOptions.TaskScheduler;
TaskScheduler scheduler = dataflowBlockOptions.TaskScheduler;
var branchTasks = new Task<int>[hasThirdSource ? 3 : 2];
branchTasks[0] = CreateChooseBranch(boxedCompleted, cts, scheduler, 0, source1, action1);
branchTasks[1] = CreateChooseBranch(boxedCompleted, cts, scheduler, 1, source2, action2);
Expand All @@ -2098,7 +2098,7 @@ private static Task<Int32> ChooseCoreByLinking<T1, T2, T3>(
// we just ignore.
List<Exception> exceptions = null;
int successfulBranchId = -1;
foreach (var task in tasks)
foreach (Task<int> task in tasks)
{
switch (task.Status)
{
Expand Down Expand Up @@ -2368,7 +2368,7 @@ internal SourceObservable(ISourceBlock<TOutput> source)
/// <returns>The aggregate exception of all errors, or null if everything completed successfully.</returns>
private AggregateException GetCompletionError()
{
var sourceCompletionTask = Common.GetPotentiallyNotSupportedCompletionTask(_source);
Task sourceCompletionTask = Common.GetPotentiallyNotSupportedCompletionTask(_source);
return sourceCompletionTask != null && sourceCompletionTask.IsFaulted ?
sourceCompletionTask.Exception : null;
}
Expand All @@ -2384,7 +2384,7 @@ IDisposable IObservable<TOutput>.Subscribe(IObserver<TOutput> observer)
Contract.EndContractBlock();
Common.ContractAssertMonitorStatus(_SubscriptionLock, held: false);

var sourceCompletionTask = Common.GetPotentiallyNotSupportedCompletionTask(_source);
Task sourceCompletionTask = Common.GetPotentiallyNotSupportedCompletionTask(_source);

// Synchronize all observers for this source.
Exception error = null;
Expand Down Expand Up @@ -2437,7 +2437,7 @@ private void Unsubscribe(IObserver<TOutput> observer)

lock (_SubscriptionLock)
{
var currentState = _observersState;
ObserversState currentState = _observersState;
Contract.Assert(currentState != null, "Observer state should never be null.");

// If the observer was already unsubscribed (or is otherwise no longer present in our list), bail.
Expand Down Expand Up @@ -2465,13 +2465,13 @@ private ImmutableList<IObserver<TOutput>> ResetObserverState()
{
Common.ContractAssertMonitorStatus(_SubscriptionLock, held: true);

var currentState = _observersState;
ObserversState currentState = _observersState;
Contract.Assert(currentState != null, "Observer state should never be null.");
Contract.Assert(currentState.Unlinker != null, "The target should be linked.");
Contract.Assert(currentState.Canceler != null, "The target should have set up continuations.");

// Replace the target with a clean one, unlink and cancel, and return the previous set of observers
var currentObservers = currentState.Observers;
ImmutableList<IObserver<TOutput>> currentObservers = currentState.Observers;
_observersState = new ObserversState(this);
currentState.Unlinker.Dispose();
currentState.Canceler.Cancel();
Expand Down Expand Up @@ -2555,7 +2555,7 @@ internal ObserversState(SourceObservable<TOutput> observable)

// When the source completes, complete the target. Then when the target completes,
// send completion messages to any observers still registered.
var sourceCompletionTask = Common.GetPotentiallyNotSupportedCompletionTask(Observable._source);
Task sourceCompletionTask = Common.GetPotentiallyNotSupportedCompletionTask(Observable._source);
if (sourceCompletionTask != null)
{
sourceCompletionTask.ContinueWith((_1, state1) =>
Expand All @@ -2582,15 +2582,15 @@ private Task ProcessItemAsync(TOutput item)
lock (Observable._SubscriptionLock) currentObservers = Observers;
try
{
foreach (var observer in currentObservers)
foreach (IObserver<TOutput> observer in currentObservers)
{
// If the observer is our own TargetObserver, we SendAsync() to it
// rather than going through IObserver.OnNext() which allows us to
// continue offering to the remaining observers without blocking.
var targetObserver = observer as TargetObserver<TOutput>;
if (targetObserver != null)
{
var sendAsyncTask = targetObserver.SendAsyncToTarget(item);
Task<bool> sendAsyncTask = targetObserver.SendAsyncToTarget(item);
if (sendAsyncTask.Status != TaskStatus.RanToCompletion)
{
// Ensure the SendAsyncTaskList is instantiated
Expand All @@ -2610,7 +2610,7 @@ private Task ProcessItemAsync(TOutput item)
if (_tempSendAsyncTaskList != null && _tempSendAsyncTaskList.Count > 0)
{
// Consolidate all SendAsync tasks into one
var allSendAsyncTasksConsolidated = Task.WhenAll(_tempSendAsyncTaskList);
Task<bool[]> allSendAsyncTasksConsolidated = Task.WhenAll(_tempSendAsyncTaskList);

// Clear the temp SendAsync task list
_tempSendAsyncTaskList.Clear();
Expand Down Expand Up @@ -2657,17 +2657,17 @@ private void NotifyObserversOfCompletion(Exception targetException = null)
if (currentObservers.Count > 0)
{
// Determine if we should fault or complete the observers
var error = targetException ?? Observable.GetCompletionError();
Exception error = targetException ?? Observable.GetCompletionError();
try
{
// Do it.
if (error != null)
{
foreach (var observer in currentObservers) observer.OnError(error);
foreach (IObserver<TOutput> observer in currentObservers) observer.OnError(error);
}
else
{
foreach (var observer in currentObservers) observer.OnCompleted();
foreach (IObserver<TOutput> observer in currentObservers) observer.OnCompleted();
}
}
catch (Exception exc)
Expand Down Expand Up @@ -2716,7 +2716,7 @@ internal TargetObserver(ITargetBlock<TInput> target)
void IObserver<TInput>.OnNext(TInput value)
{
// Send the value asynchronously...
var task = SendAsyncToTarget(value);
Task<bool> task = SendAsyncToTarget(value);

// And block until it's received.
task.GetAwaiter().GetResult(); // propagate original (non-aggregated) exception
Expand Down
2 changes: 1 addition & 1 deletion src/System.Threading.Tasks.Dataflow/src/Blocks/ActionBlock.cs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -118,7 +118,7 @@ private ActionBlock(Delegate action, ExecutionDataflowBlockOptions dataflowBlock
dataflowBlockOptions.CancellationToken, Completion, state => ((TargetCore<TInput>)state).Complete(exception: null, dropPendingMessages: true), _defaultTarget);
}
#if FEATURE_TRACING
var etwLog = DataflowEtwProvider.Log;
DataflowEtwProvider etwLog = DataflowEtwProvider.Log;
if (etwLog.IsEnabled())
{
etwLog.DataflowBlockCreated(this, dataflowBlockOptions);
Expand Down
Loading

0 comments on commit dca1705

Please sign in to comment.