Method MapAsync

Namespace

SodaFlow.Async

Assembly

SodaFlow.Async.dll

MapAsync<TInput, TResult>(Stream<TInput>, StreamSink<TResult>, StreamSink<Exception>, MapAsyncOperation<TInput, TResult>, AsyncConcurrencyStrategyBase<Unit>, Stream<Unit>?, Stream<IReadOnlyCollection<TInput>>?, bool)

For a strategy that uses only the schedule and does not read TInput. Those strategies are Parallel, Queue, and SwitchLatest. This overload changes each input to Unit before the strategy, thus it needs no converter. See the canonical MapAsync<TInput, TResult, TStrategyInput>(Stream<TInput>, StreamSink<TResult>, StreamSink<Exception>, MapAsyncOperation<TInput, TResult>, AsyncConcurrencyStrategyBase<TStrategyInput>, Func<TInput, TStrategyInput>, Stream<Unit>?, Stream<IReadOnlyCollection<TInput>>?, bool) overload for the full parameter contract.

public static AsyncMapStatus<TInput, TResult> MapAsync<TInput, TResult>(this Stream<TInput> source, StreamSink<TResult> results, StreamSink<Exception> errors, MapAsyncOperation<TInput, TResult> operation, AsyncConcurrencyStrategyBase<Unit> strategy, Stream<Unit>? cancelAll = null, Stream<IReadOnlyCollection<TInput>>? cancelMatching = null, bool cancelOnDispose = true)

Parameters

source Stream<TInput>

The stream of inputs.

results StreamSink<TResult>

The destination of the value of each run that succeeded, in the sequence of their ends.

errors StreamSink<Exception>

The destination of the exception of each run with an error.

operation MapAsyncOperation<TInput, TResult>

The asynchronous work for each input.

strategy AsyncConcurrencyStrategyBase<Unit>

The control of operations that overlap.

cancelAll Stream<Unit>

This is optional. Each send cancels each tracked operation.

cancelMatching Stream<IReadOnlyCollection<TInput>>

This is optional. Each send cancels the tracked operations with a given input value.

cancelOnDispose bool

True when a disposal also cancels the tracked items. The default is true.

Returns

AsyncMapStatus<TInput, TResult>

An AsyncMapStatus<TInput, TResult> that gives the Queued items, the Running items, and Execute. A disposal of it stops the pipeline.

Type Parameters

TInput

The type in the source stream.

TResult

The type that operation gives when it succeeds.

Exceptions

ArgumentNullException

source, results, errors, operation, or strategy is null.

MapAsync<TInput, TResult, TStrategyInput>(Stream<TInput>, StreamSink<TResult>, StreamSink<Exception>, MapAsyncOperation<TInput, TResult>, AsyncConcurrencyStrategyBase<TStrategyInput>, Stream<Unit>?, Stream<IReadOnlyCollection<TInput>>?, bool)

For a strategy that reads the input. Here TInput is the TStrategyInput of the strategy or a subtype of it, thus this overload needs no converter. See the canonical MapAsync<TInput, TResult, TStrategyInput>(Stream<TInput>, StreamSink<TResult>, StreamSink<Exception>, MapAsyncOperation<TInput, TResult>, AsyncConcurrencyStrategyBase<TStrategyInput>, Func<TInput, TStrategyInput>, Stream<Unit>?, Stream<IReadOnlyCollection<TInput>>?, bool) overload for the full parameter contract.

public static AsyncMapStatus<TInput, TResult> MapAsync<TInput, TResult, TStrategyInput>(this Stream<TInput> source, StreamSink<TResult> results, StreamSink<Exception> errors, MapAsyncOperation<TInput, TResult> operation, AsyncConcurrencyStrategyBase<TStrategyInput> strategy, Stream<Unit>? cancelAll = null, Stream<IReadOnlyCollection<TInput>>? cancelMatching = null, bool cancelOnDispose = true) where TInput : TStrategyInput

Parameters

source Stream<TInput>

The stream of inputs.

results StreamSink<TResult>

The destination of the value of each run that succeeded, in the sequence of their ends.

errors StreamSink<Exception>

The destination of the exception of each run with an error.

operation MapAsyncOperation<TInput, TResult>

The asynchronous work for each input.

strategy AsyncConcurrencyStrategyBase<TStrategyInput>

The control of operations that overlap.

cancelAll Stream<Unit>

This is optional. Each send cancels each tracked operation.

cancelMatching Stream<IReadOnlyCollection<TInput>>

This is optional. Each send cancels the tracked operations with a given input value.

cancelOnDispose bool

True when a disposal also cancels the tracked items. The default is true.

Returns

AsyncMapStatus<TInput, TResult>

An AsyncMapStatus<TInput, TResult> that gives the Queued items, the Running items, and Execute. A disposal of it stops the pipeline.

Type Parameters

TInput

The type in the source stream.

TResult

The type that operation gives when it succeeds.

TStrategyInput

