Class JsonMultiStreamLoader<TRecord>

Namespace: Wolfgang.Etl.Json

Assembly: Wolfgang.Etl.Json.dll

Loads items of type TRecord into multiple streams, writing one JSON object per stream.

public sealed class JsonMultiStreamLoader<TRecord> : LoaderBase<TRecord, JsonReport>, ILoadWithProgressAndCancellationAsync<TRecord, JsonReport>, ILoadWithProgressAsync<TRecord, JsonReport>, ILoadWithCancellationAsync<TRecord>, ILoadAsync<TRecord>, IReportsItemErrors, IAsyncDisposable, IDisposable, ISupportDryRun where TRecord : notnull

Type Parameters

TRecord: The type of items to load. Must be notnull.

Inheritance: object

LoaderBase<TRecord, JsonReport>

JsonMultiStreamLoader<TRecord>

Implements: ILoadWithProgressAndCancellationAsync<TRecord, JsonReport>

ILoadWithProgressAsync<TRecord, JsonReport>

ILoadWithCancellationAsync<TRecord>

ILoadAsync<TRecord>

IReportsItemErrors

IAsyncDisposable

IDisposable

ISupportDryRun

Inherited Members: LoaderBase<TRecord, JsonReport>.LoadAsync(IAsyncEnumerable<TRecord>)

LoaderBase<TRecord, JsonReport>.LoadAsync(IAsyncEnumerable<TRecord>, CancellationToken)

LoaderBase<TRecord, JsonReport>.LoadAsync(IAsyncEnumerable<TRecord>, IProgress<JsonReport>)

LoaderBase<TRecord, JsonReport>.LoadAsync(IAsyncEnumerable<TRecord>, IProgress<JsonReport>, CancellationToken)

LoaderBase<TRecord, JsonReport>.DisposeAsync()

LoaderBase<TRecord, JsonReport>.Dispose()

LoaderBase<TRecord, JsonReport>.ReportingInterval

LoaderBase<TRecord, JsonReport>.CurrentItemCount

LoaderBase<TRecord, JsonReport>.CurrentSkippedItemCount

LoaderBase<TRecord, JsonReport>.CurrentErrorItemCount

LoaderBase<TRecord, JsonReport>.MaximumItemCount

LoaderBase<TRecord, JsonReport>.SkipItemCount

LoaderBase<TRecord, JsonReport>.WorkerResilience

LoaderBase<TRecord, JsonReport>.ErrorPolicy

object.Equals(object)

object.Equals(object, object)

object.GetHashCode()

object.GetType()

object.ReferenceEquals(object, object)

object.ToString()

Examples

var loader = new JsonMultiStreamLoader<Person>
(
    person => new JsonNamedDestination(File.Create($"output/{person.Name}.json"), $"output/{person.Name}.json")
);
await loader.LoadAsync(items, cancellationToken);

Remarks

For each item in the input sequence, calls a factory function to obtain a stream, serializes the item as a single JSON object, and disposes the stream. Supply a JsonNamedDestination factory to surface the current destination name in progress reports via CurrentSourceName.

Constructors

JsonMultiStreamLoader(Func<TRecord, Stream>)

Initializes a new instance of the JsonMultiStreamLoader<TRecord> class.

[RequiresUnreferencedCode("JSON serialization of unknown types may require types that cannot be statically analyzed. Use the JsonTypeInfo overload for AOT compatibility.")]
[RequiresDynamicCode("JSON serialization of unknown types may require types that cannot be statically analyzed. Use the JsonTypeInfo overload for AOT compatibility.")]
public JsonMultiStreamLoader(Func<TRecord, Stream> streamFactory)

Parameters

streamFactory Func<TRecord, Stream>: A factory function that receives the item to be written and returns a Stream to write it to. The loader will dispose the stream after writing.

Exceptions

ArgumentNullException: Thrown when streamFactory is null.

JsonMultiStreamLoader(Func<TRecord, Stream>, ILogger<JsonMultiStreamLoader<TRecord>>)

Initializes a new instance of the JsonMultiStreamLoader<TRecord> class with diagnostic logging.

