BaseService

BaseService — abstract base class for all UMAA service templates.

Every service class (CommandProvider, ReportConsumer, composites, etc.) inherits from BaseService. It provides:

• A reference to the owning DDSContext (self._ctx).

• Automatic registration in the context’s service registry.

• An abstract close() method enforcing cleanup.

class rtiumaapy.base_service.BaseService(ctx: DDSContext, service_name: str | None = None)

Bases: ABC

Abstract base class for all UMAA services.

Subclasses must implement close().

Subclasses may define an async def _run(self) -> None coroutine for event-driven processing (e.g. async for on a DataReader). If present, DDSContext.run_until_shutdown() will create an asyncio.Task for it automatically. Services that are purely called by external code (e.g. a report provider’s publish()) should omit _run.

__init__(ctx: DDSContext, service_name: str | None = None) → None

Initialise the service and register it with the context.

Parameters:

• ctx – The DDSContext that owns shared DDS infrastructure.

• service_name – A unique name for this service instance (e.g. "EngineControl", "GPSReport"). Defaults to the class name if not provided.

property ctx: DDSContext

The DDSContext this service belongs to.

property service_name: str

The registered name of this service.

abstractmethod async close() → None

Perform logical cleanup for this service.

Called by DDSContext.shutdown() in reverse registration order. Subclasses should cancel tasks, dispose instances, and end sessions here but must not close DDS entities (readers/writers) — the rti.asyncio dispatcher may still reference their wait-sets. Entity destruction is handled by DDSContext.shutdown() via close_contained_entities() after the dispatcher is stopped.

Usage

All SDK service templates inherit from BaseService. You rarely instantiate it directly — instead, you subclass CommandProvider, ReportConsumer, etc.

BaseService provides:

• Auto-registration — __init__ registers the service with the DDSContext

• Lifecycle — Abstract close() enforced on every subclass

• Event loop — Optional _run() coroutine auto-started by run_until_shutdown()

from rtiumaapy import BaseService, DDSContext

class MyCustomService(BaseService):
    def __init__(self, ctx: DDSContext):
        super().__init__(ctx, "MyCustomService")

    async def _run(self):
        # Optional: event-driven processing
        while True:
            await asyncio.sleep(1.0)
            print("tick")

    async def close(self):
        print("Cleaning up")