Namespace SodaFlow.Async

Classes

The entry point, which is not generic, for the strategies in the library: Parallel, Queue, QueuePerGroup, and SwitchLatest. Parallel, Queue, and SwitchLatest use only the schedule and do not read the TInput of the call, thus their input type is Unit. QueuePerGroup reads the input to find its group, thus its input type is the TInput of the call. The schedule of each one is in one shared position, the internal AsyncConcurrencyStrategyFactory in SodaFlow.Core.Async. That factory is generic over the type of a value that a strategy does not use, because Core has no dependency on SodaFlow.Functional and thus no type of its own for it. The static methods below give Unit as that type argument and return the result. Thus, a consumer of this C# wrapper gets a version with the Unit type, which this wrapper also uses for cancelAll and for other values.

As a base class, this class has no relation to that shared factory. This class and AsyncConcurrencyStrategy<TState> are short versions of AsyncConcurrencyStrategy<TInput, TState> in SodaFlow.Core.Async, for a custom strategy that reads no input. Subclass AsyncConcurrencyStrategy<TState> for such a strategy with a state of its own, and this class for such a strategy with no state. The short version gives Unit for each type that the strategy does not use, thus the subclass does not write it.

AsyncConcurrencyStrategy.QueuePerGroupHelper<TInput>: This type is here only to let QueuePerGroup<TInput>() infer TInput, and to let Create<TGroup>(Func<TInput, TGroup>, IEqualityComparer<TGroup>?) infer TGroup. Without it, C# cannot infer two type parameters from two different calls.

AsyncConcurrencyStrategyBase<TInput>: The face of a strategy that is not generic over TState. It is the type that MapAsyncImpl<TInput, TResult, TStrategyInput>(Stream<TInput>, StreamSink<TResult>, StreamSink<Exception>, MapAsyncOperation<TInput, TResult>, AsyncConcurrencyStrategyBase<TStrategyInput>, Func<TInput, TStrategyInput>, Stream<UnitInternal>, Stream<IReadOnlyCollection<TInput>>, bool) and its overloads accept. This keeps the TState of a strategy out of each MapAsync signature. A caller and the execution engine see only AsyncConcurrencyStrategyBase<TInput>, and never AsyncConcurrencyStrategy<TInput, TState>. The internal constructor prevents an external subclass. To write a custom strategy, subclass AsyncConcurrencyStrategy<TInput, TState>.

AsyncConcurrencyStrategy<TState>: A short shape for a strategy that reads no input, because the input type is Unit.

AsyncConcurrencyStrategy<TInput, TState>: The base class of a MapAsync scheduling strategy, which is the admission and the sequence of a stream of async requests. A strategy answers two questions, and each answer is data. Admit(TState, AsyncQueuedItem<TInput>, IReadOnlyList<AsyncTrackedItem<TInput>>) answers "which items start now for this new tracked value?" and OnCompleted(TState, AsyncQueuedItem<TInput>, AsyncCompletion, IReadOnlyList<AsyncTrackedItem<TInput>>) answers "which items start next for this outcome, and does the pipeline publish it?". A strategy answers only from a TState that it controls. A strategy instance holds no state of its own, because each value that changes is in TState. CreateState() makes one instance of that state for each MapAsync call. Thus, more than one MapAsync call can use the same strategy instance safely, at the same time. The execution engine (see MapAsyncImpl<TInput, TResult, TStrategyInput>(Stream<TInput>, StreamSink<TResult>, StreamSink<Exception>, MapAsyncOperation<TInput, TResult>, AsyncConcurrencyStrategyBase<TStrategyInput>, Func<TInput, TStrategyInput>, Stream<UnitInternal>, Stream<IReadOnlyCollection<TInput>>, bool) ) holds the TState of each call, and it is the only code that gives that state back to the strategy. Admit(TState, AsyncQueuedItem<TInput>, IReadOnlyList<AsyncTrackedItem<TInput>>) and OnCompleted(TState, AsyncQueuedItem<TInput>, AsyncCompletion, IReadOnlyList<AsyncTrackedItem<TInput>>) cannot use the result sink, the error sink, or a Task, and cannot start a Task. The two methods give a description of the necessary operations, and the execution engine does them. The one imperative operation is Cancel(), which cancels an item that the strategy controls. It publishes nothing and starts nothing. It goes to the cancellation path of an external cancelAll stream, thus the item ends through OnCompleted(TState, AsyncQueuedItem<TInput>, AsyncCompletion, IReadOnlyList<AsyncTrackedItem<TInput>>) as each other item does. The identity of an item at its end is the item that the strategy got in Admit(TState, AsyncQueuedItem<TInput>, IReadOnlyList<AsyncTrackedItem<TInput>>). There is no second handle. Keep the item in TState to identify it again after this.