[RequiresUnreferencedCode("JSON serialization of unknown types may require types that cannot be statically analyzed. Use the JsonTypeInfo overload for AOT compatibility.")]
[RequiresDynamicCode("JSON serialization of unknown types may require types that cannot be statically analyzed. Use the JsonTypeInfo overload for AOT compatibility.")]
public JsonMultiStreamLoader(Func<TRecord, Stream> streamFactory, ILogger<JsonMultiStreamLoader<TRecord>> logger)

Parameters

streamFactory Func<TRecord, Stream>: A factory function that receives the item to be written and returns a Stream to write it to. The loader will dispose the stream after writing.
logger ILogger<JsonMultiStreamLoader<TRecord>>: The logger instance for diagnostic output.

Exceptions

ArgumentNullException: Thrown when streamFactory or logger is null.

JsonMultiStreamLoader(Func<TRecord, Stream>, JsonSerializerOptions?, ILogger<JsonMultiStreamLoader<TRecord>>?)

Initializes a new instance of the JsonMultiStreamLoader<TRecord> class with custom serialization options.

[RequiresUnreferencedCode("JSON serialization of unknown types may require types that cannot be statically analyzed. Use the JsonTypeInfo overload for AOT compatibility.")]
[RequiresDynamicCode("JSON serialization of unknown types may require types that cannot be statically analyzed. Use the JsonTypeInfo overload for AOT compatibility.")]
public JsonMultiStreamLoader(Func<TRecord, Stream> streamFactory, JsonSerializerOptions? options = null, ILogger<JsonMultiStreamLoader<TRecord>>? logger = null)

Parameters

streamFactory Func<TRecord, Stream>: A factory function that receives the item to be written and returns a Stream to write it to. The loader will dispose the stream after writing.
options JsonSerializerOptions: The JSON serializer options to use for serialization, or null for the serializer default.
logger ILogger<JsonMultiStreamLoader<TRecord>>: An optional logger instance for diagnostic output.

Exceptions

ArgumentNullException: Thrown when streamFactory is null.

JsonMultiStreamLoader(Func<TRecord, Stream>, JsonTypeInfo<TRecord>, ILogger<JsonMultiStreamLoader<TRecord>>?)

Initializes a new instance of the JsonMultiStreamLoader<TRecord> class with source-generated type metadata for AOT-friendly, reflection-free serialization.

public JsonMultiStreamLoader(Func<TRecord, Stream> streamFactory, JsonTypeInfo<TRecord> typeInfo, ILogger<JsonMultiStreamLoader<TRecord>>? logger = null)

Parameters

streamFactory Func<TRecord, Stream>: A factory function that receives the item to be written and returns a Stream to write it to. The loader will dispose the stream after writing.
typeInfo JsonTypeInfo<TRecord>: The source-generated type metadata for TRecord.
logger ILogger<JsonMultiStreamLoader<TRecord>>: An optional logger instance for diagnostic output.

Exceptions

ArgumentNullException: Thrown when streamFactory or typeInfo is null.

JsonMultiStreamLoader(Func<TRecord, JsonNamedDestination>)

Initializes a new instance of the JsonMultiStreamLoader<TRecord> class with a named-destination factory for progress reporting.

public JsonMultiStreamLoader(Func<TRecord, JsonNamedDestination> destinationFactory)

Parameters

destinationFactory Func<TRecord, JsonNamedDestination>: A factory function that receives the item to be written and returns a JsonNamedDestination containing the stream and an optional name. The loader will dispose the stream after writing.

Exceptions

ArgumentNullException: Thrown when destinationFactory is null.

JsonMultiStreamLoader(Func<TRecord, JsonNamedDestination>, ILogger<JsonMultiStreamLoader<TRecord>>)

Initializes a new instance of the JsonMultiStreamLoader<TRecord> class with a named-destination factory and diagnostic logging.

public JsonMultiStreamLoader(Func<TRecord, JsonNamedDestination> destinationFactory, ILogger<JsonMultiStreamLoader<TRecord>> logger)

Parameters

