Class JsonMultiStreamExtractor<TRecord>

Namespace

Wolfgang.Etl.Json

Assembly

Wolfgang.Etl.Json.dll

Extracts items of type TRecord from multiple streams, reading one JSON object per stream.

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

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

var sources = Directory.GetFiles("data/", "*.json")
    .Select(path => new JsonNamedStream(File.OpenRead(path), path));
var extractor = new JsonMultiStreamExtractor<Person>(sources);
await foreach (var person in extractor.ExtractAsync(cancellationToken))
{
    Console.WriteLine(person.Name);
}

Remarks

Iterates over an enumerable of streams, deserializing a single TRecord from each. Each stream is disposed after the item is read. Extraction stops when the enumerable is exhausted or Wolfgang.Etl.Abstractions.ExtractorBase<TSource, TProgress>.MaximumItemCount is reached. Supply JsonNamedStream sources to surface the current source name in progress reports via CurrentSourceName.

Constructors

JsonMultiStreamExtractor(IEnumerable<Stream>)

Initializes a new instance of the JsonMultiStreamExtractor<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 JsonMultiStreamExtractor(IEnumerable<Stream> streams)

Parameters

streams IEnumerable<Stream>

An enumerable of streams, each containing a single JSON object.

Exceptions

ArgumentNullException

Thrown when streams is null.

JsonMultiStreamExtractor(IEnumerable<Stream>, ILogger<JsonMultiStreamExtractor<TRecord>>)

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

Parameters

streams IEnumerable<Stream>

An enumerable of streams, each containing a single JSON object.

logger ILogger<JsonMultiStreamExtractor<TRecord>>

The logger instance for diagnostic output.

Exceptions

ArgumentNullException

Thrown when streams or logger is null.

JsonMultiStreamExtractor(IEnumerable<Stream>, JsonSerializerOptions?, ILogger<JsonMultiStreamExtractor<TRecord>>?)

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

Parameters

streams IEnumerable<Stream>

An enumerable of streams, each containing a single JSON object.

options JsonSerializerOptions

The JSON serializer options to use for deserialization, or null for the serializer default.

logger ILogger<JsonMultiStreamExtractor<TRecord>>

An optional logger instance for diagnostic output.

Exceptions

ArgumentNullException

Thrown when streams is null.

JsonMultiStreamExtractor(IEnumerable<Stream>, JsonTypeInfo<TRecord>, ILogger<JsonMultiStreamExtractor<TRecord>>?)

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

public JsonMultiStreamExtractor(IEnumerable<Stream> streams, JsonTypeInfo<TRecord> typeInfo, ILogger<JsonMultiStreamExtractor<TRecord>>? logger = null)

Parameters

streams IEnumerable<Stream>

An enumerable of streams, each containing a single JSON object.

typeInfo JsonTypeInfo<TRecord>

The source-generated type metadata for TRecord.

logger ILogger<JsonMultiStreamExtractor<TRecord>>

An optional logger instance for diagnostic output.

Exceptions

ArgumentNullException

Thrown when streams or typeInfo is null.

JsonMultiStreamExtractor(IEnumerable<JsonNamedStream>)

Initializes a new instance of the JsonMultiStreamExtractor<TRecord> class with named sources for progress reporting.

public JsonMultiStreamExtractor(IEnumerable<JsonNamedStream> sources)

Parameters

sources IEnumerable<JsonNamedStream>

An enumerable of JsonNamedStream instances, each containing a stream and an optional name.

Exceptions

ArgumentNullException

Thrown when sources is null.

JsonMultiStreamExtractor(IEnumerable<JsonNamedStream>, ILogger<JsonMultiStreamExtractor<TRecord>>)

Initializes a new instance of the JsonMultiStreamExtractor<TRecord> class with named sources and diagnostic logging.

public JsonMultiStreamExtractor(IEnumerable<JsonNamedStream> sources, ILogger<JsonMultiStreamExtractor<TRecord>> logger)

Parameters

sources IEnumerable<JsonNamedStream>

An enumerable of JsonNamedStream instances, each containing a stream and an optional name.

logger ILogger<JsonMultiStreamExtractor<TRecord>>

The logger instance for diagnostic output.

Exceptions

ArgumentNullException

Thrown when sources or logger is null.

JsonMultiStreamExtractor(IEnumerable<JsonNamedStream>, JsonSerializerOptions?, ILogger<JsonMultiStreamExtractor<TRecord>>?)

Initializes a new instance of the JsonMultiStreamExtractor<TRecord> class with named sources and custom serialization options.

public JsonMultiStreamExtractor(IEnumerable<JsonNamedStream> sources, JsonSerializerOptions? options = null, ILogger<JsonMultiStreamExtractor<TRecord>>? logger = null)

Parameters

sources IEnumerable<JsonNamedStream>

An enumerable of JsonNamedStream instances, each containing a stream and an optional name.

options JsonSerializerOptions

The JSON serializer options to use for deserialization, or null for the serializer default.

logger ILogger<JsonMultiStreamExtractor<TRecord>>

An optional logger instance for diagnostic output.

Exceptions

ArgumentNullException

Thrown when sources is null.

JsonMultiStreamExtractor(IEnumerable<JsonNamedStream>, JsonTypeInfo<TRecord>, ILogger<JsonMultiStreamExtractor<TRecord>>?)

Initializes a new instance of the JsonMultiStreamExtractor<TRecord> class with named sources and source-generated type metadata for AOT-friendly deserialization.

public JsonMultiStreamExtractor(IEnumerable<JsonNamedStream> sources, JsonTypeInfo<TRecord> typeInfo, ILogger<JsonMultiStreamExtractor<TRecord>>? logger = null)

Parameters

sources IEnumerable<JsonNamedStream>

An enumerable of JsonNamedStream instances, each containing a stream and an optional name.

typeInfo JsonTypeInfo<TRecord>

The source-generated type metadata for TRecord.

logger ILogger<JsonMultiStreamExtractor<TRecord>>

An optional logger instance for diagnostic output.

Exceptions

ArgumentNullException

Thrown when sources or typeInfo 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.

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.