Your First Processor

from dataclasses import dataclass
from microdcs.dataclass import (
    DataClassConfig,
    DataClassResponseMixin,
    DataClassValidationMixin,
)
from microdcs.models.ping_pong_mixin import PingPongDataClassMixin


@dataclass(kw_only=True)
class Ping(
    DataClassValidationMixin, DataClassResponseMixin["Pong"], PingPongDataClassMixin
):
    __request_object__: InitVar[Ping | None] = None
    message: str

    class Config(DataClassConfig):
        response_type: str = "Pong"
        type_id: str = "com.example.ping.v1"
        type_schema: str = "https://example.com/schemas/ping-pong/v1.0.0/ping"


@dataclass(kw_only=True)
class Pong(DataClassValidationMixin, PingPongDataClassMixin):
    reply: str

    class Config(DataClassConfig):
        type_id: str = "com.example.pong.v1"
        type_schema: str = "https://example.com/schemas/ping-pong/v1.0.0/pong"

import logging

from microdcs import ProcessingConfig
from microdcs.common import (
    CloudEvent,
    CloudEventProcessor,
    MessageIntent,
    ProcessorBinding,
    incoming,
    outgoing,
    processor_config,
)
from app.models.ping_pong import Ping, Pong

logger = logging.getLogger("processor.ping_pong")


@processor_config(binding=ProcessorBinding.NORTHBOUND)
class PingPongProcessor(CloudEventProcessor):
    def __init__(
        self,
        instance_id: str,
        runtime_config: ProcessingConfig,
        config_identifier: str,
    ):
        super().__init__(instance_id, runtime_config, config_identifier)

    @incoming(Ping)
    async def handle_ping(self, ping: Ping) -> Pong | None:
        logger.info("Received ping: %s", ping.message)
        # Use the response chain to create a Pong from the Ping
        return ping.response(reply=f"pong: {ping.message}")

    @outgoing(Pong)
    async def produce_pong(self, **kwargs) -> Pong | None:
        return Pong(reply=kwargs.get("reply", "unsolicited pong"))

    async def initialize(self) -> None:
        logger.info("PingPongProcessor initialized")

    async def post_start(self) -> None:
        logger.info("PingPongProcessor started")

    async def shutdown(self) -> None:
        logger.info("PingPongProcessor shutdown")

    async def process_response_cloudevent(
        self, cloudevent: CloudEvent
    ) -> list[CloudEvent] | CloudEvent | None:
        return None

    async def handle_cloudevent_expiration(
        self, cloudevent: CloudEvent, timeout: int
    ) -> list[CloudEvent] | CloudEvent | None:
        logger.warning("Event %s expired after %ds", cloudevent.id, timeout)
        return None

    async def trigger_outgoing_event(self, **kwargs) -> None:
        return None

from microdcs.core import MicroDCS
from microdcs.mqtt import MQTTHandler, MQTTProtocolBinding, OTELInstrumentedMQTTHandler
from app.processors.ping_pong import PingPongProcessor

microdcs = MicroDCS()
microdcs.register_protocol_handler(
    MQTTHandler(
        microdcs.runtime_config.mqtt,
        microdcs.redis_connection_pool,
        microdcs.redis_key_schema,
    ),
    OTELInstrumentedMQTTHandler(
        microdcs.runtime_config.mqtt,
        microdcs.redis_connection_pool,
        microdcs.redis_key_schema,
    ),
)

ping_pong_processor = PingPongProcessor(
    microdcs.runtime_config.instance_id,
    microdcs.runtime_config.processing,
    "ping-pong",
)

microdcs.register_protocol_binding(
    MQTTProtocolBinding(
        ping_pong_processor,
        microdcs.runtime_config.processing,
        microdcs.runtime_config.mqtt,
    )
)

asyncio.run(microdcs.main())

from microdcs.sfc_engine import SfcEngine

microdcs = MicroDCS()

# Register handlers and bindings as usual.
ping_pong_processor = PingPongProcessor(
    microdcs.runtime_config.instance_id,
    microdcs.runtime_config.processing,
    "ping-pong",
)

machinery_jobs_processor = MachineryJobsCloudEventProcessor(
    microdcs.runtime_config.instance_id,
    microdcs.runtime_config.processing,
    "machinery-jobs",
    microdcs.redis_connection_pool,
    microdcs.redis_key_schema,
)

sfc_engine = SfcEngine(
    microdcs.redis_connection_pool,
    microdcs.redis_key_schema,
    nb_processor=machinery_jobs_processor,
    sb_processors={"ping-pong": ping_pong_processor},
    consumer_name=microdcs.runtime_config.instance_id,
)

machinery_jobs_processor.register_scope_handler(sfc_engine.register_scope)
microdcs.add_additional_task(sfc_engine)

ping_pong_processor.register_pull_completion_handler(sfc_engine.pull_event_handler)

from microdcs.msgpack import (
    MessagePackHandler,
    MessagePackProtocolBinding,
    OTELInstrumentedMessagePackHandler,
)

microdcs.register_protocol_handler(
    MessagePackHandler(
        microdcs.runtime_config.msgpack,
        microdcs.redis_connection_pool,
        microdcs.redis_key_schema,
    ),
    OTELInstrumentedMessagePackHandler(
        microdcs.runtime_config.msgpack,
        microdcs.redis_connection_pool,
        microdcs.redis_key_schema,
    ),
)

