orbitalis 1.1.4
Event-driven micro-kernel for Python
Requires Python: >=3.12

# Orbitalis

**Orbitalis** is a **distributed**, **event-based micro-kernel** library for Python designed to simplify the construction of *loosely coupled, scalable, and modular systems*.

At its core, Orbitalis provides a lightweight yet powerful foundation for building event-driven applications using an event bus mechanism. It integrates natively with [Busline](https://github.com/orbitalis-framework/py-busline), a flexible and efficient event bus library that handles message distribution across system components, whether local or distributed.

If you don't want to read the user guide, you can skip to [Practical Example](#practical-example).

```mermaid
flowchart LR
      C1 --- EB1
      C1 --- EB2

      EB1 --- P3
      EB1 --- P1
      
      EB2 --- P2
      EB2 --- P4

      C1@{ shape: circle, label: "Core" }
      P1@{ shape: stadium, label: "Plugin" }
      P2@{ shape: stadium, label: "Plugin" }
      P3@{ shape: stadium, label: "Plugin" }
      P4@{ shape: stadium, label: "Plugin" }

      EB1@{ shape: das, label: "MQTT Broker" }
      EB2@{ shape: das, label: "Local EventBus" }
```

## User Guide

In this section we will explain how to use Orbitalis. If you want to know more details about this library, please read the [Advance Guide](#advance-guide). 

Every time-based values are expressed in seconds, because `asyncio.sleep` is used.

### Installation

```
pip install orbitalis
```

### Preliminary: Busline

If you know what [Busline](https://github.com/orbitalis-framework/py-busline) is, you can skip this section.

[Busline](https://github.com/orbitalis-framework/py-busline) is an agnostic asynchronous pub/sub library, which represents a good starting point to implement a wide set of application thanks to its pub/sub abstract layer.

It natively supports MQTT and Local event bus and you should know how to create pub/sub clients in order to allow your components to share messages.

We advice to read [Busline](https://github.com/orbitalis-framework/py-busline) documentation before continue.

### Overview

Orbitalis follows the **micro-kernel** (a.k.a. [**Plugin**](#plugin)) architectural pattern, where a minimal [**Core**](#core) (the kernel) handles only the essential system functions—like loading and managing modules, and routing events—while all additional features are implemented as independent [**Plugins**](#plugin) that interact via the event bus.

This design brings several advantages:

- **Separation of Concerns**: Each module is self-contained and focused on a specific task or domain
- **Extensibility**: New features can be added without modifying the core, reducing the risk of breaking existing functionality
- **Flexibility**: Modules can be enabled, disabled, or replaced at runtime
- **Maintainability**: Smaller code units are easier to test, debug, and reuse
- **Scalability**: Modules can be distributed across processes or machines and communicate via events


```mermaid
flowchart LR
      C1 --> P1
      C1 <--> P3

      C2 <--> P2
      P3 <--> C2
      P4 --> C2


      C1@{ shape: circle, label: "Core 1" }
      C2@{ shape: circle, label: "Core 2" }
      P1@{ shape: stadium, label: "Plugin 1" }
      P2@{ shape: stadium, label: "Plugin 2" }
      P3@{ shape: stadium, label: "Plugin 3" }
      P4@{ shape: stadium, label: "Plugin 4" }
```

Orbitalis allows to have more Cores and more Plugins, even in *different domains* thanks to [Busline](https://github.com/orbitalis-framework/py-busline) capabilities, which can be instantiate **at runtime** thanks to a powerful [handshake mechanism](#handshake). In fact, for example, we can have some plugins connected with MQTT and other with a Local event bus at the same time.
This allows you a powerful management of your components.

In other words, Orbitalis allows you to start cores and plugins, connect them together and execute plugin operations.
Cores and plugins can be started in any time and connections are created based on pre-defined policies.

Messages used by Orbitalis are **Avro messages** because we need input and output schemas.

```mermaid
flowchart LR
      C1 <--> EB1
      C1 <--> EB2

      C2 <--> EB1
      C2 <--> EB2
    
      EB1 <--> P3
      EB1 <--> P1
      
      EB2 <--> P2
      EB2 <--> P4

      C1@{ shape: circle, label: "Core 1" }
      C2@{ shape: circle, label: "Core 2" }
      P1@{ shape: stadium, label: "Plugin 1" }
      P2@{ shape: stadium, label: "Plugin 2" }
      P3@{ shape: stadium, label: "Plugin 3" }
      P4@{ shape: stadium, label: "Plugin 4" }

      EB1@{ shape: das, label: "MQTT Broker" }
      EB2@{ shape: das, label: "Local EventBus" }
```


### Orbiter

`Orbiter` is the base class which provides common capabilities to components. 
Therefore, you can use the provided methods both in [Cores](#core) and [Plugins](#plugin).

It manages _[pending requests](#pending-requests), [connections](#connections), [keepalive](#keepalive) and [connection close procedure](#connection-close-procedure)_. In addiction, it has useful _shared methods_ and main _loop_.

```mermaid
classDiagram
    Orbiter <|-- Core
    Orbiter <|-- Plugin
```

Main public attributes:

- `identifier` is the _unique_ identifier
- `eventbus_client` is a [Busline](https://github.com/orbitalis-framework/py-busline) client, used to send events
- `discover_topic` specifies topic used to send discover messages
- `raise_exceptions` if `True`, exceptions are raised, otherwise they are managed by try/catch
- `loop_interval` specifies how often the loop iterations are called (it is a minimum value, because maximum depends on weight of operations in loop)
- `with_loop` set to `False` if you don't want the loop (care about _what_ [loop](#loop) do)
- `close_connection_if_unused_after` if not None, it specifies how many seconds can pass without use a connection, then it is closed
- `pending_requests_expire_after` if not None, it specifies how many seconds can pass before that a pending request is discarded 
- `consider_others_dead_after` states how many seconds can pass before that a remote orbiter is considered dead if no keepalive arrives
- `send_keepalive_before_timelimit` states how many seconds before a keepalive message is sent that other remote orbiter considers current orbiter dead
- `graceful_close_timeout`: states how many seconds a graceful close connection can be pending
- `new_connection_added_event`: notify you when a new connection is created
- `remote_identifiers`: return all remote connected orbiter identifiers

Main hooks:

- `_get_on_close_data`: used to obtain data to send on close connection, by default None is returned
- `_on_starting`: called before starting
- `_internal_start`: actual implementation to start the orbiter
- `_on_started`: called after starting
- `_on_stopping`: called before stopping
- `_internal_stop`: actual implementation to stop the orbiter
- `_on_stopped`: called after stopping
- `_on_promote_pending_request_to_connection`: called before promotion
- `_on_keepalive_request`: called on keepalive request, before response
- `_on_keepalive`: called on inbound keepalive
- `_on_graceless_close_connection`: called before graceless close connection request is sent
- `_on_close_connection`: called when a connection is closed
- `_on_graceful_close_connection`: called before sending graceful close connection request
- `_on_loop_start`: called on loop start
- `_on_new_loop_iteration`: called before every loop iteration
- `_on_loop_iteration_end`: called at the end of every loop iteration
- `_on_loop_iteration`: called during every loop iteration

Main methods:

- `retrieve_connections`: retrieve all connections which satisfy query
- `discard_expired_pending_requests`: remove expired pending requests and return total amount of discarded requests 
- `close_unused_connections`: send a graceful close request to all remote orbiter if connection was unused based on `close_connection_if_unused_after`
- `force_close_connection_for_out_to_timeout_pending_graceful_close_connection`: send graceless close connection based on `graceful_timeout` if a connection is in close pending due to graceful close connection 
- `update_acquaintances`: update knowledge about keepalive request topics, keepalive topics and dead time
- `have_seen`: update last seen for remote orbiter
- `send_keepalive`
- `send_keepalive_request`
- `send_all_keepalive_based_on_connections`: send keepalive messages to all remote orbiters which have a connection with this orbiter
- `send_keepalive_based_on_connections_and_threshold`: send keepalive messages to all remote orbiters which have a connection with this orbiter only if `send_keepalive_before_timelimit` seconds away from being considered dead this orbiter
- `send_graceless_close_connection`: send a graceless close connection request to specified remote orbiter, therefore, self side connection will be closed immediately
- `send_graceful_close_connection`: send a graceful close connection request to specified remote orbiter, therefore self side connection is not close immediately, but ACK is waited
- `_close_self_side_connection`: close local connection with remote orbiter, therefore only this orbiter will no longer be able to use connection. Generally, a close connection request was sent before this method call.

```mermaid
flowchart TD
    n2["_on_starting()"] --> n3["_internal_start()"]
    n3 --> n4["_on_started()"]
    n13["Orbiter Activity"] -- stop() --> n16["_on_stopping()"]
    n14["Start Orbiter"] -- start() --> n2
    n16 --> n17["_internal_stop()"]
    n17 --> n18["_on_stopped()"]
    n18 --> n15["Orbiter Stopped"]
    s1["Orbiter Loop"] --> n15
    n4 -- start_loop() --> s1
    n3 --> n13
    n14@{ shape: rect}
    style n13 fill:#BBDEFB
    style n14 fill:#BBDEFB
    style n15 fill:#BBDEFB
    style s1 fill:#BBDEFB
```

#### Loop

Every orbiter has an internal **loop** which performs periodically operations. Automatic loop initialization can be avoided setting `with_loop=False`.

Loop is stopped when `stop` method is called, but you can stop it prematurely using `stop_loop` method. 
If you want to pause loop, you can use `pause_loop` and `resume_loop`.

Additionally to hooks, the following operations (*already discussed above*) are executed in parallel:

- `_on_loop_iteration`
- `close_unused_connections`
- `discard_expired_pending_requests`: 
- `force_close_connection_for_out_to_timeout_pending_graceful_close_connection`
- `send_keepalive_based_on_connections_and_threshold`

Hooks:

- `_on_loop_start`: called on loop start
- `_on_new_loop_iteration`: called before every loop iteration
- `_on_loop_iteration_end`: called at the end of every loop iteration
- `_on_loop_iteration`: called during every loop iteration

```mermaid
flowchart TD
 subgraph s1["Orbiter Loop"]
        n6["_on_loop_start()"]
        n7["_on_new_loop_iteration()"]
        n8["_on_loop_iteration()"]
        n9["close_unused_connections()"]
        n10["discard_expired_pending_requests()"]
        n11["force_close_connection_for_out_to_timeout_pending_graceful_close_connection()"]
        n12["send_keepalive_based_on_connections_and_threshold()"]
  end
    n6 --> n7
    n7 --> n8
    n8 --> n9
    n9 --> n10
    n10 --> n11
    n11 --> n12
    n12 --> n7
    n2["_on_starting()"] --> n3["_internal_start()"]
    n3 --> n4["_on_started()"]
    n4 --> n13["Orbiter Activity"]
    n13 -- stop() --> n16["_on_stopping()"]
    n14["Start Orbiter"] -- start() --> n2
    n16 --> n17["_internal_stop()"]
    n17 --> n18["_on_stopped()"]
    n18 --> n15["Orbiter Stopped"]
    s1 --> n15
    n3 -- start_loop() --> s1

    n14@{ shape: rect}
    style n13 fill:#BBDEFB
    style n14 fill:#BBDEFB
    style n15 fill:#BBDEFB
    style s1 fill:#BBDEFB
```


### Introduction to Core-Plugin Communication

Orbitalis implemented different communication protocol which are used to ensure that orbiters can share information and connect them together.

If you only use this library, you should not care about their implementation details, because every was made by us. Instead, if you want to know more or if you want to contribute, please read the [Advance Guide](#advance-guide). 

#### Handshake

To allow cores and plugins to create connections a handshake mechanism was implemented based on how DHCP works.

![DHCP](doc/assets/images/dhcp.jpg)

There are 4 steps:

1. **Discover**: message used by cores to notify plugins of their presence and to ask operations connections, core provides a full set of pluggable operations with related information
2. **Offer**: Message used by plugins to response to discover message, providing their base information and a list of offered operations. List of offered operations can be smaller than fullset provided by discover
3. **Reply**
   - **Request**: message used by core to formally request an operation. Every operation has own request. Core provides additional information to finalize the connection
   - **Reject**: message used by core to formally reject an operation (e.g., not needed anymore). Every operation has own reject
4. **Response**
   - **Confirm**: message used by plugins to confirm the connection creation
   - **OperationNoLongerAvailable**: message used by plugins to notify core that operation is no longer available


When an [Pending Request](#pending-requests) is *confirmed*, then a [Connection](#connections) is generated (and the related request is removed).

```mermaid
sequenceDiagram
    Core->>Plugin: Discover
    Plugin->>Core: Offer
    par Reply for each offered operation
        alt is still needed
            Core->>Plugin: Request
        else not needed anymore
            Core->>Plugin: Reject
        end
    end
    par Response for each requested operation
        alt is still available
            Plugin->>Core: Confirm
        else operation no longer available
            Plugin->>Core: OperationNoLongerAvailable
        end
    end
```

Orbiters which are related to the same context must use the same **discover topic** (by default `$handshake.discover`).
It can be set using `discover_topic` attribute.

Other topics are automatically generated, but generation can be modified overriding `_build_*` methods.

We must notice that discover and offer messages are also used to share information about presence of cores/plugins.
Theoretically, this means that a plugin may send an offer without an explicit discover, 
for example if a connection is closed and a slot for an operation becomes available. Anyway, this is not performed in current implementation.


#### Connections

`Connection` is a link between a core and a plugin **related to a *single* operation**, therefore more connections can be present between same core and plugin.

Regularly, connections are created after a handshake procedure, promoting a [Pending Request](#pending-requests). Generally you don't need to know how to create a connection, but if you want, you can read the [Advance Guide](#advance-guide).

```mermaid
flowchart TD
    P["Pending Request"] --> C["Connection"]
```

`Connection` class store all information about a connection:

- `operation_name`
- `remote_identifier`
- `incoming_close_connection_topic`: topic on which close connection request arrives from remote orbiter
- `close_connection_to_remote_topic`: topic which orbiter must use to close connection with remote orbiter
- `lock`: `asyncio.Lock`, used to synchronize connection uses
- `input`: acceptable `Input`
- `output`: sendable `Output`
- `input_topic`
- `output_topic`
- `soft_closed_at`: used during graceful close connection
- `created_at`: connection creation datetime
- `last_use`: datetime of last use

`last_use` must be updated **manually**, if you want to update it, using `touch` method on each connection (remember to *lock* the connection).

For example, when a new event arrives:

```python
# Following a custom plugin (MyPlugin) with an operation "my_operation"
@dataclass
class MyPlugin(Plugin):
    @operation(
        # operation name
        name="my_operation",
        
        # operation is fed with Int64Message messages (integer)
        input=Input.from_message(Int64Message),
        
        # operation doesn't send any output
        output=Output.no_output()
    )
    async def its_event_handler(self, topic: str, event: Event[...]):

        # Retrieve connections related to the input topic and the operation name
        connections = self.retrieve_connections(
            input_topic=topic, 
            operation_name="my_operation"
        )
        
        # Touch each operation to update `last_use`
        for connection in connections:
            async with connection.lock:     # lock connection to be async-safe
                connection.touch()
```

Fortunately, there is an useful method called `retrieve_and_touch_connections` which encapsulates exactly that code:

```python
# Same plugin of previous example, 
# but in which retrieve_and_touch_connections is used

@dataclass
class MyPlugin(Plugin):
    @operation(
        name="my_operation",    # operation's name, i.e. "my_operation"
        input=Input.from_message(Int64Message), # operation's input, i.e. an integer number 
        output=Output.no_output()   # operation's output, i.e. no output are produced
    )
    async def its_event_handler(self, topic: str, event: Event[...]):

        # Retrieves connections related to this operation and touches them
        connections = await self.retrieve_and_touch_connections(
            input_topic=topic, 
            operation_name="my_operation"
        )

        # ...operation's logic
```

Orbiter connections are stored in `_connections` attribute.

You can manage them using following methods:

- `_add_connection`
- `_remove_connection` (_does not lock automatically the connection_)
- `_connections_by_remote_identifier`: retrieves connection based on remote identifier
- `retrieve_connections`: to query connections
- `_find_connection_or_fail`: find _the_ connection based on `operation_name` and `input_topic` 
- `close_unused_connections` based on `close_connection_if_unused_after` and `last_use`


> [!IMPORTANT]
> Remember that potentially you can have more connections associated with the same pair input topic and operation name, based on how you have defined builder methods for input/output topics. This is the reason behind the fact that you don't receive a single connection by default as an input parameter. Anyway, if you are sure that **only one connection** is present, you can use `_find_connection_or_fail` as specified above.

#### Connection Close Procedure

An [orbiter](#orbiter) (Core or Plugin) can close a connection in every moment. There are two ways to close a connection: *Graceless* or *Graceful*.

In the following example we suppose an orbiter `"my_orbiterA"` that closes connection with another orbiter `"my_orbiterB"` related to operation `"my_operation"`.

In **Graceless** procedure the orbiter sends a `GracelessCloneConnectionMessage` to remote one and close connection immediately:

```mermaid
sequenceDiagram
    Orbiter A->>Orbiter A: Close connection
    Orbiter A->>Orbiter B: Graceless close connection
    Orbiter B->>Orbiter B: Close connection
```

```python
# `orbiterA` close connection immediately 
# with "my_orbiterB" related to operation "my_operation"
orbiterA.send_graceless_close_connection(
    remote_identifier="my_orbiterB",
    operation_name="my_operation"
)
```

In **Graceful** the orbiter sends a `GracefulCloseConnectionMessage` to remote one. Remote orbiter is able to perform some operations before connection is actually closed. Remote orbiter must send a `CloseConnectionAckMessage`, after that connection is closed. If `graceful_close_timeout` is not `None` is used to send graceless close connection if ACK is not sent. 
You can force timeout check using `force_close_connection_for_out_to_timeout_pending_graceful_close_connection` method.

```mermaid
sequenceDiagram
    Orbiter A->>Orbiter B: Graceful close connection
    activate Orbiter A
    Orbiter B->>Orbiter B: Some operations
    Orbiter B->>Orbiter B: Close connection
    Orbiter B->>Orbiter A: Close connection ACK
    deactivate Orbiter A
    Orbiter A->>Orbiter A: Close connection
```

```python
# `orbiterA` close connection gracefully 
# with "my_orbiterB" related to operation "my_operation"
orbiterA.send_graceful_close_connection(
    remote_identifier="my_orbiterB",
    operation_name="my_operation"
)
```

When all connections with a remote orbiter are closed, orbiter unsubscribes itself from topics in `_unsubscribe_on_full_close_bucket` field.

Both during graceful or graceless method call, you can provide data (`bytes`) which will be sent when connection is actually closed.
For example, considering graceful procedure:

```python
orbiterA.send_graceful_close_connection(
    remote_identifier="my_orbiterB",    # identifier of orbiter related to connection to close
    operation_name="my_operation",
    data=bytes(...)     # serialize your data
)
```

Hooks:

- `_on_graceless_close_connection`: called before graceless close connection request is sent
- `_on_close_connection`: called when a connection is closed
- `_on_graceful_close_connection`: called before sending graceful close connection request

> [!NOTE]
> These methods are also called when a connection is closed using method `close_unused_connections` (used to close unused connections based on `close_connection_if_unused_after` and `last_use`).

#### Keepalive

**Keepalive mechanism** allows orbiters to preserve connections during the time. Every orbiter must send a keepalive to all own linked orbiters.

An orbiter can _request_ a keepalive using `send_keepalive_request` method, which sends a `KeepaliveRequestMessage`.

```mermaid
sequenceDiagram
    Orbiter A->>Orbiter B: Keepalive request
    Orbiter B->>Orbiter A: Keepalive
```

Keepalive are sent using `KeepaliveMessage` messages. You can manually send a keepalive using `send_keepalive` method.

In addiction, `send_all_keepalive_based_on_connections` and `send_keepalive_based_on_connections_and_threshold` are provided.
`send_all_keepalive_based_on_connections` sends a keepalive to all remote orbiters which have an opened connection, 
instead `send_keepalive_based_on_connections_and_threshold` sends a keepalive to all remote orbiters which have an opened connection 
only if it is in range `send_keepalive_before_timelimit` seconds before remote orbiter considers it _dead_.

You can know which are dead remote orbiters thanks to `dead_remote_identifiers` property.

Main related fields:

- `_others_considers_me_dead_after`: dictionary which contains `remote_identifier => time`, used to know when a keepalive message must be sent
- `_remote_keepalive_request_topics`: dictionary which contains `remote_identifier => keepalive_request_topic`, used to send keepalive requests to remote orbiters
- `_remote_keepalive_topics`: dictionary which contains `remote_identifier => keepalive_topic`, used to send keepalive to remote orbiters
- `_last_seen`: dictionary which contains `remote_identifier => last_seen`, used to know if a remote orbiter must be considered *dead*
- `_last_keepalive_sent`: dictionary which contains `remote_identifier => last_keepalive_sent`, used to know if a keepalive message must be sent

Hooks:

- `_on_keepalive_request`: called on keepalive request, before response
- `_on_keepalive`: called on inbound keepalive



### Plugin

`Plugin` is an [Orbiter](#orbiter) and it is basically an _operations provider_. In a certain sense, plugins lent possibility to execute their operations.

In particular, every plugin has a set of operations which are exposed to other components (i.e., cores).
Only connected [Cores](#core) should execute operations, but this is not strictly ensured, you should check if there is a valid connection during operation elaboration.
You can check this using `retrieve_connections` or `_find_connection_or_fail`.

A plugin is a state machine which follows this states:

```mermaid
stateDiagram-v2
    [*] --> CREATED
    CREATED --> RUNNING: start()
    RUNNING --> STOPPED: stop()
    STOPPED --> RUNNING: start()
    STOPPED --> [*]
```

You can easily create a new plugin inheriting `Plugin` abstract class. Remember to use `@dataclass` For example:

```python
@dataclass
class MyPlugin(Plugin):
    ... # your plugin's operations and logic
```


Main hooks:

- `_on_new_discover`: called when a new discover message arrives
- `_on_reject`: called when a reject message arrives
- `_setup_operation`: called to set up operation when connection is created
- `_on_request`: called when a new request message arrives
- `_on_reply`: called when a new reply message arrives

Main methods:

- `send_offer`: send a new offer message in given topic to given core identifier (it should be used only if you want to send an offer message manually)
- `with_operation`: generally used during creation, allows you to specify additional operations (but generally we use decorator)
- `with_custom_policy`: generally used during creation, allows you to specify a custom operation policy
- `send_result_to_all`: send operation result to all connections which have an output topic

#### Operations

An **operation** (`Operation`) represents a feature of a plugin, which is exposed and can be executed remotely. 
Operations are managed by `OperationsProviderMixin` which also provides builder-like methods.

Every operation has the following attributes:

- `name`: unique name which identify the operation
- `handler`: [Busline](https://github.com/orbitalis-framework/py-busline) handler, which will be used to handle inbound events
- `policy`: specifies default operation lending rules, you can override this using `with_custom_policy` method or modifying `operations` attribute directly
- `input`: specifies which is the acceptable input
- `output`: specifies which is the sendable output

Even if output can be specified, if a `Core` doesn't need it, it should not sent. Obviously, you decide which messages must be sent in which topics, so you must ensure the compliance.

You can **add or modify manually operations** to a plugin thanks to `operations` attribute, otherwise you can use `@operation` **decorator**.

> [!IMPORTANT]
> Obviously, if the operation's input is `Input.no_input()` you **must** add manually the operation, because there isn't an event handler. You can see an example of it in [periodic operation](#periodic-operations) section.

`@operation` if you don't provide an `input` or an `output`, they are considered "no input/output" (see [input/output](#input--output)). 
If you don't specify `default_policy`, `Policy.no_constraints()` is assigned.

`@operation` automatically add to `operations` attributes the generated `Operation` object, related to (operation) `name` key.

For example, if you want to create a plugin having an operation `"lowercase"` which supports strings as input and produces strings (without lent constraints, i.e. no policy):

```python
@dataclass
class LowercaseTextProcessorPlugin(Plugin):

    @operation(
        name="lowercase",       # operation's name
        input=Input.from_message(StringMessage),    # operation's input
        output=Output.from_message(StringMessage)   # operation's output
        # no policy is specified => Policy.no_constraints()
    )
    async def lowercase_event_handler(self, topic: str, event: Event[StringMessage]):
        # NOTE: input message specified in @operation should be the same of 
        # what is specified as type hint of event parameter

        # Retrieve input string value, remember that it is wrapped into StringMessage
        input_str = event.payload.value

        lowercase_text = input_str.lower()  # process the string

        # Retrieve and touch related connections
        connections = await self.retrieve_and_touch_connections(
            input_topic=topic, 
            operation_name="lowercase"
        )

        tasks = []
        for connection in connections:

            # Only if the connection expects an output
            # it is published in the related topic
            # specified by `connection.output_topic`
            if connection.has_output:
                tasks.append(
                    asyncio.create_task(
                        self.eventbus_client.publish(
                            connection.output_topic,
                            lowercase_text  # will be wrapped into StringMessage
                        )
                    )
                )

        await asyncio.gather(*tasks)    # wait publishes
```

> [!NOTE]
> Method name is not related to operation's name.


##### Periodic operations

If you want to elaborate something periodically, you can use provided orbiter's loop. This allows you to avoid [custom loop](#custom-loop), even if you can create them.

We advice to use an auxiliary field which contains last elaboration datetime.

For example, we suppose to have a [plugin](#plugin) which has an operation `randint` which periodically send a random number to connected cores.

```python
@dataclass
class RandomNumberPlugin(Plugin):
    """
    This plugin has only one operation which sends periodically a random number
    """

    last_sent: Optional[datetime] = field(default=None)     # will be used to check if a new value must be sent

    def __post_init__(self):
        super().__post_init__()

        # Manually define "randint" operation
        self.operations["randint"] = Operation(
            name="randint",     # operation's name
            input=Input.no_input(),     # no inputs are expected
            handler=None,               # handler is not needed, given that no inputs must be processed
            output=Output.from_message(Int64Message),   # integers will be sent to cores
            policy=Policy.no_constraints()          # no constraint during handshake
        )

    @override
    async def _on_loop_iteration(self):     # we use loop iteration hook to provide periodic operations

        now = datetime.now()

        # Only if the enough time is elapsed the operation will be executed
        if self.last_sent is None or (now - self.last_sent).seconds > 2:    # send a new random int every 2 seconds
            self.last_sent = now    # update timer for next iteration
            await self.__send_randint()


    async def __send_randint(self):
        random_number = random.randint(0, 100)      # generate a new random number, it will be sent

        connections = await self.retrieve_and_touch_connections(operation_name="randint")  # retrieve current core connections

        tasks = []
        for connection in connections:
            if connection.has_output:   # check if output is expected (it should be...)
                tasks.append(
                    asyncio.create_task(
                        self.eventbus_client.publish(   # send random int
                            connection.output_topic,
                            random_number       # message (output)
                        )
                    )
                )

        await asyncio.gather(*tasks)
```

> [!IMPORTANT]
> Remember to enable loop (*avoid* `with_loop=False`) during plugin instantiation.

##### Policy

`Policy` allows you to specify for an operation:

- `maximum` amount of connections
- `allowlist` of remote orbiters
- `blocklist` of remote orbiters

Obviously, you can specify `allowlist` or `blocklist`, not both.

If you don't want constraints: `Policy.no_constraints()`.

If you use `@operation`, you can specified a *default policy*, which is what is used if you don't override it during plugin initialization.
Anyway, you could manage manually operations' policies, but we advise against.

```python
@dataclass
class LowercaseTextProcessorPlugin(Plugin):

    @operation(
        name="lowercase",
        input=Input.from_message(StringMessage),
        output=Output.from_message(StringMessage),
        default_policy=Policy(
            ... # insert here you constraints (e.g. allowlist or maximum)
        )
    )
    async def lowercase_event_handler(self, topic: str, event: Event[StringMessage]):
        ...     # operation's logic
```

As already mentioned, if you want to override default policy for a plugin's operation you can use `with_custom_policy`.

For example, if you want to add an allowlist (within core identifier `"my_core"`) for the above plugin, related to operation `"lowercase"`:

```python
plugin = LowercaseTextProcessorPlugin(...)
        .with_custom_policy(        # default policy specified with @operation will be override
            operation_name="lowercase",     # target operation
            policy=Policy(allowlist=["my_core"])    # new custom policy
        )
```

##### Input & Output

`Input` and `Output` are both `SchemaSpec`, i.e. the way to specify a schema set.

In a `SchemaSpec` we can specify `support_empty_schema` if we want to support payload-empty events, 
`support_undefined_schema` if we want to accept every schema and/or a schema list (`schemas`). A schema is a string.

We do have an input or an output only if:

```python
support_undefined_schema or support_empty_schema or has_some_explicit_schemas
```

In other words, we must always specify an `Input`/`Output` even if it is "no input/output". 
We can easily generate a "no input/output" thanks to: 

```python
input = Input.no_input()    # create new "no input" 

# `has_input` property is used to check if input is expected
assert not input.has_input

output = Output.no_output()    # create new "no output" 

# `has_output` property is used to check if output is expected
assert not output.has_output
```

If we want to generate a filled `Input`/`Output`:

```python
# Input related with MyMessage
Input.from_message(MyMessage)

# Input related with MyMessage, same as above
Input.from_schema(MyMessage.avro_schema())

# Input which accepts empty events
Input.empty()

# Input which accepts any payload
Input.undefined()

# Input which accepts related primitive data 
Input.int32()
Input.int64()
Input.float32()
Input.float64()
Input.string()

# Manual initialization of an Input object
Input(schemas=[...], support_empty_schema=True, support_undefined_schema=False)
```

By default, given that we use Avro JSON schemas, two schemas are compatible if the dictionary version of both is equal or
if the string version of both is equal.

For this reason, you must avoid time-based default value in your classes, because Avro set as default a time-variant value. 
Therefore, in this way, two same-class schema are **different**, even if they are related to the same class.

```python
created_at: datetime = field(default_factory=lambda: datetime.now())    # AVOID !!!
```



#### Plugins inheritance example

In this example, we will show you how to implement a **hierarchy of plugins**.

We will define `LampPlugin` abstract class plugin which provides a common operation `"get_status"`, then it will be inherited by `LampXPlugin` and `LampYPlugin` which add more operations.

```python
class LampStatus(StrEnum):
    """
    Utility enum to define possible plugin statuses
    """

    ON = "on"
    OFF = "off"


@dataclass
class StatusMessage(AvroMessageMixin):
    """
    Custom message defined to share status of lamps
    """

    lamp_identifier: str
    status: str     # "on" or "off"
    created_at: datetime = None # it is an Avro message, avoid default of time-variant fields

    def __post_init__(self):
        if self.created_at is None:
            self.created_at = datetime.now()    # default created_at value


@dataclass
class LampPlugin(Plugin, ABC):
    """
    Plugin to control a smart plugin which has an energy-meter
    """

    # Custom plugin attributes
    kw: float  # plugin energy consumption
    status: LampStatus = field(default=LampStatus.OFF)
    on_at: Optional[datetime] = field(default=None) # datetime of on request
    total_kwh: float = field(default=0.0)   # total consumption history

    @property
    def is_on(self) -> bool:
        return self.status == LampStatus.ON

    @property
    def is_off(self) -> bool:
        return self.status == LampStatus.OFF

    def turn_on(self):
        self.status = LampStatus.ON

        if self.on_at is None:
            self.on_at = datetime.now()

    def turn_off(self):
        """
        Turn off this plugin and update consumption history
        """

        self.status = LampStatus.OFF

        if self.on_at is not None:
            # Update total consumption:
            self.total_kwh += self.kw * (datetime.now() - self.on_at).total_seconds() / 3600

            self.on_at = None

    @operation(
        name="get_status",
        input=Input.empty(),
        output=Output.from_message(StatusMessage)
    )
    async def get_status_event_handler(self, topic: str, event: Event):
        connections = await self.retrieve_and_touch_connections(input_topic=topic, operation_name="get_status")

        # Only one connection should be present on inbound topic
        assert len(connections) == 1

        connection = connections[0]

        assert connection.output_topic is not None
        assert connection.output.has_output

        # Manually touch the connection
        async with connection.lock:
            connection.touch()

        # Send output to core
        await self.eventbus_client.publish(
            connection.output_topic,
            StatusMessage(self.identifier, str(self.status))
        )

    @abstractmethod
    async def turn_on_event_handler(self, topic: str, event: Event):
        raise NotImplemented()

    @abstractmethod
    async def turn_off_event_handler(self, topic: str, event: Event):
        raise NotImplemented()
```

```python
@dataclass
class LampXPlugin(LampPlugin):
    """
    Specific plugin related to brand X of smart lamps.
    This type of lamps doesn't have additional features
    """

    @operation(     # add new operation with name: "turn_on"
        name="turn_on",
        input=Input.empty()     # accepts empty events
    )
    async def turn_on_event_handler(self, topic: str, event: Event):
        self.turn_on()

    @operation(     # add new operation with name: "turn_off"
        name="turn_off",
        input=Input.empty()     # accepts empty events
    )
    async def turn_off_event_handler(self, topic: str, event: Event):
        self.turn_off()
```

```python
# Create new plugin X plugin
lamp_x_plugin = LampXPlugin(
    identifier="lamp_x_plugin",
    eventbus_client=...,    # provide Busline client
    raise_exceptions=True,
    with_loop=False,

    kw=24      # LampPlugin-specific attribute
).with_custom_policy(   # override custom policy related to operation "turn_on"
    operation_name="turn_on",
    policy=Policy(allowlist=["smart_home"])
)
```

```python
@dataclass(frozen=True)
class TurnOnLampYMessage(AvroMessageMixin):
    """
    Custom message to turn on plugin of brand Y.
    You can provide a "power" value which will be used to
    control brightness (and energy consumption)
    """

    power: float = field(default=1)

    def __post_init__(self):
        assert 0 < self.power <= 1

@dataclass(frozen=True)
class TurnOffLampYMessage(AvroMessageMixin):
    """
    Custom message to turn off plugin of brand Y.
    You can reset energy-meter setting True the flag
    """
    reset_consumption: bool = field(default=False)



@dataclass(kw_only=True)
class LampYPlugin(LampPlugin):
    """
    Specific plugin related to brand Y of smart lamps.
    These lamps are able to manage brightness level
    thanks to "power" attribute
    """

    power: float = field(default=1)

    @override
    def turn_off(self):
        """
        Overridden version to turn off the plugin and compute energy consumption 
        also based on power field
        """

        self.status = LampStatus.OFF

        if self.on_at is not None:
            self.total_kwh += self.power * s