Class JsonLineExtractor<TRecord>

Namespace: Wolfgang.Etl.Json

Assembly: Wolfgang.Etl.Json.dll

Extracts items of type TRecord from a JSONL (JSON Lines / NDJSON) stream.

public sealed class JsonLineExtractor<TRecord> : ExtractorBase<TRecord, JsonReport>, IExtractWithProgressAndCancellationAsync<TRecord, JsonReport>, IExtractWithCancellationAsync<TRecord>, IExtractWithProgressAsync<TRecord, JsonReport>, IExtractAsync<TRecord>, IReportsItemErrors, IAsyncDisposable, IDisposable where TRecord : notnull

Type Parameters

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

Inheritance: object

ExtractorBase<TRecord, JsonReport>

JsonLineExtractor<TRecord>

Implements: IExtractWithProgressAndCancellationAsync<TRecord, JsonReport>

IExtractWithCancellationAsync<TRecord>

IExtractWithProgressAsync<TRecord, JsonReport>

IExtractAsync<TRecord>

IReportsItemErrors

IAsyncDisposable

IDisposable

Inherited Members: ExtractorBase<TRecord, JsonReport>.ExtractAsync()

ExtractorBase<TRecord, JsonReport>.ExtractAsync(CancellationToken)

ExtractorBase<TRecord, JsonReport>.ExtractAsync(IProgress<JsonReport>)

ExtractorBase<TRecord, JsonReport>.ExtractAsync(IProgress<JsonReport>, CancellationToken)

ExtractorBase<TRecord, JsonReport>.DisposeAsync()

ExtractorBase<TRecord, JsonReport>.Dispose()

ExtractorBase<TRecord, JsonReport>.ReportingInterval

ExtractorBase<TRecord, JsonReport>.CurrentItemCount

ExtractorBase<TRecord, JsonReport>.CurrentSkippedItemCount

ExtractorBase<TRecord, JsonReport>.CurrentErrorItemCount

ExtractorBase<TRecord, JsonReport>.MaximumItemCount

ExtractorBase<TRecord, JsonReport>.SkipItemCount

ExtractorBase<TRecord, JsonReport>.WorkerResilience

ExtractorBase<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.OpenRead("data.jsonl");
var extractor = new JsonLineExtractor<Person>(stream);
await foreach (var person in extractor.ExtractAsync(cancellationToken))
{
    Console.WriteLine(person.Name);
}

Remarks

Reads a stream, line by line, deserializing each non-empty line as a single JSON object. Blank lines are skipped with a warning logged. Compatible with both JSONL and NDJSON formats.

Constructors

JsonLineExtractor(Stream)

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

[RequiresUnreferencedCode("JSON deserialization of unknown types may require types that cannot be statically analyzed. Use the JsonTypeInfo overload for AOT compatibility.")]
[RequiresDynamicCode("JSON deserialization of unknown types may require types that cannot be statically analyzed. Use the JsonTypeInfo overload for AOT compatibility.")]
public JsonLineExtractor(Stream stream)

Parameters

stream Stream: The stream containing JSONL data to read from.

Exceptions

ArgumentNullException: Thrown when stream is null.

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

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

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

Parameters

stream Stream: The stream containing JSONL data to read from.
logger ILogger<JsonLineExtractor<TRecord>>: The logger instance for diagnostic output.

Exceptions

ArgumentNullException: Thrown when stream or logger is null.

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

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

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

Parameters

stream Stream: The stream containing JSONL data to read from.
options JsonSerializerOptions: The JSON serializer options to use for deserialization, or null for the serializer default.
logger ILogger<JsonLineExtractor<TRecord>>: An optional logger instance for diagnostic output.

Exceptions

ArgumentNullException: Thrown when stream is null.

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

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

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

Parameters

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

Exceptions

ArgumentNullException: Thrown when stream or typeInfo is null.

JsonLineExtractor(string, JsonSerializerOptions?, ILogger<JsonLineExtractor<TRecord>>?)

Initializes a new instance of the JsonLineExtractor<TRecord> class that opens and owns the JSONL file at path. The file is closed when extraction completes or the extractor is disposed.

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

Parameters

path string: The path of the JSONL file to read.
options JsonSerializerOptions: The JSON serializer options to use, or null for the default.
logger ILogger<JsonLineExtractor<TRecord>>: An optional logger instance for diagnostic output.

Exceptions

ArgumentNullException: Thrown when path is null.

Properties

CurrentByteOffset

Gets the byte offset of the start of the next unread line in the stream. Capture this value after each yield to create a resumable checkpoint.

public long CurrentByteOffset { get; }

Property Value

long

Exceptions

InvalidOperationException: Thrown when EnableCheckpointing is false. Byte offsets are only tracked when checkpointing is enabled.

EnableCheckpointing

Gets or sets a value indicating whether the extractor tracks the byte offset of each line so that CurrentByteOffset can be captured as a resumable checkpoint. Default is false.

public bool EnableCheckpointing { get; set; }

Property Value

bool

Remarks

Tracking requires computing the byte length of every line, which adds measurable per-line overhead on the extraction hot path. It is therefore opt-in: leave this false unless you intend to read CurrentByteOffset to create checkpoints. Resuming a prior run via StartByteOffset does not by itself require this flag — set it only when you also need to capture new checkpoints during the resumed run.

Encoding

Gets or sets the character encoding to use when reading the JSONL stream. When null (the default), the encoding is inferred from the stream's byte-order mark (BOM), falling back to UTF-8.

public Encoding? Encoding { get; set; }

Property Value

Encoding

StartByteOffset

Gets or sets the byte offset within the stream at which extraction begins. Set this before calling ExtractAsync(CancellationToken) to resume from a saved checkpoint. The stream must be seekable when this value is greater than zero. Default is 0 (start of stream).

public long StartByteOffset { get; set; }

Property Value

long

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 extraction 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.

Dispose(bool)

Releases resources held by this extractor. Override in a derived class to dispose resources it owns (streams, connections, etc.), then call base.Dispose(disposing). The base implementation only marks the instance disposed and is idempotent.

protected override void Dispose(bool disposing)

Parameters

disposing bool: true when called from Wolfgang.Etl.Abstractions.ExtractorBase<TSource, TProgress>.Dispose() or Wolfgang.Etl.Abstractions.ExtractorBase<TSource, TProgress>.DisposeAsync() (dispose managed resources); false when called from a finalizer.

ExtractWorkerAsync(CancellationToken)

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

protected override IAsyncEnumerable<TRecord> ExtractWorkerAsync(CancellationToken token)

Parameters

token CancellationToken: A CancellationToken to observe while waiting for the task to complete.

Returns

IAsyncEnumerable<TRecord>: IAsyncEnumerable<TSource> The result may be an empty sequence if no data is available or if the extraction fails.