Drasi Reactions SDK

import { DrasiReaction, ChangeEvent } from '@drasi/reaction-sdk';

let myReaction = new DrasiReaction(async (event: ChangeEvent) => {
    console.log(`Received change sequence: ${event.sequence} for query ${event.queryId}`);
    // 处理添加的结果
    for (let added of event.addedResults) {
        console.log(`Added result: ${JSON.stringify(added)}`);
    }
    // 处理删除的结果
    for (let deleted of event.deletedResults) {
        console.log(`Removed result: ${JSON.stringify(deleted)}`);
    }
    // 处理更新的结果
    for (let updated of event.updatedResults) {
        console.log(`Updated result - before: ${JSON.stringify(updated.before)}, after: ${JSON.stringify(updated.after)}`);
    }
});

myReaction.start();

from drasi.reaction.models.ChangeEvent import ChangeEvent
from drasi.reaction.sdk import DrasiReaction

async def change_event(event: ChangeEvent, query_configs: dict[str, Any] | None = None):
    print(f"Received change sequence {event.sequence} for query {event.queryId}")
    if event.addedResults:
        print(f"Added result: {event.addedResults}")
    if event.deletedResults:
        print(f"Removed result: {event.deletedResults}")
    if event.updatedResults:
        print(f"Updated result - before: {event.updatedResults[0].before}, after {event.updatedResults[0].after}")

reaction = DrasiReaction(on_change_event=change_event)
reaction.start()

var reaction = new ReactionBuilder()
    .UseChangeEventHandler(async (evt, queryConfig) => {
        Console.WriteLine($"Received change event from query {evt.QueryId} sequence {evt.Sequence}");
        
        foreach (var item in evt.AddedResults)
            Console.WriteLine($"Added result: {item}");

        foreach (var item in evt.UpdatedResults)
            Console.WriteLine($"Updated result, before {item.Before}, after {item.After}");

        foreach (var item in evt.DeletedResults)
            Console.WriteLine($"Deleted result: {item}");
    })
    .Build();

await reaction.StartAsync();

class MyQueryConfig
{
    [JsonPropertyName("greeting")]
    public string? Greeting { get; set; }
}

class MyService
{
    private readonly string _connectionString;

    public MyService(IConfiguration configuration)
    {
        _connectionString = configuration["MyConnectionString"];
    }

    public void DoSomething()
    {
        Console.WriteLine("Doing something");
    }
}

class MyChangeHandler : IChangeEventHandler<MyQueryConfig>
{   
    private readonly MyService _service;
    private readonly ILogger<MyChangeHandler> _logger;

    public MyChangeHandler(MyService service, ILogger<MyChangeHandler> logger)
    {
        _service = service;
        _logger = logger;
    }

    public async Task HandleChange(ChangeEvent evt, MyQueryConfig? queryConfig)
    {
        _logger.LogInformation($"Received change event from query {evt.QueryId} sequence {evt.Sequence}. Query greeting is {queryConfig?.Greeting}");
        _logger.LogInformation($"Full event: {evt.ToJson()}");
        _service.DoSomething();
    }
}

// 控制事件处理器（MyControlSignalHandler） 
class MyControlSignalHandler : IControlEventHandler<MyQueryConfig>
{
    private readonly ILogger<MyControlSignalHandler> _logger;

    public MyControlSignalHandler(ILogger<MyControlSignalHandler> logger)
    {
        _logger = logger;
    }

    public async Task HandleControlSignal(ControlEvent evt, MyQueryConfig? queryConfig)
    {
        _logger.LogWarning($"Received control signal: {evt.ControlSignal?.Kind} for query {evt.QueryId}. Query greeting is {queryConfig?.Greeting}");
    }
}

var reaction = new ReactionBuilder<MyQueryConfig>()
    .UseChangeEventHandler<MyChangeHandler>()               // Use your custom change handler
    .UseControlEventHandler<MyControlSignalHandler>()       // Use your custom control signal handler
    .UseYamlQueryConfig()                                   // Parse the per query configuration from Yaml
    .ConfigureServices((services) =>                        // Register your own services
    {
        services.AddSingleton<MyService>();
    })
    .Build();

// Start the reaction
await reaction.StartAsync();
注册你的ReactionProvider创建好Reaction后，需要通过ReactionProvider将它注册到Drasi系统中。 下面是一个ReactionProvider配置：apiVersion: v1

kind: ReactionProvider

name: Advanced

spec:
   services:
     reaction:
       image: reaction-advanced
       externalImage: true
   config_schema:
     type: object
     properties: 
       MyConnectionString:
         type: string
     required:
       - MyConnectionString
Reaction使用示例

kind: Reaction

apiVersion: v1

name: test-advanced

spec:
   kind: Advanced
   properties:
     MyConnectionString: "some connection string"
   queries:
     query1: |
       greeting: "Hello, World!"
     query2: |
       greeting: "Howdy!"

# 先部署ReactionProvider
drasi apply -f reaction-provider.yaml

# 然后部署Reaction
drasi apply -f reaction.yaml