# Luna Streams v.0.6.21 documentation

## Plugins

The service supports the system of plugins. Plugins must be written in the Python programming language.

## Plugin types

There are two sorts of plugins:

> - On event plugin. The plugin is triggered when an event occurs. The plugin should implement a callback function. This function is called on each event of the corresponding type. The set of event types is defined by the service developers.

>   Supported event types:

>   > event type | description
>   > --- | ---
>   > monitoring_event | Event contains monitoring points for sending to a custom monitoring system

>   Plugin example:

>   >     """
>   >     Module request monitoring plugin example
>   >     """
>   >     from abc import abstractmethod
>   >     from logging import Logger
>   >     from typing import List, Optional

>   >     from aiohttp import ClientSession
>   >     from luna_plugins.base.plugins_meta.base_plugins import BaseEventPlugin

>   >     class BaseRequestMonitoringPlugin(BaseEventPlugin):
>   >         """
>   >         Base class for requests monitoring.
>   >         """

>   >         # event name for triggering callback
>   >         eventName = "monitoring_event"

>   >         @abstractmethod
>   >         async def flushPointToMonitoring(self, points: List["BaseRequestMonitoringPoint"], logger: Logger) -> None:
>   >             """
>   >             All plugins must realize this method.

>   >             This function call after end of request

>   >             Args:
>   >                 points: points for monitoring which corresponding the request
>   >                 logger: logger
>   >             """

>   >         async def handleEvent(  # pylint: disable-msg=C0204,W0221
>   >             self, points: List["BaseRequestMonitoringPoint"], logger: Logger
>   >         ):
>   >             await self.flushPointToMonitoring(points=points, logger=logger)

>   >     class RequestMonitoringPlugin(BaseRequestMonitoringPlugin):
>   >         """
>   >         Example plugin sends a request data for monitoring to third-party source.
>   >         Only one instance of this class exist during the program execution.
>   >         """

>   >         def __init__(self, app: "LunaApplication"):
>   >             super().__init__(app)
>   >             self.url = "http://127.0.0.1:5020/1/buckets"
>   >             self.session: Optional[ClientSession] = None
>   >             self.bucket = "plugin_test_bucket"

>   >         async def close(self):
>   >             """
>   >             Stop plugin.

>   >             Close all open connections and ect
>   >             """
>   >             if self.session:
>   >                 await self.session.close()

>   >         async def initialize(self):
>   >             """
>   >             Initialize plugin.

>   >             Close all open connections and ect
>   >             """
>   >             self.session = ClientSession()
>   >             async with self.session.post(f"{self.url}?bucket={self.bucket}") as resp:
>   >                 if resp.status not in (201, 409):
>   >                     response = await resp.json()
>   >                     raise RuntimeError(f"failed create bucket, {self.bucket}, response: {response}")

>   >         async def flushPointToMonitoring(self, points: List["BaseRequestMonitoringPoint"], logger: Logger) -> None:
>   >             """
>   >             Callback for sending a request monitoring data.

>   >             Args:
>   >                 points: point for monitoring which corresponding the request
>   >                 logger: logger
>   >             """
>   >             if not points:
>   >                 return
>   >             point = points[0]
>   >             logger.debug(
>   >                 f"Plugin 'flushPointToMonitoring' get point, request_id: {point.requestId}, "
>   >                 f"start time: {point.eventTime}"
>   >             )
>   >             msg = {"tags": point.tags, "fields": point.fields}
>   >             async with self.session.post(f"{self.url}/{self.bucket}/objects", json=msg) as resp:
>   >                 logger.info(resp.status)
>   >                 logger.info(await resp.text())

>   This plugin demonstrates the sending of a request monitoring data to another service. All monitoring plugins must implement the BaseRequestMonitoringPlugin abstract class.

> - Background plugin. This sort of plugin is intended for background work.

>   The background plugin can implement:

>   - custom route

>   - background monitoring of service resources

>   - collaboration of an event plugin and a background plugin (batching monitoring points)

>   - connection to other data sources (Redis, RabbitMQ) and their data processing

>   Plugin example:

>   >     """
>   >     Module realizes background plugin example
>   >     """
>   >     import asyncio
>   >     from asyncio import Task
>   >     from typing import Optional

>   >     from luna_plugins.base.plugins_meta.base_plugins import BaseBackgroundPlugin
>   >     from sanic.response import HTTPResponse

>   >     from ...web.handlers import BaseHandler

>   >     class HandlerExample(BaseHandler):
>   >         """
>   >         Handler example
>   >         """

>   >         async def get(self) -> HTTPResponse:
>   >             """
>   >             Method get example.

>   >             Returns:
>   >                 response
>   >             """
>   >             return self.success(body="I am teapot", contentType="text/plain", statusCode=418)

>   >         @property
>   >         def app(self):
>   >             """
>   >             Get app
>   >             Abstract method of `BaseHandler`
>   >             Returns:
>   >                 app
>   >             """
>   >             return self.request.app

>   >         @property
>   >         def config(self):
>   >             """
>   >             Get app config
>   >             Abstract method of `BaseHandler`
>   >             Returns:
>   >                 app config
>   >             """
>   >             return self.request.app.ctx.serviceConfig

>   >     class BackgroundPluginExample(BaseBackgroundPlugin):
>   >         """
>   >         Background plugin example.

>   >         Create background task and add a route.
>   >         """

>   >         def __init__(self, app: "LunaApplication"):
>   >             super().__init__(app)
>   >             app.router.reset()
>   >             app.addRoutes([("/teapot", HandlerExample)])
>   >             app.finalize()
>   >             self.task: Optional[Task] = None
>   >             self.temperature = 0

>   >         async def initialize(self):
>   >             """
>   >             Initialize plugin
>   >             """

>   >         async def close(self):
>   >             """
>   >             Stop background process
>   >             Returns:

>   >             """
>   >             if self.task:
>   >                 self.task.cancel()

>   >         async def usefulJob(self):
>   >             """
>   >             Some useful  async work
>   >             """
>   >             while True:
>   >                 await asyncio.sleep(1)
>   >                 self.temperature = min(100, self.temperature + 1)
>   >                 if self.temperature < 100:
>   >                     self.app.ctx.logger.info(f"I boil water, temperature: {self.temperature}")
>   >                 else:
>   >                     self.app.ctx.logger.info("boiling water is ready, would you care for a cup of tea?")

>   >         async def start(self):
>   >             """
>   >             Run background process

>   >             .. warning::
>   >                 The function suppose that the process is handle in this coroutine. The coroutine must start
>   >                 the process only without awaiting a end of the process
>   >             """
>   >             self.task = asyncio.create_task(self.usefulJob())

>   This plugin demonstrates background work and implements a route. All background plugins must implement the BaseRequestMonitoringPlugin abstract class.

## Enable plugin

If the user implements a plugin, the file with the plugin should be added to the luna_streams/plugins directory of the service. The plugin filename should be added to the LUNA_STREAMS_ACTIVE_PLUGINS configuration setting.