Custom Actions

Consume Custom Action Messages

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.

To ensure the Custom Action being sent is handled properly, the app.yaml outputs needs to be declared:

Note

You can choose any name for the type.

This is how the Custom Action Manager chooses which Consumer Application (Executor) will receive the Custom Action object.

Warning

For each Type declared in any Publisher App, only ONE Consumer Application (Executor) can be deployed on the Kelvin Platform to receive that Type.

Multiple Consumer Application (Executor) with the same TYPE is not supported.

custom_actions:
  inputs:
    - type: custom-action-name

The Custom Action Object in the main.py script supports the following attributes :

Attribute	Required	Default Value	Description
resource	required	N/A	The KRNAsset that this Custom Action is meant for.
type	required	N/A	The name of Custom Action.
title	required	N/A	Title of the Custom Action
description	required	N/A	Description details of the Custom Action
expiration_date	required	N/A	Absolute datetime or a timedelta (from now) when the Control Change will expire.
payload	required	N/A	The custom information of the Custom Action that will be required by the Consumer Application
trace_id	optional	N/A	A custom id for tracking the Custom Action status

The Custom Action Result Object supports the following attributes :

Attribute	Required	Default Value	Description
success	required	N/A	Whether the Custom Actions were completed successfully. Boolean True or False.
message	optional	N/A	Any message to return to the Publishing Application.
metadata	optional	N/A	Any additional metadata that needs to be returned to the Publishing Application
action_id	optional	N/A	The id of the Custom Action Object
resource	optional	N/A	The KRNAsset that this Custom Action is meant for.

Example

In this example we will create a Consumer Application (Executor) that will;

1. Listen and receive any new Custom Action objects with the type email.
2. Extract the relevant email information from the payload
3. Connect to an SMTP and send the email (This function is defined but not fully coded)
4. Return the status of the Custom Action (True or False)

Check out the Produce Custom Actions documentation here to see how to send this Custom Action from the Publisher Application.

app.yaml

spec_version: 5.0.0
type: app            # Any app type can handle and/or publish custom actions.

name: hello-app
title: Hello App
description: Lorem ipsum dolor sit amet, consectetur adipiscing elit
version: 1.0.0

custom_actions:
  inputs:
    - type: email

  ...

Consumer (Executor) Application

import asyncio

from kelvin.application import KelvinApp
from kelvin.message import CustomAction, CustomActionResult

app = None

async def send_mail(recipient, subject, body):
    # Specific code for sending an email.
    print (f'Sending email to {recipient}')

async def on_custom_action(action: CustomAction) -> None:

    # React on right type of action (if Consumer handles multiple Custom Action Types)
    if action.type == "email":
        try:
            await send_mail(
                action.payload.get("recipient"),
                action.payload.get("subject"),
                action.payload.get("body")
            )

            print("Received Custom Action: ", action)

            # Send confirmation of results of Action to Producer Application
            await app.publish(CustomActionResult(
                success=True,
                action_id=action._msg.id,
                resource=action.resource,
                metadata={}
            ))

            #
            # or
            #

            # Alternative method to send confirmation of results of Action to Producer Application
            await app.publish(CustomActionResult(
                success=True,
                action_id=action._msg.id,
                resource=action.resource,
                metadata={}
            ))

        except Exception as e:

            # Send confirmation of error of Action to Producer Application
            await app.publish(action.result(success=False, message=str(e)))


async def main() -> None:

    global app

    app = KelvinApp()
    app.on_custom_action = on_custom_action

    await app.connect()

    while True:
        await asyncio.sleep(1)


if __name__ == "__main__":
    asyncio.run(main())