Load a topology from configuration

Build and start an engine from JSON or IConfiguration.

  3 minute read  

Engine.FromConfigAsync builds an engine from a document of sources, queries, and reactions, then starts it. Missing id defaults to drasi.

This matches Python and Node fromConfig. Use CreateAsync when you want to add components in code after construction.

JSON document

var config = JsonNode.Parse("""
{
  "id": "demo",
  "sources": [{ "id": "orders", "autoStart": true }],
  "queries": [{
    "id": "open",
    "query": "MATCH (o:Order) WHERE o.status = 'open' RETURN o.id AS id, o.total AS total",
    "sources": ["orders"],
    "language": "cypher"
  }]
}
""")!;

await using var drasi = await Engine.FromConfigAsync(config);

Document keys:

KeyMeaning
idEngine id (default drasi)
secrets, stateStore, indexStore, walProvider, identitySame as EngineOptions. Merged from the options argument when the document omits them.
pluginsDirDirectory passed to LoadPluginsAsync before start
sourcesIn-process (id only) or plugin (kind, id, config, autoStart, bootstrap)
queriesid, query, sources, optional language, joins
reactionsPlugin reactions (kind, id, queries, config)

C# callback reactions cannot be expressed in JSON. Add them after FromConfigAsync, or use AddDrasi with a DrasiBuilder callback.

appsettings.json

{
  "Drasi": {
    "id": "orders-app",
    "sources": [{ "id": "orders" }],
    "queries": [{
      "id": "open",
      "query": "MATCH (o:Order) WHERE o.status = 'open' RETURN o.id AS id, o.total AS total",
      "sources": ["orders"]
    }]
  }
}

await using var drasi = await Engine.FromConfigAsync(
    configuration.GetSection("Drasi"));

In a hosted app:

builder.Services.AddDrasi(builder.Configuration.GetSection("Drasi"), drasi =>
{
    drasi.AddReaction("watch", ["open"], evt =>
    {
        foreach (var diff in evt.Results)
        {
            Console.WriteLine(diff.Type);
        }
    });
});

The configuration section supplies sources and queries (in-process or plugin). The DrasiBuilder callback supplies in-process reactions and seeds.

Engine options

Pass EngineOptions to CreateAsync or FromConfigAsync:

await Engine.CreateAsync("orders-app", new EngineOptions
{
    LoggerFactory = logs,
    PluginsDir = "./plugins",
    Secrets = new Dictionary<string, string> { ["db"] = password },
    StateStore = new StateStoreOptions { Path = "./state.redb" },
    IndexStore = new IndexStoreOptions { Path = "./index" },
    WalProvider = new WalProviderOptions { Path = "./wal" },
    Identity = new IdentityOptions { Kind = "token", Token = token },
});

The equivalent configuration object is:

{
  "walProvider": {
    "kind": "redb",
    "path": "./wal"
  }
}

RocksDB holds a process-exclusive lock on IndexStore.Path. Only one engine may use a given path at a time. ShutdownAsync releases that lock.

WalProvider.Path is a directory. The default redb provider creates one WAL file per transient source that explicitly enables durability. Configuring the provider does not enable durability for every source. ShutdownAsync releases the WAL file handles so another engine can reopen the same directory without disposing the first Engine object.

The WAL provider records transient-source events for replay. It is not a PostgreSQL replication slot and does not resume a PostgreSQL CDC source’s logical replication slot.