AsyncMapBase: The shared base of the two parts of a MapAsync pipeline: the strategy ( AsyncConcurrencyStrategy<TInput, TState>) and the engine that runs it ( SodaFlow.Async.AsyncMapExecutionManager<TInput, TResult, TStrategyInput>). Its only purpose is to hold the small data types that the two parts send to each other: AsyncMapBase.AsyncQueuedItem<TInput>, AsyncMapBase.AsyncToStart<TInput> , AsyncMapBase.AsyncOutcome<TResult>, and AsyncMapBase.AsyncStrategyResult<TInput>. They are nested types here, and not public top-level types. One class is not a subtype of the other. Without this shared base, one part or the two parts must make these types fully public to name them. With this base, the two parts get them through usual inheritance, and the types stay off the public surface of the library. Only code in this assembly, and code that subclasses AsyncConcurrencyStrategy<TInput, TState> to write a custom strategy, can see them. This base is not generic, because TInput and TResult belong to the nested types that use them, and not to each user of this base.

AsyncMapBase.AsyncCompletion

How an item ended, which is what a strategy reads. It carries no result value: the pipeline makes the result of a MapAsync operation after the strategy decides, and only for an item that it publishes. See Construct(Func<TResult>). A value here thus obliges the pipeline to make each result, also the results that no code reads.

Succeeded says that the operation returned. The pipeline publishes an error for such an item when the function that makes the result throws.

A cancellation is not one of these ends. It goes to OnCanceled(TState, IReadOnlyList<AsyncQueuedItem<TInput>>, IReadOnlyList<AsyncTrackedItem<TInput>>), which reads the items alone, because a canceled end carries nothing to read.

AsyncMapBase.AsyncQueuedItem<TInput>: A value that a MapAsync pipeline tracks, from its admission until its promotion, its end, or its cancellation. It is also the one object that identifies the value in AsyncMapBase.AsyncToStart<TInput> and in OnCompleted(TState, AsyncQueuedItem<TInput>, AsyncCompletion, IReadOnlyList<AsyncTrackedItem<TInput>>). It is opaque. A strategy can keep one, usually in the state of its call, to promote it after this or to identify it again at its end, and can read its Value. A strategy cannot make one, because the constructor is internal and the class is sealed. Thus, each instance comes from an Admit(TState, AsyncQueuedItem<TInput>, IReadOnlyList<AsyncTrackedItem<TInput>>) call. A strategy always receives the same instance that it got before. Thus, ReferenceEquals, or ==, is sufficient to identify an admitted value in its state. An equal Id gives the same answer, and it is also correct across the different instances that the execution engine keeps for each admitted value. For that cause the SwitchLatest strategy in the library compares the Id.

AsyncMapBase.AsyncStrategyResult<TInput>: The answer of a strategy for one end that can send: if the pipeline publishes that outcome, and which items to start next. A cancellation has no such answer, because it publishes nothing. See OnCanceled(TState, IReadOnlyList<AsyncQueuedItem<TInput>>, IReadOnlyList<AsyncTrackedItem<TInput>>), which gives the items to start and nothing else.

AsyncMapBase.AsyncToStart<TInput>: A tracked item to start now, or to promote now from Queued to Running.

