Sweep several source folders in one extractor

Change an extractor that sweeps one source folder so it sweeps several, each through the pipeline you route it to.

Before you begin

• Read Systems and topology for what a Port is
• Have an extractor and the system host that declares it. The examples use erp-extractor from the distribution tutorial, which sweeps the Port erp-extractor-source, and add a Port erp-extractor-cancellations
• Know whether the new folder carries different data (another kind of record, or the same record meaning something else) or the same data delivered in a second place. Step 5 differs between the two.

Each source Port is handled by exactly one route, and each route has its own pipeline. The framework sweeps the Ports one after another, in the order you declare them.

Steps

1. Update the framework packages

Routes need Intropy.Framework 2.0.0-beta.1 or later. In src/ and both test projects, set every Intropy.Framework.* package reference to that version:

src/ErpExtractor.csproj

<PackageReference Include="Intropy.Framework.Adapters" Version="2.0.0-beta.1" />
<PackageReference Include="Intropy.Framework.Blocks" Version="2.0.0-beta.1" />
<PackageReference Include="Intropy.Framework.Hosting" Version="2.0.0-beta.1" />

An extractor scaffolded from intropy-templates 2.0.0 or later already references it.

2. Declare the Port in the System

In the system host, declare the Port, read it from the extractor, and resolve it to a local folder for development:

order-notifications/Ports.cs

public static readonly PortRef ErpExtractorSource = PortRef.Define("erp-extractor-source");

public static readonly PortRef ErpExtractorCancellations = PortRef.Define("erp-extractor-cancellations");

order-notifications/OrderNotificationsSystem.cs: Define

builder.AddExtractor("erp-extractor")
    .From(Ports.ErpExtractorSource)
    .From(Ports.ErpExtractorCancellations)
    .Publishes(Messages.OrderPlaced)

order-notifications/OrderNotificationsDevelopment.cs: Define

development.Files(Ports.ErpExtractorSource).RootPath("./test/erp-extractor-source");
development.Files(Ports.ErpExtractorCancellations).RootPath("./test/erp-extractor-cancellations");

Step 4 publishes the new Port’s files as order-cancelled. If the System does not declare that message yet, declare it in Messages.cs and add .Publishes(Messages.OrderCancelled) next to the other Publishes call, as in Route within a Component.

3. Name the Port in the extractor and choose its adapter

Add the Port’s name below SourcePort in src/Constants.cs. The name must match the PortRef exactly. The cancellations are published as order-cancelled, so add that type below EventType:

src/Constants.cs

public const string CancellationsPort = "erp-extractor-cancellations";
public const string CancelledEventType = "order-cancelled";

Each Port’s file adapter is chosen by its Kind. Select the local folder adapter for development:

src/appsettings.Development.json

{
  "Ports": {
    "erp-extractor-source": { "Kind": "Local" },
    "erp-extractor-cancellations": { "Kind": "Local" }
  }
}

A deployment sets Ports__erp-extractor-cancellations__Kind to match the binding it renders for the Port, as it does for the first one.

4. Register the adapters and declare the routes

In src/Composition/Composition.cs, register one file adapter per Port, keyed by its name. On a scaffold from before intropy-templates 2.0.0, this replaces the AddSourcePort call:

src/Composition/Composition.cs: ConfigureServices

services.AddFileAdapter(Constants.SourcePort, configuration.GetSection(
    $"{PortServiceCollectionExtensions.PortsSectionName}:{Constants.SourcePort}"));
services.AddFileAdapter(Constants.CancellationsPort, configuration.GetSection(
    $"{PortServiceCollectionExtensions.PortsSectionName}:{Constants.CancellationsPort}"));

Do not keep AddSourcePort next to the routed registration. sources declares the Ports, and registering one twice fails at startup.

Then replace the AddExtractor call. sources lists the Ports to sweep, and routes sends each Port’s files through a pipeline. The original Port keeps the scaffold’s ConfigurePipeline; the new Port’s pipeline is declared inline:

src/Composition/Composition.cs: ConfigureServices

services.AddExtractor(
    sources => sources
        .Port(Constants.SourcePort)
        .Port(Constants.CancellationsPort),
    routes => routes
        .On<SourceOrderPlaced, OrderPlaced>([Constants.SourcePort], ConfigurePipeline)
        .On<SourceOrderPlaced, OrderPlaced>([Constants.CancellationsPort],
            (builder, provider) => builder
                .WithDeserializer(new Deserializer())
                .WithValidator(new Validator())
                .WithIdempotency(
                    idExtractor: (SourceOrderPlaced order, Context _) => order.OrderId,
                    dateExtractor: (SourceOrderPlaced order, Context _) => order.OrderDate)
                .WithTransformer(new Transformer())
                .WithSerializer(new CloudEventSerializeStep<OrderPlaced, Context>(
                    subject: order => order.OrderId,
                    time: order => order.OrderDate,
                    type: (_, _) => Constants.CancelledEventType))
                .WithSender(provider.GetRequiredService<SendStep<Context>>())
                .WithBusinessIncidents(
                    messageIdExtractor: ctx => ctx.Metadata.GetValueOrDefault(Constants.ContextKeyOrderId,
                        ctx.Metadata.GetValueOrDefault(Constants.ContextKeyFileName, "unknown")),
                    subjectExtractor: ctx => ctx.Metadata.GetValueOrDefault(Constants.ContextKeyOrderId,
                        ctx.Metadata.GetValueOrDefault(Constants.ContextKeyFileName, "unknown")))));

