◆ TransformManyBlock() [5/5]

System.Threading.Tasks.Dataflow.TransformManyBlock< TInput, TOutput >.TransformManyBlock	(	Func< TInput, IEnumerable< TOutput > >	transformSync,
		Func< TInput, Task< IEnumerable< TOutput > > >	transformAsync,
		ExecutionDataflowBlockOptions	dataflowBlockOptions )
inlineprivate
Definition at line 93 of file TransformManyBlock.cs.
    {
        TransformManyBlock<TInput, TOutput> transformManyBlock = this;
        if (transformSync == null && transformAsync == null)
        {
            throw new ArgumentNullException("transform");
        }
        if (dataflowBlockOptions == null)
        {
            throw new ArgumentNullException("dataflowBlockOptions");
        }
        dataflowBlockOptions = dataflowBlockOptions.DefaultOrClone();
        Action<ISourceBlock<TOutput>, int> itemsRemovedAction = null;
        if (dataflowBlockOptions.BoundedCapacity > 0)
        {
            itemsRemovedAction = delegate(ISourceBlock<TOutput> owningSource, int count)
            {
                ((TransformManyBlock<TInput, TOutput>)owningSource)._target.ChangeBoundingCount(-count);
            };
        }
        _source = new SourceCore<TOutput>(this, dataflowBlockOptions, delegate(ISourceBlock<TOutput> owningSource)
        {
            ((TransformManyBlock<TInput, TOutput>)owningSource)._target.Complete(null, dropPendingMessages: true);
        }, itemsRemovedAction);
        if (dataflowBlockOptions.SupportsParallelExecution && dataflowBlockOptions.EnsureOrdered)
        {
            _reorderingBuffer = new ReorderingBuffer<IEnumerable<TOutput>>(this, delegate(object source, IEnumerable<TOutput> messages)
            {
                ((TransformManyBlock<TInput, TOutput>)source)._source.AddMessages(messages);
            });
        }
        if (transformSync != null)
        {
            _target = new TargetCore<TInput>(this, delegate(KeyValuePair<TInput, long> messageWithId)
            {
                transformManyBlock.ProcessMessage(transformSync, messageWithId);
            }, _reorderingBuffer, dataflowBlockOptions, TargetCoreOptions.None);
        }
        else
        {
            _target = new TargetCore<TInput>(this, delegate(KeyValuePair<TInput, long> messageWithId)
            {
                transformManyBlock.ProcessMessageWithTask(transformAsync, messageWithId);
            }, _reorderingBuffer, dataflowBlockOptions, TargetCoreOptions.UsesAsyncCompletion);
        }
        _target.Completion.ContinueWith(delegate(Task completed, object state)
        {
            SourceCore<TOutput> sourceCore = (SourceCore<TOutput>)state;
            if (completed.IsFaulted)
            {
                sourceCore.AddAndUnwrapAggregateException(completed.Exception);
            }
            sourceCore.Complete();
        }, _source, CancellationToken.None, Common.GetContinuationOptions(), TaskScheduler.Default);
        _source.Completion.ContinueWith(delegate(Task completed, object state)
        {
            IDataflowBlock dataflowBlock = (TransformManyBlock<TInput, TOutput>)state;
            dataflowBlock.Fault(completed.Exception);
        }, this, CancellationToken.None, Common.GetContinuationOptions() | TaskContinuationOptions.OnlyOnFaulted, TaskScheduler.Default);
        Common.WireCancellationToComplete(dataflowBlockOptions.CancellationToken, Completion, delegate(object state)
        {
            ((TargetCore<TInput>)state).Complete(null, dropPendingMessages: true);
        }, _target);
        DataflowEtwProvider log = DataflowEtwProvider.Log;
        if (log.IsEnabled())
        {
            log.DataflowBlockCreated(this, dataflowBlockOptions);
        }
    }