Commit 984da2eb authored by George Papathanail's avatar George Papathanail
Browse files

feat: send x-correlator header on qod-status-changed callbacks

parent fa847797

src/open_exposure_gateway/adapters/http/qod_callback_client.py

+7 −2
Changes for src/open_exposure_gateway/adapters/http/qod_callback_client.py: 7 added lines, 2 removed lines.
Original line number Diff line number Diff line
@@ -9,11 +9,16 @@ class HttpQodCallbackClient: 
        settings = get_settings()

        self.timeout = settings.callback_settings.timeout



    async def deliver(self, sink: str, event: QosStatusChangedCloudEvent) -> None:

    async def deliver(

        self, sink: str, event: QosStatusChangedCloudEvent, x_correlator: str

    ) -> None:

        async with httpx.AsyncClient(timeout=self.timeout) as client:

            response = await client.post(

                sink,

                content=event.model_dump_json(exclude_none=True),

                headers={"Content-Type": "application/cloudevents+json"},

                headers={

                    "Content-Type": "application/cloudevents+json",

                    "x-correlator": x_correlator,

                },

            )

        response.raise_for_status()

src/open_exposure_gateway/application/services/quality_on_demand_service.py

+7 −1
Changes for src/open_exposure_gateway/application/services/quality_on_demand_service.py: 7 added lines, 1 removed line.
Original line number Diff line number Diff line
@@ -233,6 +233,7 @@ class QualityOnDemandService: 
    async def _deliver_qos_status_changed(

        self,

        operation_id: UUID,

        correlation_id: str,

        session_id: UUID,

        qos_status: Literal["AVAILABLE", "UNAVAILABLE"],

        status_info: Literal["DURATION_EXPIRED", "NETWORK_TERMINATED", "DELETE_REQUESTED"] | None,
@@ -260,7 +261,9 @@ class QualityOnDemandService: 
        )

        last_error: Optional[str] = None

        try:

            await self._callback_delivery_port.deliver(registration.sink, cloud_event)

            await self._callback_delivery_port.deliver(

                registration.sink, cloud_event, correlation_id

            )

            state = "delivered"

        except Exception as exc:

            state = "failed"
@@ -340,6 +343,7 @@ class QualityOnDemandService: 


        await self._deliver_qos_status_changed(

            operation_id=operation_id,

            correlation_id=operation.correlation_id,

            session_id=qod_session.session_id,

            qos_status=qos_status,

            status_info=status_info,
@@ -407,6 +411,7 @@ class QualityOnDemandService: 


        await self._deliver_qos_status_changed(

            operation_id=operation_id,

            correlation_id=operation.correlation_id,

            session_id=qod_session.session_id,

            qos_status=qos_status,

            status_info=status_info,
@@ -480,6 +485,7 @@ class QualityOnDemandService: 
        if qod_session.state == QodSessionState.AVAILABLE:

            await self._deliver_qos_status_changed(

                operation_id=qod_session.operation_id,

                correlation_id=correlation_id,

                session_id=qod_session.session_id,

                qos_status="UNAVAILABLE",

                status_info="DELETE_REQUESTED",

src/open_exposure_gateway/ports/qod_callback_port.py

+3 −1
Changes for src/open_exposure_gateway/ports/qod_callback_port.py: 3 added lines, 1 removed line.
Original line number Diff line number Diff line
@@ -4,4 +4,6 @@ from open_exposure_gateway.domain.quality_on_demand import QosStatusChangedCloud 




class QodCallbackDeliveryPort(Protocol):

    async def deliver(self, sink: str, event: QosStatusChangedCloudEvent) -> None: ...
 
    async def deliver(

        self, sink: str, event: QosStatusChangedCloudEvent, x_correlator: str

    ) -> None: ...

tests/unit/fakes.py

+5 −1
Changes for tests/unit/fakes.py: 5 added lines, 1 removed line.
Original line number Diff line number Diff line
@@ -635,6 +635,10 @@ class FakeCallbackDeliveryPort: 
class FakeQodCallbackDeliveryPort:

    def __init__(self) -> None:

        self.delivered: list[tuple[str, QosStatusChangedCloudEvent]] = []

        self.correlators: list[str] = []



    async def deliver(self, sink: str, event: QosStatusChangedCloudEvent) -> None:

    async def deliver(

        self, sink: str, event: QosStatusChangedCloudEvent, x_correlator: str

    ) -> None:

        self.delivered.append((sink, event))

        self.correlators.append(x_correlator)