Class JsonSingleStreamExtractor<TRecord>

Namespace: Wolfgang.Etl.Json

Assembly: Wolfgang.Etl.Json.dll

Extracts items of type TRecord from a single JSON array stream.

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

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

Remarks

Reads a JSON array (e.g. [{"name":"Alice"},{"name":"Bob"}]) from a Stream and yields each deserialized object as an item in the async enumerable sequence. Uses DeserializeAsyncEnumerable<TValue>(Stream, JsonSerializerOptions, CancellationToken) for true streaming deserialization.

Constructors

JsonSingleStreamExtractor(Stream)

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

Parameters

stream Stream: The stream containing a JSON array to read from.

Exceptions

ArgumentNullException: Thrown when stream is null.

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

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

Parameters

stream Stream: The stream containing a JSON array to read from.
logger ILogger<JsonSingleStreamExtractor<TRecord>>: The logger instance for diagnostic output.

Exceptions

ArgumentNullException: Thrown when stream or logger is null.

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

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

Parameters

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

Exceptions

ArgumentNullException: Thrown when stream is null.

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

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

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

Parameters

stream Stream: The stream containing a JSON array to read from.
typeInfo JsonTypeInfo<TRecord>: The source-generated type metadata for TRecord.
logger ILogger<JsonSingleStreamExtractor<TRecord>>: An optional logger instance for diagnostic output.

Exceptions

ArgumentNullException: Thrown when stream or typeInfo is null.

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

Initializes a new instance of the JsonSingleStreamExtractor<TRecord> class that opens and owns the JSON array 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 JsonSingleStreamExtractor(string path, JsonSerializerOptions? options = null, ILogger<JsonSingleStreamExtractor<TRecord>>? logger = null)

Parameters

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

Exceptions

ArgumentNullException: Thrown when path is null.

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.