Skip to content

feat: Implement composite provider support in ConfigurationFileValueConnectorParser and add IDeferredProviderAdapter interface; update version to 2.5.6 #556

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
paillave merged 1 commit into from
Jun 17, 2026
Merged

feat: Implement composite provider support in ConfigurationFileValueConnectorParser and add IDeferredProviderAdapter interface; update version to 2.5.6 #556

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
31 changes: 31 additions & 0 deletions src/Paillave.Etl.FromConfigurationConnectors.Tests/ConnectorsMergeIntegrationTests.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -167,4 +167,35 @@ public void GetConnectors_SingleSftpDefinition_ReturnsProvider()
Assert.Contains("INPUTPOSITIONS", real.Providers);
Assert.IsType<SftpFileValueProvider>(real.GetProvider("INPUTPOSITIONS"));
}

// -----------------------------------------------------------------------
// Composite provider defined in the same JSON as the sub-connectors.
// -----------------------------------------------------------------------
private const string DefinitionComposite = """
{
"AllFiles": {
"Type": "Composite",
"Providers": {
"ALL_INPUTS": {
"ProviderCodes": ["INPUTPOSITIONS", "INPUTDOCUMENTS"]
}
}
}
}
""";

[Fact]
public void GetConnectors_CompositeDefinitionMergedWithSftp_ReturnsCompositeProvider()
{
var merged = MergeDefinitions(Definition1, Definition2, DefinitionComposite);

var parser = new ConfigurationFileValueConnectorParser(new SftpProviderProcessorAdapter());
var connectors = parser.GetConnectors(merged, PassThroughResolver);

var real = Assert.IsType<FileValueConnectors>(connectors);
Assert.Contains("INPUTPOSITIONS", real.Providers);
Assert.Contains("INPUTDOCUMENTS", real.Providers);
Assert.Contains("ALL_INPUTS", real.Providers);
Assert.IsType<Paillave.Etl.Core.CompositeFileValueProvider>(real.GetProvider("ALL_INPUTS"));
}
}
78 changes: 61 additions & 17 deletions src/Paillave.Etl.FromConfigurationConnectors/ConfigurationFileValueConnectorParser.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,10 @@ namespace Paillave.Etl.FromConfigurationConnectors;

