diff --git a/CHANGELOG.md b/CHANGELOG.md index fb2385a..7971934 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -63,6 +63,14 @@ - Non-generic definition metadata now exposes defensive read-only collections and rejects duplicate stage IDs with the shared structural topology validator. +### JSON definition integration + +- Added source-generated-metadata JSON definition builders and runtime-owned + source, transform, and sink components with lazy per-run activation. +- Added direct, trimmed, NativeAOT, and DependencyInjection composition + consumers for the canonical JSON definitions while preserving the existing + facade-source and 2.1.2 binary compatibility scenarios. + ### Build and package infrastructure - Central package management, lock-file reconciliation, package graph and diff --git a/benchmarks/SmartPipe.Benchmarks/JsonPipelineBenchmarks.cs b/benchmarks/SmartPipe.Benchmarks/JsonPipelineBenchmarks.cs new file mode 100644 index 0000000..58dfb82 --- /dev/null +++ b/benchmarks/SmartPipe.Benchmarks/JsonPipelineBenchmarks.cs @@ -0,0 +1,337 @@ +#nullable enable + +using System.Text.Json; +using System.Text.Json.Serialization; +using BenchmarkDotNet.Attributes; +using Microsoft.Extensions.Logging.Abstractions; +using SmartPipe.Core; +using SmartPipe.Extensions; +using SmartPipe.Extensions.Json; +using SmartPipe.Extensions.Selectors; +using SmartPipe.Extensions.Sinks; + +namespace SmartPipe.Benchmarks; + +[MemoryDiagnoser] +[BenchmarkCategory("JSON")] +public class JsonPipelineBenchmarks +{ + private const int BatchItemCount = 32; + private const int SinkItemCount = 32; + private const int OversizedLimit = 128; + private readonly Dictionary _rootPaths = []; + private readonly Dictionary _ndjsonPaths = []; + private readonly Dictionary _boundaryPaths = []; + private readonly Dictionary _boundaryLimits = []; + private readonly Dictionary _oversizedPaths = []; + private readonly List _definitionPaths = []; + private string _directory = null!; + private string _batchPath = null!; + private string _partialPath = null!; + private string _sinkPath = null!; + + [GlobalSetup] + public async Task Setup() + { + _directory = Path.Combine(Path.GetTempPath(), $"smartpipe-json-bench-{Guid.NewGuid():N}"); + Directory.CreateDirectory(_directory); + _batchPath = Path.Combine(_directory, "batch.jsonl"); + _partialPath = Path.Combine(_directory, "partial.jsonl"); + _sinkPath = Path.Combine(_directory, "sink.jsonl"); + + foreach (var itemCount in new[] { 1_000, 100_000 }) + { + var path = Path.Combine(_directory, $"root-{itemCount}.json"); + var items = Enumerable.Range(0, itemCount) + .Select(static value => new JsonBenchmarkItem(value, "root")) + .ToList(); + await File.WriteAllTextAsync( + path, + JsonSerializer.Serialize(items, BenchmarkJsonContext.Default.ListJsonBenchmarkItem)); + _rootPaths.Add(itemCount, path); + } + + foreach (var size in new[] { 64, 1_024, 65_536 }) + { + var item = CreateItem(size, value: size); + var record = JsonSerializer.Serialize(item, BenchmarkJsonContext.Default.JsonBenchmarkItem); + var ndjsonPath = Path.Combine(_directory, $"ndjson-{size}.jsonl"); + await File.WriteAllTextAsync(ndjsonPath, string.Join('\n', Enumerable.Repeat(record, 4)) + "\n"); + _ndjsonPaths.Add(size, ndjsonPath); + + var boundaryPath = Path.Combine(_directory, $"boundary-{size}.jsonl"); + var boundaryBytes = JsonSerializer.SerializeToUtf8Bytes(item, BenchmarkJsonContext.Default.JsonBenchmarkItem); + await File.WriteAllBytesAsync(boundaryPath, [.. boundaryBytes, (byte)'\n']); + _boundaryPaths.Add(size, boundaryPath); + _boundaryLimits.Add(size, boundaryBytes.Length + 1); + + } + + foreach (var size in new[] { 256, 4_096, 65_536 }) + { + var oversizedPath = Path.Combine(_directory, $"oversized-{size}.jsonl"); + var validRecord = JsonSerializer.Serialize( + new JsonBenchmarkItem(size, "ok"), + BenchmarkJsonContext.Default.JsonBenchmarkItem); + var oversized = new string('x', size) + "\n" + validRecord + "\n"; + await File.WriteAllTextAsync(oversizedPath, oversized); + _oversizedPaths.Add(size, oversizedPath); + } + + var batchRecords = Enumerable.Range(0, BatchItemCount) + .Select(static value => new JsonBenchmarkItem(value, "batch")) + .ToList(); + var batchJson = JsonSerializer.Serialize(batchRecords, BenchmarkJsonContext.Default.ListJsonBenchmarkItem); + await File.WriteAllTextAsync(_batchPath, batchJson + "\n" + batchJson + "\n"); + await File.WriteAllTextAsync(_partialPath, string.Join('\n', Enumerable.Repeat( + JsonSerializer.Serialize(new JsonBenchmarkItem(1, "partial"), BenchmarkJsonContext.Default.JsonBenchmarkItem), + 256)) + "\n"); + + for (var index = 0; index < 32; index++) + { + var path = Path.Combine(_directory, $"definition-{index}.json"); + await File.WriteAllTextAsync( + path, + JsonSerializer.Serialize( + new[] { new JsonBenchmarkItem(index, "definition") }, + BenchmarkJsonContext.Default.JsonBenchmarkItemArray)); + _definitionPaths.Add(path); + } + + if (await ReadFileAsync(_rootPaths[1_000], new JsonFileSourceOptions { Format = JsonFileFormat.Array }) != 1_000) + throw new InvalidOperationException("JSON root-array benchmark setup failed."); + if (await ReadFileAsync(_batchPath, new JsonFileSourceOptions { Format = JsonFileFormat.BatchJsonLines }) != BatchItemCount * 2) + throw new InvalidOperationException("JSON batch benchmark setup failed."); + if (await ReadFileAsync( + _boundaryPaths[64], + new JsonFileSourceOptions + { + Format = JsonFileFormat.Ndjson, + MaxRecordSizeBytes = _boundaryLimits[64], + }) != 1) + throw new InvalidOperationException("JSON boundary benchmark setup failed."); + if (await ReadFileAsync( + _oversizedPaths[256], + new JsonFileSourceOptions + { + Format = JsonFileFormat.Ndjson, + InvalidRecordBehavior = InvalidJsonRecordBehavior.SkipAndLog, + MaxRecordSizeBytes = OversizedLimit, + }) != 1) + throw new InvalidOperationException("JSON oversized-discard benchmark setup failed."); + + var definition = JsonPipelineDefinitionBuilder + .FromJsonFile( + new PipelineKey("json-benchmark-setup"), + _rootPaths[1_000], + BenchmarkJsonContext.Default.JsonBenchmarkItem, + BenchmarkJsonContext.Default.ListJsonBenchmarkItem, + new JsonFileSourceOptions { Format = JsonFileFormat.Array }) + .Build(); + if (await RunDefinitionAsync(definition) != 1_000) + throw new InvalidOperationException("JSON definition benchmark setup failed."); + } + + [GlobalCleanup] + public void Cleanup() => Directory.Delete(_directory, recursive: true); + + [Benchmark] + [Arguments(1_000)] + [Arguments(100_000)] + public Task RootArray_Read(int itemCount) => ReadFileAsync( + _rootPaths[itemCount], + new JsonFileSourceOptions { Format = JsonFileFormat.Array }); + + [Benchmark] + [Arguments(64)] + [Arguments(1_024)] + [Arguments(65_536)] + public Task Ndjson_Read_RecordSize(int recordSize) => ReadFileAsync( + _ndjsonPaths[recordSize], + new JsonFileSourceOptions { Format = JsonFileFormat.Ndjson }); + + [Benchmark] + public Task BatchJsonLines_Read() => ReadFileAsync( + _batchPath, + new JsonFileSourceOptions { Format = JsonFileFormat.BatchJsonLines }); + + [Benchmark] + [Arguments(64)] + [Arguments(1_024)] + [Arguments(65_536)] + public Task MaxRecord_Boundary(int recordSize) => ReadFileAsync( + _boundaryPaths[recordSize], + new JsonFileSourceOptions + { + Format = JsonFileFormat.Ndjson, + MaxRecordSizeBytes = _boundaryLimits[recordSize], + }); + + [Benchmark] + [Arguments(256)] + [Arguments(4_096)] + [Arguments(65_536)] + public Task OversizedDiscard_Scaling(int oversizedSize) => ReadFileAsync( + _oversizedPaths[oversizedSize], + new JsonFileSourceOptions + { + Format = JsonFileFormat.Ndjson, + InvalidRecordBehavior = InvalidJsonRecordBehavior.SkipAndLog, + MaxRecordSizeBytes = OversizedLimit, + }); + + [Benchmark] + public async Task ThirtyTwo_IndependentDefinitionsAndFiles() + { + var total = 0; + for (var index = 0; index < _definitionPaths.Count; index++) + { + var definition = JsonPipelineDefinitionBuilder + .FromJsonFile( + new PipelineKey($"json-benchmark-definition-{index}"), + _definitionPaths[index], + BenchmarkJsonContext.Default.JsonBenchmarkItem, + BenchmarkJsonContext.Default.ListJsonBenchmarkItem, + new JsonFileSourceOptions { Format = JsonFileFormat.Array }) + .Build(); + total += await RunDefinitionAsync(definition).ConfigureAwait(false); + } + + return total; + } + + [Benchmark] + public async Task PartialEnumeration_DisposesSource() + { + await using var source = new JsonFileSource( + _partialPath, + BenchmarkJsonContext.Default.JsonBenchmarkItem, + BenchmarkJsonContext.Default.ListJsonBenchmarkItem, + new JsonFileSourceOptions { Format = JsonFileFormat.Ndjson }); + await source.InitializeAsync().ConfigureAwait(false); + await using var enumerator = source.ReadEnvelopesAsync().GetAsyncEnumerator(); + return await enumerator.MoveNextAsync().ConfigureAwait(false) ? 1 : 0; + } + + [Benchmark] + public async Task CancellationAndDisposal_Interaction() + { + using var cancellation = new CancellationTokenSource(); + var source = new JsonFileSource( + _partialPath, + BenchmarkJsonContext.Default.JsonBenchmarkItem, + BenchmarkJsonContext.Default.ListJsonBenchmarkItem, + new JsonFileSourceOptions { Format = JsonFileFormat.Ndjson }); + try + { + await source.InitializeAsync().ConfigureAwait(false); + var readTask = ConsumeSourceAsync(source, cancellation.Token); + await Task.Yield(); + var disposalTask = source.DisposeAsync().AsTask(); + await cancellation.CancelAsync().ConfigureAwait(false); + try + { + await Task.WhenAll(readTask, disposalTask).ConfigureAwait(false); + return 0; + } + catch (OperationCanceledException) when (cancellation.IsCancellationRequested) + { + return 1; + } + } + finally + { + await source.DisposeAsync().ConfigureAwait(false); + } + } + + [Benchmark] + [Arguments(1, 64)] + [Arguments(1, 1_024)] + [Arguments(1, 65_536)] + [Arguments(1_000, 64)] + [Arguments(1_000, 1_024)] + [Arguments(1_000, 65_536)] + public async Task Sink_AllocationFlushPayload(int flushInterval, int payloadSize) + { + var payload = new string('p', payloadSize); + await using var sink = new JsonFileSink( + _sinkPath, + BenchmarkJsonContext.Default.JsonBenchmarkItem, + BenchmarkJsonContext.Default.ListJsonBenchmarkItem, + new JsonFileSinkOptions + { + Format = JsonFileFormat.Ndjson, + OpenMode = JsonFileOpenMode.Create, + FlushInterval = flushInterval, + }); + await sink.InitializeAsync().ConfigureAwait(false); + for (var index = 0; index < SinkItemCount; index++) + { + await sink.WriteAsync( + ProcessingEnvelope.Create(new JsonBenchmarkItem(index, payload))) + .ConfigureAwait(false); + } + + return SinkItemCount; + } + + private static JsonBenchmarkItem CreateItem(int targetSize, int value) + { + var padding = new string('x', Math.Max(0, targetSize - 32)); + return new JsonBenchmarkItem(value, padding); + } + + private static async Task ReadFileAsync( + string path, + JsonFileSourceOptions options, + CancellationToken cancellationToken = default) + { + var logger = options.InvalidRecordBehavior == InvalidJsonRecordBehavior.SkipAndLog + ? NullLogger>.Instance + : null; + await using var source = new JsonFileSource( + path, + BenchmarkJsonContext.Default.JsonBenchmarkItem, + BenchmarkJsonContext.Default.ListJsonBenchmarkItem, + options, + logger); + await source.InitializeAsync(cancellationToken).ConfigureAwait(false); + var count = 0; + await foreach (var _ in source.ReadEnvelopesAsync(cancellationToken).ConfigureAwait(false)) + count++; + return count; + } + + private static async Task ConsumeSourceAsync( + JsonFileSource source, + CancellationToken cancellationToken) + { + var count = 0; + await foreach (var _ in source.ReadEnvelopesAsync(cancellationToken).ConfigureAwait(false)) + count++; + return count; + } + + private static async Task RunDefinitionAsync( + PipelineDefinition definition) + { + await using var run = await definition.StartAsync().ConfigureAwait(false); + var count = 0; + await foreach (var output in run.Outputs.ReadAllAsync().ConfigureAwait(false)) + { + if (output.Result.IsSuccess) + count++; + } + + await run.Completion.ConfigureAwait(false); + return count; + } +} + +internal sealed record JsonBenchmarkItem(int Value, string Payload); + +[JsonSerializable(typeof(JsonBenchmarkItem))] +[JsonSerializable(typeof(JsonBenchmarkItem[]))] +[JsonSerializable(typeof(List))] +internal sealed partial class BenchmarkJsonContext : JsonSerializerContext; diff --git a/benchmarks/SmartPipe.Benchmarks/Program.cs b/benchmarks/SmartPipe.Benchmarks/Program.cs index 0d14f2a..c9a0467 100644 --- a/benchmarks/SmartPipe.Benchmarks/Program.cs +++ b/benchmarks/SmartPipe.Benchmarks/Program.cs @@ -1,8 +1,3 @@ -using BenchmarkDotNet.Configs; -using BenchmarkDotNet.Jobs; using BenchmarkDotNet.Running; -using BenchmarkDotNet.Toolchains.InProcess.NoEmit; -var config = DefaultConfig.Instance.AddJob(Job.Default.WithToolchain(InProcessNoEmitToolchain.Instance)); - -BenchmarkSwitcher.FromAssembly(typeof(Program).Assembly).Run(args, config); +BenchmarkSwitcher.FromAssembly(typeof(Program).Assembly).Run(args); diff --git a/benchmarks/SmartPipe.Benchmarks/SmartPipe.Benchmarks.csproj b/benchmarks/SmartPipe.Benchmarks/SmartPipe.Benchmarks.csproj index e5d661c..edae74a 100644 --- a/benchmarks/SmartPipe.Benchmarks/SmartPipe.Benchmarks.csproj +++ b/benchmarks/SmartPipe.Benchmarks/SmartPipe.Benchmarks.csproj @@ -5,6 +5,7 @@ + diff --git a/benchmarks/SmartPipe.Benchmarks/packages.lock.json b/benchmarks/SmartPipe.Benchmarks/packages.lock.json index 7da8b63..3199933 100644 --- a/benchmarks/SmartPipe.Benchmarks/packages.lock.json +++ b/benchmarks/SmartPipe.Benchmarks/packages.lock.json @@ -151,6 +151,13 @@ "SmartPipe.Core": "[2.2.0, )" } }, + "smartpipe.extensions.json": { + "type": "Project", + "dependencies": { + "Microsoft.Extensions.Logging.Abstractions": "[10.0.11, )", + "SmartPipe.Core": "[2.2.0, )" + } + }, "smartpipe.extensions.logging": { "type": "Project", "dependencies": { @@ -200,4 +207,4 @@ } } } -} \ No newline at end of file +} diff --git a/docs/aot-compatibility.md b/docs/aot-compatibility.md index bc20c23..f3e608f 100644 --- a/docs/aot-compatibility.md +++ b/docs/aot-compatibility.md @@ -55,6 +55,12 @@ These five integrations are implemented by `SmartPipe.Extensions.Json`. `JsonLinesDeadLetterSerializer` remains in `SmartPipe.Core` and also exposes a source-generated metadata constructor. +For canonical typed definitions, use `JsonPipelineDefinitionBuilder.FromJsonFile` +or `FromJsonDeadLetterFile`, then `TransformJson` and `ToJsonFile` from +`SmartPipe.Extensions.Json`. Those adapters require source-generated metadata, +snapshot it into private resolver-backed options, and keep component activation +and logger creation lazy and trimming-safe. + The legacy `JsonFileSink` batch-metadata constructors and the default `BatchJsonLines` format write one JSON array per flushed line. Explicit `Ndjson` writes one value per line, while `Array` writes one root array. The diff --git a/docs/architecture/pipeline-definition-model.md b/docs/architecture/pipeline-definition-model.md index 5d45bf2..79755a1 100644 --- a/docs/architecture/pipeline-definition-model.md +++ b/docs/architecture/pipeline-definition-model.md @@ -190,3 +190,27 @@ CPU, .NET SDK/runtime, BenchmarkDotNet job/configuration, raw artifact, and allocation shape from the same environment. No absolute timing or percentage threshold is a correctness gate; deterministic tests prove compile-once, resource-free compilation, activation order, and bounded cleanup counts. + +## JSON Line Framing + +`SmartPipe.Extensions.Json` compiles the internal +`src/Shared/JsonFraming/Utf8LineRecordReader.cs` as a linked source. This +transport-neutral helper is BCL-only and handles only bounded LF/CRLF framing, +UTF-8 bytes, BOM handling, cancellation, and discard-through-boundary for +oversized records. It is not a public API or a separate package. JSON stream +probing, unframed-input limits, record validation, path diagnostics, and +invalid-record policy remain owned by the JSON package. + +## JSON Definition Adapters + +`SmartPipe.Extensions.Json` exposes `JsonPipelineComponents` plus the typed +`JsonPipelineDefinitionBuilder` and `JsonPipelineDefinitionBuilderExtensions` +adapters. File sources, file sinks, transforms, and dead-letter components are +created through Core `RuntimeOwned` descriptors, so definition construction is +resource-free and each activation receives a fresh component. Source-generated +metadata is accepted only when its resolver can be re-run from a private, +read-only `JsonSerializerOptions` snapshot; item and batch metadata must share +one caller options instance, while transform input and output metadata are +snapshotted independently. Logger factories are borrowed and logger instances +are created at activation for policies that require logging; JSON never disposes +the factory. diff --git a/docs/migration/2.2.0-core-definition-model.md b/docs/migration/2.2.0-core-definition-model.md index 8bc7e82..32a7883 100644 --- a/docs/migration/2.2.0-core-definition-model.md +++ b/docs/migration/2.2.0-core-definition-model.md @@ -66,3 +66,13 @@ components; those paths bypass readiness, ownership, and shared disposal. The existing non-generic `PipelineDefinition` and `PipelineExecutionPlan` remain callable metadata compatibility types. Their collections are now defensive read-only copies, and duplicate stage IDs fail compilation before runtime work. + +The `SmartPipe.Extensions.Json` package also provides the typed +`JsonPipelineDefinitionBuilder` entry points and `JsonPipelineDefinitionBuilderExtensions` +for source-generated JSON metadata. `FromJsonFile` or +`FromJsonDeadLetterFile` creates a runtime-owned source; `TransformJson` appends +a runtime-owned transform, and `ToJsonFile` adds a runtime-owned sink. These +adapters snapshot serializer metadata and option records while building the +descriptor, then create fresh components on each activation. A logger factory +is borrowed and is used only during activation when a skip-or-log policy needs +it; the factory is never disposed by the pipeline. diff --git a/docs/package-ownership.md b/docs/package-ownership.md index 9d20294..ca7b533 100644 --- a/docs/package-ownership.md +++ b/docs/package-ownership.md @@ -11,6 +11,7 @@ The machine-readable authority is `eng/package-ownership.json`. | Composite, conditional, compression, and filter transforms | `SmartPipe.Extensions.Transforms` | `SmartPipe.Extensions` | type forwarding | | `LoggerSink` | `SmartPipe.Extensions.Logging` | `SmartPipe.Extensions` | type forwarding | | `ValidationTransform` and `ToFilter` | `SmartPipe.Extensions.DataAnnotations` | `SmartPipe.Extensions` | type forwarding | +| Canonical JSON pipeline definitions | `SmartPipe.Extensions.Json` | none | new 2.2 API | The HealthChecks leaf depends only on Core, DependencyInjection, DI abstractions, Diagnostics.HealthChecks, and Options. It does not depend on Hosting, ASP.NET Core, or the broad facade. diff --git a/eng/consumer-scenarios.json b/eng/consumer-scenarios.json index 35161e3..d5bb34f 100644 --- a/eng/consumer-scenarios.json +++ b/eng/consumer-scenarios.json @@ -87,7 +87,7 @@ "id": "json-nativeaot", "set": "current", "mode": "publish-native-aot", - "templatePath": "tests/Consumers/Scenarios/json-nativeaot/Consumer.csproj", + "templatePath": "tests/Consumers/Scenarios/json-direct/Consumer.csproj", "packageIds": ["SmartPipe.Core", "SmartPipe.Extensions.Json"], "expectedSmartPipeDependencies": ["SmartPipe.Core", "SmartPipe.Extensions.Json"], "forbiddenDependencies": ["SmartPipe.Extensions", "CsvHelper", "Dapper", "Microsoft.EntityFrameworkCore", "Mapster", "Polly"], @@ -95,6 +95,30 @@ "timeout": "00:15:00", "runSecondLockedRestore": true }, + { + "id": "json-trim", + "set": "current", + "mode": "publish-trimmed", + "templatePath": "tests/Consumers/Scenarios/json-direct/Consumer.csproj", + "packageIds": ["SmartPipe.Core", "SmartPipe.Extensions.Json"], + "expectedSmartPipeDependencies": ["SmartPipe.Core", "SmartPipe.Extensions.Json"], + "forbiddenDependencies": ["SmartPipe.Extensions", "CsvHelper", "Dapper", "Microsoft.EntityFrameworkCore", "Mapster", "Polly"], + "baselineVersion": null, + "timeout": "00:10:00", + "runSecondLockedRestore": true + }, + { + "id": "json-dependency-injection-direct", + "set": "current", + "mode": "build-and-run", + "templatePath": "tests/Consumers/Scenarios/json-dependency-injection-direct/Consumer.csproj", + "packageIds": ["SmartPipe.Core", "SmartPipe.Extensions.DependencyInjection", "SmartPipe.Extensions.Json"], + "expectedSmartPipeDependencies": ["SmartPipe.Core", "SmartPipe.Extensions.DependencyInjection", "SmartPipe.Extensions.Json"], + "forbiddenDependencies": ["SmartPipe.Extensions", "SmartPipe.Extensions.Hosting", "SmartPipe.Extensions.HealthChecks", "Microsoft.Extensions.Hosting.Abstractions", "Microsoft.Extensions.Diagnostics.HealthChecks", "Microsoft.Extensions.Options"], + "baselineVersion": null, + "timeout": "00:05:00", + "runSecondLockedRestore": true + }, { "id": "dependency-injection-direct", "set": "current", diff --git a/eng/consumer-scenarios.schema.json b/eng/consumer-scenarios.schema.json index 88b75bb..4443ca8 100644 --- a/eng/consumer-scenarios.schema.json +++ b/eng/consumer-scenarios.schema.json @@ -13,8 +13,8 @@ }, "scenarios": { "type": "array", - "minItems": 33, - "maxItems": 33, + "minItems": 35, + "maxItems": 35, "items": { "$ref": "#/$defs/scenario" } } }, diff --git a/eng/package-graph.json b/eng/package-graph.json index 56f577b..b50d6fb 100644 --- a/eng/package-graph.json +++ b/eng/package-graph.json @@ -39,6 +39,8 @@ "core-trim", "core-nativeaot", "json-nativeaot", + "json-trim", + "json-dependency-injection-direct", "dependency-injection-direct", "dependency-injection-keyed", "dependency-injection-from-keyed-services", @@ -206,6 +208,8 @@ "json-direct", "legacy-binary-2.1.2", "json-nativeaot", + "json-trim", + "json-dependency-injection-direct", "dependency-injection-facade-binary-2.1.2", "hosting-facade-binary-2.1.2" ] @@ -548,6 +552,7 @@ "temporaryAllowances": [], "consumerScenarios": [ "dependency-injection-direct", + "json-dependency-injection-direct", "dependency-injection-keyed", "dependency-injection-from-keyed-services", "dependency-injection-facade-source", diff --git a/eng/package-ownership.json b/eng/package-ownership.json index e00a18b..d1bd810 100644 --- a/eng/package-ownership.json +++ b/eng/package-ownership.json @@ -89,6 +89,17 @@ "namespacePreserved": true, "evidence": "2.1.2 Json assembly implementation" }, + { + "typePattern": "SmartPipe.Extensions.Json.JsonPipeline*", + "baselineAssembly": "SmartPipe.Extensions.Json", + "currentImplementationAssembly": "SmartPipe.Extensions.Json", + "targetImplementationAssembly": "SmartPipe.Extensions.Json", + "compatibilityAssembly": null, + "strategy": "stay", + "migrationEpic": "SP220-08", + "namespacePreserved": true, + "evidence": "Canonical source-generated JSON definition adapters introduced by SP220-08" + }, { "typePattern": "SmartPipe.Extensions.Selectors.CsvFileSource*", "baselineAssembly": "SmartPipe.Extensions", diff --git a/eng/tests/workflow_contract_tests.py b/eng/tests/workflow_contract_tests.py index 7ba0634..528d2f6 100644 --- a/eng/tests/workflow_contract_tests.py +++ b/eng/tests/workflow_contract_tests.py @@ -619,7 +619,8 @@ def assert_consumer_contract() -> None: current = [scenario for scenario in document["scenarios"] if scenario["set"] == "current"] expected = { "core-direct", "json-direct", "extensions-meta", "legacy-binary-2.1.2", - "core-trim", "core-nativeaot", "json-nativeaot", + "core-trim", "core-nativeaot", "json-nativeaot", "json-trim", + "json-dependency-injection-direct", "dependency-injection-direct", "dependency-injection-keyed", "dependency-injection-from-keyed-services", "dependency-injection-facade-source", "dependency-injection-facade-binary-2.1.2", "dependency-injection-trim", @@ -633,8 +634,8 @@ def assert_consumer_contract() -> None: "channels-direct", "transforms-direct", "logging-direct", "data-annotations-direct", "data-annotations-runtime", } - require(len(current) == 33 and {scenario["id"] for scenario in current} == expected, - "Current consumer set must contain the exact thirty-three scenarios.") + require(len(current) == 35 and {scenario["id"] for scenario in current} == expected, + "Current consumer set must contain the exact thirty-five scenarios.") hosting = [scenario for scenario in current if scenario.get("category") == "hosting"] require({scenario["id"] for scenario in hosting} == { "hosting-direct", "hosting-facade-source", "hosting-facade-binary-2.1.2", diff --git a/src/SmartPipe.Extensions.Json/Utf8LineRecordReader.cs b/src/Shared/JsonFraming/Utf8LineRecordReader.cs similarity index 99% rename from src/SmartPipe.Extensions.Json/Utf8LineRecordReader.cs rename to src/Shared/JsonFraming/Utf8LineRecordReader.cs index 4b925fa..c8d2a4f 100644 --- a/src/SmartPipe.Extensions.Json/Utf8LineRecordReader.cs +++ b/src/Shared/JsonFraming/Utf8LineRecordReader.cs @@ -1,4 +1,4 @@ -namespace SmartPipe.Extensions; +namespace SmartPipe.Shared.JsonFraming; internal readonly record struct Utf8LineRecord(byte[] Bytes, bool TooLarge); diff --git a/src/SmartPipe.Extensions.Json/DeadLetterRecordReader.cs b/src/SmartPipe.Extensions.Json/DeadLetterRecordReader.cs index b553837..b488859 100644 --- a/src/SmartPipe.Extensions.Json/DeadLetterRecordReader.cs +++ b/src/SmartPipe.Extensions.Json/DeadLetterRecordReader.cs @@ -2,6 +2,7 @@ using System.Text.Json; using Microsoft.Extensions.Logging; using SmartPipe.Core; +using SmartPipe.Shared.JsonFraming; namespace SmartPipe.Extensions; diff --git a/src/SmartPipe.Extensions.Json/JsonInputOptionsValidator.cs b/src/SmartPipe.Extensions.Json/JsonInputOptionsValidator.cs index a679ba5..9ef0470 100644 --- a/src/SmartPipe.Extensions.Json/JsonInputOptionsValidator.cs +++ b/src/SmartPipe.Extensions.Json/JsonInputOptionsValidator.cs @@ -5,10 +5,13 @@ namespace SmartPipe.Extensions; internal static class JsonInputOptionsValidator { public static JsonFileSourceOptions Validate(JsonFileSourceOptions? options, ILogger? logger) + => Validate(options, logger is not null); + + internal static JsonFileSourceOptions Validate(JsonFileSourceOptions? options, bool loggerAvailable) { ArgumentNullException.ThrowIfNull(options); ValidateCommon(options.Format, options.InvalidRecordBehavior, options.MaxDepth, - options.MaxRecordSizeBytes, options.MaxUnframedInputSizeBytes, logger, nameof(options)); + options.MaxRecordSizeBytes, options.MaxUnframedInputSizeBytes, loggerAvailable, nameof(options)); if (options.InvalidRecordBehavior == InvalidJsonRecordBehavior.SkipAndLog && options.Format is not (JsonFileFormat.Ndjson or JsonFileFormat.BatchJsonLines)) throw new ArgumentException( @@ -18,10 +21,13 @@ public static JsonFileSourceOptions Validate(JsonFileSourceOptions? options, ILo } public static DeadLetterSourceOptions Validate(DeadLetterSourceOptions? options, ILogger? logger) + => Validate(options, logger is not null); + + internal static DeadLetterSourceOptions Validate(DeadLetterSourceOptions? options, bool loggerAvailable) { ArgumentNullException.ThrowIfNull(options); ValidateCommon(options.Format, options.InvalidRecordBehavior, options.MaxDepth, - options.MaxRecordSizeBytes, options.MaxUnframedInputSizeBytes, logger, nameof(options)); + options.MaxRecordSizeBytes, options.MaxUnframedInputSizeBytes, loggerAvailable, nameof(options)); if (options.Format == JsonFileFormat.BatchJsonLines) throw new ArgumentException("BatchJsonLines is not supported by DeadLetterSource.", nameof(options)); if (options.Format == JsonFileFormat.Array @@ -32,13 +38,43 @@ public static DeadLetterSourceOptions Validate(DeadLetterSourceOptions? options, return options with { }; } + internal static JsonFileSinkOptions Validate(JsonFileSinkOptions? options) + { + ArgumentNullException.ThrowIfNull(options); + if (!Enum.IsDefined(options.Format)) + throw new ArgumentOutOfRangeException(nameof(options), options.Format, "The JSON format is not defined."); + if (!Enum.IsDefined(options.OpenMode)) + throw new ArgumentOutOfRangeException(nameof(options), options.OpenMode, "The JSON open mode is not defined."); + if (options.Format == JsonFileFormat.Auto) + throw new ArgumentException("Auto format is valid only for JSON sources.", nameof(options)); + if (options.Format == JsonFileFormat.Array && options.OpenMode == JsonFileOpenMode.Append) + throw new ArgumentException("A root JSON array cannot be appended safely.", nameof(options)); + if (options.FlushInterval <= 0) + throw new ArgumentOutOfRangeException(nameof(options), options.FlushInterval, "Flush interval must be greater than zero."); + return options with { }; + } + + internal static DeadLetterSinkOptions Validate(DeadLetterSinkOptions? options, bool loggerAvailable) + { + ArgumentNullException.ThrowIfNull(options); + ArgumentNullException.ThrowIfNull(options.RetryDelays); + ArgumentNullException.ThrowIfNull(options.TimeProvider); + if (!Enum.IsDefined(options.FailureMode)) + throw new ArgumentOutOfRangeException(nameof(options), options.FailureMode, "The dead-letter failure mode is not defined."); + if (options.RetryDelays.Any(static delay => delay < TimeSpan.Zero)) + throw new ArgumentOutOfRangeException(nameof(options), "Retry delays cannot be negative."); + if (options.FailureMode == Sinks.DeadLetterWriteFailureMode.LogAndDrop && !loggerAvailable) + throw new ArgumentException("LogAndDrop requires a logger factory.", nameof(options)); + return options with { RetryDelays = options.RetryDelays.ToArray() }; + } + private static void ValidateCommon( JsonFileFormat format, InvalidJsonRecordBehavior invalidRecordBehavior, int maxDepth, int maxRecordSizeBytes, long maxUnframedInputSizeBytes, - ILogger? logger, + bool loggerAvailable, string parameterName) { if (!Enum.IsDefined(format)) @@ -51,7 +87,7 @@ private static void ValidateCommon( throw new ArgumentOutOfRangeException(parameterName, maxRecordSizeBytes, "MaxRecordSizeBytes must be greater than zero."); if (maxUnframedInputSizeBytes <= 0) throw new ArgumentOutOfRangeException(parameterName, maxUnframedInputSizeBytes, "MaxUnframedInputSizeBytes must be greater than zero."); - if (invalidRecordBehavior == InvalidJsonRecordBehavior.SkipAndLog && logger == null) + if (invalidRecordBehavior == InvalidJsonRecordBehavior.SkipAndLog && !loggerAvailable) throw new ArgumentException("SkipAndLog requires a logger.", parameterName); } } diff --git a/src/SmartPipe.Extensions.Json/JsonMetadataSnapshot.cs b/src/SmartPipe.Extensions.Json/JsonMetadataSnapshot.cs new file mode 100644 index 0000000..94ecbf9 --- /dev/null +++ b/src/SmartPipe.Extensions.Json/JsonMetadataSnapshot.cs @@ -0,0 +1,87 @@ +#nullable enable + +using System.Text.Json; +using System.Text.Json.Serialization.Metadata; +using SmartPipe.Core; + +namespace SmartPipe.Extensions.Json; + +internal static class JsonMetadataSnapshot +{ + public static (JsonTypeInfo Item, JsonTypeInfo> Batch) ForFile( + JsonTypeInfo? itemTypeInfo, + JsonTypeInfo>? batchTypeInfo, + int? maxDepth = null) + { + ArgumentNullException.ThrowIfNull(itemTypeInfo); + ArgumentNullException.ThrowIfNull(batchTypeInfo); + if (itemTypeInfo.Type != typeof(T) || batchTypeInfo.Type != typeof(List)) + throw new ArgumentException("JSON type metadata does not match the source item and batch types."); + if (!ReferenceEquals(itemTypeInfo.Options, batchTypeInfo.Options)) + throw new ArgumentException("Item and batch JSON type metadata must come from the same serializer context."); + + var options = CloneOptions(itemTypeInfo.Options, maxDepth); + return (Resolve(options), Resolve>(options)); + } + + public static JsonTypeInfo ForValue(JsonTypeInfo? typeInfo) + { + ArgumentNullException.ThrowIfNull(typeInfo); + if (typeInfo.Type != typeof(T)) + throw new ArgumentException("JSON type metadata does not match the requested value type."); + + return Resolve(CloneOptions(typeInfo.Options, maxDepth: null)); + } + + public static JsonTypeInfo> ForDeadLetterEnvelope( + JsonTypeInfo>? typeInfo, + int? maxDepth = null) + { + ArgumentNullException.ThrowIfNull(typeInfo); + if (typeInfo.Type != typeof(DeadLetterEnvelope)) + throw new ArgumentException("JSON type metadata does not match the dead-letter envelope type."); + + return Resolve>(CloneOptions(typeInfo.Options, maxDepth)); + } + + private static JsonSerializerOptions CloneOptions(JsonSerializerOptions source, int? maxDepth) + { + ArgumentNullException.ThrowIfNull(source); + if (source.TypeInfoResolver is null) + throw new ArgumentException("Source-generated JSON metadata must provide a type-info resolver."); + + var clone = new JsonSerializerOptions(source) + { + TypeInfoResolver = source.TypeInfoResolver, + }; + if (maxDepth.HasValue) + clone.MaxDepth = maxDepth.Value; + + clone.MakeReadOnly(); + return clone; + } + + private static JsonTypeInfo Resolve(JsonSerializerOptions options) + { + try + { + if (options.GetTypeInfo(typeof(T)) is JsonTypeInfo typeInfo) + return typeInfo; + } + catch (NotSupportedException exception) + { + throw new ArgumentException( + $"The JSON metadata resolver cannot resolve '{typeof(T)}'.", + exception); + } + catch (InvalidOperationException exception) + { + throw new ArgumentException( + $"The JSON metadata resolver cannot resolve '{typeof(T)}'.", + exception); + } + + throw new ArgumentException( + $"The JSON metadata resolver returned incompatible metadata for '{typeof(T)}'."); + } +} diff --git a/src/SmartPipe.Extensions.Json/JsonPipelineComponents.cs b/src/SmartPipe.Extensions.Json/JsonPipelineComponents.cs new file mode 100644 index 0000000..9cb0f3d --- /dev/null +++ b/src/SmartPipe.Extensions.Json/JsonPipelineComponents.cs @@ -0,0 +1,122 @@ +#nullable enable + +using System.Text.Json.Serialization.Metadata; +using Microsoft.Extensions.Logging; +using SmartPipe.Core; +using SmartPipe.Extensions.Selectors; +using SmartPipe.Extensions.Sinks; +using SmartPipe.Extensions.Transforms; + +namespace SmartPipe.Extensions.Json; + +/// Creates runtime-owned JSON pipeline components. +public static class JsonPipelineComponents +{ + /// Creates a lazy, per-run JSON file source component. + public static PipelineComponent> FileSource( + string path, + JsonTypeInfo itemTypeInfo, + JsonTypeInfo> batchTypeInfo, + JsonFileSourceOptions options, + ILoggerFactory? loggerFactory = null) + { + var validatedPath = ValidatePath(path); + var validatedOptions = JsonInputOptionsValidator.Validate(options, loggerFactory is not null); + var metadata = JsonMetadataSnapshot.ForFile(itemTypeInfo, batchTypeInfo, validatedOptions.MaxDepth); + + return PipelineComponent.RuntimeOwned>( + (_, _) => ValueTask.FromResult>( + new JsonFileSource( + validatedPath, + metadata.Item, + metadata.Batch, + validatedOptions, + loggerFactory?.CreateLogger>()))); + } + + /// Creates a lazy, per-run JSON file sink component. + public static PipelineComponent> FileSink( + string path, + JsonTypeInfo itemTypeInfo, + JsonTypeInfo> batchTypeInfo, + JsonFileSinkOptions options) + { + var validatedPath = ValidatePath(path); + var validatedOptions = JsonInputOptionsValidator.Validate(options); + var metadata = JsonMetadataSnapshot.ForFile(itemTypeInfo, batchTypeInfo); + + return PipelineComponent.RuntimeOwned>( + (_, _) => ValueTask.FromResult>( + new JsonFileSink( + validatedPath, + metadata.Item, + metadata.Batch, + validatedOptions))); + } + + /// Creates a lazy, per-run JSON transform component. + public static PipelineComponent> Transform( + JsonTypeInfo inputTypeInfo, + JsonTypeInfo outputTypeInfo) + { + var inputMetadata = JsonMetadataSnapshot.ForValue(inputTypeInfo); + var outputMetadata = JsonMetadataSnapshot.ForValue(outputTypeInfo); + + return PipelineComponent.RuntimeOwned>( + (_, _) => ValueTask.FromResult>( + new JsonTransform(inputMetadata, outputMetadata))); + } + + /// Creates a lazy, per-run dead-letter JSON source component. + public static PipelineComponent> DeadLetterSource( + string path, + JsonTypeInfo> envelopeTypeInfo, + DeadLetterSourceOptions options, + ILoggerFactory? loggerFactory = null) + { + var validatedPath = ValidatePath(path); + var validatedOptions = JsonInputOptionsValidator.Validate(options, loggerFactory is not null); + var metadata = JsonMetadataSnapshot.ForDeadLetterEnvelope( + envelopeTypeInfo, + validatedOptions.MaxDepth); + + return PipelineComponent.RuntimeOwned>( + (_, _) => ValueTask.FromResult>( + loggerFactory is null + ? new DeadLetterSource(validatedPath, metadata, validatedOptions) + : new DeadLetterSource( + validatedPath, + metadata, + validatedOptions, + loggerFactory.CreateLogger>()))); + } + + /// Creates a lazy, per-run dead-letter JSON sink component. + public static PipelineComponent>> DeadLetterSink( + string path, + JsonTypeInfo> envelopeTypeInfo, + DeadLetterSinkOptions options, + ILoggerFactory? loggerFactory = null) + { + var validatedPath = ValidatePath(path); + var validatedOptions = JsonInputOptionsValidator.Validate(options, loggerFactory is not null); + var metadata = JsonMetadataSnapshot.ForDeadLetterEnvelope(envelopeTypeInfo); + + return PipelineComponent.RuntimeOwned>>( + (_, _) => ValueTask.FromResult>>( + new DeadLetterSink( + validatedPath, + new JsonLinesDeadLetterSerializer(metadata), + validatedOptions, + loggerFactory?.CreateLogger>(), + stream: null))); + } + + private static string ValidatePath(string? path) + { + ArgumentNullException.ThrowIfNull(path); + if (string.IsNullOrWhiteSpace(path)) + throw new ArgumentException("Path cannot be empty or whitespace.", nameof(path)); + return path; + } +} diff --git a/src/SmartPipe.Extensions.Json/JsonPipelineDefinitionBuilder.cs b/src/SmartPipe.Extensions.Json/JsonPipelineDefinitionBuilder.cs new file mode 100644 index 0000000..e4ffe07 --- /dev/null +++ b/src/SmartPipe.Extensions.Json/JsonPipelineDefinitionBuilder.cs @@ -0,0 +1,111 @@ +#nullable enable + +using System.Text.Json.Serialization.Metadata; +using Microsoft.Extensions.Logging; +using SmartPipe.Core; + +namespace SmartPipe.Extensions.Json; + +/// Starts typed pipeline definitions backed by JSON file components. +public static class JsonPipelineDefinitionBuilder +{ + /// Starts a typed definition with a JSON file source. + public static PipelineDefinitionBuilder FromJsonFile( + PipelineKey pipelineKey, + string path, + JsonTypeInfo itemTypeInfo, + JsonTypeInfo> batchTypeInfo, + JsonFileSourceOptions options, + ILoggerFactory? loggerFactory = null) => + SmartPipe.Core.PipelineDefinitionBuilder.From( + pipelineKey, + JsonPipelineComponents.FileSource( + path, + itemTypeInfo, + batchTypeInfo, + options, + loggerFactory)); + + /// Starts a typed definition with a dead-letter JSON source. + public static PipelineDefinitionBuilder FromJsonDeadLetterFile( + PipelineKey pipelineKey, + string path, + JsonTypeInfo> envelopeTypeInfo, + DeadLetterSourceOptions options, + ILoggerFactory? loggerFactory = null) => + SmartPipe.Core.PipelineDefinitionBuilder.From( + pipelineKey, + JsonPipelineComponents.DeadLetterSource( + path, + envelopeTypeInfo, + options, + loggerFactory)); +} + +/// Adds JSON transforms and file sinks to typed definitions. +public static class JsonPipelineDefinitionBuilderExtensions +{ +#pragma warning disable RS0026 // The canonical JSON builder intentionally mirrors Core's typed overload families. + /// Appends a JSON transform to a source-only definition. + public static PipelineDefinitionBuilder TransformJson( + this PipelineDefinitionBuilder builder, + PipelineStageKey stageKey, + JsonTypeInfo inputTypeInfo, + JsonTypeInfo outputTypeInfo, + StageFailureOptions? failureOptions = null, + StageDeadLetterOptions? deadLetterOptions = null, + string? stageName = null) + { + ArgumentNullException.ThrowIfNull(builder); + return builder.Transform( + stageKey, + JsonPipelineComponents.Transform(inputTypeInfo, outputTypeInfo), + failureOptions, + deadLetterOptions, + stageName); + } + + /// Appends a JSON transform to a multi-stage definition. + public static PipelineDefinitionBuilder TransformJson( + this PipelineDefinitionBuilder builder, + PipelineStageKey stageKey, + JsonTypeInfo inputTypeInfo, + JsonTypeInfo outputTypeInfo, + StageFailureOptions? failureOptions = null, + StageDeadLetterOptions? deadLetterOptions = null, + string? stageName = null) + { + ArgumentNullException.ThrowIfNull(builder); + return builder.Transform( + stageKey, + JsonPipelineComponents.Transform(inputTypeInfo, outputTypeInfo), + failureOptions, + deadLetterOptions, + stageName); + } + + /// Completes a source-only definition with a JSON file sink. + public static PipelineDefinition ToJsonFile( + this PipelineDefinitionBuilder builder, + string path, + JsonTypeInfo itemTypeInfo, + JsonTypeInfo> batchTypeInfo, + JsonFileSinkOptions options) + { + ArgumentNullException.ThrowIfNull(builder); + return builder.To(JsonPipelineComponents.FileSink(path, itemTypeInfo, batchTypeInfo, options)); + } + + /// Completes a multi-stage definition with a JSON file sink. + public static PipelineDefinition ToJsonFile( + this PipelineDefinitionBuilder builder, + string path, + JsonTypeInfo itemTypeInfo, + JsonTypeInfo> batchTypeInfo, + JsonFileSinkOptions options) + { + ArgumentNullException.ThrowIfNull(builder); + return builder.To(JsonPipelineComponents.FileSink(path, itemTypeInfo, batchTypeInfo, options)); + } +#pragma warning restore RS0026 +} diff --git a/src/SmartPipe.Extensions.Json/PublicAPI.Unshipped.txt b/src/SmartPipe.Extensions.Json/PublicAPI.Unshipped.txt index 3182dee..c50cb74 100644 --- a/src/SmartPipe.Extensions.Json/PublicAPI.Unshipped.txt +++ b/src/SmartPipe.Extensions.Json/PublicAPI.Unshipped.txt @@ -1,4 +1,18 @@ #nullable enable +SmartPipe.Extensions.Json.JsonPipelineComponents +static SmartPipe.Extensions.Json.JsonPipelineComponents.DeadLetterSink(string! path, System.Text.Json.Serialization.Metadata.JsonTypeInfo!>! envelopeTypeInfo, SmartPipe.Extensions.DeadLetterSinkOptions! options, Microsoft.Extensions.Logging.ILoggerFactory? loggerFactory = null) -> SmartPipe.Core.PipelineComponent!>!>! +static SmartPipe.Extensions.Json.JsonPipelineComponents.DeadLetterSource(string! path, System.Text.Json.Serialization.Metadata.JsonTypeInfo!>! envelopeTypeInfo, SmartPipe.Extensions.DeadLetterSourceOptions! options, Microsoft.Extensions.Logging.ILoggerFactory? loggerFactory = null) -> SmartPipe.Core.PipelineComponent!>! +static SmartPipe.Extensions.Json.JsonPipelineComponents.FileSink(string! path, System.Text.Json.Serialization.Metadata.JsonTypeInfo! itemTypeInfo, System.Text.Json.Serialization.Metadata.JsonTypeInfo!>! batchTypeInfo, SmartPipe.Extensions.JsonFileSinkOptions! options) -> SmartPipe.Core.PipelineComponent!>! +static SmartPipe.Extensions.Json.JsonPipelineComponents.FileSource(string! path, System.Text.Json.Serialization.Metadata.JsonTypeInfo! itemTypeInfo, System.Text.Json.Serialization.Metadata.JsonTypeInfo!>! batchTypeInfo, SmartPipe.Extensions.JsonFileSourceOptions! options, Microsoft.Extensions.Logging.ILoggerFactory? loggerFactory = null) -> SmartPipe.Core.PipelineComponent!>! +static SmartPipe.Extensions.Json.JsonPipelineComponents.Transform(System.Text.Json.Serialization.Metadata.JsonTypeInfo! inputTypeInfo, System.Text.Json.Serialization.Metadata.JsonTypeInfo! outputTypeInfo) -> SmartPipe.Core.PipelineComponent!>! +SmartPipe.Extensions.Json.JsonPipelineDefinitionBuilder +static SmartPipe.Extensions.Json.JsonPipelineDefinitionBuilder.FromJsonDeadLetterFile(SmartPipe.Core.PipelineKey pipelineKey, string! path, System.Text.Json.Serialization.Metadata.JsonTypeInfo!>! envelopeTypeInfo, SmartPipe.Extensions.DeadLetterSourceOptions! options, Microsoft.Extensions.Logging.ILoggerFactory? loggerFactory = null) -> SmartPipe.Core.PipelineDefinitionBuilder! +static SmartPipe.Extensions.Json.JsonPipelineDefinitionBuilder.FromJsonFile(SmartPipe.Core.PipelineKey pipelineKey, string! path, System.Text.Json.Serialization.Metadata.JsonTypeInfo! itemTypeInfo, System.Text.Json.Serialization.Metadata.JsonTypeInfo!>! batchTypeInfo, SmartPipe.Extensions.JsonFileSourceOptions! options, Microsoft.Extensions.Logging.ILoggerFactory? loggerFactory = null) -> SmartPipe.Core.PipelineDefinitionBuilder! +SmartPipe.Extensions.Json.JsonPipelineDefinitionBuilderExtensions +static SmartPipe.Extensions.Json.JsonPipelineDefinitionBuilderExtensions.ToJsonFile(this SmartPipe.Core.PipelineDefinitionBuilder! builder, string! path, System.Text.Json.Serialization.Metadata.JsonTypeInfo! itemTypeInfo, System.Text.Json.Serialization.Metadata.JsonTypeInfo!>! batchTypeInfo, SmartPipe.Extensions.JsonFileSinkOptions! options) -> SmartPipe.Core.PipelineDefinition! +static SmartPipe.Extensions.Json.JsonPipelineDefinitionBuilderExtensions.ToJsonFile(this SmartPipe.Core.PipelineDefinitionBuilder! builder, string! path, System.Text.Json.Serialization.Metadata.JsonTypeInfo! itemTypeInfo, System.Text.Json.Serialization.Metadata.JsonTypeInfo!>! batchTypeInfo, SmartPipe.Extensions.JsonFileSinkOptions! options) -> SmartPipe.Core.PipelineDefinition! +static SmartPipe.Extensions.Json.JsonPipelineDefinitionBuilderExtensions.TransformJson(this SmartPipe.Core.PipelineDefinitionBuilder! builder, SmartPipe.Core.PipelineStageKey stageKey, System.Text.Json.Serialization.Metadata.JsonTypeInfo! inputTypeInfo, System.Text.Json.Serialization.Metadata.JsonTypeInfo! outputTypeInfo, SmartPipe.Core.StageFailureOptions? failureOptions = null, SmartPipe.Core.StageDeadLetterOptions? deadLetterOptions = null, string? stageName = null) -> SmartPipe.Core.PipelineDefinitionBuilder! +static SmartPipe.Extensions.Json.JsonPipelineDefinitionBuilderExtensions.TransformJson(this SmartPipe.Core.PipelineDefinitionBuilder! builder, SmartPipe.Core.PipelineStageKey stageKey, System.Text.Json.Serialization.Metadata.JsonTypeInfo! inputTypeInfo, System.Text.Json.Serialization.Metadata.JsonTypeInfo! outputTypeInfo, SmartPipe.Core.StageFailureOptions? failureOptions = null, SmartPipe.Core.StageDeadLetterOptions? deadLetterOptions = null, string? stageName = null) -> SmartPipe.Core.PipelineDefinitionBuilder! SmartPipe.Extensions.Selectors.DeadLetterSource SmartPipe.Extensions.Selectors.DeadLetterSource.DeadLetterSource(string! path, System.Text.Json.Serialization.Metadata.JsonTypeInfo! valueTypeInfo) -> void SmartPipe.Extensions.Selectors.DeadLetterSource.DeadLetterSource(string! path, System.Text.Json.Serialization.Metadata.JsonTypeInfo! valueTypeInfo, SmartPipe.Extensions.DeadLetterSourceOptions! options) -> void diff --git a/src/SmartPipe.Extensions.Json/README.md b/src/SmartPipe.Extensions.Json/README.md index e7c2d4a..500ff1e 100644 --- a/src/SmartPipe.Extensions.Json/README.md +++ b/src/SmartPipe.Extensions.Json/README.md @@ -23,6 +23,8 @@ health-check, or Newtonsoft.Json dependencies. - `JsonTransform` - `DeadLetterSource` - `DeadLetterSink` +- `JsonPipelineComponents`, `JsonPipelineDefinitionBuilder`, and + `JsonPipelineDefinitionBuilderExtensions` The related `JsonLinesDeadLetterSerializer` remains part of `SmartPipe.Core`. @@ -33,6 +35,12 @@ The package uses `System.Text.Json` from the .NET 10 shared framework, so an additional `System.Text.Json` NuGet dependency is neither required nor pinned. Newtonsoft.Json is not a dependency and is not selected at runtime. +Line-framed input uses the internal, bounded UTF-8 reader linked from +`src/Shared/JsonFraming/Utf8LineRecordReader.cs`. The reader is BCL-only and +knows only about LF/CRLF boundaries, BOM bytes, and the configured record-size +limit. JSON validation, path diagnostics, and invalid-record policy remain in +this package; the framer is not a public API or a separate package. + ## Trimming and NativeAOT Reflection-based constructors are annotated for trimming and NativeAOT risk. @@ -53,9 +61,9 @@ source-generated paths. Explicitly line-framed records (`Ndjson`, (`MaxRecordSizeBytes`); root arrays and auto-detected legacy top-level value sequences use a 256 MiB unframed input limit (`MaxUnframedInputSizeBytes`). `SkipAndLog` requires a logger and is supported only when the source is reading -independently line-framed records. `JsonFileSource` requires explicit -`Ndjson` or `BatchJsonLines`; dead-letter `Auto` recovery depends on whether it -detects a framed stream rather than a root array. +independently line-framed records. `JsonFileSource` defaults to `Auto` and +also accepts explicit `Ndjson` or `BatchJsonLines`; dead-letter `Auto` recovery +depends on whether it detects a framed stream rather than a root array. Append mode preserves existing bytes. If a non-empty destination has no final LF, the sink inserts one before the next record; an existing partial row is not diff --git a/src/SmartPipe.Extensions.Json/Selectors/DeadLetterSource.cs b/src/SmartPipe.Extensions.Json/Selectors/DeadLetterSource.cs index a536f63..a6fbe63 100644 --- a/src/SmartPipe.Extensions.Json/Selectors/DeadLetterSource.cs +++ b/src/SmartPipe.Extensions.Json/Selectors/DeadLetterSource.cs @@ -5,6 +5,7 @@ using System.Text.Json.Serialization.Metadata; using Microsoft.Extensions.Logging; using SmartPipe.Core; +using SmartPipe.Shared.JsonFraming; namespace SmartPipe.Extensions.Selectors; diff --git a/src/SmartPipe.Extensions.Json/Selectors/JsonFileSource.cs b/src/SmartPipe.Extensions.Json/Selectors/JsonFileSource.cs index 91528b8..496179a 100644 --- a/src/SmartPipe.Extensions.Json/Selectors/JsonFileSource.cs +++ b/src/SmartPipe.Extensions.Json/Selectors/JsonFileSource.cs @@ -5,6 +5,8 @@ using System.Text.Json.Serialization.Metadata; using Microsoft.Extensions.Logging; using SmartPipe.Core; +using SmartPipe.Extensions.Json; +using SmartPipe.Shared.JsonFraming; namespace SmartPipe.Extensions.Selectors; @@ -111,7 +113,10 @@ public JsonFileSource( ArgumentNullException.ThrowIfNull(listTypeInfo); _options = JsonInputOptionsValidator.Validate(options, logger); _logger = logger; - var frozenTypeInfo = FreezeSourceGeneratedOptions(itemTypeInfo, listTypeInfo, _options.MaxDepth); + var frozenTypeInfo = JsonMetadataSnapshot.ForFile( + itemTypeInfo, + listTypeInfo, + _options.MaxDepth); _deserializeItems = (stream, topLevelValues, token) => WrapJsonErrors( JsonSerializer.DeserializeAsyncEnumerable(stream, frozenTypeInfo.Item, topLevelValues, token), @@ -119,11 +124,11 @@ public JsonFileSource( "document"); _deserializeBatches = (stream, token) => WrapJsonErrors( - JsonSerializer.DeserializeAsyncEnumerable(stream, frozenTypeInfo.List, topLevelValues: true, token), + JsonSerializer.DeserializeAsyncEnumerable(stream, frozenTypeInfo.Batch, topLevelValues: true, token), _path, "document"); _deserializeItemRecord = bytes => JsonSerializer.Deserialize(bytes, frozenTypeInfo.Item); - _deserializeBatchRecord = bytes => JsonSerializer.Deserialize(bytes, frozenTypeInfo.List); + _deserializeBatchRecord = bytes => JsonSerializer.Deserialize(bytes, frozenTypeInfo.Batch); } [RequiresUnreferencedCode("Reflection-based JSON file reading is not trimming-safe.")] @@ -353,30 +358,6 @@ private static JsonSerializerOptions FreezeOptions(JsonSerializerOptions? option return clone; } - private static (JsonTypeInfo Item, JsonTypeInfo> List) FreezeSourceGeneratedOptions( - JsonTypeInfo itemTypeInfo, - JsonTypeInfo> listTypeInfo, - int maxDepth) - { - if (itemTypeInfo.Type != typeof(T) || listTypeInfo.Type != typeof(List)) - throw new ArgumentException("JSON type metadata does not match the source item and batch types."); - if (!ReferenceEquals(itemTypeInfo.Options, listTypeInfo.Options)) - throw new ArgumentException("Item and list JSON type metadata must come from the same serializer context."); - if (itemTypeInfo.Options.TypeInfoResolver == null) - throw new ArgumentException("Source-generated JSON metadata must provide a type-info resolver."); - - var clone = new JsonSerializerOptions(itemTypeInfo.Options) - { - MaxDepth = maxDepth, - TypeInfoResolver = itemTypeInfo.Options.TypeInfoResolver, - }; - if (clone.GetTypeInfo(typeof(T)) is not JsonTypeInfo frozenItem - || clone.GetTypeInfo(typeof(List)) is not JsonTypeInfo> frozenList) - throw new ArgumentException("The JSON metadata resolver cannot resolve both item and batch types."); - clone.MakeReadOnly(); - return (frozenItem, frozenList); - } - private static async IAsyncEnumerable WrapJsonErrors( IAsyncEnumerable values, string path, diff --git a/src/SmartPipe.Extensions.Json/Sinks/DeadLetterSink.cs b/src/SmartPipe.Extensions.Json/Sinks/DeadLetterSink.cs index 003f15d..6de1725 100644 --- a/src/SmartPipe.Extensions.Json/Sinks/DeadLetterSink.cs +++ b/src/SmartPipe.Extensions.Json/Sinks/DeadLetterSink.cs @@ -324,14 +324,7 @@ private static string ValidatePath(string? path) } private static DeadLetterSinkOptions ValidateOptions(DeadLetterSinkOptions? options) - { - ArgumentNullException.ThrowIfNull(options); - ArgumentNullException.ThrowIfNull(options.RetryDelays); - ArgumentNullException.ThrowIfNull(options.TimeProvider); - if (options.RetryDelays.Any(static delay => delay < TimeSpan.Zero)) - throw new ArgumentOutOfRangeException(nameof(options), "Retry delays cannot be negative."); - return options with { RetryDelays = options.RetryDelays.ToArray() }; - } + => JsonInputOptionsValidator.Validate(options, loggerAvailable: true); [LoggerMessage(1, LogLevel.Warning, "IOException on attempt {Attempt}/{MaxAttempts} writing to dead letter file {Path}. Retrying in {DelayMilliseconds}ms.")] private static partial void LogRetry( diff --git a/src/SmartPipe.Extensions.Json/Sinks/JsonFileSink.cs b/src/SmartPipe.Extensions.Json/Sinks/JsonFileSink.cs index edc4769..26fb76d 100644 --- a/src/SmartPipe.Extensions.Json/Sinks/JsonFileSink.cs +++ b/src/SmartPipe.Extensions.Json/Sinks/JsonFileSink.cs @@ -356,16 +356,7 @@ private static async Task WriteTransactionalAsync(Stream stream, byte[] bytes, b } private static JsonFileSinkOptions ValidateOptions(JsonFileSinkOptions? options) - { - ArgumentNullException.ThrowIfNull(options); - if (options.Format == JsonFileFormat.Auto) - throw new ArgumentException("Auto format is valid only for JSON sources.", nameof(options)); - if (options.Format == JsonFileFormat.Array && options.OpenMode == JsonFileOpenMode.Append) - throw new ArgumentException("A root JSON array cannot be appended safely.", nameof(options)); - if (options.FlushInterval <= 0) - throw new ArgumentOutOfRangeException(nameof(options), options.FlushInterval, "Flush interval must be greater than zero."); - return options with { }; - } + => JsonInputOptionsValidator.Validate(options); private static string ValidatePath(string? path) { diff --git a/src/SmartPipe.Extensions.Json/SmartPipe.Extensions.Json.csproj b/src/SmartPipe.Extensions.Json/SmartPipe.Extensions.Json.csproj index e67fc7a..bf537e9 100644 --- a/src/SmartPipe.Extensions.Json/SmartPipe.Extensions.Json.csproj +++ b/src/SmartPipe.Extensions.Json/SmartPipe.Extensions.Json.csproj @@ -19,6 +19,10 @@ + + + + diff --git a/tests/Consumers/Scenarios/json-dependency-injection-direct/Consumer.csproj b/tests/Consumers/Scenarios/json-dependency-injection-direct/Consumer.csproj new file mode 100644 index 0000000..b1cb74a --- /dev/null +++ b/tests/Consumers/Scenarios/json-dependency-injection-direct/Consumer.csproj @@ -0,0 +1,16 @@ + + + Exe + net10.0 + enable + enable + true + false + + + + + + + + diff --git a/tests/Consumers/Scenarios/json-dependency-injection-direct/Program.cs b/tests/Consumers/Scenarios/json-dependency-injection-direct/Program.cs new file mode 100644 index 0000000..0e9f3bc --- /dev/null +++ b/tests/Consumers/Scenarios/json-dependency-injection-direct/Program.cs @@ -0,0 +1,53 @@ +using JsonDependencyInjectionConsumer; +using System.Text.Json.Serialization; +using Microsoft.Extensions.DependencyInjection; +using SmartPipe.Core; +using SmartPipe.Extensions; +using SmartPipe.Extensions.DependencyInjection; +using SmartPipe.Extensions.Json; + +var inputPath = Path.Combine(Path.GetTempPath(), $"smartpipe-json-di-{Guid.NewGuid():N}-input.json"); +try +{ + await File.WriteAllTextAsync(inputPath, "[{\"Value\":13}]\n"); + var key = new PipelineKey("json-dependency-injection-direct"); + var definition = JsonPipelineDefinitionBuilder + .FromJsonFile( + key, + inputPath, + ConsumerJsonContext.Default.ConsumerModel, + ConsumerJsonContext.Default.ListConsumerModel, + new JsonFileSourceOptions { Format = JsonFileFormat.Array }) + .Build(); + + var services = new ServiceCollection(); + services.AddSmartPipe().AddPipeline(definition); + await using var provider = services.BuildServiceProvider(new ServiceProviderOptions + { + ValidateScopes = true, + ValidateOnBuild = true, + }); + var factory = provider + .GetRequiredService() + .GetFactory(key); + await using var run = await factory.StartAsync(); + var output = await run.Outputs.ReadAsync(); + await run.Completion; + if (!output.Result.IsSuccess || output.Result.Value?.Value != 13 || run.Outputs.TryRead(out _)) return 1; +} +finally +{ + File.Delete(inputPath); +} + +Console.WriteLine("CONSUMER_OK json-dependency-injection-direct"); +return 0; + +namespace JsonDependencyInjectionConsumer +{ + internal sealed record ConsumerModel(int Value); + + [JsonSerializable(typeof(ConsumerModel))] + [JsonSerializable(typeof(List))] + internal sealed partial class ConsumerJsonContext : JsonSerializerContext; +} diff --git a/tests/Consumers/Scenarios/json-direct/Consumer.csproj b/tests/Consumers/Scenarios/json-direct/Consumer.csproj index 8ebf482..594d4b2 100644 --- a/tests/Consumers/Scenarios/json-direct/Consumer.csproj +++ b/tests/Consumers/Scenarios/json-direct/Consumer.csproj @@ -1,4 +1,4 @@ - Exenet10.0enableenabletrue + Exenet10.0enableenabletruefalse diff --git a/tests/Consumers/Scenarios/json-direct/Program.cs b/tests/Consumers/Scenarios/json-direct/Program.cs index e88264f..92f0c45 100644 --- a/tests/Consumers/Scenarios/json-direct/Program.cs +++ b/tests/Consumers/Scenarios/json-direct/Program.cs @@ -1,17 +1,56 @@ -using System.Text.Json; +using JsonDirectConsumer; using System.Text.Json.Serialization; using SmartPipe.Core; -using SmartPipe.Extensions.Transforms; +using SmartPipe.Extensions; +using SmartPipe.Extensions.Json; + +var inputPath = Path.Combine(Path.GetTempPath(), $"smartpipe-json-direct-{Guid.NewGuid():N}-input.json"); +var outputPath = Path.Combine(Path.GetTempPath(), $"smartpipe-json-direct-{Guid.NewGuid():N}-output.jsonl"); +try +{ + await File.WriteAllTextAsync(inputPath, "[{\"Value\":42}]\n"); + var key = new PipelineKey("json-direct"); + var definition = JsonPipelineDefinitionBuilder + .FromJsonFile( + key, + inputPath, + ConsumerJsonContext.Default.ConsumerModel, + ConsumerJsonContext.Default.ListConsumerModel, + new JsonFileSourceOptions { Format = JsonFileFormat.Array }) + .TransformJson( + new PipelineStageKey("json-round-trip"), + ConsumerJsonContext.Default.ConsumerModel, + ConsumerJsonContext.Default.ConsumerModel) + .ToJsonFile( + outputPath, + ConsumerJsonContext.Default.ConsumerModel, + ConsumerJsonContext.Default.ListConsumerModel, + new JsonFileSinkOptions + { + Format = JsonFileFormat.BatchJsonLines, + OpenMode = JsonFileOpenMode.Create, + FlushInterval = 1, + }); + + await using var run = await definition.StartAsync(); + await run.Completion; + var output = await File.ReadAllTextAsync(outputPath); + if (!output.Contains("42", StringComparison.Ordinal)) return 1; +} +finally +{ + File.Delete(inputPath); + File.Delete(outputPath); +} -var model = new ConsumerModel(42); -var json = JsonSerializer.Serialize(model, ConsumerJsonContext.Default.ConsumerModel); -var transformed = await new JsonTransform(ConsumerJsonContext.Default.ConsumerModel, ConsumerJsonContext.Default.ConsumerModel) - .TransformAsync(ProcessingEnvelope.Create(model)); -var roundTrip = JsonSerializer.Deserialize(json, ConsumerJsonContext.Default.ConsumerModel); -if (roundTrip?.Value != 42 || transformed.IsSuccess is false || transformed.Value?.Value != 42) return 1; Console.WriteLine("CONSUMER_OK json-direct"); return 0; -internal sealed record ConsumerModel(int Value); -[JsonSerializable(typeof(ConsumerModel))] -internal sealed partial class ConsumerJsonContext : JsonSerializerContext; +namespace JsonDirectConsumer +{ + internal sealed record ConsumerModel(int Value); + + [JsonSerializable(typeof(ConsumerModel))] + [JsonSerializable(typeof(List))] + internal sealed partial class ConsumerJsonContext : JsonSerializerContext; +} diff --git a/tests/Consumers/Scenarios/json-nativeaot/Consumer.csproj b/tests/Consumers/Scenarios/json-nativeaot/Consumer.csproj deleted file mode 100644 index 8ebf482..0000000 --- a/tests/Consumers/Scenarios/json-nativeaot/Consumer.csproj +++ /dev/null @@ -1,4 +0,0 @@ - - Exenet10.0enableenabletrue - - diff --git a/tests/Consumers/Scenarios/json-nativeaot/Program.cs b/tests/Consumers/Scenarios/json-nativeaot/Program.cs deleted file mode 100644 index 268ec80..0000000 --- a/tests/Consumers/Scenarios/json-nativeaot/Program.cs +++ /dev/null @@ -1,14 +0,0 @@ -using System.Text.Json; -using System.Text.Json.Serialization; -using SmartPipe.Extensions.Transforms; - -var info = ConsumerJsonContext.Default.ConsumerModel; -var transform = new JsonTransform(info, info); -var json = JsonSerializer.Serialize(new ConsumerModel(7), info); -if (json.Length == 0 || transform is null) return 1; -Console.WriteLine("CONSUMER_OK json-nativeaot"); -return 0; - -internal sealed record ConsumerModel(int Value); -[JsonSerializable(typeof(ConsumerModel))] -internal sealed partial class ConsumerJsonContext : JsonSerializerContext; diff --git a/tests/SmartPipe.Extensions.Json.Tests/JsonPipelineDefinitionContractTests.cs b/tests/SmartPipe.Extensions.Json.Tests/JsonPipelineDefinitionContractTests.cs new file mode 100644 index 0000000..75f24e2 --- /dev/null +++ b/tests/SmartPipe.Extensions.Json.Tests/JsonPipelineDefinitionContractTests.cs @@ -0,0 +1,463 @@ +using System.Reflection; +using System.Text.Json; +using System.Text.Json.Serialization; +using System.Text.Json.Serialization.Metadata; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Logging.Abstractions; +using SmartPipe.Core; +using SmartPipe.Extensions; +using SmartPipe.Extensions.Sinks; + +namespace SmartPipe.Extensions.Tests; + +public sealed class JsonPipelineDefinitionContractTests +{ + [Fact] + public void CanonicalComponentFactories_ExposeThePlannedPublicSurface() + { + var type = typeof(JsonFileSourceOptions).Assembly.GetType( + "SmartPipe.Extensions.Json.JsonPipelineComponents"); + + Assert.NotNull(type); + Assert.True(type!.IsAbstract && type.IsSealed); + + AssertFactory(type, "FileSource", typeof(IPipelineSource<>), 5); + AssertFactory(type, "FileSink", typeof(IPipelineSink<>), 4); + AssertFactory(type, "Transform", typeof(IPipelineTransformer<,>), 2); + AssertFactory(type, "DeadLetterSource", typeof(IPipelineSource<>), 4); + AssertFactory(type, "DeadLetterSink", typeof(IPipelineSink<>), 4); + } + + [Fact] + public void CanonicalBuilders_ExposeThePlannedPublicSurface() + { + var assembly = typeof(JsonFileSourceOptions).Assembly; + var builder = assembly.GetType("SmartPipe.Extensions.Json.JsonPipelineDefinitionBuilder"); + var extensions = assembly.GetType("SmartPipe.Extensions.Json.JsonPipelineDefinitionBuilderExtensions"); + + Assert.NotNull(builder); + Assert.NotNull(extensions); + Assert.True(builder!.IsAbstract && builder.IsSealed); + Assert.True(extensions!.IsAbstract && extensions.IsSealed); + + AssertMethod(builder, "FromJsonFile", parameterCount: 6); + AssertMethod(builder, "FromJsonDeadLetterFile", parameterCount: 5); + Assert.Equal( + 2, + extensions.GetMethods(BindingFlags.Public | BindingFlags.Static) + .Count(method => method.Name == "TransformJson" && method.IsGenericMethodDefinition)); + Assert.Equal( + 2, + extensions.GetMethods(BindingFlags.Public | BindingFlags.Static) + .Count(method => method.Name == "ToJsonFile" && method.IsGenericMethodDefinition)); + } + + [Fact] + public async Task CanonicalFileSource_IsLazyRuntimeOwnedAndFreshPerActivation() + { + var components = RequireComponentsType(); + var loggerFactory = new TrackingLoggerFactory(); + var options = new JsonFileSourceOptions + { + Format = JsonFileFormat.Ndjson, + MaxDepth = 8, + }; + var descriptor = InvokeFactory( + components, + "FileSource", + typeof(DefinitionItem), + "this-file-does-not-exist.json", + DefinitionJsonContext.Default.DefinitionItem, + DefinitionJsonContext.Default.ListDefinitionItem, + options, + loggerFactory); + + Assert.Equal(PipelineComponentOwnership.RuntimeOwned, GetProperty(descriptor, "Ownership")); + Assert.True((bool)GetProperty(descriptor, "Initialize")!); + Assert.True((bool)GetProperty(descriptor, "IsPerRun")!); + Assert.Equal(0, loggerFactory.CreateLoggerCalls); + Assert.Equal(JsonFileFormat.Ndjson, options.Format); + Assert.Equal(8, options.MaxDepth); + + var firstContext = new PipelineActivationContext(new PipelineKey("json"), Guid.NewGuid()); + var secondContext = new PipelineActivationContext(new PipelineKey("json"), Guid.NewGuid()); + using var firstCts = CancellationTokenSource.CreateLinkedTokenSource(TestContext.Current.CancellationToken); + using var secondCts = CancellationTokenSource.CreateLinkedTokenSource(TestContext.Current.CancellationToken); + var first = await InvokeActivatorAsync(descriptor, firstContext, firstCts.Token); + var second = await InvokeActivatorAsync(descriptor, secondContext, secondCts.Token); + + Assert.NotSame(first, second); + Assert.Equal(2, loggerFactory.CreateLoggerCalls); + + await ((IAsyncDisposable)first).DisposeAsync(); + await ((IAsyncDisposable)second).DisposeAsync(); + Assert.Equal(0, loggerFactory.DisposeCalls); + } + + [Fact] + public void CanonicalSkipPolicies_RequireAnExplicitLoggerFactory() + { + var components = RequireComponentsType(); + var sourceException = Assert.ThrowsAny(() => InvokeFactory( + components, + "FileSource", + typeof(DefinitionItem), + "input.json", + DefinitionJsonContext.Default.DefinitionItem, + DefinitionJsonContext.Default.ListDefinitionItem, + new JsonFileSourceOptions + { + Format = JsonFileFormat.Ndjson, + InvalidRecordBehavior = InvalidJsonRecordBehavior.SkipAndLog, + }, + null)); + AssertPolicyException(sourceException); + + var deadLetterSinkException = Assert.ThrowsAny(() => InvokeFactory( + components, + "DeadLetterSink", + typeof(DefinitionItem), + "dead-letter.json", + DefinitionJsonContext.Default.DeadLetterEnvelopeDefinitionItem, + new DeadLetterSinkOptions { FailureMode = DeadLetterWriteFailureMode.LogAndDrop }, + null)); + AssertPolicyException(deadLetterSinkException); + } + + [Fact] + public void CanonicalDeadLetterSourceSkipPolicy_RequiresAnExplicitLoggerFactory() + { + var exception = Assert.ThrowsAny(() => InvokeFactory( + RequireComponentsType(), + "DeadLetterSource", + typeof(DefinitionItem), + "dead-letter.json", + DefinitionJsonContext.Default.DeadLetterEnvelopeDefinitionItem, + new DeadLetterSourceOptions + { + Format = JsonFileFormat.Ndjson, + InvalidRecordBehavior = InvalidJsonRecordBehavior.SkipAndLog, + }, + null)); + + AssertPolicyException(exception); + } + + [Fact] + public void CanonicalMetadataFactories_RejectMissingOrUnresolvableResolvers() + { + var noResolverOptions = new JsonSerializerOptions(); + var noResolverItem = JsonTypeInfo.CreateJsonTypeInfo(noResolverOptions); + var noResolverBatch = JsonTypeInfo.CreateJsonTypeInfo>(noResolverOptions); + var noResolverException = Assert.ThrowsAny(() => InvokeFactory( + RequireComponentsType(), + "FileSource", + typeof(DefinitionItem), + "input.json", + noResolverItem, + noResolverBatch, + new JsonFileSourceOptions { Format = JsonFileFormat.Ndjson }, + null)); + AssertArgumentException(noResolverException); + + var unresolvableOptions = new JsonSerializerOptions + { + TypeInfoResolver = new NullTypeInfoResolver(), + }; + var unresolvableItem = JsonTypeInfo.CreateJsonTypeInfo(unresolvableOptions); + var unresolvableBatch = JsonTypeInfo.CreateJsonTypeInfo>(unresolvableOptions); + var unresolvableException = Assert.ThrowsAny(() => InvokeFactory( + RequireComponentsType(), + "FileSource", + typeof(DefinitionItem), + "input.json", + unresolvableItem, + unresolvableBatch, + new JsonFileSourceOptions { Format = JsonFileFormat.Ndjson }, + null)); + AssertArgumentException(unresolvableException); + } + + [Fact] + public void CanonicalMetadataFactories_RejectMismatchedContexts() + { + var first = new DefinitionJsonContext(new JsonSerializerOptions()); + var second = new DefinitionJsonContext(new JsonSerializerOptions()); + var exception = Assert.ThrowsAny(() => InvokeFactory( + RequireComponentsType(), + "FileSource", + typeof(DefinitionItem), + "input.json", + first.DefinitionItem, + second.ListDefinitionItem, + new JsonFileSourceOptions { Format = JsonFileFormat.Ndjson }, + null)); + + AssertArgumentException(exception); + } + + [Fact] + public void CanonicalFileSource_DoesNotMutateCallerSerializerOptions() + { + var serializerOptions = new JsonSerializerOptions + { + PropertyNameCaseInsensitive = true, + }; + var context = new DefinitionJsonContext(serializerOptions); + var originalMaxDepth = serializerOptions.MaxDepth; + var originalReadOnly = serializerOptions.IsReadOnly; + var originalResolver = serializerOptions.TypeInfoResolver; + var originalItemTypeInfoReadOnly = context.DefinitionItem.IsReadOnly; + var originalBatchTypeInfoReadOnly = context.ListDefinitionItem.IsReadOnly; + + _ = InvokeFactory( + RequireComponentsType(), + "FileSource", + typeof(DefinitionItem), + "input.json", + context.DefinitionItem, + context.ListDefinitionItem, + new JsonFileSourceOptions { Format = JsonFileFormat.Ndjson, MaxDepth = 13 }, + null); + + Assert.Equal(originalMaxDepth, serializerOptions.MaxDepth); + Assert.Equal(originalReadOnly, serializerOptions.IsReadOnly); + Assert.Same(originalResolver, serializerOptions.TypeInfoResolver); + Assert.Same(serializerOptions, context.DefinitionItem.Options); + Assert.Equal(originalItemTypeInfoReadOnly, context.DefinitionItem.IsReadOnly); + Assert.Equal(originalBatchTypeInfoReadOnly, context.ListDefinitionItem.IsReadOnly); + } + + [Fact] + public async Task CanonicalDeadLetterSink_CopiesRetryDelaysAndCreatesLoggerAtActivation() + { + var loggerFactory = new TrackingLoggerFactory(); + var retryDelays = new List + { + TimeSpan.FromMilliseconds(11), + TimeSpan.FromMilliseconds(22), + }; + var descriptor = InvokeFactory( + RequireComponentsType(), + "DeadLetterSink", + typeof(DefinitionItem), + Path.Combine(Path.GetTempPath(), $"smartpipe-dl-{Guid.NewGuid():N}.json"), + DefinitionJsonContext.Default.DeadLetterEnvelopeDefinitionItem, + new DeadLetterSinkOptions + { + FailureMode = DeadLetterWriteFailureMode.LogAndDrop, + RetryDelays = retryDelays, + }, + loggerFactory); + + Assert.Equal(0, loggerFactory.CreateLoggerCalls); + retryDelays[0] = TimeSpan.FromHours(1); + var sink = await InvokeActivatorAsync( + descriptor, + new PipelineActivationContext(new PipelineKey("json-dead-letter"), Guid.NewGuid()), + TestContext.Current.CancellationToken); + + Assert.Equal(1, loggerFactory.CreateLoggerCalls); + var delaysField = sink.GetType().GetField("_retryDelays", BindingFlags.NonPublic | BindingFlags.Instance); + Assert.NotNull(delaysField); + var capturedDelays = Assert.IsType(delaysField!.GetValue(sink)); + Assert.Equal(TimeSpan.FromMilliseconds(11), capturedDelays[0]); + Assert.Equal(TimeSpan.FromMilliseconds(22), capturedDelays[1]); + + await ((IAsyncDisposable)sink).DisposeAsync(); + Assert.Equal(0, loggerFactory.DisposeCalls); + } + + [Fact] + public void CanonicalBuilderExtensions_ChainTypedDefinitionWithoutActivation() + { + var assembly = typeof(JsonFileSourceOptions).Assembly; + var builderType = assembly.GetType("SmartPipe.Extensions.Json.JsonPipelineDefinitionBuilder"); + var extensionsType = assembly.GetType("SmartPipe.Extensions.Json.JsonPipelineDefinitionBuilderExtensions"); + Assert.NotNull(builderType); + Assert.NotNull(extensionsType); + + var itemTypeInfo = DefinitionJsonContext.Default.DefinitionItem; + var listTypeInfo = DefinitionJsonContext.Default.ListDefinitionItem; + var inputPath = Path.Combine(Path.GetTempPath(), $"smartpipe-definition-input-{Guid.NewGuid():N}.json"); + var outputPath = Path.Combine(Path.GetTempPath(), $"smartpipe-definition-output-{Guid.NewGuid():N}.json"); + Assert.False(File.Exists(inputPath)); + Assert.False(File.Exists(outputPath)); + var sourceFactory = builderType!.GetMethods(BindingFlags.Public | BindingFlags.Static) + .Single(method => method.Name == "FromJsonFile" && method.IsGenericMethodDefinition); + var builder = sourceFactory.MakeGenericMethod(typeof(DefinitionItem)).Invoke(null, + [ + new PipelineKey("json-builder"), + inputPath, + itemTypeInfo, + listTypeInfo, + new JsonFileSourceOptions { Format = JsonFileFormat.Ndjson }, + null, + ]); + Assert.NotNull(builder); + + var transform = extensionsType!.GetMethods(BindingFlags.Public | BindingFlags.Static) + .Single(method => method.Name == "TransformJson" + && method.IsGenericMethodDefinition + && method.GetGenericArguments().Length == 2); + var transformed = transform.MakeGenericMethod(typeof(DefinitionItem), typeof(DefinitionItem)).Invoke(null, + [ + builder, + new PipelineStageKey("json-transform"), + itemTypeInfo, + itemTypeInfo, + null, + null, + null, + ]); + Assert.NotNull(transformed); + + var sinkFactory = extensionsType.GetMethods(BindingFlags.Public | BindingFlags.Static) + .Single(method => method.Name == "ToJsonFile" + && method.IsGenericMethodDefinition + && method.GetGenericArguments().Length == 2); + var definition = sinkFactory.MakeGenericMethod(typeof(DefinitionItem), typeof(DefinitionItem)).Invoke(null, + [ + transformed, + outputPath, + itemTypeInfo, + listTypeInfo, + new JsonFileSinkOptions { Format = JsonFileFormat.BatchJsonLines }, + ]); + + var typedDefinition = Assert.IsType>(definition); + Assert.True(typedDefinition.HasSink); + Assert.Single(typedDefinition.Stages); + Assert.Equal("json-transform", typedDefinition.Stages[0].Key.Value); + Assert.False(File.Exists(inputPath)); + Assert.False(File.Exists(outputPath)); + } + + [Fact] + public void JsonPackage_DoesNotReferenceDependencyInjectionOrFacade() + { + var references = typeof(JsonFileSourceOptions).Assembly + .GetReferencedAssemblies() + .Select(static assembly => assembly.Name) + .Where(static name => name is not null) + .ToHashSet(StringComparer.OrdinalIgnoreCase); + + Assert.DoesNotContain("SmartPipe.Extensions", references); + Assert.DoesNotContain("SmartPipe.Extensions.DependencyInjection", references); + } + + private static Type RequireComponentsType() + { + var type = typeof(JsonFileSourceOptions).Assembly.GetType( + "SmartPipe.Extensions.Json.JsonPipelineComponents"); + Assert.NotNull(type); + return type!; + } + + private static object InvokeFactory( + Type components, + string name, + Type genericType, + params object?[] arguments) + { + var method = components.GetMethods(BindingFlags.Public | BindingFlags.Static) + .Single(candidate => candidate.Name == name && candidate.IsGenericMethodDefinition); + return method.MakeGenericMethod(genericType).Invoke(null, arguments)!; + } + + private static object? GetProperty(object value, string name) + { + var property = value.GetType().GetProperty( + name, + BindingFlags.Public | BindingFlags.NonPublic | BindingFlags.Instance); + Assert.NotNull(property); + return property!.GetValue(value); + } + + private static async Task InvokeActivatorAsync( + object descriptor, + PipelineActivationContext context, + CancellationToken cancellationToken) + { + var property = descriptor.GetType().GetProperty( + "Activator", + BindingFlags.NonPublic | BindingFlags.Instance); + Assert.NotNull(property); + var activator = Assert.IsAssignableFrom(property!.GetValue(descriptor)); + var valueTask = activator.DynamicInvoke(context, cancellationToken); + Assert.NotNull(valueTask); + var asTask = valueTask!.GetType().GetMethod("AsTask", Type.EmptyTypes); + Assert.NotNull(asTask); + var task = Assert.IsAssignableFrom(asTask!.Invoke(valueTask, null)); + await task; + return task.GetType().GetProperty("Result")!.GetValue(task)!; + } + + private static void AssertPolicyException(Exception exception) + { + var actual = UnwrapInvocationException(exception); + Assert.True( + actual is ArgumentException or InvalidOperationException, + $"Expected a policy validation exception, got {actual.GetType().FullName}: {actual.Message}"); + } + + private static void AssertArgumentException(Exception exception) + { + var actual = UnwrapInvocationException(exception); + Assert.IsType(actual); + } + + private static Exception UnwrapInvocationException(Exception exception) => + exception is TargetInvocationException { InnerException: not null } invocation + ? invocation.InnerException! + : exception; + + private static void AssertFactory(Type type, string name, Type resultDefinition, int parameterCount) + { + var method = type.GetMethods(BindingFlags.Public | BindingFlags.Static) + .SingleOrDefault(candidate => candidate.Name == name); + + Assert.NotNull(method); + Assert.True(method!.IsGenericMethodDefinition); + Assert.Equal(resultDefinition, method.ReturnType.GetGenericArguments()[0].GetGenericTypeDefinition()); + Assert.Equal(parameterCount, method.GetParameters().Length); + } + + private static void AssertMethod(Type type, string name, int parameterCount) + { + var method = type.GetMethods(BindingFlags.Public | BindingFlags.Static) + .SingleOrDefault(candidate => candidate.Name == name); + + Assert.NotNull(method); + Assert.True(method!.IsGenericMethodDefinition); + Assert.Equal(parameterCount, method.GetParameters().Length); + } + + private sealed class NullTypeInfoResolver : IJsonTypeInfoResolver + { + public JsonTypeInfo? GetTypeInfo(Type type, JsonSerializerOptions options) => null; + } + + private sealed class TrackingLoggerFactory : ILoggerFactory + { + public int CreateLoggerCalls { get; private set; } + public int DisposeCalls { get; private set; } + + public ILogger CreateLogger(string categoryName) + { + CreateLoggerCalls++; + return NullLogger.Instance; + } + + public void AddProvider(ILoggerProvider provider) { } + + public void Dispose() => DisposeCalls++; + } +} + +public sealed record DefinitionItem(int Id); + +[JsonSerializable(typeof(DefinitionItem))] +[JsonSerializable(typeof(List))] +[JsonSerializable(typeof(DeadLetterEnvelope))] +internal sealed partial class DefinitionJsonContext : JsonSerializerContext; diff --git a/tests/SmartPipe.Extensions.Json.Tests/Utf8LineRecordReaderTests.cs b/tests/SmartPipe.Extensions.Json.Tests/Utf8LineRecordReaderTests.cs index c7d7dcc..6d1c6c7 100644 --- a/tests/SmartPipe.Extensions.Json.Tests/Utf8LineRecordReaderTests.cs +++ b/tests/SmartPipe.Extensions.Json.Tests/Utf8LineRecordReaderTests.cs @@ -1,5 +1,5 @@ using System.Text; -using SmartPipe.Extensions; +using SmartPipe.Shared.JsonFraming; namespace SmartPipe.Extensions.Tests; @@ -117,6 +117,22 @@ public async Task PartialMultibyteUtf8_IsPreservedWithoutDecoding() Assert.Equal(expected, record.Bytes); } + [Fact] + public async Task BomBlankLineAndCrLfSplitAcrossReads_AreFramedCorrectly() + { + var bytes = Encoding.UTF8.GetPreamble() + .Concat(" \r\n\"first\"\r\n\"second\"\n"u8.ToArray()) + .ToArray(); + await using var stream = new OneByteReadStream(bytes); + + var records = await ReadAllAsync(stream, maxRecordSizeBytes: 16); + + Assert.Equal(2, records.Count); + Assert.Equal("\"first\""u8.ToArray(), records[0].Bytes); + Assert.Equal("\"second\""u8.ToArray(), records[1].Bytes); + Assert.All(records, static record => Assert.False(record.TooLarge)); + } + private static async Task> ReadAllAsync(Stream stream, int maxRecordSizeBytes) { var records = new List(); diff --git a/tests/SmartPipe.RepositoryChecks.Tests/Consumers/ConsumerScenarioSchemaTests.cs b/tests/SmartPipe.RepositoryChecks.Tests/Consumers/ConsumerScenarioSchemaTests.cs index 6be6542..91ad360 100644 --- a/tests/SmartPipe.RepositoryChecks.Tests/Consumers/ConsumerScenarioSchemaTests.cs +++ b/tests/SmartPipe.RepositoryChecks.Tests/Consumers/ConsumerScenarioSchemaTests.cs @@ -9,12 +9,12 @@ namespace SmartPipe.RepositoryChecks.Tests.Consumers; public sealed class ConsumerScenarioSchemaTests { [Fact] - public async Task CurrentManifest_HasExactlyThirtyThreeStrictScenarios() + public async Task CurrentManifest_HasExactlyThirtyFiveStrictScenarios() { var root = RepositoryRoot(); var graph = await new PackageGraphLoader().LoadAsync(root, "eng/package-graph.json", TestContext.Current.CancellationToken); var document = await new ConsumerScenarioLoader().LoadAsync(root, "eng/consumer-scenarios.json", graph, TestContext.Current.CancellationToken); - Assert.Equal(33, document.Scenarios.Count); + Assert.Equal(35, document.Scenarios.Count); Assert.Equal( [ "core-direct", @@ -24,6 +24,8 @@ public async Task CurrentManifest_HasExactlyThirtyThreeStrictScenarios() "core-trim", "core-nativeaot", "json-nativeaot", + "json-trim", + "json-dependency-injection-direct", "dependency-injection-direct", "dependency-injection-keyed", "dependency-injection-from-keyed-services",