AsyncMapBase.AsyncTrackedItem<TInput>: One item that a MapAsync pipeline tracks now, with its status. A strategy reads a list of these for the queue of the pipeline. See Admit(TState, AsyncQueuedItem<TInput>, IReadOnlyList<AsyncTrackedItem<TInput>>) and OnCompleted(TState, AsyncQueuedItem<TInput>, AsyncCompletion, IReadOnlyList<AsyncTrackedItem<TInput>>).

AsyncMapStatus: The status of a MapAsync pipeline, without the part that the input type decides. It gives the operation of the pipeline, and it is the only handle to stop the pipeline. See Dispose(). A caller that does not read Items can hold this type and does not have to name the input type. AsyncMapStatus<TInput> is the type that MapAsync returns.

AsyncMapStatusExtensions: The Execute overloads of AsyncMapStatus<TInput, TResult> that use Maybe<T>. SodaFlow.Core.Async cannot refer to Maybe, thus these overloads are here and not on the type.

AsyncMapStatus<TInput>: The status of a MapAsync pipeline. It gives the operation of the pipeline, and each input value that the pipeline tracks now with the status of that value. It is also the only handle to stop the pipeline. See Dispose(). This type adds Items to AsyncMapStatus, which is what makes it generic: a caller that reads only IsRunning or stops the pipeline can hold the base type. AsyncMapStatus<TInput, TResult> extends this one with Execute, thus the count of the type parameters a caller keeps selects what that caller can do.

AsyncMapStatus<TInput, TResult>

The status of a MapAsync pipeline, with the Execute methods. MapAsync answers with this type. A caller that wants only IsRunning and the disposal holds AsyncMapStatus. One that also wants Items holds AsyncMapStatus<TInput>, and one that also wants an Execute method holds this type. Thus, the count of the type parameters a caller keeps says what that caller does with the pipeline.

Execute has one purpose. The operation of a MapAsync pipeline calls a second MapAsync pipeline with it, and waits for the result of that one value. See Execute(TInput).

AsyncStreamExtensions

Extension methods that connect an impure asynchronous operation to the FRP graph. They listen on a Stream<TInput>, run an async operation for each send, put the result into a StreamSink<TResult>, and give the Queued items and the Running items. They can also connect to streams that cancel Queued work and Running work. The AsyncMapStatus<TInput, TResult> that they return is IDisposable, and a disposal of it stops the full pipeline.

The overloads are different only in the path from the TInput of the call to the input type of the strategy argument. Select the overload for your strategy. A strategy that only schedules, such as Parallel, Queue, or SwitchLatest, does not read the input, thus it needs no converter. A strategy that reads the input, such as QueuePerGroup, needs a TInput that is its input type or a subtype of it. Where that is not true, it needs a converter that changes TInput into its input type. Each other overload is the last overload with a fixed converter. The last overload takes the converter explicitly and needs no relation between the types.

MapAsyncResult<TResult>: The answer of a MapAsyncOperation<TInput, TResult>: a result, or a function that makes one. The pipeline calls such a function in the transaction that sends the result. Make one with ResultFactory<TResult>, which the operation receives.

ResultFactory<TResult>: Makes the answer of a MapAsyncOperation<TInput, TResult>. The pipeline gives one to each call of the operation. It holds no state, and a caller cannot make one.

Structs

AsyncItem<TInput>: An input value that a MapAsync pipeline tracks, and its current status.

Enums

AsyncItemStatus: The two states of a tracked item: a wait for a slot, or execution.

Delegates

MapAsyncOperation<TInput, TResult>: The work that a MapAsync pipeline does for one value from the source stream. It gives a MapAsyncResult<TResult> and not a result, because the two answers are not the same: FromValue(TResult) carries a value that the operation has, and Construct(Func<TResult>) carries a function that the pipeline calls in the transaction that sends the result. Use the second one when the result contains a cell, a stream, or another part of a SodaFlow graph, because such a part must come into existence in that transaction.

Table of Contents

Namespace SodaFlow.Async

Classes

Structs

Enums

Delegates