Class JsonSingleStreamLoader<TRecord>

Namespace: Wolfgang.Etl.Json

Assembly: Wolfgang.Etl.Json.dll

Loads items of type TRecord into a single JSON array stream.

public sealed class JsonSingleStreamLoader<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>

JsonSingleStreamLoader<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

using var stream = File.Create("output.json");
var loader = new JsonSingleStreamLoader<Person>(stream);
await loader.LoadAsync(items, cancellationToken);

Remarks

Writes a JSON array (e.g. [{"name":"Alice"},{"name":"Bob"}]) to a Stream by serializing each item from the input async enumerable sequence.

Constructors

JsonSingleStreamLoader(Stream)

Initializes a new instance of the JsonSingleStreamLoader<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 JsonSingleStreamLoader(Stream stream)

Parameters

stream Stream: The stream to write the JSON array to.

Exceptions

ArgumentNullException: Thrown when stream is null.

JsonSingleStreamLoader(Stream, ILogger<JsonSingleStreamLoader<TRecord>>)

Initializes a new instance of the JsonSingleStreamLoader<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 JsonSingleStreamLoader(Stream stream, ILogger<JsonSingleStreamLoader<TRecord>> logger)

Parameters

stream Stream: The stream to write the JSON array to.
logger ILogger<JsonSingleStreamLoader<TRecord>>: The logger instance for diagnostic output.

Exceptions

ArgumentNullException: Thrown when stream or logger is null.

JsonSingleStreamLoader(Stream, JsonSerializerOptions?, ILogger<JsonSingleStreamLoader<TRecord>>?)

Initializes a new instance of the JsonSingleStreamLoader<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 JsonSingleStreamLoader(Stream stream, JsonSerializerOptions? options = null, ILogger<JsonSingleStreamLoader<TRecord>>? logger = null)

Parameters

stream Stream: The stream to write the JSON array to.
options JsonSerializerOptions: The JSON serializer options to use for serialization, or null for the serializer default.
logger ILogger<JsonSingleStreamLoader<TRecord>>: An optional logger instance for diagnostic output.

Exceptions

ArgumentNullException: Thrown when stream is null.

JsonSingleStreamLoader(Stream, JsonTypeInfo<TRecord>, ILogger<JsonSingleStreamLoader<TRecord>>?)

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

public JsonSingleStreamLoader(Stream stream, JsonTypeInfo<TRecord> typeInfo, ILogger<JsonSingleStreamLoader<TRecord>>? logger = null)

Parameters

stream Stream: The stream to write the JSON array to.
typeInfo JsonTypeInfo<TRecord>: The source-generated type metadata for TRecord.
logger ILogger<JsonSingleStreamLoader<TRecord>>: An optional logger instance for diagnostic output.

Exceptions

ArgumentNullException: Thrown when stream 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 does not write any JSON to the output stream.

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