The input type of strategy. The compiler infers it from the type of strategy, and no caller gives it explicitly. It is usually TInput. The where TInput : TStrategyInput constraint also lets more than one MapAsync call share a strategy that uses a base class or an interface, with its own TInput in each call. The strategy then uses only TStrategyInput.

Exceptions

ArgumentNullException

source, results, errors, operation, or strategy is null.

MapAsync<TInput, TResult, TStrategyInput>(Stream<TInput>, StreamSink<TResult>, StreamSink<Exception>, MapAsyncOperation<TInput, TResult>, AsyncConcurrencyStrategyBase<TStrategyInput>, Func<TInput, TStrategyInput>, Stream<Unit>?, Stream<IReadOnlyCollection<TInput>>?, bool)

The fully general shape. Each other MapAsync overload is this overload with a fixed converter. It is the only overload with no necessary relation between TInput and TStrategyInput, because the caller gives inputConverter explicitly. Use a narrower overload where one applies, because those overloads let most calls omit that converter. Use this overload when the input type of strategy has no inheritance relation to the TInput of the call.

public static AsyncMapStatus<TInput, TResult> MapAsync<TInput, TResult, TStrategyInput>(this Stream<TInput> source, StreamSink<TResult> results, StreamSink<Exception> errors, MapAsyncOperation<TInput, TResult> operation, AsyncConcurrencyStrategyBase<TStrategyInput> strategy, Func<TInput, TStrategyInput> inputConverter, Stream<Unit>? cancelAll = null, Stream<IReadOnlyCollection<TInput>>? cancelMatching = null, bool cancelOnDispose = true)

Parameters

source Stream<TInput>

The stream of inputs. The pipeline gives each send to strategy, and the strategy starts it immediately or makes it wait. After a disposal of the status, the pipeline ignores each subsequent send.

results StreamSink<TResult>

This is necessary. The pipeline sends the return value of each operation that succeeded here, in the sequence of their ends and not in the sequence of the inputs. The pipeline does not send a result from a run that a different run replaced, or from a run that a cancellation stopped. See strategy.

errors StreamSink<Exception>

This is necessary. The pipeline sends each operation with an error here. There is no call of this method with no destination for the errors.

operation MapAsyncOperation<TInput, TResult>

The asynchronous work for each input. The pipeline calls it inline, thus it does not go to a thread pool before its own await. It receives a CancellationToken that combines the cancellation of this item with each token from the strategy. An operation that obeys that token lets cancelAll, cancelMatching, and cancelOnDispose stop work that started. An operation that ignores the token continues to its end, and the cancellation then only stops the publication of its result. There is no overload with no token.

strategy AsyncConcurrencyStrategyBase<TStrategyInput>

The control of requests that overlap. A strategy instance holds no state of its own, and more than one MapAsync call can use the same instance safely, at the same time. Each call gets its own new state manager, thus two pipelines never share a scheduling state.

inputConverter Func<TInput, TStrategyInput>

Changes each TInput value to the TStrategyInput of strategy, before the admission.

cancelAll Stream<Unit>

This is optional. Each send cancels each tracked operation, Queued or Running. The pipeline does not start a canceled Queued item at its turn. A cancellation stops a Running operation only if that operation monitors its CancellationToken.

cancelMatching Stream<IReadOnlyCollection<TInput>>

This is optional. Each send cancels the tracked operations, Queued or Running, whose input value is in the collection of that send. This uses the default equality comparer for TInput. The limits of cancelAll also apply here.

cancelOnDispose bool

True when a disposal of the AsyncMapStatus also cancels each item that the pipeline tracks at that time, Queued or Running. The default is true. At each value, a disposal always stops the admission of more values. This code sets the value here, at the setup, and not as a parameter of Dispose, because IDisposable.Dispose() is the only path to a disposal. A disposal does not stop the output. Each operation that runs continues to its end and publishes to results or to errors after the call returns. With true that output is usually not important, because the pipeline never publishes a canceled outcome and an operation that obeys its token makes none. An operation that ignores its token makes one, and with false that output is the purpose.

Returns

AsyncMapStatus<TInput, TResult>

An AsyncMapStatus<TInput, TResult>. IsRunning is a Cell< bool> that is true while one call or more has the Running status, and a Queued item does not make it true. It updates with no glitch, in the transaction of the event that changes it. Items gives each tracked value with its status. Execute puts one value in, for an operation that drives a second pipeline. A disposal of the status stops the pipeline.

Type Parameters

TInput

The type in the source stream. It is the input for each call of operation, the type in Items, and the type that cancelMatching compares.

TResult

The type that operation gives when it succeeds. The pipeline sends it to results.

TStrategyInput

The input type of strategy. The compiler infers it from the type of strategy, and no caller gives it explicitly. It needs no inheritance relation to TInput, because inputConverter makes it explicitly.

Exceptions

ArgumentNullException

source, results, errors, operation, or strategy is null.