microdcs.register_protocol_binding(
    MessagePackProtocolBinding(
        ping_pong_processor,
        microdcs.runtime_config.processing,
        microdcs.runtime_config.msgpack,
    )
)

# Ping inherits DataClassResponseMixin["Pong"]
pong = ping.response(reply="hello back")
# pong is a Pong instance

@dataclass(kw_only=True)
class StoreCall(DataClassResponseMixin["StoreResponse"], ...):
    job_order_id: str
    job_order: ISA95JobOrderDataType


@dataclass(kw_only=True)
class StoreResponse(...):
    job_order_id: str
    return_status: MethodReturnStatus


# In the processor:
response = store_call.response(
    takeover=["job_order_id"],
    return_status=MethodReturnStatus(status_code=0),
)
# response.job_order_id is copied from store_call.job_order_id

@dataclass(kw_only=True)
class Hello(DataClassResponseMixin["Hello"], GreetingsDataClassMixin):
    __request_object__: InitVar[Hello | None] = None
    _hidden_str: str | None = None
    name: str

from typing import TYPE_CHECKING, Any
from microdcs.dataclass import DataClassMixin

if TYPE_CHECKING:
    from app.models.ping_pong import PingPong


class PingPongDataClassMixin(DataClassMixin):
    def __post_init__(
        self,
        __request_object__: PingPong | None = None,
        __custom_metadata__: dict[str, Any] | None = None,
    ) -> None:
        # Call parent __post_init__ if it exists (important for MRO chain)
        super_post_init = getattr(super(), "__post_init__", None)
        if super_post_init is not None:
            super_post_init()

        # --- Deserialization path ---
        # When the framework deserializes a CloudEvent payload, custom metadata
        # from the CloudEvent envelope is injected via __custom_metadata__ InitVar
        # or stored in a context var consumed by _consume_custom_metadata().
        if __custom_metadata__ is None:
            __custom_metadata__ = self._consume_custom_metadata()
        if __custom_metadata__ is not None:
            if hasattr(self, "_my_hidden_field"):
                self._my_hidden_field = __custom_metadata__.get("x-my-hidden-field")

        # --- Response creation path ---
        # When .response() is called and the response class has __request_object__,
        # the request instance is passed here so hidden fields can be copied over.
        if __request_object__ is not None:
            if hasattr(self, "_my_hidden_field"):
                self._my_hidden_field = getattr(
                    __request_object__, "_my_hidden_field", None
                )

    def __get_custom_metadata__(self) -> dict[str, str]:
        # --- Serialization path ---
        # Called by CloudEvent.serialize_payload() to extract hidden fields
        # back into the CloudEvent custommetadata dict.
        custom_metadata = {}
        value = getattr(self, "_my_hidden_field", None)
        if value is not None:
            custom_metadata["x-my-hidden-field"] = value
        return custom_metadata

class MyProcessor(CloudEventProcessor):
    def __init__(self, ...):
        super().__init__(...)
        self._event_attributes.extend([
            CloudeventAttributeTuple("subject", "subject"),
            CloudeventAttributeTuple("correlationid", "correlationid"),
        ])

    @incoming(Ping)
    async def handle_ping(self, ping: Ping, *, subject: str, correlationid: str) -> ...:
        ...

@scope_from_subject
@incoming(Ping)
async def handle_ping(self, ping: Ping, *, scope: str) -> ...:
    # scope is the part of subject before the first "/"
    ...

async def __pre_outgoing_callback__(self, cloudevent, **kwargs):
    if some_condition:
        return False, None  # skip the main callback
    return True, None  # proceed normally

async def __post_outgoing_callback__(self, responses, cloudevent, **kwargs):
    # responses is the list of dataclass objects returned by @incoming handler
    # modify or filter them before the framework wraps them into CloudEvents
    return responses

@processor_config(binding=ProcessorBinding.NORTHBOUND)
class PingPongProcessor(CloudEventProcessor):
    async def initialize(self) -> None:
        # Runs before message handlers are started.
        # Safe to call async setup code here (e.g. database index creation).
        logger.info("Initializing PingPongProcessor")

    async def post_start(self) -> None:
        # Runs after all handlers are registered and the transport is live.
        # Use this to publish initial state or metadata events.
        self.publish_event(self.create_event(), intent=MessageIntent.META)

    async def shutdown(self) -> None:
        # Runs during graceful shutdown after the task group has exited.
        logger.info("Shutting down PingPongProcessor")

@processor_config(binding=ProcessorBinding.NORTHBOUND)
class PingPongProcessor(CloudEventProcessor):
    post_start_singleton = True  # only one replica across the set runs post_start()

    async def post_start(self) -> None:
        # This runs on exactly one replica. Other replicas skip it silently.
        self.publish_event(self.create_event(), intent=MessageIntent.META)

import pytest
from unittest.mock import AsyncMock, MagicMock

from microdcs import ProcessingConfig
from app.processors.ping_pong import PingPongProcessor
from app.models.ping_pong import Ping


@pytest.fixture
def processor():
    config = MagicMock(spec=ProcessingConfig)
    config.cloudevent_source = "test-source"
    config.message_expiry_interval = 60
    return PingPongProcessor("test-instance", config, "ping-pong")


class TestPingPongProcessor:
    @pytest.mark.asyncio
    async def test_handle_ping(self, processor):
        ping = Ping(message="hello")
        result = await processor.handle_ping(ping)
        assert result.reply == "pong: hello"