destinationFactory Func<TRecord, JsonNamedDestination>: A factory function that receives the item to be written and returns a JsonNamedDestination containing the stream and an optional name. The loader will dispose the stream after writing.
logger ILogger<JsonMultiStreamLoader<TRecord>>: The logger instance for diagnostic output.

Exceptions

ArgumentNullException: Thrown when destinationFactory or logger is null.

JsonMultiStreamLoader(Func<TRecord, JsonNamedDestination>, JsonSerializerOptions?, ILogger<JsonMultiStreamLoader<TRecord>>?)

Initializes a new instance of the JsonMultiStreamLoader<TRecord> class with a named-destination factory and custom serialization options.

public JsonMultiStreamLoader(Func<TRecord, JsonNamedDestination> destinationFactory, JsonSerializerOptions? options = null, ILogger<JsonMultiStreamLoader<TRecord>>? logger = null)

Parameters

destinationFactory Func<TRecord, JsonNamedDestination>: A factory function that receives the item to be written and returns a JsonNamedDestination containing the stream and an optional name. The loader will dispose the stream after writing.
options JsonSerializerOptions: The JSON serializer options to use for serialization, or null for the serializer default.
logger ILogger<JsonMultiStreamLoader<TRecord>>: An optional logger instance for diagnostic output.

Exceptions

ArgumentNullException: Thrown when destinationFactory is null.

JsonMultiStreamLoader(Func<TRecord, JsonNamedDestination>, JsonTypeInfo<TRecord>, ILogger<JsonMultiStreamLoader<TRecord>>?)

Initializes a new instance of the JsonMultiStreamLoader<TRecord> class with a named-destination factory and source-generated type metadata for AOT-friendly serialization.

public JsonMultiStreamLoader(Func<TRecord, JsonNamedDestination> destinationFactory, JsonTypeInfo<TRecord> typeInfo, ILogger<JsonMultiStreamLoader<TRecord>>? logger = null)

Parameters

destinationFactory Func<TRecord, JsonNamedDestination>: A factory function that receives the item to be written and returns a JsonNamedDestination containing the stream and an optional name. The loader will dispose the stream after writing.
typeInfo JsonTypeInfo<TRecord>: The source-generated type metadata for TRecord.
logger ILogger<JsonMultiStreamLoader<TRecord>>: An optional logger instance for diagnostic output.

Exceptions

ArgumentNullException: Thrown when destinationFactory or typeInfo is null.

Properties

IsDryRun

Gets or sets a value indicating whether the stage runs in dry-run mode.

public bool IsDryRun { get; set; }

Property Value

bool: When true, the stage runs the full pipeline but skips the external side effect that mutates the destination or source. Defaults to false.

Remarks

When true, the loader enumerates the source and increments progress counters as usual but skips calling the stream factory and writing any JSON to output streams.

Methods

CreateProgressReport()

Creates a progress report of type TProgress. This gives the derived class the opportunity to implement a custom progress report that is specific to the loading process.

protected override JsonReport CreateProgressReport()

Returns

JsonReport: Progress of type TProgress

CreateProgressTimer(IProgress<JsonReport>)

Creates the Wolfgang.Etl.Abstractions.IProgressTimer used to drive progress callbacks. Override this method in a derived class to inject a custom timer (for example, a custom implementation that allows manual control in unit tests).

protected override IProgressTimer CreateProgressTimer(IProgress<JsonReport> progress)

Parameters

progress IProgress<JsonReport>: The progress sink that will receive callbacks.

Returns

IProgressTimer: A started Wolfgang.Etl.Abstractions.IProgressTimer instance.

LoadWorkerAsync(IAsyncEnumerable<TRecord>, CancellationToken)

This method is the core implementation of the loading logic and should be overridden by derived classes.

protected override Task LoadWorkerAsync(IAsyncEnumerable<TRecord> items, CancellationToken token)

Parameters

items IAsyncEnumerable<TRecord>: The items to be loaded to the destination.
token CancellationToken: A CancellationToken to observe while waiting for the task to complete.

Returns

Task: A task representing the asynchronous operation.

Remarks

Items may be an empty sequence if no data is available or if the loading fails.

Exceptions

ArgumentNullException: Argument items is null