Custom Actions¶

In the Kelvin SDK there is a kelvin-publisher test tool to help you ingest data to the inputs and debug the outputs to test the performance of your Kelvin SmartApp™.

It is a CLI utility for exercising your application during development. It is not imported into your code and is used only as a standalone testing tool.

You can use Custom Actions to enable communication between two Applications on either the same cluster or across different clusters.

Once you have created a Consumer Application (Executor), you may want to test it first locally on your development computer before uploading it to the Kelvin Platform.

Generator Application¶

To test your Consumer Application (Executor), you will need to first create a local Generator Application.

In our documentation examples in the Produce - Custom Actions and Consume - Custom Actions sections, we showed you the code to create Publisher and Consumer (Executor) Applications.

Here is a typical Generator Application you would create to send Custom Actions with the type email.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33

# mygenerator.py
import asyncio
from datetime import timedelta
from typing import AsyncGenerator

from kelvin.message import Message, CustomAction
from kelvin.publisher import DataGenerator
from kelvin.krn import KRNAsset

class CustomActionGenerator(DataGenerator):
    def __init__(self) -> None:
        print("Hello from MyGenerator")

    async def run(self) -> AsyncGenerator[Message, None]:
        print("Running MyGenerator")

        for i in range(10):
            print("Create and Send Custom Action message")
            msg = CustomAction(
                    resource=KRNAsset("test-asset-1"),
                    expiration_date=timedelta(seconds=30),
                    type="email",
                    title="Send Emails",
                    payload={
                    "recipient": "operations@example.com",
                    "subject": "Recommendation to reduce speed",
                    "body": "This is the email body",
                }
            )
            yield msg

            print ("Sent custom action message: ", msg)
            await asyncio.sleep(3)

Test Consumer Application¶

With your Generator Application ready, you can now test your Consumer Application (Executor).

First in the terminal run the Generator Test Application.

1

kelvin-publisher generator --entrypoint mygenerator.py:CustomActionGenerator

Then run your Consumer Application (Executor).

1

python main.py

The Custom Action Objects created in the Generator Test Application will now be sent to the Consumer Application (Executor).