Class JsonLineLoader<TRecord>

Namespace: Wolfgang.Etl.Json

Assembly: Wolfgang.Etl.Json.dll

Loads items of type TRecord into a JSONL (JSON Lines / NDJSON) stream.

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

JsonLineLoader<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.jsonl");
var loader = new JsonLineLoader<Person>(stream);
await loader.LoadAsync(items, cancellationToken);

Remarks

Writes each item as a single line of JSON to the output stream, with each line separated by a newline character. Compatible with both JSONL and NDJSON consumers.

Constructors

JsonLineLoader(Stream)

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

Parameters

stream Stream: The stream to write JSONL data to.

Exceptions

ArgumentNullException: Thrown when stream is null.

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

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

Parameters

stream Stream: The stream to write JSONL data to.
logger ILogger<JsonLineLoader<TRecord>>: The logger instance for diagnostic output.

Exceptions

ArgumentNullException: Thrown when stream or logger is null.

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

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

Parameters

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

Exceptions

ArgumentNullException: Thrown when stream is null.

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

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

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

Parameters

stream Stream: The stream to write JSONL data to.
typeInfo JsonTypeInfo<TRecord>: The source-generated type metadata for TRecord.
logger ILogger<JsonLineLoader<TRecord>>: An optional logger instance for diagnostic output.

Exceptions

ArgumentNullException: Thrown when stream or typeInfo is null.

Properties

Encoding

Gets or sets the character encoding to use when writing the JSONL stream. When null (the default), UTF-8 is used.

public Encoding? Encoding { get; set; }

Property Value

Encoding

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