Custom Actions

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

Note

To understand the purpose of Custom Actions or view the overall structure of how they work, check out the documentation in the overview page here.

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.

Note

This Application is only for local testing and does not form part of your code to upload to the Kelvin Platform.

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.

# 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).

Note

In this example you would use the code from the Consumer Application example of the Custom Actions.

First in the terminal run the Generator Test Application.

kelvin app test generator --entrypoint mygenerator.py:CustomActionGenerator

Then run your Consumer Application (Executor).

Warning

This is to test the Custom Action consumer application NOT the Custom Action publisher application.

The test generator is simulating a Custom Action publisher application.

Success

You can add Recommendation, Control Changes, etc. in your Consumer application. They will be received by the test generator and displayed in the terminal like a normal test.

python main.py

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