public class ConfigurationFileValueConnectorParser(params IProviderProcessorAdapter[] providerProcessorAdapters)
{
// Composite is a framework built-in — callers never register it explicitly.
private readonly IProviderProcessorAdapter[] _adapters =
[new CompositeFileValueProviderAdapter(), .. providerProcessorAdapters];

public IFileValueConnectors GetConnectors(string jsonConfig, Func<string, string> resolveSensitiveValue)
{
if (string.IsNullOrWhiteSpace(jsonConfig))
Expand All @@ -27,19 +31,31 @@ public IFileValueConnectors GetConnectors(string jsonConfig, Func<string, string
{
return new NoFileValueConnectors();
}
Dictionary<string, IProviderProcessorAdapter> adapterDictionary = providerProcessorAdapters.ToDictionary(i => i.Name);
Dictionary<string, IProviderProcessorAdapter> adapterDictionary = _adapters.ToDictionary(i => i.Name);
JObject o = JObject.Parse(jsonConfig);
var elts = o.Properties()
.Where(p => p.Path != "$schema")
.Select(i => ParseConnections(i, adapterDictionary, resolveSensitiveValue));
var processors = elts.SelectMany(i => i.processors).ToDictionary(i => i.Code);
var providers = elts.SelectMany(i => i.providers).ToDictionary(i => i.Code);
return new FileValueConnectors(providers, processors);
var connectionProperties = o.Properties().Where(p => p.Path != "$schema").ToList();

// Pass 1: build all regular (non-deferred) providers and processors.
var regularConnections = connectionProperties
.Where(p => GetAdapterType(p, adapterDictionary) is { } a && a is not IDeferredProviderAdapter)
.Select(p => ParseConnections(p, adapterDictionary, resolveSensitiveValue));
var processors = regularConnections.SelectMany(i => i.processors).ToDictionary(i => i.Code);
var providers = regularConnections.SelectMany(i => i.providers).ToDictionary(i => i.Code);
var connectors = new FileValueConnectors(providers, processors);

// Pass 2: build deferred providers (e.g. composites) now that the registry is populated.
var deferredProviders = connectionProperties
.Where(p => GetAdapterType(p, adapterDictionary) is IDeferredProviderAdapter)
.SelectMany(p => ParseDeferredConnection(p, adapterDictionary, connectors));
foreach (var provider in deferredProviders)
connectors.Register(provider);

return connectors;
}
public IFileValueProvider GetProvider(IConfigurationSection configurationSection)
{
var type = configurationSection.GetSection("Type").Value;
var providerProcessorAdapter = providerProcessorAdapters.Single(a => a.Name.Equals(type, StringComparison.InvariantCultureIgnoreCase));
var providerProcessorAdapter = _adapters.Single(a => a.Name.Equals(type, StringComparison.InvariantCultureIgnoreCase));
var connectionConfiguration = configurationSection.GetSection("Connection");
var connectionParameters = connectionConfiguration.Get(providerProcessorAdapter.ConnectionParametersType);
var providerSection = configurationSection.GetSection("Provider");
Expand All @@ -49,13 +65,38 @@ public IFileValueProvider GetProvider(IConfigurationSection configurationSection
public IFileValueProcessor GetProcessor(IConfigurationSection configurationSection)
{
var type = configurationSection.GetSection("Type").Value;
var providerProcessorAdapter = providerProcessorAdapters.Single(a => a.Name.Equals(type, StringComparison.InvariantCultureIgnoreCase));
var providerProcessorAdapter = _adapters.Single(a => a.Name.Equals(type, StringComparison.InvariantCultureIgnoreCase));
var connectionConfiguration = configurationSection.GetSection("Connection");
var connectionParameters = connectionConfiguration.Get(providerProcessorAdapter.ConnectionParametersType);
var processorSection = configurationSection.GetSection("Processor");
var processorParameters = processorSection.Get(providerProcessorAdapter.ProcessorParametersType) ?? Activator.CreateInstance(providerProcessorAdapter.ProcessorParametersType);
return providerProcessorAdapter.CreateProcessor("code", "name", "cnx", connectionParameters, processorParameters);
}
private static IProviderProcessorAdapter? GetAdapterType(JProperty property, Dictionary<string, IProviderProcessorAdapter> adapterDictionary)
{
var typeCode = (string?)((JObject)property.Value)["Type"];
return typeCode != null && adapterDictionary.TryGetValue(typeCode, out var adapter) ? adapter : null;
}
private static IEnumerable<IFileValueProvider> ParseDeferredConnection(JProperty property, Dictionary<string, IProviderProcessorAdapter> adapterDictionary, IFileValueConnectors connectors)
{
var connectionNode = (JObject)property.Value;
string connectionName = property.Name;
// Type is guaranteed non-null — this method is only called after GetAdapterType filtering.
string connectionTypeCode = ((string?)connectionNode["Type"])!;
var adapter = adapterDictionary[connectionTypeCode];
var deferredAdapter = (IDeferredProviderAdapter)adapter;
var providersNode = (JObject?)connectionNode["Providers"];
if (providersNode == null || adapter.ProviderParametersType == null) yield break;
foreach (var providerProp in providersNode.Properties())
{
var connectorNode = (JObject)providerProp.Value;
var code = providerProp.Name;
var name = ((string?)connectorNode["Name"]) ?? code;
var parameters = connectorNode.ToObject(adapter.ProviderParametersType)
?? Activator.CreateInstance(adapter.ProviderParametersType)!;
yield return deferredAdapter.CreateDeferredProvider(code, name, connectionName, parameters, connectors);
}
}
private (List<IFileValueProvider> providers, List<IFileValueProcessor> processors) ParseConnections(JProperty property, Dictionary<string, IProviderProcessorAdapter> adapterDictionary, Func<string, string> resolveSensitiveValue)
{
var connectionNode = (JObject)property.Value;
Expand Down Expand Up @@ -135,15 +176,18 @@ private static JsonSchema CreateAddonAdapter(JsonSchema docSchema, IProviderProc
// Multiple adapters referencing the same type (e.g. an enum) would otherwise produce
// independent instances; only one would land in Definitions, leaving the others as
// dangling references that ToJson() cannot resolve.
var connectionParameterSchema = new JsonSchema
{
Reference = generator.Generate(inputOutputAdapter.ConnectionParametersType, resolver)
};

var adapterSchema = JsonSchemaEx
.CreateObject($"Adapter_{inputOutputAdapter.Name}")
.AddProperty("Type", new JsonSchemaProperty { Pattern = $"^{inputOutputAdapter.Name}$" }, true)
.AddProperty("Connection", connectionParameterSchema, false);
.AddProperty("Type", new JsonSchemaProperty { Pattern = $"^{inputOutputAdapter.Name}$" }, true);

if (inputOutputAdapter.ConnectionParametersType != null)
{
var connectionParameterSchema = new JsonSchema
{
Reference = generator.Generate(inputOutputAdapter.ConnectionParametersType, resolver)
};
adapterSchema.AddProperty("Connection", connectionParameterSchema, false);
}

if (inputOutputAdapter.ProviderParametersType != null)
{
Expand Down Expand Up @@ -178,7 +222,7 @@ public JsonSchema GetConnectorsSchema()
var generator = new JsonSchemaGenerator(settings);
var resolver = new JsonSchemaResolver(docSchema, settings);

var connectionSchema = CreateAddonAdapters(docSchema, "Connection", providerProcessorAdapters, generator, resolver)
var connectionSchema = CreateAddonAdapters(docSchema, "Connection", _adapters, generator, resolver)
.AddAsDefinition(docSchema);

docSchema
Expand Down
49 changes: 19 additions & 30 deletions src/Paillave.Etl.Tests/CompositeFileValueProviderTests.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -66,53 +66,50 @@ public async IAsyncEnumerable<FileReference> EnumerateAsync(object? input = null
[Fact]
public void Adapter_ProcessorParametersType_IsNull()
{
var adapter = new CompositeFileValueProviderAdapter(new FileValueConnectors());
var adapter = new CompositeFileValueProviderAdapter();
Assert.Null(adapter.ProcessorParametersType);
}

[Fact]
public void Adapter_ProviderParametersType_IsCompositeProviderParameters()
{
var adapter = new CompositeFileValueProviderAdapter(new FileValueConnectors());
var adapter = new CompositeFileValueProviderAdapter();
Assert.Equal(typeof(CompositeProviderParameters), adapter.ProviderParametersType);
}

[Fact]
public void Adapter_Name_IsComposite()
{
var adapter = new CompositeFileValueProviderAdapter(new FileValueConnectors());
var adapter = new CompositeFileValueProviderAdapter();
Assert.Equal("Composite", adapter.Name);
}

// --- Provider created by adapter ---
// --- CompositeFileValueProvider behavior ---

[Fact]
public void Adapter_CreateProvider_Test_CallsTestOnAllUnderlyingProviders()
public void Provider_Test_CallsTestOnAllUnderlyingProviders()
{
var p1 = new FakeProvider("A", ProcessImpact.Light, "a1.csv");
var p2 = new FakeProvider("B", ProcessImpact.Light, "b1.csv");
var connectors = new FileValueConnectors();
connectors.Register(p1).Register(p2);

var adapter = new CompositeFileValueProviderAdapter(connectors);
var composite = adapter.CreateProvider("C", "Composite", "cnx", null, new CompositeProviderParameters { ProviderCodes = ["A", "B"] });

var composite = new CompositeFileValueProvider("C", connectors, "A", "B");
composite.Test();

Assert.True(p1.WasTested);
Assert.True(p2.WasTested);
}

[Fact]
public void Adapter_CreateProvider_Provide_AggregatesFilesInOrder()
public void Provider_Provide_AggregatesFilesInOrder()
{
var connectors = new FileValueConnectors();
connectors
.Register(new FakeProvider("A", ProcessImpact.Light, "a1.csv", "a2.csv"))
.Register(new FakeProvider("B", ProcessImpact.Light, "b1.csv"));

var adapter = new CompositeFileValueProviderAdapter(connectors);
var composite = adapter.CreateProvider("C", "Composite", "cnx", null, new CompositeProviderParameters { ProviderCodes = ["A", "B"] });
var composite = new CompositeFileValueProvider("C", connectors, "A", "B");

var collected = new List<string>();
composite.Provide(null, (fv, _) => collected.Add(fv.Name), CancellationToken.None);
Expand All @@ -121,15 +118,14 @@ public void Adapter_CreateProvider_Provide_AggregatesFilesInOrder()
}

[Fact]
public async Task Adapter_CreateProvider_ProvideAsync_AggregatesFilesInOrder()
public async Task Provider_ProvideAsync_AggregatesFilesInOrder()
{
var connectors = new FileValueConnectors();
connectors
.Register(new FakeProvider("A", ProcessImpact.Light, "a1.csv", "a2.csv"))
.Register(new FakeProvider("B", ProcessImpact.Light, "b1.csv"));

var adapter = new CompositeFileValueProviderAdapter(connectors);
var composite = adapter.CreateProvider("C", "Composite", "cnx", null, new CompositeProviderParameters { ProviderCodes = ["A", "B"] });
var composite = new CompositeFileValueProvider("C", connectors, "A", "B");

var collected = new List<string>();
await foreach (var fv in composite.ProvideAsync())
Expand All @@ -139,29 +135,27 @@ public async Task Adapter_CreateProvider_ProvideAsync_AggregatesFilesInOrder()
}

[Fact]
public void Adapter_CreateProvider_PerformanceImpact_IsMaxOfUnderlying()
public void Provider_PerformanceImpact_IsMaxOfUnderlying()
{
var connectors = new FileValueConnectors();
connectors
.Register(new FakeProvider("A", ProcessImpact.Light))
.Register(new FakeProvider("B", ProcessImpact.Heavy));

var adapter = new CompositeFileValueProviderAdapter(connectors);
var composite = adapter.CreateProvider("C", "Composite", "cnx", null, new CompositeProviderParameters { ProviderCodes = ["A", "B"] });
var composite = new CompositeFileValueProvider("C", connectors, "A", "B");

Assert.Equal(ProcessImpact.Heavy, composite.PerformanceImpact);
}

[Fact]
public void Adapter_CreateProvider_FileReferences_PointToUnderlyingProviderCodes()
public void Provider_FileReferences_PointToUnderlyingProviderCodes()
{
var connectors = new FileValueConnectors();
connectors
.Register(new FakeProvider("A", ProcessImpact.Light, "a1.csv"))
.Register(new FakeProvider("B", ProcessImpact.Light, "b1.csv"));

var adapter = new CompositeFileValueProviderAdapter(connectors);
var composite = adapter.CreateProvider("C", "Composite", "cnx", null, new CompositeProviderParameters { ProviderCodes = ["A", "B"] });
var composite = new CompositeFileValueProvider("C", connectors, "A", "B");

var refs = new List<FileReference>();
composite.Provide(null, (_, fr) => refs.Add(fr), CancellationToken.None);
Expand All @@ -171,7 +165,7 @@ public void Adapter_CreateProvider_FileReferences_PointToUnderlyingProviderCodes
}

[Fact]
public void Adapter_CreateProvider_Provide_StopsOnCancellation()
public void Provider_Provide_StopsOnCancellation()
{
var cts = new CancellationTokenSource();
cts.Cancel();
Expand All @@ -181,27 +175,22 @@ public void Adapter_CreateProvider_Provide_StopsOnCancellation()
.Register(new FakeProvider("A", ProcessImpact.Light, "a1.csv"))
.Register(new FakeProvider("B", ProcessImpact.Light, "b1.csv"));

var adapter = new CompositeFileValueProviderAdapter(connectors);
var composite = adapter.CreateProvider("C", "Composite", "cnx", null, new CompositeProviderParameters { ProviderCodes = ["A", "B"] });
var composite = new CompositeFileValueProvider("C", connectors, "A", "B");

var collected = new List<string>();
composite.Provide(null, (fv, _) => collected.Add(fv.Name), cts.Token);

Assert.Empty(collected);
}

// --- Lazy resolution: providers can be registered after adapter creation ---
// --- Lazy resolution: underlying providers registered after composite is created ---

[Fact]
public void Adapter_CreateProvider_ResolvesProvidersLazily()
public void Provider_ResolvesProvidersLazily()
{
var connectors = new FileValueConnectors();
var adapter = new CompositeFileValueProviderAdapter(connectors);

// composite created before underlying providers are registered
var composite = adapter.CreateProvider("C", "Composite", "cnx", null, new CompositeProviderParameters { ProviderCodes = ["A", "B"] });
var composite = new CompositeFileValueProvider("C", connectors, "A", "B");

// register underlying providers afterwards
connectors
.Register(new FakeProvider("A", ProcessImpact.Light, "a1.csv"))
.Register(new FakeProvider("B", ProcessImpact.Light, "b1.csv"));
Expand Down
28 changes: 28 additions & 0 deletions src/Paillave.Etl.sln
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -61,6 +61,10 @@ Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Paillave.Etl.JsonFile", "Pa
EndProject
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Paillave.Etl.GraphApi.Tests", "Paillave.Etl.GraphApi.Tests\Paillave.Etl.GraphApi.Tests.csproj", "{82E4637C-A0AB-4AFB-A6BC-FEC3892CB7C9}"
EndProject
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Paillave.Etl.FromConfigurationConnectors.Tests", "Paillave.Etl.FromConfigurationConnectors.Tests\Paillave.Etl.FromConfigurationConnectors.Tests.csproj", "{449ADA7B-C1FA-4D4D-9C0B-A3B0973EC141}"
EndProject
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Paillave.Etl.Tests", "Paillave.Etl.Tests\Paillave.Etl.Tests.csproj", "{03F552CF-D994-4580-9166-C53BF5692F9A}"
EndProject
Global
GlobalSection(SolutionConfigurationPlatforms) = preSolution
Debug|Any CPU = Debug|Any CPU
Expand Down Expand Up @@ -419,6 +423,30 @@ Global
{82E4637C-A0AB-4AFB-A6BC-FEC3892CB7C9}.Release|x64.Build.0 = Release|Any CPU
{82E4637C-A0AB-4AFB-A6BC-FEC3892CB7C9}.Release|x86.ActiveCfg = Release|Any CPU
{82E4637C-A0AB-4AFB-A6BC-FEC3892CB7C9}.Release|x86.Build.0 = Release|Any CPU
{449ADA7B-C1FA-4D4D-9C0B-A3B0973EC141}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
{449ADA7B-C1FA-4D4D-9C0B-A3B0973EC141}.Debug|Any CPU.Build.0 = Debug|Any CPU
{449ADA7B-C1FA-4D4D-9C0B-A3B0973EC141}.Debug|x64.ActiveCfg = Debug|Any CPU
{449ADA7B-C1FA-4D4D-9C0B-A3B0973EC141}.Debug|x64.Build.0 = Debug|Any CPU
{449ADA7B-C1FA-4D4D-9C0B-A3B0973EC141}.Debug|x86.ActiveCfg = Debug|Any CPU
{449ADA7B-C1FA-4D4D-9C0B-A3B0973EC141}.Debug|x86.Build.0 = Debug|Any CPU
{449ADA7B-C1FA-4D4D-9C0B-A3B0973EC141}.Release|Any CPU.ActiveCfg = Release|Any CPU
{449ADA7B-C1FA-4D4D-9C0B-A3B0973EC141}.Release|Any CPU.Build.0 = Release|Any CPU
{449ADA7B-C1FA-4D4D-9C0B-A3B0973EC141}.Release|x64.ActiveCfg = Release|Any CPU
{449ADA7B-C1FA-4D4D-9C0B-A3B0973EC141}.Release|x64.Build.0 = Release|Any CPU
{449ADA7B-C1FA-4D4D-9C0B-A3B0973EC141}.Release|x86.ActiveCfg = Release|Any CPU
{449ADA7B-C1FA-4D4D-9C0B-A3B0973EC141}.Release|x86.Build.0 = Release|Any CPU
{03F552CF-D994-4580-9166-C53BF5692F9A}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
{03F552CF-D994-4580-9166-C53BF5692F9A}.Debug|Any CPU.Build.0 = Debug|Any CPU
{03F552CF-D994-4580-9166-C53BF5692F9A}.Debug|x64.ActiveCfg = Debug|Any CPU
{03F552CF-D994-4580-9166-C53BF5692F9A}.Debug|x64.Build.0 = Debug|Any CPU
{03F552CF-D994-4580-9166-C53BF5692F9A}.Debug|x86.ActiveCfg = Debug|Any CPU
{03F552CF-D994-4580-9166-C53BF5692F9A}.Debug|x86.Build.0 = Debug|Any CPU
{03F552CF-D994-4580-9166-C53BF5692F9A}.Release|Any CPU.ActiveCfg = Release|Any CPU
{03F552CF-D994-4580-9166-C53BF5692F9A}.Release|Any CPU.Build.0 = Release|Any CPU
{03F552CF-D994-4580-9166-C53BF5692F9A}.Release|x64.ActiveCfg = Release|Any CPU
{03F552CF-D994-4580-9166-C53BF5692F9A}.Release|x64.Build.0 = Release|Any CPU
{03F552CF-D994-4580-9166-C53BF5692F9A}.Release|x86.ActiveCfg = Release|Any CPU
{03F552CF-D994-4580-9166-C53BF5692F9A}.Release|x86.Build.0 = Release|Any CPU
EndGlobalSection
GlobalSection(SolutionProperties) = preSolution
HideSolutionNode = FALSE
Expand Down
Loading
Loading