Port takes an optional second argument for what happens to a file once it is handled. The default is FileCompletion.Delete; FileCompletion.Archive(basePath) copies the file to basePath first.

5. Choose the new route’s pipeline

The type arguments of On are the route’s deserialized input and published payload. Its first argument lists the Ports whose files the route handles. The second configures the route’s pipeline: a method such as ConfigurePipeline, or a lambda over the builder and the file’s service provider.

Different data. Give the route its own pipeline, as in step 4. There, both Ports carry the same record and the new route reuses the scaffold’s steps, but publishes the cancellations as order-cancelled. The publisher stamps Constants.EventType on every event; the type: function on the serializer overrides it. A Port carrying another kind of record gets its own type arguments, and its own deserializer, validator and transformer.

The same data in several places. Use one route that names every Port in place of the two in step 4, and share its idempotency records across them. Here OtherSourcePort stands for the second Port:

src/Composition/Composition.cs: ConfigureServices

routes => routes.On<SourceOrderPlaced, OrderPlaced>(
    [Constants.SourcePort, Constants.OtherSourcePort], ConfigurePipeline)

src/Composition/Composition.cs: ConfigurePipeline

.WithIdempotency(
    idExtractor: (SourceOrderPlaced order, Context _) =>
        order.OrderId,
    dateExtractor: (SourceOrderPlaced order, Context _) =>
        order.OrderDate,
    scope: IdempotencyScope.Entity)

By default (IdempotencyScope.Route) each Port keeps its own records, keyed {port}:{id}, so a record from one Port never makes a file on another look like a duplicate. IdempotencyScope.Entity keys on the id alone, so an order seen on one Port is a duplicate on every other.

Existing idempotency records

An extractor without routes keys its records on the id alone. After this change, the original Port’s route keys them on {port}:{id}, so the first delivery of each already-recorded id is treated as unseen and published again. To keep the existing records in effect, pass scope: IdempotencyScope.Entity on the original Port’s route.

6. Fake the new Port in the integration tests

The integration suite swaps every external for a fake. Fake the new Port’s adapter next to the first one, or the sweep tests reach for a real Dapr binding:

test/ErpExtractor.Test.Integration/FakeEdges.cs

public InMemoryFileAdapter Files { get; } = new();
public InMemoryFileAdapter CancellationFiles { get; } = new();

// in CreateServices:
var services = Composition.Composition.ConfigureServices(new ServiceCollection(), "Development");
services.RemoveAllKeyed<IFileAdapter>(Constants.SourcePort);
services.AddKeyedSingleton<IFileAdapter>(Constants.SourcePort, (_, _) => Files);
services.RemoveAllKeyed<IFileAdapter>(Constants.CancellationsPort);
services.AddKeyedSingleton<IFileAdapter>(Constants.CancellationsPort, (_, _) => CancellationFiles);

With routes, the pipeline is no longer registered in the service provider, so ResolvePipeline builds it from ConfigurePipeline instead. On a scaffold from before intropy-templates 2.0.0, replace it:

test/ErpExtractor.Test.Integration/FakeEdges.cs: ResolvePipeline

public static Extractor<SourceOrderPlaced, OrderPlaced, Context> ResolvePipeline(IServiceProvider provider)
{
    var scope = provider.CreateScope().ServiceProvider;
    var builder = ExtractorBuilder<SourceOrderPlaced, OrderPlaced, Context>.Create(
        $"{Constants.ComponentName}.Process.{Constants.SourcePort}", scope);
    return Composition.Composition.ConfigurePipeline(builder, scope).Build();
}

See Test a Component for adding tests that cover the new route.

Verify

From the extractor, run both suites:

task test

From the system host, validate the topology and render the deployment files:

task check
task generate
ls out/components

out/components/ holds one binding per source Port, here erp-extractor-source.yaml and erp-extractor-cancellations.yaml.

Then put a file in each folder and start the host with task run. Both folders are empty once the extractor has run. In the Aspire dashboard, each Port’s sweep is a trace named after the Port, such as process erp-extractor-cancellations. Its pipeline span is named after the route, Pipeline.erp-extractor.Process.erp-extractor-cancellations, with pipeline.name set to erp-extractor.Process.erp-extractor-cancellations. Every published event carries the Port in the intropysourceport CloudEvent extension.

Symptoms and where to go back to:

• Startup fails with Ports:erp-extractor-cancellations (key 'erp-extractor-cancellations'): Not configured. Set Kind to Local, Sftp or AzureBlob — the Port has no Kind. Revisit step 3.
• Startup fails with Extractor composition failed: no route handles source port '…', … is handled by more than one route, or a route names source port '…', which is not declared — sources and routes disagree. Every declared Port needs exactly one route. Revisit step 4.
• Sweep tests fail with Binding operation failed: the Dapr endpoint indicated a failure — the new Port’s adapter is not faked. Revisit step 6.
• Already-processed orders are published again after the change — the records are now keyed per Port. See the caution in step 5.