From c073cf5fd0e9c94700c9c1072b878953c878ce77 Mon Sep 17 00:00:00 2001 From: Shiying Chen Date: Sat, 28 Feb 2026 14:19:33 +0800 Subject: [PATCH 1/8] support invoke upstream event --- .../azure-messaging-webpubsubclient/README.md | 16 + .../messaging/webpubsubclient/_client.py | 219 +++++++++ .../messaging/webpubsubclient/aio/_client.py | 221 +++++++++ .../webpubsubclient/models/__init__.py | 6 + .../webpubsubclient/models/_enums.py | 2 + .../webpubsubclient/models/_models.py | 432 +++++++++++++++++- .../tests/test_invocation_manager.py | 87 ++++ .../tests/test_invocation_manager_async.py | 86 ++++ .../tests/test_unit.py | 77 ++++ 9 files changed, 1145 insertions(+), 1 deletion(-) create mode 100644 sdk/webpubsub/azure-messaging-webpubsubclient/tests/test_invocation_manager.py create mode 100644 sdk/webpubsub/azure-messaging-webpubsubclient/tests/test_invocation_manager_async.py diff --git a/sdk/webpubsub/azure-messaging-webpubsubclient/README.md b/sdk/webpubsub/azure-messaging-webpubsubclient/README.md index 5f01b7ec18ed..e1f905d05a19 100644 --- a/sdk/webpubsub/azure-messaging-webpubsubclient/README.md +++ b/sdk/webpubsub/azure-messaging-webpubsubclient/README.md @@ -77,6 +77,22 @@ client.send_to_group(group_name, "hello world", WebPubSubDataType.TEXT); # In the Console tab of your developer tools found in your browser, you should see the message printed there. ``` +### 5. Invoke upstream events (preview) + +`invoke_event` sends an `invoke` request to the service, awaits the correlated `invokeResponse`, and returns the payload. + +```python +from azure.messaging.webpubsubclient import WebPubSubClient +from azure.messaging.webpubsubclient.models import WebPubSubDataType + +client = WebPubSubClient("") +with client: + result = client.invoke_event("processOrder", {"orderId": 1}, WebPubSubDataType.JSON) + print(f"Invocation result: {result.data}") +``` + +_Streaming and service-initiated invocations are not yet supported._ + --- ## Examples ### Add callbacks for connected, disconnected and stopped events diff --git a/sdk/webpubsub/azure-messaging-webpubsubclient/azure/messaging/webpubsubclient/_client.py b/sdk/webpubsub/azure-messaging-webpubsubclient/azure/messaging/webpubsubclient/_client.py index c92f5d85ad48..b6d53887606f 100644 --- a/sdk/webpubsub/azure-messaging-webpubsubclient/azure/messaging/webpubsubclient/_client.py +++ b/sdk/webpubsub/azure-messaging-webpubsubclient/azure/messaging/webpubsubclient/_client.py @@ -44,6 +44,12 @@ OpenClientError, ReconnectError, RecoverError, + InvokeMessage, + InvokeResponseMessage, + CancelInvocationMessage, + InvokeEventResult, + InvocationError, + InvocationManager, ) from .models._enums import ( WebPubSubDataType, @@ -355,6 +361,7 @@ def __init__( ) self._group_map_lock = threading.Lock() self._ack_map: AckMap = AckMap() + self._invocation_map: InvocationManager = InvocationManager() self._ws: Optional[WebSocketAppSync] = None self._thread_seq_ack: Optional[threading.Thread] = None self._thread: Optional[threading.Thread] = None @@ -690,6 +697,204 @@ def send_to_group_attempt(): self._retry(send_to_group_attempt) + @overload + def invoke_event( + self, + event_name: str, + content: str, + data_type: Literal[WebPubSubDataType.TEXT], + **kwargs: Any, + ) -> InvokeEventResult: + """Invoke an upstream event and wait for the correlated response. + + :param event_name: The event name. Required. + :type event_name: str. + :param content: The data content. Required. + :type content: str. + :param data_type: The data type. Required. + :type data_type: ~azure.messaging.webpubsubclient.models.WebPubSubDataType.TEXT + :keyword str invocation_id: The optional invocation id. If not specified, client will generate one. + :keyword float timeout: Time limit in seconds to wait for the invoke response. If None (the default), + waits indefinitely. + :return: The invocation result. + :rtype: ~azure.messaging.webpubsubclient.models.InvokeEventResult + """ + + @overload + def invoke_event( + self, + event_name: str, + content: memoryview, + data_type: Literal[WebPubSubDataType.BINARY, WebPubSubDataType.PROTOBUF], + **kwargs: Any, + ) -> InvokeEventResult: + """Invoke an upstream event and wait for the correlated response. + + :param event_name: The event name. Required. + :type event_name: str. + :param content: The data content. Required. + :type content: memoryview. + :param data_type: The data type. Required. + :type data_type: ~azure.messaging.webpubsubclient.models.WebPubSubDataType.BINARY or + ~azure.messaging.webpubsubclient.models.WebPubSubDataType.PROTOBUF + :keyword str invocation_id: The optional invocation id. If not specified, client will generate one. + :keyword float timeout: Time limit in seconds to wait for the invoke response. If None (the default), + waits indefinitely. + :return: The invocation result. + :rtype: ~azure.messaging.webpubsubclient.models.InvokeEventResult + """ + + @overload + def invoke_event( + self, + event_name: str, + content: Dict[str, Any], + data_type: Literal[WebPubSubDataType.JSON], + **kwargs: Any, + ) -> InvokeEventResult: + """Invoke an upstream event and wait for the correlated response. + + :param event_name: The event name. Required. + :type event_name: str. + :param content: The data content. Required. + :type content: Dict[str, Any]. + :param data_type: The data type. Required. + :type data_type: ~azure.messaging.webpubsubclient.models.WebPubSubDataType.JSON + :keyword str invocation_id: The optional invocation id. If not specified, client will generate one. + :keyword float timeout: Time limit in seconds to wait for the invoke response. If None (the default), + waits indefinitely. + :return: The invocation result. + :rtype: ~azure.messaging.webpubsubclient.models.InvokeEventResult + """ + + def invoke_event( + self, + event_name: str, + content: Union[str, memoryview, Dict[str, Any]], + data_type: WebPubSubDataType, + **kwargs: Any, + ) -> InvokeEventResult: + """Invoke an upstream event and wait for the correlated response. + + :param event_name: The event name. Required. + :type event_name: str. + :param content: The data content. Required. + :type content: Union[str, memoryview, Dict[str, Any]]. + :param data_type: The data type. Required. + :type data_type: ~azure.messaging.webpubsubclient.models.WebPubSubDataType or str. + :keyword str invocation_id: The optional invocation id. If not specified, client will generate one. + :keyword float timeout: Time limit in seconds to wait for the invoke response. If None (the default), + waits indefinitely. + :return: The invocation result. + :rtype: ~azure.messaging.webpubsubclient.models.InvokeEventResult + """ + + def invoke_event_attempt() -> InvokeEventResult: + return self._invoke_event_core(event_name, content, data_type, **kwargs) + + return self._retry_with_result(invoke_event_attempt) + + def _invoke_event_core( + self, + event_name: str, + content: Union[str, memoryview, Dict[str, Any]], + data_type: WebPubSubDataType, + **kwargs: Any, + ) -> InvokeEventResult: + invocation_id_opt = kwargs.pop("invocation_id", None) + timeout = kwargs.pop("timeout", None) + invocation_id, entry = self._invocation_map.register(invocation_id_opt) + + invoke_message = InvokeMessage( + invocation_id=invocation_id, + target="event", + event=event_name, + data_type=data_type, + data=content, + ) + + try: + with entry.cv: + try: + self._send_message(invoke_message, **kwargs) + except Exception as e: + invocation_error = ( + e + if isinstance(e, InvocationError) + else InvocationError( + str(e) if str(e) else "Failed to send invocation message.", + invocation_id=invocation_id, + ) + ) + self._invocation_map.reject(invocation_id, invocation_error) + raise invocation_error from e + + notified = entry.cv.wait(timeout) + + if entry.error: + raise entry.error + + if entry.result is None: + raise InvocationError( + "Timeout while waiting for invoke response." + if not notified + else "No invoke response received.", + invocation_id=invocation_id, + ) + + return self._map_invoke_response(entry.result) + except Exception as e: # pylint: disable=broad-except + should_cancel = isinstance(e, InvocationError) and e.error_detail is None + if should_cancel: + self._send_cancel_invocation(invocation_id) + raise + finally: + self._invocation_map.discard(invocation_id) + + def _map_invoke_response(self, message: InvokeResponseMessage) -> InvokeEventResult: + if message.success is not True: + if message.success is False: + raise InvocationError( + message.error.message if message.error else "Invocation failed.", + invocation_id=message.invocation_id, + error_detail=message.error, + ) + raise InvocationError( + "Unsupported invoke response frame.", + invocation_id=message.invocation_id, + ) + + return InvokeEventResult( + invocation_id=message.invocation_id, + data_type=message.data_type, + data=message.data, + ) + + def _send_cancel_invocation(self, invocation_id: str) -> None: + try: + self._send_message(CancelInvocationMessage(invocation_id=invocation_id)) + except Exception as e: # pylint: disable=broad-except + _LOGGER.debug("Failed to send cancelInvocation for %s: %s", invocation_id, e) + + def _retry_with_result(self, func: Callable[[], InvokeEventResult]) -> InvokeEventResult: + retry_attempt = 0 + while True: + try: + return func() + except InvocationError: + raise + except Exception as e: # pylint: disable=broad-except + retry_attempt = retry_attempt + 1 + delay_seconds = self._message_retry_policy.next_retry_delay(retry_attempt) + if delay_seconds is None: + raise e + _LOGGER.debug( + "will retry %sth times after %s seconds", + retry_attempt, + delay_seconds, + ) + time.sleep(delay_seconds) + def _retry(self, func: Callable[[], None]): retry_attempt = 0 while True: @@ -890,6 +1095,13 @@ def on_message(_: Any, data: str): sequence_id=message.sequence_id, ), ) + elif message.kind == "invokeResponse": + resolved = self._invocation_map.resolve(message) + if not resolved: + _LOGGER.debug( + "Received invokeResponse for unknown invocationId: %s", + message.invocation_id, + ) else: _LOGGER.warning("unknown message type: %s", message.kind) @@ -909,6 +1121,13 @@ def on_close( # clean ack cache self._ack_map.clear() + self._invocation_map.reject_all( + lambda inv_id: InvocationError( + "Connection is disconnected before receiving invoke response from the service", + invocation_id=inv_id, + ) + ) + if self._is_stopping: _LOGGER.info("The client is stopping state. Stop recovery.") self._handle_connection_close_and_no_recovery(ws_instance.reconnect_tried_times) diff --git a/sdk/webpubsub/azure-messaging-webpubsubclient/azure/messaging/webpubsubclient/aio/_client.py b/sdk/webpubsub/azure-messaging-webpubsubclient/azure/messaging/webpubsubclient/aio/_client.py index b62021c6fbe4..eb0b86edda49 100644 --- a/sdk/webpubsub/azure-messaging-webpubsubclient/azure/messaging/webpubsubclient/aio/_client.py +++ b/sdk/webpubsub/azure-messaging-webpubsubclient/azure/messaging/webpubsubclient/aio/_client.py @@ -51,6 +51,12 @@ OpenClientError, ReconnectError, RecoverError, + InvokeMessage, + InvokeResponseMessage, + CancelInvocationMessage, + InvokeEventResult, + InvocationError, + InvocationManagerAsync, ) from ..models._enums import ( WebPubSubDataType, @@ -275,6 +281,7 @@ def __init__( **kwargs, ) self._ack_map: AckMapAsync = AckMapAsync() + self._invocation_map: InvocationManagerAsync = InvocationManagerAsync() self._ws: Optional[WebSocketAppAsync] = None self._event: asyncio.Event = asyncio.Event() self._task_seq_ack: Optional[asyncio.Task] = None @@ -605,6 +612,206 @@ async def send_to_group_attempt(): await self._retry(send_to_group_attempt) + @overload + async def invoke_event( + self, + event_name: str, + content: str, + data_type: Literal[WebPubSubDataType.TEXT], + **kwargs: Any, + ) -> InvokeEventResult: + """Invoke an upstream event and wait for the correlated response. + + :param event_name: The event name. Required. + :type event_name: str. + :param content: The data content. Required. + :type content: str. + :param data_type: The data type. Required. + :type data_type: ~azure.messaging.webpubsubclient.models.WebPubSubDataType.TEXT + :keyword str invocation_id: The optional invocation id. If not specified, client will generate one. + :keyword float timeout: Time limit in seconds to wait for the invoke response. If None (the default), + waits indefinitely. + :return: The invocation result. + :rtype: ~azure.messaging.webpubsubclient.models.InvokeEventResult + """ + + @overload + async def invoke_event( + self, + event_name: str, + content: memoryview, + data_type: Literal[WebPubSubDataType.BINARY, WebPubSubDataType.PROTOBUF], + **kwargs: Any, + ) -> InvokeEventResult: + """Invoke an upstream event and wait for the correlated response. + + :param event_name: The event name. Required. + :type event_name: str. + :param content: The data content. Required. + :type content: memoryview. + :param data_type: The data type. Required. + :type data_type: ~azure.messaging.webpubsubclient.models.WebPubSubDataType.BINARY or + ~azure.messaging.webpubsubclient.models.WebPubSubDataType.PROTOBUF + :keyword str invocation_id: The optional invocation id. If not specified, client will generate one. + :keyword float timeout: Time limit in seconds to wait for the invoke response. If None (the default), + waits indefinitely. + :return: The invocation result. + :rtype: ~azure.messaging.webpubsubclient.models.InvokeEventResult + """ + + @overload + async def invoke_event( + self, + event_name: str, + content: Dict[str, Any], + data_type: Literal[WebPubSubDataType.JSON], + **kwargs: Any, + ) -> InvokeEventResult: + """Invoke an upstream event and wait for the correlated response. + + :param event_name: The event name. Required. + :type event_name: str. + :param content: The data content. Required. + :type content: Dict[str, Any]. + :param data_type: The data type. Required. + :type data_type: ~azure.messaging.webpubsubclient.models.WebPubSubDataType.JSON + :keyword str invocation_id: The optional invocation id. If not specified, client will generate one. + :keyword float timeout: Time limit in seconds to wait for the invoke response. If None (the default), + waits indefinitely. + :return: The invocation result. + :rtype: ~azure.messaging.webpubsubclient.models.InvokeEventResult + """ + + async def invoke_event( + self, + event_name: str, + content: Union[str, memoryview, Dict[str, Any]], + data_type: WebPubSubDataType, + **kwargs: Any, + ) -> InvokeEventResult: + """Invoke an upstream event and wait for the correlated response. + + :param event_name: The event name. Required. + :type event_name: str. + :param content: The data content. Required. + :type content: Union[str, memoryview, Dict[str, Any]]. + :param data_type: The data type. Required. + :type data_type: ~azure.messaging.webpubsubclient.models.WebPubSubDataType or str. + :keyword str invocation_id: The optional invocation id. If not specified, client will generate one. + :keyword float timeout: Time limit in seconds to wait for the invoke response. If None (the default), + waits indefinitely. + :return: The invocation result. + :rtype: ~azure.messaging.webpubsubclient.models.InvokeEventResult + """ + + async def invoke_event_attempt() -> InvokeEventResult: + return await self._invoke_event_core(event_name, content, data_type, **kwargs) + + return await self._retry_with_result(invoke_event_attempt) + + async def _invoke_event_core( + self, + event_name: str, + content: Union[str, memoryview, Dict[str, Any]], + data_type: WebPubSubDataType, + **kwargs: Any, + ) -> InvokeEventResult: + invocation_id_opt = kwargs.pop("invocation_id", None) + timeout = kwargs.pop("timeout", None) + invocation_id, entry = self._invocation_map.register(invocation_id_opt) + + invoke_message = InvokeMessage( + invocation_id=invocation_id, + target="event", + event=event_name, + data_type=data_type, + data=content, + ) + + try: + await self._send_message(invoke_message, **kwargs) + except Exception as e: + invocation_error = ( + e + if isinstance(e, InvocationError) + else InvocationError( + str(e) if str(e) else "Failed to send invocation message.", + invocation_id=invocation_id, + ) + ) + self._invocation_map.reject(invocation_id, invocation_error) + raise invocation_error from e + + try: + await asyncio.wait_for(entry.event.wait(), timeout=timeout) + + if entry.error: + raise entry.error + + if entry.result is None: + raise InvocationError( + "No invoke response received.", + invocation_id=invocation_id, + ) + + return self._map_invoke_response(entry.result) + except asyncio.TimeoutError as e: + raise InvocationError( + "Timeout while waiting for invoke response.", + invocation_id=invocation_id, + ) from e + except Exception as e: # pylint: disable=broad-except + should_cancel = isinstance(e, InvocationError) and e.error_detail is None + if should_cancel: + await self._send_cancel_invocation(invocation_id) + raise + finally: + self._invocation_map.discard(invocation_id) + + def _map_invoke_response(self, message: InvokeResponseMessage) -> InvokeEventResult: + if message.success is not True: + if message.success is False: + raise InvocationError( + message.error.message if message.error else "Invocation failed.", + invocation_id=message.invocation_id, + error_detail=message.error, + ) + raise InvocationError( + "Unsupported invoke response frame.", + invocation_id=message.invocation_id, + ) + + return InvokeEventResult( + invocation_id=message.invocation_id, + data_type=message.data_type, + data=message.data, + ) + + async def _send_cancel_invocation(self, invocation_id: str) -> None: + try: + await self._send_message(CancelInvocationMessage(invocation_id=invocation_id)) + except Exception as e: # pylint: disable=broad-except + _LOGGER.debug("Failed to send cancelInvocation for %s: %s", invocation_id, e) + + async def _retry_with_result(self, func: Callable[[], Awaitable[InvokeEventResult]]) -> InvokeEventResult: + retry_attempt = 0 + while True: + try: + return await func() + except InvocationError: + raise + except Exception as e: # pylint: disable=broad-except + retry_attempt = retry_attempt + 1 + delay_seconds = self._message_retry_policy.next_retry_delay(retry_attempt) + if delay_seconds is None: + raise e + _LOGGER.debug( + "will retry %sth times after %s seconds", + retry_attempt, + delay_seconds, + ) + await asyncio.sleep(delay_seconds) + async def _retry(self, func: Callable[[], Awaitable[None]]): retry_attempt = 0 while True: @@ -790,6 +997,13 @@ async def on_message(data: str): sequence_id=message.sequence_id, ), ) + elif message.kind == "invokeResponse": + resolved = self._invocation_map.resolve(message) + if not resolved: + _LOGGER.debug( + "Received invokeResponse for unknown invocationId: %s", + message.invocation_id, + ) else: _LOGGER.warning("unknown message type: %s", message.kind) @@ -809,6 +1023,13 @@ async def on_close( # clean ack cache self._ack_map.clear() + self._invocation_map.reject_all( + lambda inv_id: InvocationError( + "Connection is disconnected before receiving invoke response from the service", + invocation_id=inv_id, + ) + ) + if self._is_stopping: _LOGGER.warning("The client is stopping state. Stop recovery.") await self._handle_connection_close_and_no_recovery(ws_instance.reconnect_tried_times) diff --git a/sdk/webpubsub/azure-messaging-webpubsubclient/azure/messaging/webpubsubclient/models/__init__.py b/sdk/webpubsub/azure-messaging-webpubsubclient/azure/messaging/webpubsubclient/models/__init__.py index e825a0249210..56dddabb8405 100644 --- a/sdk/webpubsub/azure-messaging-webpubsubclient/azure/messaging/webpubsubclient/models/__init__.py +++ b/sdk/webpubsub/azure-messaging-webpubsubclient/azure/messaging/webpubsubclient/models/__init__.py @@ -13,6 +13,9 @@ SendMessageError, OpenClientError, AckMessageError, + InvocationError, + InvokeEventResult, + InvokeResponseError, ) from ._enums import WebPubSubDataType, WebPubSubProtocolType, CallbackType @@ -30,4 +33,7 @@ "SendMessageError", "OpenClientError", "AckMessageError", + "InvocationError", + "InvokeEventResult", + "InvokeResponseError", ] diff --git a/sdk/webpubsub/azure-messaging-webpubsubclient/azure/messaging/webpubsubclient/models/_enums.py b/sdk/webpubsub/azure-messaging-webpubsubclient/azure/messaging/webpubsubclient/models/_enums.py index 4e3dc9e59a11..fed0abdc140d 100644 --- a/sdk/webpubsub/azure-messaging-webpubsubclient/azure/messaging/webpubsubclient/models/_enums.py +++ b/sdk/webpubsub/azure-messaging-webpubsubclient/azure/messaging/webpubsubclient/models/_enums.py @@ -22,6 +22,8 @@ class UpstreamMessageType(str, Enum, metaclass=CaseInsensitiveEnumMeta): SEND_TO_GROUP = "sendToGroup" SEND_EVENT = "sendEvent" SEQUENCE_ACK = "sequenceAck" + INVOKE = "invoke" + CANCEL_INVOCATION = "cancelInvocation" class WebPubSubDataType(str, Enum, metaclass=CaseInsensitiveEnumMeta): diff --git a/sdk/webpubsub/azure-messaging-webpubsubclient/azure/messaging/webpubsubclient/models/_models.py b/sdk/webpubsub/azure-messaging-webpubsubclient/azure/messaging/webpubsubclient/models/_models.py index 6c83528f9c6e..a978c1f9383b 100644 --- a/sdk/webpubsub/azure-messaging-webpubsubclient/azure/messaging/webpubsubclient/models/_models.py +++ b/sdk/webpubsub/azure-messaging-webpubsubclient/azure/messaging/webpubsubclient/models/_models.py @@ -7,7 +7,7 @@ import sys import asyncio import logging -from typing import Any, Mapping, overload, Union, Optional, TypeVar, Tuple, Dict, Literal +from typing import Any, Callable, Mapping, overload, Union, Optional, TypeVar, Tuple, Dict, Literal import json import math import threading @@ -291,6 +291,76 @@ def __init__(self, *args, **kwargs): # pylint: disable=useless-super-delegation super().__init__(*args, **kwargs) +class InvokeData(_model_base.Model): + """Data for invoke message + + :ivar type: The type of the message. Required. Default value is "invoke". + :vartype type: str + :ivar invocation_id: The invocation id. Required. + :vartype invocation_id: str + :ivar target: The invocation target type. + :vartype target: str + :ivar event: The event name. + :vartype event: str + :ivar data_type: The data type of the message. + :vartype data_type: ~azure.messaging.webpubsubclient.models.WebPubSubDataType or str + :ivar data: The data of the message. + :vartype data: Any + """ + + type: Literal["invoke"] = rest_field(default="invoke") + invocation_id: str = rest_field(name="invocationId") + target: Optional[str] = rest_field() + event: Optional[str] = rest_field() + data_type: Optional[Union[WebPubSubDataType, str]] = rest_field(name="dataType") + data: Any = rest_field() + + @overload + def __init__( + self, + *, + type: Literal["invoke"] = "invoke", # pylint: disable=redefined-builtin + invocation_id: str, + target: Optional[str] = None, + event: Optional[str] = None, + data_type: Optional[WebPubSubDataType] = None, + data: Any = None, + ) -> None: ... + + @overload + def __init__(self, mapping: Mapping[str, Any]): ... + + def __init__(self, *args, **kwargs): # pylint: disable=useless-super-delegation + super().__init__(*args, **kwargs) + + +class CancelInvocationData(_model_base.Model): + """Data for cancel invocation message + + :ivar type: The type of the message. Required. Default value is "cancelInvocation". + :vartype type: str + :ivar invocation_id: The invocation id. Required. + :vartype invocation_id: str + """ + + type: Literal["cancelInvocation"] = rest_field(default="cancelInvocation") + invocation_id: str = rest_field(name="invocationId") + + @overload + def __init__( + self, + *, + type: Literal["cancelInvocation"] = "cancelInvocation", # pylint: disable=redefined-builtin + invocation_id: str, + ) -> None: ... + + @overload + def __init__(self, mapping: Mapping[str, Any]): ... + + def __init__(self, *args, **kwargs): # pylint: disable=useless-super-delegation + super().__init__(*args, **kwargs) + + class SequenceAckMessage: """Message for sequence ack @@ -427,6 +497,119 @@ def __init__( self.ack_id = ack_id +class InvokeResponseError: + """Error details of an invoke response + + :ivar name: The error name. Required. + :vartype name: str + :ivar message: The error message. Required. + :vartype message: str + """ + + def __init__(self, *, name: str, message: str): + self.name = name + self.message = message + + +class InvokeMessage: + """Message for invoking an upstream event + + :ivar invocation_id: The invocation id. Required. + :vartype invocation_id: str + :ivar target: The invocation target type. Currently, only upstream events are supported. + :vartype target: str + :ivar event: The event name when targeting upstream events. + :vartype event: str + :ivar data_type: Data type of the payload. + :vartype data_type: ~azure.messaging.webpubsubclient.models.WebPubSubDataType or str + :ivar data: Payload data. + :vartype data: Any + """ + + def __init__( + self, + invocation_id: str, + *, + target: Optional[str] = "event", + event: Optional[str] = None, + data_type: Optional[WebPubSubDataType] = None, + data: Any = None, + ) -> None: + self.kind: Literal["invoke"] = "invoke" + self.invocation_id = invocation_id + self.target = target + self.event = event + self.data_type = data_type + self.data = data + + +class InvokeResponseMessage: + """Message for invoke response + + :ivar invocation_id: The invocation ID that this response is for. Required. + :vartype invocation_id: str + :ivar success: Indicates whether the invocation was successful. + :vartype success: bool + :ivar data_type: Data type of the payload. + :vartype data_type: ~azure.messaging.webpubsubclient.models.WebPubSubDataType or str + :ivar data: Payload data. + :vartype data: Any + :ivar error: Error details if the invocation failed. + :vartype error: ~azure.messaging.webpubsubclient.models.InvokeResponseError + """ + + def __init__( + self, + invocation_id: str, + *, + success: Optional[bool] = None, + data_type: Optional[WebPubSubDataType] = None, + data: Any = None, + error: Optional[InvokeResponseError] = None, + ) -> None: + self.kind: Literal["invokeResponse"] = "invokeResponse" + self.invocation_id = invocation_id + self.success = success + self.data_type = data_type + self.data = data + self.error = error + + +class CancelInvocationMessage: + """Message for canceling an invocation + + :ivar invocation_id: The invocation ID to cancel. Required. + :vartype invocation_id: str + """ + + def __init__(self, invocation_id: str) -> None: + self.kind: Literal["cancelInvocation"] = "cancelInvocation" + self.invocation_id = invocation_id + + +class InvokeEventResult: + """Result of invokeEvent + + :ivar invocation_id: Invocation identifier correlated with the response. Required. + :vartype invocation_id: str + :ivar data_type: The response payload data type. + :vartype data_type: ~azure.messaging.webpubsubclient.models.WebPubSubDataType or str + :ivar data: The response payload. + :vartype data: Any + """ + + def __init__( + self, + invocation_id: str, + *, + data_type: Optional[WebPubSubDataType] = None, + data: Any = None, + ) -> None: + self.invocation_id = invocation_id + self.data_type = data_type + self.data = data + + WebPubSubMessage = TypeVar( "WebPubSubMessage", GroupDataMessage, @@ -439,6 +622,9 @@ def __init__( SendEventMessage, SequenceAckMessage, AckMessage, + InvokeMessage, + InvokeResponseMessage, + CancelInvocationMessage, ) SendMessageType = TypeVar( @@ -573,6 +759,22 @@ def parse_messages( AckMessageError(name=error["name"], message=error["message"]) if isinstance(error, dict) else None ), ) + if message["type"] == "invokeResponse": + data = None + if message.get("dataType") is not None: + data = parse_payload(message.get("data"), message["dataType"]) + error = message.get("error") + return InvokeResponseMessage( + invocation_id=message["invocationId"], + success=message.get("success"), + data_type=message.get("dataType"), + data=data, + error=( + InvokeResponseError(name=error["name"], message=error["message"]) + if isinstance(error, dict) + else None + ), + ) _LOGGER.error("wrong message type: %s", message["type"]) return None @@ -607,6 +809,18 @@ def write_message(message: WebPubSubMessage) -> str: ) elif message.kind == UpstreamMessageType.SEQUENCE_ACK: data = SequenceAckData(sequence_id=message.sequence_id) + elif message.kind == UpstreamMessageType.INVOKE: + invoke_data = InvokeData( + invocation_id=message.invocation_id, + target=message.target, + event=message.event, + ) + if message.data_type is not None and message.data is not None: + invoke_data.data_type = message.data_type + invoke_data.data = get_pay_load(message.data, message.data_type) + data = invoke_data + elif message.kind == UpstreamMessageType.CANCEL_INVOCATION: + data = CancelInvocationData(invocation_id=message.invocation_id) else: raise TypeError(f"Unsupported type: {message.kind}") @@ -709,6 +923,29 @@ def __init__( self.error_detail = error_detail +class InvocationError(AzureError): + """Exception raised when an invocation fails or is cancelled + + :ivar message: The error message. Required. + :vartype message: str + :ivar invocation_id: The invocation id of the request. + :vartype invocation_id: str + :ivar error_detail: Error details from the service if available. + :vartype error_detail: ~azure.messaging.webpubsubclient.models.InvokeResponseError + """ + + def __init__( + self, + message: str, + invocation_id: str, + error_detail: Optional[InvokeResponseError] = None, + ) -> None: + super().__init__(message) + self.name = "InvocationError" + self.invocation_id = invocation_id + self.error_detail = error_detail + + class OnGroupDataMessageArgs: """Arguments for group data message @@ -1016,6 +1253,199 @@ def clear(self) -> None: self.ack_map.clear() +class _InvocationEntry: + """Entry for a pending invocation (sync version)""" + + def __init__(self, invocation_id: str) -> None: + self.invocation_id = invocation_id + self.cv = threading.Condition() + self.result: Optional[InvokeResponseMessage] = None + self.error: Optional[Exception] = None + + +class _InvocationEntryAsync: + """Entry for a pending invocation (async version)""" + + def __init__(self, invocation_id: str) -> None: + self.invocation_id = invocation_id + self.event = asyncio.Event() + self.result: Optional[InvokeResponseMessage] = None + self.error: Optional[Exception] = None + + +class InvocationManager: + """Manages pending invocations awaiting invokeResponse frames (sync version)""" + + def __init__(self) -> None: + self._entries: Dict[str, _InvocationEntry] = {} + self._next_id = 0 + self._lock = threading.Lock() + + def _generate_invocation_id(self) -> str: + self._next_id += 1 + return str(self._next_id) + + def register(self, invocation_id: Optional[str] = None) -> Tuple[str, _InvocationEntry]: + """Register a new invocation and return the invocation id and entry. + + :param invocation_id: Optional invocation id. If not provided, one will be generated. + :type invocation_id: str + :return: Tuple of invocation id and entry. + :rtype: Tuple[str, _InvocationEntry] + :raises InvocationError: If invocation id is already registered. + """ + with self._lock: + resolved_id = invocation_id if invocation_id else self._generate_invocation_id() + if resolved_id in self._entries: + raise InvocationError( + "Invocation id is already registered.", + invocation_id=resolved_id, + ) + entry = _InvocationEntry(resolved_id) + self._entries[resolved_id] = entry + return resolved_id, entry + + def resolve(self, message: InvokeResponseMessage) -> bool: + """Resolve a pending invocation with a response message. + + :param message: The invoke response message. + :type message: InvokeResponseMessage + :return: True if invocation was found and resolved. + :rtype: bool + """ + with self._lock: + entry = self._entries.pop(message.invocation_id, None) + if not entry: + return False + with entry.cv: + entry.result = message + entry.cv.notify() + return True + + def reject(self, invocation_id: str, error: Exception) -> bool: + """Reject a pending invocation with an error. + + :param invocation_id: The invocation id to reject. + :type invocation_id: str + :param error: The error to reject with. + :type error: Exception + :return: True if invocation was found and rejected. + :rtype: bool + """ + with self._lock: + entry = self._entries.pop(invocation_id, None) + if not entry: + return False + with entry.cv: + entry.error = error + entry.cv.notify() + return True + + def discard(self, invocation_id: str) -> None: + """Discard a pending invocation without resolving or rejecting it. + + :param invocation_id: The invocation id to discard. + :type invocation_id: str + """ + with self._lock: + self._entries.pop(invocation_id, None) + + def reject_all(self, create_error: Callable[[str], Exception]) -> None: + """Reject all pending invocations with errors. + + :param create_error: Factory function to create error for each invocation id. + :type create_error: Callable[[str], Exception] + """ + with self._lock: + for invocation_id, entry in list(self._entries.items()): + self._entries.pop(invocation_id, None) + with entry.cv: + entry.error = create_error(invocation_id) + entry.cv.notify() + + +class InvocationManagerAsync: + """Manages pending invocations awaiting invokeResponse frames (async version)""" + + def __init__(self) -> None: + self._entries: Dict[str, _InvocationEntryAsync] = {} + self._next_id = 0 + + def _generate_invocation_id(self) -> str: + self._next_id += 1 + return str(self._next_id) + + def register(self, invocation_id: Optional[str] = None) -> Tuple[str, _InvocationEntryAsync]: + """Register a new invocation and return the invocation id and entry. + + :param invocation_id: Optional invocation id. If not provided, one will be generated. + :type invocation_id: str + :return: Tuple of invocation id and entry. + :rtype: Tuple[str, _InvocationEntryAsync] + :raises InvocationError: If invocation id is already registered. + """ + resolved_id = invocation_id if invocation_id else self._generate_invocation_id() + if resolved_id in self._entries: + raise InvocationError( + "Invocation id is already registered.", + invocation_id=resolved_id, + ) + entry = _InvocationEntryAsync(resolved_id) + self._entries[resolved_id] = entry + return resolved_id, entry + + def resolve(self, message: InvokeResponseMessage) -> bool: + """Resolve a pending invocation with a response message. + + :param message: The invoke response message. + :type message: InvokeResponseMessage + :return: True if invocation was found and resolved. + :rtype: bool + """ + entry = self._entries.pop(message.invocation_id, None) + if not entry: + return False + entry.result = message + entry.event.set() + return True + + def reject(self, invocation_id: str, error: Exception) -> bool: + """Reject a pending invocation with an error. + + :param invocation_id: The invocation id to reject. + :type invocation_id: str + :param error: The error to reject with. + :type error: Exception + :return: True if invocation was found and rejected. + :rtype: bool + """ + entry = self._entries.pop(invocation_id, None) + if not entry: + return False + entry.error = error + entry.event.set() + return True + + def discard(self, invocation_id: str) -> None: + """Discard a pending invocation without resolving or rejecting it. + + :param invocation_id: The invocation id to discard. + :type invocation_id: str + """ + self._entries.pop(invocation_id, None) + + def reject_all(self, create_error: Callable[[str], Exception]) -> None: + """Reject all pending invocations with errors. + + :param create_error: Factory function to create error for each invocation id. + :type create_error: Callable[[str], Exception] + """ + for invocation_id, entry in list(self._entries.items()): + self._entries.pop(invocation_id, None) + entry.error = create_error(invocation_id) + entry.event.set() + + class OpenClientError(AzureError): """Exception raised when fail to start the client""" diff --git a/sdk/webpubsub/azure-messaging-webpubsubclient/tests/test_invocation_manager.py b/sdk/webpubsub/azure-messaging-webpubsubclient/tests/test_invocation_manager.py new file mode 100644 index 000000000000..dde289ad9250 --- /dev/null +++ b/sdk/webpubsub/azure-messaging-webpubsubclient/tests/test_invocation_manager.py @@ -0,0 +1,87 @@ +# coding: utf-8 +# ------------------------------------------------------------------------- +# Copyright (c) Microsoft Corporation. All rights reserved. +# Licensed under the MIT License. See License.txt in the project root for +# license information. +# ------------------------------------------------------------------------- +import pytest +from azure.messaging.webpubsubclient.models._models import ( + InvokeResponseMessage, + InvocationManager, + InvocationError, +) +from azure.messaging.webpubsubclient.models import WebPubSubDataType + + +class TestInvocationManager: + """Test InvocationManager functionality""" + + def test_register_and_resolve(self): + """Test registering and resolving invocations""" + inv_mgr = InvocationManager() + inv_id, entry = inv_mgr.register() + + assert inv_id == "1" + assert entry is not None + assert inv_mgr._entries.get(inv_id) is not None + + response = InvokeResponseMessage( + invocation_id=inv_id, + success=True, + data_type=WebPubSubDataType.TEXT, + data="result", + ) + + resolved = inv_mgr.resolve(response) + assert resolved is True + assert entry.result == response + + def test_register_custom_id(self): + """Test registering with custom invocation ID""" + inv_mgr = InvocationManager() + inv_id, entry = inv_mgr.register("custom-id") + + assert inv_id == "custom-id" + assert inv_mgr._entries.get("custom-id") is not None + + def test_register_duplicate_raises(self): + """Test that registering duplicate ID raises InvocationError""" + inv_mgr = InvocationManager() + inv_mgr.register("same-id") + + with pytest.raises(InvocationError) as exc_info: + inv_mgr.register("same-id") + + assert exc_info.value.invocation_id == "same-id" + + def test_reject(self): + """Test rejecting an invocation""" + inv_mgr = InvocationManager() + inv_id, entry = inv_mgr.register() + + error = Exception("test error") + rejected = inv_mgr.reject(inv_id, error) + + assert rejected is True + assert entry.error == error + + def test_discard(self): + """Test discarding an invocation""" + inv_mgr = InvocationManager() + inv_id, _ = inv_mgr.register() + + assert inv_mgr._entries.get(inv_id) is not None + inv_mgr.discard(inv_id) + assert inv_mgr._entries.get(inv_id) is None + + def test_reject_all(self): + """Test rejecting all pending invocations""" + inv_mgr = InvocationManager() + inv_id1, entry1 = inv_mgr.register() + inv_id2, entry2 = inv_mgr.register() + + inv_mgr.reject_all(lambda inv_id: InvocationError("disconnected", invocation_id=inv_id)) + + assert entry1.error is not None + assert entry2.error is not None + assert len(inv_mgr._entries) == 0 diff --git a/sdk/webpubsub/azure-messaging-webpubsubclient/tests/test_invocation_manager_async.py b/sdk/webpubsub/azure-messaging-webpubsubclient/tests/test_invocation_manager_async.py new file mode 100644 index 000000000000..98c3ff761484 --- /dev/null +++ b/sdk/webpubsub/azure-messaging-webpubsubclient/tests/test_invocation_manager_async.py @@ -0,0 +1,86 @@ +# coding: utf-8 +# ------------------------------------------------------------------------- +# Copyright (c) Microsoft Corporation. All rights reserved. +# Licensed under the MIT License. See License.txt in the project root for +# license information. +# ------------------------------------------------------------------------- +import pytest +from azure.messaging.webpubsubclient.models._models import ( + InvokeResponseMessage, + InvocationManagerAsync, + InvocationError, +) + + +class TestInvocationManagerAsync: + """Test InvocationManagerAsync functionality""" + + def test_register_and_resolve(self): + """Test registering and resolving invocations""" + inv_mgr = InvocationManagerAsync() + inv_id, entry = inv_mgr.register() + + assert inv_id == "1" + assert entry is not None + + response = InvokeResponseMessage( + invocation_id=inv_id, + success=True, + ) + + resolved = inv_mgr.resolve(response) + assert resolved is True + assert entry.result == response + assert entry.event.is_set() + + def test_register_custom_id(self): + """Test registering with custom invocation ID""" + inv_mgr = InvocationManagerAsync() + inv_id, entry = inv_mgr.register("custom-id") + + assert inv_id == "custom-id" + assert inv_mgr._entries.get("custom-id") is not None + + def test_register_duplicate_raises(self): + """Test that registering duplicate ID raises InvocationError""" + inv_mgr = InvocationManagerAsync() + inv_mgr.register("same-id") + + with pytest.raises(InvocationError) as exc_info: + inv_mgr.register("same-id") + + assert exc_info.value.invocation_id == "same-id" + + def test_reject_sets_event(self): + """Test that rejecting sets the event""" + inv_mgr = InvocationManagerAsync() + inv_id, entry = inv_mgr.register() + + error = Exception("test error") + inv_mgr.reject(inv_id, error) + + assert entry.error == error + assert entry.event.is_set() + + def test_discard(self): + """Test discarding an invocation""" + inv_mgr = InvocationManagerAsync() + inv_id, _ = inv_mgr.register() + + assert inv_mgr._entries.get(inv_id) is not None + inv_mgr.discard(inv_id) + assert inv_mgr._entries.get(inv_id) is None + + def test_reject_all(self): + """Test rejecting all pending invocations""" + inv_mgr = InvocationManagerAsync() + inv_id1, entry1 = inv_mgr.register() + inv_id2, entry2 = inv_mgr.register() + + inv_mgr.reject_all(lambda inv_id: InvocationError("disconnected", invocation_id=inv_id)) + + assert entry1.error is not None + assert entry2.error is not None + assert entry1.event.is_set() + assert entry2.event.is_set() + assert len(inv_mgr._entries) == 0 diff --git a/sdk/webpubsub/azure-messaging-webpubsubclient/tests/test_unit.py b/sdk/webpubsub/azure-messaging-webpubsubclient/tests/test_unit.py index 1174dfe33b96..c01cf717cf5c 100644 --- a/sdk/webpubsub/azure-messaging-webpubsubclient/tests/test_unit.py +++ b/sdk/webpubsub/azure-messaging-webpubsubclient/tests/test_unit.py @@ -13,7 +13,10 @@ SendEventMessage, SequenceAckMessage, WebPubSubJsonReliableProtocol, + InvokeMessage, + CancelInvocationMessage, ) +from azure.messaging.webpubsubclient.models import WebPubSubDataType from testcase import WebpubsubClientPowerShellPreparer @@ -160,6 +163,50 @@ def compare_dict(dict1, dict2): "sequenceId": 123456, }, ), + ( + "invoke1", + InvokeMessage( + invocation_id="test-123", + target="event", + event="processOrder", + data_type=WebPubSubDataType.JSON, + data={"orderId": 1}, + ), + { + "type": "invoke", + "invocationId": "test-123", + "target": "event", + "event": "processOrder", + "dataType": "json", + "data": {"orderId": 1}, + }, + ), + ( + "invoke2", + InvokeMessage( + invocation_id="test-456", + target="event", + event="echo", + data_type=WebPubSubDataType.TEXT, + data="hello world", + ), + { + "type": "invoke", + "invocationId": "test-456", + "target": "event", + "event": "echo", + "dataType": "text", + "data": "hello world", + }, + ), + ( + "cancelInvocation1", + CancelInvocationMessage(invocation_id="test-789"), + { + "type": "cancelInvocation", + "invocationId": "test-789", + }, + ), ], ) def test_write_message(testname, message, expect): @@ -356,6 +403,36 @@ def test_write_message(testname, message, expect): }, lambda msg: msg.kind == "disconnected" and msg.message == "message", ), + ( + "invokeResponse1", + { + "type": "invokeResponse", + "invocationId": "test-123", + "success": True, + "dataType": "json", + "data": {"result": "ok"}, + }, + lambda msg: msg.kind == "invokeResponse" + and msg.invocation_id == "test-123" + and msg.success is True + and msg.data_type == "json" + and msg.data == {"result": "ok"}, + ), + ( + "invokeResponse2", + { + "type": "invokeResponse", + "invocationId": "test-456", + "success": False, + "error": {"name": "BadRequest", "message": "Invalid request"}, + }, + lambda msg: msg.kind == "invokeResponse" + and msg.invocation_id == "test-456" + and msg.success is False + and msg.error is not None + and msg.error.name == "BadRequest" + and msg.error.message == "Invalid request", + ), ], ) def test_parse_message(testname, message, assert_func): From 625a97205d78144f8098687d5299687346c9062d Mon Sep 17 00:00:00 2001 From: Shiying Chen Date: Sat, 28 Feb 2026 16:54:13 +0800 Subject: [PATCH 2/8] fix ci and add more mock tests --- .../messaging/webpubsubclient/_client.py | 8 +- .../messaging/webpubsubclient/aio/_client.py | 5 +- .../webpubsubclient/models/_models.py | 6 +- .../tests/test_invoke_event.py | 253 ++++++++++++++++ .../tests/test_invoke_event_async.py | 278 ++++++++++++++++++ 5 files changed, 542 insertions(+), 8 deletions(-) create mode 100644 sdk/webpubsub/azure-messaging-webpubsubclient/tests/test_invoke_event.py create mode 100644 sdk/webpubsub/azure-messaging-webpubsubclient/tests/test_invoke_event_async.py diff --git a/sdk/webpubsub/azure-messaging-webpubsubclient/azure/messaging/webpubsubclient/_client.py b/sdk/webpubsub/azure-messaging-webpubsubclient/azure/messaging/webpubsubclient/_client.py index b6d53887606f..55ad33c5192b 100644 --- a/sdk/webpubsub/azure-messaging-webpubsubclient/azure/messaging/webpubsubclient/_client.py +++ b/sdk/webpubsub/azure-messaging-webpubsubclient/azure/messaging/webpubsubclient/_client.py @@ -157,7 +157,7 @@ class WebPubSubClientBase: # pylint: disable=client-accepts-api-version-keyword retry will sleep for [0.0s, 0.2s, 0.4s, ...] between retries. The default value is 0.8. :keyword float reconnect_retry_backoff_max: The maximum back off time. Default value is 120.0 seconds :keyword ~azure.messaging.webpubsubclient.RetryMode reconnect_retry_mode: Fixed or exponential delay - between attemps, default is exponential. + between attempts, default is exponential. :keyword int message_retry_total: total number of retries to allow for sending message. Default is 3. :keyword float message_retry_backoff_factor: A backoff factor to apply between attempts after the second try (most errors are resolved immediately by a second try without a delay). In fixed mode, retry policy will always @@ -165,7 +165,7 @@ class WebPubSubClientBase: # pylint: disable=client-accepts-api-version-keyword "{backoff factor} * (2 ** ({number of retries} - 1))" seconds. If the backoff_factor is 0.1, then the retry will sleep for [0.0s, 0.2s, 0.4s, ...] between retries. The default value is 0.8. :keyword float message_retry_backoff_max: The maximum back off time. Default value is 120.0 seconds - :keyword RetryMode message_retry_mode: Fixed or exponential delay between attemps, default is exponential. + :keyword RetryMode message_retry_mode: Fixed or exponential delay between attempts, default is exponential. :keyword bool auto_rejoin_groups: auto_rejoin_groups, default is True :keyword bool logging_enable: Whether to output network trace logs to the logger. Default is `False`. :keyword float ack_timeout: Time limit to wait for ack message from server. The default value is 30.0 seconds. @@ -297,7 +297,7 @@ class WebPubSubClient( retry will sleep for [0.0s, 0.2s, 0.4s, ...] between retries. The default value is 0.8. :keyword float reconnect_retry_backoff_max: The maximum back off time. Default value is 120.0 seconds :keyword ~azure.messaging.webpubsubclient.RetryMode reconnect_retry_mode: Fixed or exponential delay - between attemps, default is exponential. + between attempts, default is exponential. :keyword int message_retry_total: total number of retries to allow for sending message. Default is 3. :keyword float message_retry_backoff_factor: A backoff factor to apply between attempts after the second try (most errors are resolved immediately by a second try without a delay). In fixed mode, retry policy will always @@ -305,7 +305,7 @@ class WebPubSubClient( "{backoff factor} * (2 ** ({number of retries} - 1))" seconds. If the backoff_factor is 0.1, then the retry will sleep for [0.0s, 0.2s, 0.4s, ...] between retries. The default value is 0.8. :keyword float message_retry_backoff_max: The maximum back off time. Default value is 120.0 seconds - :keyword RetryMode message_retry_mode: Fixed or exponential delay between attemps, default is exponential. + :keyword RetryMode message_retry_mode: Fixed or exponential delay between attempts, default is exponential. :keyword bool auto_rejoin_groups: auto_rejoin_groups, default is True :keyword bool logging_enable: Whether to output network trace logs to the logger. Default is `False`. :keyword float ack_timeout: Time limit to wait for ack message from server. The default value is 30.0 seconds. diff --git a/sdk/webpubsub/azure-messaging-webpubsubclient/azure/messaging/webpubsubclient/aio/_client.py b/sdk/webpubsub/azure-messaging-webpubsubclient/azure/messaging/webpubsubclient/aio/_client.py index eb0b86edda49..1d6b0bd4aabd 100644 --- a/sdk/webpubsub/azure-messaging-webpubsubclient/azure/messaging/webpubsubclient/aio/_client.py +++ b/sdk/webpubsub/azure-messaging-webpubsubclient/azure/messaging/webpubsubclient/aio/_client.py @@ -219,7 +219,7 @@ class WebPubSubClient( retry will sleep for [0.0s, 0.2s, 0.4s, ...] between retries. The default value is 0.8. :keyword float reconnect_retry_backoff_max: The maximum back off time. Default value is 120.0 seconds :keyword ~azure.messaging.webpubsubclient.RetryMode reconnect_retry_mode: Fixed or exponential delay - between attemps, default is exponential. + between attempts, default is exponential. :keyword int message_retry_total: total number of retries to allow for sending message. Default is 3. :keyword float message_retry_backoff_factor: A backoff factor to apply between attempts after the second try (most errors are resolved immediately by a second try without a delay). In fixed mode, retry policy will always @@ -227,7 +227,7 @@ class WebPubSubClient( "{backoff factor} * (2 ** ({number of retries} - 1))" seconds. If the backoff_factor is 0.1, then the retry will sleep for [0.0s, 0.2s, 0.4s, ...] between retries. The default value is 0.8. :keyword float message_retry_backoff_max: The maximum back off time. Default value is 120.0 seconds - :keyword RetryMode message_retry_mode: Fixed or exponential delay between attemps, default is exponential. + :keyword RetryMode message_retry_mode: Fixed or exponential delay between attempts, default is exponential. :keyword bool auto_rejoin_groups: auto_rejoin_groups, default is True :keyword bool logging_enable: Whether to output network trace logs to the logger. Default is `False`. :keyword float ack_timeout: Time limit to wait for ack message from server. The default value is 30.0 seconds. @@ -756,6 +756,7 @@ async def _invoke_event_core( return self._map_invoke_response(entry.result) except asyncio.TimeoutError as e: + await self._send_cancel_invocation(invocation_id) raise InvocationError( "Timeout while waiting for invoke response.", invocation_id=invocation_id, diff --git a/sdk/webpubsub/azure-messaging-webpubsubclient/azure/messaging/webpubsubclient/models/_models.py b/sdk/webpubsub/azure-messaging-webpubsubclient/azure/messaging/webpubsubclient/models/_models.py index a978c1f9383b..2b81394433dc 100644 --- a/sdk/webpubsub/azure-messaging-webpubsubclient/azure/messaging/webpubsubclient/models/_models.py +++ b/sdk/webpubsub/azure-messaging-webpubsubclient/azure/messaging/webpubsubclient/models/_models.py @@ -588,7 +588,7 @@ def __init__(self, invocation_id: str) -> None: class InvokeEventResult: - """Result of invokeEvent + """Result of invoke_event :ivar invocation_id: Invocation identifier correlated with the response. Required. :vartype invocation_id: str @@ -704,6 +704,7 @@ def parse_messages( GroupDataMessage, ServerDataMessage, AckMessage, + InvokeResponseMessage, None, ]: """Parse messages from raw message @@ -711,7 +712,8 @@ def parse_messages( :param raw_message: The raw message. Required. :type raw_message: str :return: The parsed message. - :rtype: Union[ConnectedMessage, DisconnectedMessage, GroupDataMessage, ServerDataMessage, AckMessage, None] + :rtype: Union[ConnectedMessage, DisconnectedMessage, GroupDataMessage, ServerDataMessage, AckMessage, + InvokeResponseMessage, None] :raises ValueError: If raw_message is None or raw_message type is not string. """ if raw_message is None: diff --git a/sdk/webpubsub/azure-messaging-webpubsubclient/tests/test_invoke_event.py b/sdk/webpubsub/azure-messaging-webpubsubclient/tests/test_invoke_event.py new file mode 100644 index 000000000000..8dd3146f1fca --- /dev/null +++ b/sdk/webpubsub/azure-messaging-webpubsubclient/tests/test_invoke_event.py @@ -0,0 +1,253 @@ +# coding: utf-8 +# ------------------------------------------------------------------------- +# Copyright (c) Microsoft Corporation. All rights reserved. +# Licensed under the MIT License. See License.txt in the project root for +# license information. +# ------------------------------------------------------------------------- +"""Client-level mock tests for invoke_event (sync) exercising request/response +correlation, timeout, error mapping and disconnect rejection. +""" + +import threading +import pytest +from unittest.mock import patch, MagicMock + +from azure.messaging.webpubsubclient import WebPubSubClient +from azure.messaging.webpubsubclient.models import WebPubSubDataType +from azure.messaging.webpubsubclient.models._models import ( + InvokeResponseMessage, + InvokeResponseError, + InvocationError, +) + + +def _make_client() -> WebPubSubClient: + client = WebPubSubClient( + "wss://fake.webpubsub.azure.com", + message_retry_total=0, + ) + return client + + +class TestInvokeEventSync: + """Client-level sync invoke_event tests with mocked websocket.""" + + def test_request_response_correlation_text(self): + """invoke_event returns the correct InvokeEventResult for a successful text response.""" + client = _make_client() + + def fake_send(message, **kwargs): + # Simulate the service replying with invokeResponse on another thread + def reply(): + client._invocation_map.resolve( + InvokeResponseMessage( + invocation_id=message.invocation_id, + success=True, + data_type=WebPubSubDataType.TEXT, + data="pong", + ) + ) + + threading.Thread(target=reply, daemon=True).start() + + with patch.object(client, "_send_message", side_effect=fake_send): + result = client.invoke_event("echo", "ping", WebPubSubDataType.TEXT) + + assert result.data == "pong" + assert result.data_type == WebPubSubDataType.TEXT + assert result.invocation_id is not None + + def test_request_response_correlation_json(self): + """invoke_event returns the correct InvokeEventResult for a successful JSON response.""" + client = _make_client() + + def fake_send(message, **kwargs): + def reply(): + client._invocation_map.resolve( + InvokeResponseMessage( + invocation_id=message.invocation_id, + success=True, + data_type=WebPubSubDataType.JSON, + data={"result": "ok"}, + ) + ) + + threading.Thread(target=reply, daemon=True).start() + + with patch.object(client, "_send_message", side_effect=fake_send): + result = client.invoke_event( + "processOrder", {"orderId": 1}, WebPubSubDataType.JSON + ) + + assert result.data == {"result": "ok"} + assert result.data_type == WebPubSubDataType.JSON + + def test_custom_invocation_id(self): + """When a custom invocation_id is provided it is used in the result.""" + client = _make_client() + + def fake_send(message, **kwargs): + assert message.invocation_id == "my-custom-id" + + def reply(): + client._invocation_map.resolve( + InvokeResponseMessage( + invocation_id=message.invocation_id, + success=True, + data_type=WebPubSubDataType.TEXT, + data="ok", + ) + ) + + threading.Thread(target=reply, daemon=True).start() + + with patch.object(client, "_send_message", side_effect=fake_send): + result = client.invoke_event( + "echo", "hi", WebPubSubDataType.TEXT, invocation_id="my-custom-id" + ) + + assert result.invocation_id == "my-custom-id" + + def test_timeout_raises_invocation_error(self): + """invoke_event raises InvocationError when the response does not arrive within timeout.""" + client = _make_client() + + # _send_message succeeds but no invokeResponse arrives + with patch.object(client, "_send_message"): + with pytest.raises(InvocationError) as exc_info: + client.invoke_event( + "slowEvent", "data", WebPubSubDataType.TEXT, timeout=0.1 + ) + + assert "Timeout" in str(exc_info.value) + + def test_error_response_success_false(self): + """invoke_event raises InvocationError with error_detail when success == False.""" + client = _make_client() + + def fake_send(message, **kwargs): + def reply(): + client._invocation_map.resolve( + InvokeResponseMessage( + invocation_id=message.invocation_id, + success=False, + error=InvokeResponseError( + name="BadRequest", message="Invalid payload" + ), + ) + ) + + threading.Thread(target=reply, daemon=True).start() + + with patch.object(client, "_send_message", side_effect=fake_send): + with pytest.raises(InvocationError) as exc_info: + client.invoke_event("fail", "data", WebPubSubDataType.TEXT) + + err = exc_info.value + assert err.error_detail is not None + assert err.error_detail.name == "BadRequest" + assert err.error_detail.message == "Invalid payload" + + def test_error_response_success_false_no_error_detail(self): + """invoke_event raises InvocationError with default message when success == False and no error detail.""" + client = _make_client() + + def fake_send(message, **kwargs): + def reply(): + client._invocation_map.resolve( + InvokeResponseMessage( + invocation_id=message.invocation_id, + success=False, + error=None, + ) + ) + + threading.Thread(target=reply, daemon=True).start() + + with patch.object(client, "_send_message", side_effect=fake_send): + with pytest.raises(InvocationError) as exc_info: + client.invoke_event("fail", "data", WebPubSubDataType.TEXT) + + assert "Invocation failed" in str(exc_info.value) + + def test_send_failure_raises_invocation_error(self): + """invoke_event raises InvocationError when _send_message fails.""" + client = _make_client() + + with patch.object( + client, "_send_message", side_effect=Exception("connection lost") + ): + with pytest.raises(InvocationError) as exc_info: + client.invoke_event("event", "data", WebPubSubDataType.TEXT) + + assert "connection lost" in str(exc_info.value) + + def test_disconnect_rejects_pending_invocation(self): + """Pending invocations are rejected when reject_all is called (simulating disconnect).""" + client = _make_client() + errors_received = [] + + def fake_send(message, **kwargs): + # Simulate the connection dropping after the message is sent + def disconnect(): + client._invocation_map.reject_all( + lambda inv_id: InvocationError( + "Connection is disconnected", + invocation_id=inv_id, + ) + ) + + threading.Thread(target=disconnect, daemon=True).start() + + with patch.object(client, "_send_message", side_effect=fake_send): + with pytest.raises(InvocationError) as exc_info: + client.invoke_event("event", "data", WebPubSubDataType.TEXT) + + assert "disconnected" in str(exc_info.value).lower() + + def test_invocation_entry_cleaned_up_after_success(self): + """The invocation entry is discarded from the map after a successful call.""" + client = _make_client() + + def fake_send(message, **kwargs): + def reply(): + client._invocation_map.resolve( + InvokeResponseMessage( + invocation_id=message.invocation_id, + success=True, + data_type=WebPubSubDataType.TEXT, + data="ok", + ) + ) + + threading.Thread(target=reply, daemon=True).start() + + with patch.object(client, "_send_message", side_effect=fake_send): + client.invoke_event("echo", "hi", WebPubSubDataType.TEXT) + + # Entry should have been discarded + assert len(client._invocation_map._entries) == 0 + + def test_invocation_entry_cleaned_up_after_error(self): + """The invocation entry is discarded from the map after an error.""" + client = _make_client() + + def fake_send(message, **kwargs): + def reply(): + client._invocation_map.resolve( + InvokeResponseMessage( + invocation_id=message.invocation_id, + success=False, + error=InvokeResponseError( + name="Error", message="fail" + ), + ) + ) + + threading.Thread(target=reply, daemon=True).start() + + with patch.object(client, "_send_message", side_effect=fake_send): + with pytest.raises(InvocationError): + client.invoke_event("fail", "data", WebPubSubDataType.TEXT) + + assert len(client._invocation_map._entries) == 0 diff --git a/sdk/webpubsub/azure-messaging-webpubsubclient/tests/test_invoke_event_async.py b/sdk/webpubsub/azure-messaging-webpubsubclient/tests/test_invoke_event_async.py new file mode 100644 index 000000000000..6fe5638a5746 --- /dev/null +++ b/sdk/webpubsub/azure-messaging-webpubsubclient/tests/test_invoke_event_async.py @@ -0,0 +1,278 @@ +# coding: utf-8 +# ------------------------------------------------------------------------- +# Copyright (c) Microsoft Corporation. All rights reserved. +# Licensed under the MIT License. See License.txt in the project root for +# license information. +# ------------------------------------------------------------------------- +"""Client-level mocktests for invoke_event (async) exercising request/response +correlation, timeout, error mapping and disconnect rejection. +""" + +import asyncio +import pytest +from unittest.mock import patch, AsyncMock + +from azure.messaging.webpubsubclient.aio import WebPubSubClient as WebPubSubClientAsync +from azure.messaging.webpubsubclient.models import WebPubSubDataType +from azure.messaging.webpubsubclient.models._models import ( + InvokeResponseMessage, + InvokeResponseError, + InvocationError, +) + + +def _make_client() -> WebPubSubClientAsync: + client = WebPubSubClientAsync( + "wss://fake.webpubsub.azure.com", + message_retry_total=0, + ) + return client + + +@pytest.mark.asyncio +class TestInvokeEventAsync: + """Client-level async invoke_event tests with mocked websocket.""" + + async def test_request_response_correlation_text(self): + """invoke_event returns the correct InvokeEventResult for a successful text response.""" + client = _make_client() + + async def fake_send(message, **kwargs): + # Schedule the reply so it arrives after invoke_event starts waiting + async def reply(): + client._invocation_map.resolve( + InvokeResponseMessage( + invocation_id=message.invocation_id, + success=True, + data_type=WebPubSubDataType.TEXT, + data="pong", + ) + ) + + asyncio.get_event_loop().call_soon(lambda: asyncio.ensure_future(reply())) + + with patch.object(client, "_send_message", side_effect=fake_send): + result = await client.invoke_event("echo", "ping", WebPubSubDataType.TEXT) + + assert result.data == "pong" + assert result.data_type == WebPubSubDataType.TEXT + assert result.invocation_id is not None + + async def test_request_response_correlation_json(self): + """invoke_event returns the correct InvokeEventResult for a successful JSON response.""" + client = _make_client() + + async def fake_send(message, **kwargs): + async def reply(): + client._invocation_map.resolve( + InvokeResponseMessage( + invocation_id=message.invocation_id, + success=True, + data_type=WebPubSubDataType.JSON, + data={"result": "ok"}, + ) + ) + + asyncio.get_event_loop().call_soon(lambda: asyncio.ensure_future(reply())) + + with patch.object(client, "_send_message", side_effect=fake_send): + result = await client.invoke_event( + "processOrder", {"orderId": 1}, WebPubSubDataType.JSON + ) + + assert result.data == {"result": "ok"} + assert result.data_type == WebPubSubDataType.JSON + + async def test_custom_invocation_id(self): + """When a custom invocation_id is provided it is used in the result.""" + client = _make_client() + + async def fake_send(message, **kwargs): + assert message.invocation_id == "my-custom-id" + + async def reply(): + client._invocation_map.resolve( + InvokeResponseMessage( + invocation_id=message.invocation_id, + success=True, + data_type=WebPubSubDataType.TEXT, + data="ok", + ) + ) + + asyncio.get_event_loop().call_soon(lambda: asyncio.ensure_future(reply())) + + with patch.object(client, "_send_message", side_effect=fake_send): + result = await client.invoke_event( + "echo", "hi", WebPubSubDataType.TEXT, invocation_id="my-custom-id" + ) + + assert result.invocation_id == "my-custom-id" + + async def test_timeout_raises_invocation_error(self): + """invoke_event raises InvocationError when the response does not arrive within timeout.""" + client = _make_client() + + async def fake_send(message, **kwargs): + pass # No reply — simulates a timeout + + with patch.object(client, "_send_message", side_effect=fake_send): + with pytest.raises(InvocationError) as exc_info: + await client.invoke_event( + "slowEvent", "data", WebPubSubDataType.TEXT, timeout=0.1 + ) + + assert "Timeout" in str(exc_info.value) + + async def test_timeout_sends_cancel_invocation(self): + """invoke_event sends cancelInvocation when timeout occurs.""" + client = _make_client() + cancel_ids = [] + + original_send_cancel = client._send_cancel_invocation + + async def track_cancel(invocation_id): + cancel_ids.append(invocation_id) + await original_send_cancel(invocation_id) + + async def fake_send(message, **kwargs): + pass # No reply — simulates a timeout + + with patch.object(client, "_send_message", side_effect=fake_send): + with patch.object(client, "_send_cancel_invocation", side_effect=track_cancel): + with pytest.raises(InvocationError): + await client.invoke_event( + "slowEvent", "data", WebPubSubDataType.TEXT, timeout=0.1 + ) + + assert len(cancel_ids) == 1 + + async def test_error_response_success_false(self): + """invoke_event raises InvocationError with error_detail when success == False.""" + client = _make_client() + + async def fake_send(message, **kwargs): + async def reply(): + client._invocation_map.resolve( + InvokeResponseMessage( + invocation_id=message.invocation_id, + success=False, + error=InvokeResponseError( + name="BadRequest", message="Invalid payload" + ), + ) + ) + + asyncio.get_event_loop().call_soon(lambda: asyncio.ensure_future(reply())) + + with patch.object(client, "_send_message", side_effect=fake_send): + with pytest.raises(InvocationError) as exc_info: + await client.invoke_event("fail", "data", WebPubSubDataType.TEXT) + + err = exc_info.value + assert err.error_detail is not None + assert err.error_detail.name == "BadRequest" + assert err.error_detail.message == "Invalid payload" + + async def test_error_response_success_false_no_error_detail(self): + """invoke_event raises InvocationError with default message when success == False and no error detail.""" + client = _make_client() + + async def fake_send(message, **kwargs): + async def reply(): + client._invocation_map.resolve( + InvokeResponseMessage( + invocation_id=message.invocation_id, + success=False, + error=None, + ) + ) + + asyncio.get_event_loop().call_soon(lambda: asyncio.ensure_future(reply())) + + with patch.object(client, "_send_message", side_effect=fake_send): + with pytest.raises(InvocationError) as exc_info: + await client.invoke_event("fail", "data", WebPubSubDataType.TEXT) + + assert "Invocation failed" in str(exc_info.value) + + async def test_send_failure_raises_invocation_error(self): + """invoke_event raises InvocationError when _send_message fails.""" + client = _make_client() + + async def fake_send(message, **kwargs): + raise Exception("connection lost") + + with patch.object(client, "_send_message", side_effect=fake_send): + with pytest.raises(InvocationError) as exc_info: + await client.invoke_event("event", "data", WebPubSubDataType.TEXT) + + assert "connection lost" in str(exc_info.value) + + async def test_disconnect_rejects_pending_invocation(self): + """Pending invocations are rejected when reject_all is called (simulating disconnect).""" + client = _make_client() + + async def fake_send(message, **kwargs): + # Simulate the connection dropping after the message is sent + async def disconnect(): + client._invocation_map.reject_all( + lambda inv_id: InvocationError( + "Connection is disconnected", + invocation_id=inv_id, + ) + ) + + asyncio.get_event_loop().call_soon(lambda: asyncio.ensure_future(disconnect())) + + with patch.object(client, "_send_message", side_effect=fake_send): + with pytest.raises(InvocationError) as exc_info: + await client.invoke_event("event", "data", WebPubSubDataType.TEXT) + + assert "disconnected" in str(exc_info.value).lower() + + async def test_invocation_entry_cleaned_up_after_success(self): + """The invocation entry is discarded from the map after a successful call.""" + client = _make_client() + + async def fake_send(message, **kwargs): + async def reply(): + client._invocation_map.resolve( + InvokeResponseMessage( + invocation_id=message.invocation_id, + success=True, + data_type=WebPubSubDataType.TEXT, + data="ok", + ) + ) + + asyncio.get_event_loop().call_soon(lambda: asyncio.ensure_future(reply())) + + with patch.object(client, "_send_message", side_effect=fake_send): + await client.invoke_event("echo", "hi", WebPubSubDataType.TEXT) + + assert len(client._invocation_map._entries) == 0 + + async def test_invocation_entry_cleaned_up_after_error(self): + """The invocation entry is discarded from the map after an error.""" + client = _make_client() + + async def fake_send(message, **kwargs): + async def reply(): + client._invocation_map.resolve( + InvokeResponseMessage( + invocation_id=message.invocation_id, + success=False, + error=InvokeResponseError( + name="Error", message="fail" + ), + ) + ) + + asyncio.get_event_loop().call_soon(lambda: asyncio.ensure_future(reply())) + + with patch.object(client, "_send_message", side_effect=fake_send): + with pytest.raises(InvocationError): + await client.invoke_event("fail", "data", WebPubSubDataType.TEXT) + + assert len(client._invocation_map._entries) == 0 From f7fefcd6143f49962d2b7f74d569b27ede98dcda Mon Sep 17 00:00:00 2001 From: Shiying Chen Date: Sat, 28 Feb 2026 16:55:30 +0800 Subject: [PATCH 3/8] add changelog --- sdk/webpubsub/azure-messaging-webpubsubclient/CHANGELOG.md | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/sdk/webpubsub/azure-messaging-webpubsubclient/CHANGELOG.md b/sdk/webpubsub/azure-messaging-webpubsubclient/CHANGELOG.md index f2a69dcc0265..69eb72ea9869 100644 --- a/sdk/webpubsub/azure-messaging-webpubsubclient/CHANGELOG.md +++ b/sdk/webpubsub/azure-messaging-webpubsubclient/CHANGELOG.md @@ -1,11 +1,15 @@ # Release History -## 1.1.1 (2024-XX-XX) +## 1.2.0 (2026-XX-XX) ### Other Changes - Clean useless warnings in log +### Features Added + +- Add preview API `invoke_event` + ## 1.1.0 (2024-04-24) ### Features Added From 1cf9ead57e3a3dd3638d1c77ee13ba579127f7c0 Mon Sep 17 00:00:00 2001 From: Shiying Chen Date: Mon, 2 Mar 2026 10:47:03 +0800 Subject: [PATCH 4/8] modify change log --- .../azure-messaging-webpubsubclient/CHANGELOG.md | 12 +++++++----- 1 file changed, 7 insertions(+), 5 deletions(-) diff --git a/sdk/webpubsub/azure-messaging-webpubsubclient/CHANGELOG.md b/sdk/webpubsub/azure-messaging-webpubsubclient/CHANGELOG.md index 69eb72ea9869..a7772f802989 100644 --- a/sdk/webpubsub/azure-messaging-webpubsubclient/CHANGELOG.md +++ b/sdk/webpubsub/azure-messaging-webpubsubclient/CHANGELOG.md @@ -1,15 +1,17 @@ # Release History -## 1.2.0 (2026-XX-XX) - -### Other Changes - -- Clean useless warnings in log +## 1.2.0 (Unreleased) ### Features Added - Add preview API `invoke_event` +## 1.1.1 (Unreleased) + +### Other Changes + +- Clean useless warnings in log + ## 1.1.0 (2024-04-24) ### Features Added From 72ea3e5a20835a6ad9d877884adc942217991e3f Mon Sep 17 00:00:00 2001 From: Shiying Chen Date: Mon, 9 Mar 2026 16:23:21 +0800 Subject: [PATCH 5/8] add cancel_invocation --- .../azure-messaging-webpubsubclient/README.md | 42 +++++++++++++++++++ .../messaging/webpubsubclient/_client.py | 14 +++++++ .../messaging/webpubsubclient/aio/_client.py | 14 +++++++ 3 files changed, 70 insertions(+) diff --git a/sdk/webpubsub/azure-messaging-webpubsubclient/README.md b/sdk/webpubsub/azure-messaging-webpubsubclient/README.md index e1f905d05a19..ad1d5816b977 100644 --- a/sdk/webpubsub/azure-messaging-webpubsubclient/README.md +++ b/sdk/webpubsub/azure-messaging-webpubsubclient/README.md @@ -91,6 +91,48 @@ with client: print(f"Invocation result: {result.data}") ``` +You can pass a `timeout` (in seconds) to limit how long the client waits for the response. An `InvocationError` is raised if the timeout elapses. + +```python +from azure.messaging.webpubsubclient import WebPubSubClient +from azure.messaging.webpubsubclient.models import WebPubSubDataType, InvocationError + +client = WebPubSubClient("") +with client: + try: + result = client.invoke_event( + "processOrder", {"orderId": 1}, WebPubSubDataType.JSON, timeout=5.0, + ) + print(f"Invocation result: {result.data}") + except InvocationError as e: + print(f"Invocation timed out: {e}") +``` + +To cancel a pending invocation by its invocation ID, you can use `cancel_invocation` to send a cancel message to the server. + +```python +import asyncio +from azure.messaging.webpubsubclient.aio import WebPubSubClient +from azure.messaging.webpubsubclient.models import WebPubSubDataType, InvocationError + +client = WebPubSubClient("") +async with client: + invocation_id = "my-invocation" + + async def invoke(): + try: + result = await client.invoke_event( + "processOrder", {"orderId": 1}, WebPubSubDataType.JSON, invocation_id=invocation_id, + ) + except InvocationError as e: + print(f"Invocation cancelled: {e}") + + task = asyncio.create_task(invoke()) + + await client.cancel_invocation(invocation_id) + await task +``` + _Streaming and service-initiated invocations are not yet supported._ --- diff --git a/sdk/webpubsub/azure-messaging-webpubsubclient/azure/messaging/webpubsubclient/_client.py b/sdk/webpubsub/azure-messaging-webpubsubclient/azure/messaging/webpubsubclient/_client.py index 55ad33c5192b..17aee2bd9314 100644 --- a/sdk/webpubsub/azure-messaging-webpubsubclient/azure/messaging/webpubsubclient/_client.py +++ b/sdk/webpubsub/azure-messaging-webpubsubclient/azure/messaging/webpubsubclient/_client.py @@ -876,6 +876,20 @@ def _send_cancel_invocation(self, invocation_id: str) -> None: except Exception as e: # pylint: disable=broad-except _LOGGER.debug("Failed to send cancelInvocation for %s: %s", invocation_id, e) + def cancel_invocation(self, invocation_id: str) -> None: + """Cancel a pending invocation. + + Sends a cancelInvocation message to the server so that the waiting thread is unblocked. + + :param invocation_id: The invocation id to cancel. Required. + :type invocation_id: str + """ + self._invocation_map.reject( + invocation_id, + InvocationError("Invocation cancelled by the user.", invocation_id=invocation_id), + ) + self._send_cancel_invocation(invocation_id) + def _retry_with_result(self, func: Callable[[], InvokeEventResult]) -> InvokeEventResult: retry_attempt = 0 while True: diff --git a/sdk/webpubsub/azure-messaging-webpubsubclient/azure/messaging/webpubsubclient/aio/_client.py b/sdk/webpubsub/azure-messaging-webpubsubclient/azure/messaging/webpubsubclient/aio/_client.py index 1d6b0bd4aabd..d2c7d675aa10 100644 --- a/sdk/webpubsub/azure-messaging-webpubsubclient/azure/messaging/webpubsubclient/aio/_client.py +++ b/sdk/webpubsub/azure-messaging-webpubsubclient/azure/messaging/webpubsubclient/aio/_client.py @@ -794,6 +794,20 @@ async def _send_cancel_invocation(self, invocation_id: str) -> None: except Exception as e: # pylint: disable=broad-except _LOGGER.debug("Failed to send cancelInvocation for %s: %s", invocation_id, e) + async def cancel_invocation(self, invocation_id: str) -> None: + """Cancel a pending invocation. + + Sends a cancelInvocation message to the server so that the waiting task is unblocked. + + :param invocation_id: The invocation id to cancel. Required. + :type invocation_id: str + """ + self._invocation_map.reject( + invocation_id, + InvocationError("Invocation cancelled by the user.", invocation_id=invocation_id), + ) + await self._send_cancel_invocation(invocation_id) + async def _retry_with_result(self, func: Callable[[], Awaitable[InvokeEventResult]]) -> InvokeEventResult: retry_attempt = 0 while True: From 7a3fee90a77690d685f68ffe0ad156d315e8adf7 Mon Sep 17 00:00:00 2001 From: Shiying Chen Date: Mon, 9 Mar 2026 17:10:29 +0800 Subject: [PATCH 6/8] remove cancel_invocation --- .../azure-messaging-webpubsubclient/README.md | 25 ------------------- .../messaging/webpubsubclient/_client.py | 14 ----------- .../messaging/webpubsubclient/aio/_client.py | 14 ----------- 3 files changed, 53 deletions(-) diff --git a/sdk/webpubsub/azure-messaging-webpubsubclient/README.md b/sdk/webpubsub/azure-messaging-webpubsubclient/README.md index ad1d5816b977..cf935dc5f38d 100644 --- a/sdk/webpubsub/azure-messaging-webpubsubclient/README.md +++ b/sdk/webpubsub/azure-messaging-webpubsubclient/README.md @@ -108,31 +108,6 @@ with client: print(f"Invocation timed out: {e}") ``` -To cancel a pending invocation by its invocation ID, you can use `cancel_invocation` to send a cancel message to the server. - -```python -import asyncio -from azure.messaging.webpubsubclient.aio import WebPubSubClient -from azure.messaging.webpubsubclient.models import WebPubSubDataType, InvocationError - -client = WebPubSubClient("") -async with client: - invocation_id = "my-invocation" - - async def invoke(): - try: - result = await client.invoke_event( - "processOrder", {"orderId": 1}, WebPubSubDataType.JSON, invocation_id=invocation_id, - ) - except InvocationError as e: - print(f"Invocation cancelled: {e}") - - task = asyncio.create_task(invoke()) - - await client.cancel_invocation(invocation_id) - await task -``` - _Streaming and service-initiated invocations are not yet supported._ --- diff --git a/sdk/webpubsub/azure-messaging-webpubsubclient/azure/messaging/webpubsubclient/_client.py b/sdk/webpubsub/azure-messaging-webpubsubclient/azure/messaging/webpubsubclient/_client.py index 17aee2bd9314..55ad33c5192b 100644 --- a/sdk/webpubsub/azure-messaging-webpubsubclient/azure/messaging/webpubsubclient/_client.py +++ b/sdk/webpubsub/azure-messaging-webpubsubclient/azure/messaging/webpubsubclient/_client.py @@ -876,20 +876,6 @@ def _send_cancel_invocation(self, invocation_id: str) -> None: except Exception as e: # pylint: disable=broad-except _LOGGER.debug("Failed to send cancelInvocation for %s: %s", invocation_id, e) - def cancel_invocation(self, invocation_id: str) -> None: - """Cancel a pending invocation. - - Sends a cancelInvocation message to the server so that the waiting thread is unblocked. - - :param invocation_id: The invocation id to cancel. Required. - :type invocation_id: str - """ - self._invocation_map.reject( - invocation_id, - InvocationError("Invocation cancelled by the user.", invocation_id=invocation_id), - ) - self._send_cancel_invocation(invocation_id) - def _retry_with_result(self, func: Callable[[], InvokeEventResult]) -> InvokeEventResult: retry_attempt = 0 while True: diff --git a/sdk/webpubsub/azure-messaging-webpubsubclient/azure/messaging/webpubsubclient/aio/_client.py b/sdk/webpubsub/azure-messaging-webpubsubclient/azure/messaging/webpubsubclient/aio/_client.py index d2c7d675aa10..1d6b0bd4aabd 100644 --- a/sdk/webpubsub/azure-messaging-webpubsubclient/azure/messaging/webpubsubclient/aio/_client.py +++ b/sdk/webpubsub/azure-messaging-webpubsubclient/azure/messaging/webpubsubclient/aio/_client.py @@ -794,20 +794,6 @@ async def _send_cancel_invocation(self, invocation_id: str) -> None: except Exception as e: # pylint: disable=broad-except _LOGGER.debug("Failed to send cancelInvocation for %s: %s", invocation_id, e) - async def cancel_invocation(self, invocation_id: str) -> None: - """Cancel a pending invocation. - - Sends a cancelInvocation message to the server so that the waiting task is unblocked. - - :param invocation_id: The invocation id to cancel. Required. - :type invocation_id: str - """ - self._invocation_map.reject( - invocation_id, - InvocationError("Invocation cancelled by the user.", invocation_id=invocation_id), - ) - await self._send_cancel_invocation(invocation_id) - async def _retry_with_result(self, func: Callable[[], Awaitable[InvokeEventResult]]) -> InvokeEventResult: retry_attempt = 0 while True: From 14ea3a609cb7b1ee8a8b75420e7a8407bb6ae825 Mon Sep 17 00:00:00 2001 From: shiyingchen Date: Fri, 17 Jul 2026 17:40:57 +0800 Subject: [PATCH 7/8] update for copilot comments --- .../azure-messaging-webpubsubclient/README.md | 2 +- .../messaging/webpubsubclient/_client.py | 39 +++++++++++-------- .../messaging/webpubsubclient/aio/_client.py | 4 +- .../tests/test_invoke_event.py | 19 +++++++++ .../tests/test_invoke_event_async.py | 2 +- 5 files changed, 46 insertions(+), 20 deletions(-) diff --git a/sdk/webpubsub/azure-messaging-webpubsubclient/README.md b/sdk/webpubsub/azure-messaging-webpubsubclient/README.md index cf935dc5f38d..fc078253218c 100644 --- a/sdk/webpubsub/azure-messaging-webpubsubclient/README.md +++ b/sdk/webpubsub/azure-messaging-webpubsubclient/README.md @@ -105,7 +105,7 @@ with client: ) print(f"Invocation result: {result.data}") except InvocationError as e: - print(f"Invocation timed out: {e}") + print(f"Invocation failed: {e}") ``` _Streaming and service-initiated invocations are not yet supported._ diff --git a/sdk/webpubsub/azure-messaging-webpubsubclient/azure/messaging/webpubsubclient/_client.py b/sdk/webpubsub/azure-messaging-webpubsubclient/azure/messaging/webpubsubclient/_client.py index 55ad33c5192b..a6c0d6b96dd3 100644 --- a/sdk/webpubsub/azure-messaging-webpubsubclient/azure/messaging/webpubsubclient/_client.py +++ b/sdk/webpubsub/azure-messaging-webpubsubclient/azure/messaging/webpubsubclient/_client.py @@ -814,22 +814,25 @@ def _invoke_event_core( ) try: - with entry.cv: - try: - self._send_message(invoke_message, **kwargs) - except Exception as e: - invocation_error = ( - e - if isinstance(e, InvocationError) - else InvocationError( - str(e) if str(e) else "Failed to send invocation message.", - invocation_id=invocation_id, - ) + try: + self._send_message(invoke_message, **kwargs) + except Exception as e: + invocation_error = ( + e + if isinstance(e, InvocationError) + else InvocationError( + str(e) if str(e) else "Failed to send invocation message.", + invocation_id=invocation_id, ) - self._invocation_map.reject(invocation_id, invocation_error) - raise invocation_error from e + ) + self._invocation_map.reject(invocation_id, invocation_error) + raise invocation_error from e - notified = entry.cv.wait(timeout) + with entry.cv: + completed = entry.cv.wait_for( + lambda: entry.result is not None or entry.error is not None, + timeout, + ) if entry.error: raise entry.error @@ -837,14 +840,16 @@ def _invoke_event_core( if entry.result is None: raise InvocationError( "Timeout while waiting for invoke response." - if not notified + if not completed else "No invoke response received.", invocation_id=invocation_id, ) return self._map_invoke_response(entry.result) - except Exception as e: # pylint: disable=broad-except - should_cancel = isinstance(e, InvocationError) and e.error_detail is None + except Exception as e: # pylint: disable=broad-except + should_cancel = ( + entry.result is None and isinstance(e, InvocationError) and e.error_detail is None + ) if should_cancel: self._send_cancel_invocation(invocation_id) raise diff --git a/sdk/webpubsub/azure-messaging-webpubsubclient/azure/messaging/webpubsubclient/aio/_client.py b/sdk/webpubsub/azure-messaging-webpubsubclient/azure/messaging/webpubsubclient/aio/_client.py index 1d6b0bd4aabd..b5d8e64ef8ed 100644 --- a/sdk/webpubsub/azure-messaging-webpubsubclient/azure/messaging/webpubsubclient/aio/_client.py +++ b/sdk/webpubsub/azure-messaging-webpubsubclient/azure/messaging/webpubsubclient/aio/_client.py @@ -762,7 +762,9 @@ async def _invoke_event_core( invocation_id=invocation_id, ) from e except Exception as e: # pylint: disable=broad-except - should_cancel = isinstance(e, InvocationError) and e.error_detail is None + should_cancel = ( + entry.result is None and isinstance(e, InvocationError) and e.error_detail is None + ) if should_cancel: await self._send_cancel_invocation(invocation_id) raise diff --git a/sdk/webpubsub/azure-messaging-webpubsubclient/tests/test_invoke_event.py b/sdk/webpubsub/azure-messaging-webpubsubclient/tests/test_invoke_event.py index 8dd3146f1fca..1239b5fad444 100644 --- a/sdk/webpubsub/azure-messaging-webpubsubclient/tests/test_invoke_event.py +++ b/sdk/webpubsub/azure-messaging-webpubsubclient/tests/test_invoke_event.py @@ -57,6 +57,25 @@ def reply(): assert result.data_type == WebPubSubDataType.TEXT assert result.invocation_id is not None + def test_response_received_during_send(self): + """invoke_event handles a response that arrives before it starts waiting.""" + client = _make_client() + + def fake_send(message, **kwargs): + client._invocation_map.resolve( + InvokeResponseMessage( + invocation_id=message.invocation_id, + success=True, + data_type=WebPubSubDataType.TEXT, + data="pong", + ) + ) + + with patch.object(client, "_send_message", side_effect=fake_send): + result = client.invoke_event("echo", "ping", WebPubSubDataType.TEXT) + + assert result.data == "pong" + def test_request_response_correlation_json(self): """invoke_event returns the correct InvokeEventResult for a successful JSON response.""" client = _make_client() diff --git a/sdk/webpubsub/azure-messaging-webpubsubclient/tests/test_invoke_event_async.py b/sdk/webpubsub/azure-messaging-webpubsubclient/tests/test_invoke_event_async.py index 6fe5638a5746..913e0caa6b56 100644 --- a/sdk/webpubsub/azure-messaging-webpubsubclient/tests/test_invoke_event_async.py +++ b/sdk/webpubsub/azure-messaging-webpubsubclient/tests/test_invoke_event_async.py @@ -4,7 +4,7 @@ # Licensed under the MIT License. See License.txt in the project root for # license information. # ------------------------------------------------------------------------- -"""Client-level mocktests for invoke_event (async) exercising request/response +"""Client-level mock tests for invoke_event (async) exercising request/response correlation, timeout, error mapping and disconnect rejection. """ From 820e800f46f5e0d73517beceb045c449909e2c2a Mon Sep 17 00:00:00 2001 From: shiyingchen Date: Fri, 17 Jul 2026 19:33:20 +0800 Subject: [PATCH 8/8] fix api consistency --- .../CHANGELOG.md | 2 - .../azure-messaging-webpubsubclient/api.md | 596 ++++++++++++++++++ .../api.metadata.yml | 3 + .../webpubsubclient/models/_models.py | 34 +- .../tests/test_invocation_manager.py | 12 + .../tests/test_invocation_manager_async.py | 12 + 6 files changed, 645 insertions(+), 14 deletions(-) create mode 100644 sdk/webpubsub/azure-messaging-webpubsubclient/api.md create mode 100644 sdk/webpubsub/azure-messaging-webpubsubclient/api.metadata.yml diff --git a/sdk/webpubsub/azure-messaging-webpubsubclient/CHANGELOG.md b/sdk/webpubsub/azure-messaging-webpubsubclient/CHANGELOG.md index a7772f802989..e794beee5d98 100644 --- a/sdk/webpubsub/azure-messaging-webpubsubclient/CHANGELOG.md +++ b/sdk/webpubsub/azure-messaging-webpubsubclient/CHANGELOG.md @@ -6,8 +6,6 @@ - Add preview API `invoke_event` -## 1.1.1 (Unreleased) - ### Other Changes - Clean useless warnings in log diff --git a/sdk/webpubsub/azure-messaging-webpubsubclient/api.md b/sdk/webpubsub/azure-messaging-webpubsubclient/api.md new file mode 100644 index 000000000000..c4623cf647eb --- /dev/null +++ b/sdk/webpubsub/azure-messaging-webpubsubclient/api.md @@ -0,0 +1,596 @@ +```py +namespace azure.messaging.webpubsubclient + + class azure.messaging.webpubsubclient.WebPubSubClient(WebPubSubClientBase): implements ContextManager + + def __init__( + self, + credential: Union[WebPubSubClientCredential, str], + *, + ack_timeout: float = _ACK_TIMEOUT, + auto_rejoin_groups: bool = True, + logging_enable: bool = False, + message_retry_backoff_factor: float = _RETRY_BACKOFF_FACTOR, + message_retry_backoff_max: float = _RETRY_BACKOFF_MAX, + message_retry_mode: RetryMode = RetryMode.Exponential, + message_retry_total: int = _RETRY_TOTAL, + protocol_type: WebPubSubProtocolType = WebPubSubProtocolType.JSON_RELIABLE, + reconnect_retry_backoff_factor: float = _RETRY_BACKOFF_FACTOR, + reconnect_retry_backoff_max: float = _RETRY_BACKOFF_MAX, + reconnect_retry_mode: RetryMode = RetryMode.Exponential, + reconnect_retry_total: int = _RETRY_TOTAL, + start_timeout: float = _START_TIMEOUT, + user_agent: Optional[str] = ..., + **kwargs: Any + ) -> None: ... + + def close(self) -> None: ... + + @overload + def invoke_event( + self, + event_name: str, + content: str, + data_type: Literal[WebPubSubDataType.TEXT], + **kwargs: Any + ) -> InvokeEventResult: ... + + @overload + def invoke_event( + self, + event_name: str, + content: memoryview, + data_type: Literal[WebPubSubDataType.BINARY, WebPubSubDataType.PROTOBUF], + **kwargs: Any + ) -> InvokeEventResult: ... + + @overload + def invoke_event( + self, + event_name: str, + content: Dict[str, Any], + data_type: Literal[WebPubSubDataType.JSON], + **kwargs: Any + ) -> InvokeEventResult: ... + + def is_connected(self) -> bool: ... + + def join_group( + self, + group_name: str, + *, + ack_id: Optional[int] = ..., + **kwargs: Any + ) -> None: ... + + def leave_group( + self, + group_name: str, + *, + ack_id: Optional[int] = ..., + **kwargs: Any + ) -> None: ... + + def open(self) -> None: ... + + @overload + def send_event( + self, + event_name: str, + content: str, + data_type: Literal[WebPubSubDataType.TEXT], + **kwargs: Any + ) -> None: ... + + @overload + def send_event( + self, + event_name: str, + content: memoryview, + data_type: Literal[WebPubSubDataType.BINARY, WebPubSubDataType.PROTOBUF], + **kwargs: Any + ) -> None: ... + + @overload + def send_event( + self, + event_name: str, + content: Dict[str, Any], + data_type: Literal[WebPubSubDataType.JSON], + **kwargs: Any + ) -> None: ... + + @overload + def send_to_group( + self, + group_name: str, + content: str, + data_type: Literal[WebPubSubDataType.TEXT], + **kwargs: Any + ) -> None: ... + + @overload + def send_to_group( + self, + group_name: str, + content: Dict[str, Any], + data_type: Literal[WebPubSubDataType.JSON], + **kwargs: Any + ) -> None: ... + + @overload + def send_to_group( + self, + group_name: str, + content: memoryview, + data_type: Literal[WebPubSubDataType.BINARY, WebPubSubDataType.PROTOBUF], + **kwargs: Any + ) -> None: ... + + @overload + def subscribe( + self, + event: Literal[CallbackType.CONNECTED], + listener: Callable[[OnConnectedArgs], None] + ) -> None: ... + + @overload + def subscribe( + self, + event: Literal[CallbackType.DISCONNECTED], + listener: Callable[[OnDisconnectedArgs], None] + ) -> None: ... + + @overload + def subscribe( + self, + event: Literal[CallbackType.STOPPED], + listener: Callable[[], None] + ) -> None: ... + + @overload + def subscribe( + self, + event: Literal[CallbackType.SERVER_MESSAGE], + listener: Callable[[OnServerDataMessageArgs], None] + ) -> None: ... + + @overload + def subscribe( + self, + event: Literal[CallbackType.GROUP_MESSAGE], + listener: Callable[[OnGroupDataMessageArgs], None] + ) -> None: ... + + @overload + def subscribe( + self, + event: Literal[CallbackType.REJOIN_GROUP_FAILED], + listener: Callable[[OnRejoinGroupFailedArgs], None] + ) -> None: ... + + @overload + def unsubscribe( + self, + event: Literal[CallbackType.CONNECTED], + listener: Callable[[OnConnectedArgs], None] + ) -> None: ... + + @overload + def unsubscribe( + self, + event: Literal[CallbackType.DISCONNECTED], + listener: Callable[[OnDisconnectedArgs], None] + ) -> None: ... + + @overload + def unsubscribe( + self, + event: Literal[CallbackType.STOPPED], + listener: Callable[[], None] + ) -> None: ... + + @overload + def unsubscribe( + self, + event: Literal[CallbackType.SERVER_MESSAGE], + listener: Callable[[OnServerDataMessageArgs], None] + ) -> None: ... + + @overload + def unsubscribe( + self, + event: Literal[CallbackType.GROUP_MESSAGE], + listener: Callable[[OnGroupDataMessageArgs], None] + ) -> None: ... + + @overload + def unsubscribe( + self, + event: Literal[CallbackType.REJOIN_GROUP_FAILED], + listener: Callable[[OnRejoinGroupFailedArgs], None] + ) -> None: ... + + + class azure.messaging.webpubsubclient.WebPubSubClientCredential: + + def __init__(self, client_access_url_provider: Union[str, Callable]) -> None: ... + + def get_client_access_url(self) -> str: ... + + +namespace azure.messaging.webpubsubclient.aio + + class azure.messaging.webpubsubclient.aio.WebPubSubClient(WebPubSubClientBase): implements AsyncContextManager + + def __init__( + self, + credential: Union[WebPubSubClientCredential, str], + *, + ack_timeout: float = _ACK_TIMEOUT, + auto_rejoin_groups: bool = True, + logging_enable: bool = False, + message_retry_backoff_factor: float = _RETRY_BACKOFF_FACTOR, + message_retry_backoff_max: float = _RETRY_BACKOFF_MAX, + message_retry_mode: RetryMode = RetryMode.Exponential, + message_retry_total: int = _RETRY_TOTAL, + protocol_type: WebPubSubProtocolType = WebPubSubProtocolType.JSON_RELIABLE, + reconnect_retry_backoff_factor: float = _RETRY_BACKOFF_FACTOR, + reconnect_retry_backoff_max: float = _RETRY_BACKOFF_MAX, + reconnect_retry_mode: RetryMode = RetryMode.Exponential, + reconnect_retry_total: int = _RETRY_TOTAL, + start_timeout: float = _START_TIMEOUT, + user_agent: Optional[str] = ..., + **kwargs: Any + ) -> None: ... + + async def close(self) -> None: ... + + @overload + async def invoke_event( + self, + event_name: str, + content: str, + data_type: Literal[WebPubSubDataType.TEXT], + **kwargs: Any + ) -> InvokeEventResult: ... + + @overload + async def invoke_event( + self, + event_name: str, + content: memoryview, + data_type: Literal[WebPubSubDataType.BINARY, WebPubSubDataType.PROTOBUF], + **kwargs: Any + ) -> InvokeEventResult: ... + + @overload + async def invoke_event( + self, + event_name: str, + content: Dict[str, Any], + data_type: Literal[WebPubSubDataType.JSON], + **kwargs: Any + ) -> InvokeEventResult: ... + + def is_connected(self) -> bool: ... + + async def join_group( + self, + group_name: str, + *, + ack_id: Optional[int] = ..., + **kwargs: Any + ) -> None: ... + + async def leave_group( + self, + group_name: str, + *, + ack_id: Optional[int] = ..., + **kwargs: Any + ) -> None: ... + + async def open(self) -> None: ... + + @overload + async def send_event( + self, + event_name: str, + content: str, + data_type: Literal[WebPubSubDataType.TEXT], + **kwargs: Any + ) -> None: ... + + @overload + async def send_event( + self, + event_name: str, + content: memoryview, + data_type: Literal[WebPubSubDataType.BINARY, WebPubSubDataType.PROTOBUF], + **kwargs: Any + ) -> None: ... + + @overload + async def send_event( + self, + event_name: str, + content: Dict[str, Any], + data_type: Literal[WebPubSubDataType.JSON], + **kwargs: Any + ) -> None: ... + + @overload + async def send_to_group( + self, + group_name: str, + content: str, + data_type: Literal[WebPubSubDataType.TEXT], + **kwargs: Any + ) -> None: ... + + @overload + async def send_to_group( + self, + group_name: str, + content: Dict[str, Any], + data_type: Literal[WebPubSubDataType.JSON], + **kwargs: Any + ) -> None: ... + + @overload + async def send_to_group( + self, + group_name: str, + content: memoryview, + data_type: Literal[WebPubSubDataType.BINARY, WebPubSubDataType.PROTOBUF], + **kwargs: Any + ) -> None: ... + + @overload + async def subscribe( + self, + event: Literal[CallbackType.CONNECTED], + listener: Callable[[OnConnectedArgs], Awaitable[None]] + ) -> None: ... + + @overload + async def subscribe( + self, + event: Literal[CallbackType.DISCONNECTED], + listener: Callable[[OnDisconnectedArgs], Awaitable[None]] + ) -> None: ... + + @overload + async def subscribe( + self, + event: Literal[CallbackType.STOPPED], + listener: Callable[[], Awaitable[None]] + ) -> None: ... + + @overload + async def subscribe( + self, + event: Literal[CallbackType.SERVER_MESSAGE], + listener: Callable[[OnServerDataMessageArgs], Awaitable[None]] + ) -> None: ... + + @overload + async def subscribe( + self, + event: Literal[CallbackType.GROUP_MESSAGE], + listener: Callable[[OnGroupDataMessageArgs], Awaitable[None]] + ) -> None: ... + + @overload + async def subscribe( + self, + event: Literal[CallbackType.REJOIN_GROUP_FAILED], + listener: Callable[[OnRejoinGroupFailedArgs], Awaitable[None]] + ) -> None: ... + + @overload + async def unsubscribe( + self, + event: Literal[CallbackType.CONNECTED], + listener: Callable[[OnConnectedArgs], Awaitable[None]] + ) -> None: ... + + @overload + async def unsubscribe( + self, + event: Literal[CallbackType.DISCONNECTED], + listener: Callable[[OnDisconnectedArgs], Awaitable[None]] + ) -> None: ... + + @overload + async def unsubscribe( + self, + event: Literal[CallbackType.STOPPED], + listener: Callable[[], Awaitable[None]] + ) -> None: ... + + @overload + async def unsubscribe( + self, + event: Literal[CallbackType.SERVER_MESSAGE], + listener: Callable[[OnServerDataMessageArgs], Awaitable[None]] + ) -> None: ... + + @overload + async def unsubscribe( + self, + event: Literal[CallbackType.GROUP_MESSAGE], + listener: Callable[[OnGroupDataMessageArgs], Awaitable[None]] + ) -> None: ... + + @overload + async def unsubscribe( + self, + event: Literal[CallbackType.REJOIN_GROUP_FAILED], + listener: Callable[[OnRejoinGroupFailedArgs], Awaitable[None]] + ) -> None: ... + + + class azure.messaging.webpubsubclient.aio.WebPubSubClientCredential: + + def __init__(self, client_access_url_provider: Union[str, Callable[[], Coroutine[Any, Any, str]]]) -> None: ... + + async def get_client_access_url(self) -> str: ... + + +namespace azure.messaging.webpubsubclient.models + + class azure.messaging.webpubsubclient.models.AckMessageError: + message: str + name: str + + def __init__( + self, + *, + message: str, + name: str + ): ... + + + class azure.messaging.webpubsubclient.models.CallbackType(str, Enum, metaclass=CaseInsensitiveEnumMeta): + CONNECTED = "connected" + DISCONNECTED = "disconnected" + GROUP_MESSAGE = "group-message" + REJOIN_GROUP_FAILED = "rejoin-group-failed" + SERVER_MESSAGE = "server-message" + STOPPED = "stopped" + + + class azure.messaging.webpubsubclient.models.InvocationError(AzureError): + error_detail: InvokeResponseError + invocation_id: str + message: str + + def __init__( + self, + message: str, + invocation_id: str, + error_detail: Optional[InvokeResponseError] = None + ) -> None: ... + + + class azure.messaging.webpubsubclient.models.InvokeEventResult: + data: Any + data_type: Union[WebPubSubDataType, str] + invocation_id: str + + def __init__( + self, + invocation_id: str, + *, + data: Any = ..., + data_type: Optional[WebPubSubDataType] = ... + ) -> None: ... + + + class azure.messaging.webpubsubclient.models.InvokeResponseError: + message: str + name: str + + def __init__( + self, + *, + message: str, + name: str + ): ... + + + class azure.messaging.webpubsubclient.models.OnConnectedArgs: + connection_id: str + user_id: str + + def __init__( + self, + connection_id: str, + user_id: Optional[str] = None + ) -> None: ... + + + class azure.messaging.webpubsubclient.models.OnDisconnectedArgs: + connection_id: str + message: str + + def __init__( + self, + connection_id: Optional[str] = None, + message: Optional[str] = None + ) -> None: ... + + + class azure.messaging.webpubsubclient.models.OnGroupDataMessageArgs: + data: Any + data_type: Union[WebPubSubDataType, str] + from_user_id: str + group: str + sequence_id: int + + def __init__( + self, + *, + data: Any, + data_type: WebPubSubDataType, + from_user_id: Optional[str] = ..., + group: str, + sequence_id: Optional[int] = ... + ) -> None: ... + + + class azure.messaging.webpubsubclient.models.OnRejoinGroupFailedArgs: + error: Exception + group: str + + def __init__( + self, + group: str, + error: Exception + ) -> None: ... + + + class azure.messaging.webpubsubclient.models.OnServerDataMessageArgs: + data: Any + data_type: Union[WebPubSubDataType, str] + sequence_id: int + + def __init__( + self, + data_type: WebPubSubDataType, + data: Any, + sequence_id: Optional[int] = None + ) -> None: ... + + + class azure.messaging.webpubsubclient.models.OpenClientError(AzureError): + + + class azure.messaging.webpubsubclient.models.SendMessageError(AzureError): + ack_id: int + error_detail: AckMessageError + message: str + + def __init__( + self, + message: str, + ack_id: Optional[int] = None, + error_detail: Optional[AckMessageError] = None + ) -> None: ... + + + class azure.messaging.webpubsubclient.models.WebPubSubDataType(str, Enum, metaclass=CaseInsensitiveEnumMeta): + BINARY = "binary" + JSON = "json" + PROTOBUF = "protobuf" + TEXT = "text" + + + class azure.messaging.webpubsubclient.models.WebPubSubProtocolType(str, Enum, metaclass=CaseInsensitiveEnumMeta): + JSON = "json.webpubsub.azure.v1" + JSON_RELIABLE = "json.reliable.webpubsub.azure.v1" + + +``` \ No newline at end of file diff --git a/sdk/webpubsub/azure-messaging-webpubsubclient/api.metadata.yml b/sdk/webpubsub/azure-messaging-webpubsubclient/api.metadata.yml new file mode 100644 index 000000000000..d9e4ccf47e99 --- /dev/null +++ b/sdk/webpubsub/azure-messaging-webpubsubclient/api.metadata.yml @@ -0,0 +1,3 @@ +apiMdSha256: c29e087869974bb1f3475718c27bc413c3fb0800fe0264b97bafc90afa0a69ec +parserVersion: 0.3.28 +pythonVersion: 3.14.4 diff --git a/sdk/webpubsub/azure-messaging-webpubsubclient/azure/messaging/webpubsubclient/models/_models.py b/sdk/webpubsub/azure-messaging-webpubsubclient/azure/messaging/webpubsubclient/models/_models.py index 2b81394433dc..c41cb0afc2be 100644 --- a/sdk/webpubsub/azure-messaging-webpubsubclient/azure/messaging/webpubsubclient/models/_models.py +++ b/sdk/webpubsub/azure-messaging-webpubsubclient/azure/messaging/webpubsubclient/models/_models.py @@ -1297,12 +1297,17 @@ def register(self, invocation_id: Optional[str] = None) -> Tuple[str, _Invocatio :raises InvocationError: If invocation id is already registered. """ with self._lock: - resolved_id = invocation_id if invocation_id else self._generate_invocation_id() - if resolved_id in self._entries: - raise InvocationError( - "Invocation id is already registered.", - invocation_id=resolved_id, - ) + if invocation_id is not None: + resolved_id = invocation_id + if resolved_id in self._entries: + raise InvocationError( + "Invocation id is already registered.", + invocation_id=resolved_id, + ) + else: + resolved_id = self._generate_invocation_id() + while resolved_id in self._entries: + resolved_id = self._generate_invocation_id() entry = _InvocationEntry(resolved_id) self._entries[resolved_id] = entry return resolved_id, entry @@ -1386,12 +1391,17 @@ def register(self, invocation_id: Optional[str] = None) -> Tuple[str, _Invocatio :rtype: Tuple[str, _InvocationEntryAsync] :raises InvocationError: If invocation id is already registered. """ - resolved_id = invocation_id if invocation_id else self._generate_invocation_id() - if resolved_id in self._entries: - raise InvocationError( - "Invocation id is already registered.", - invocation_id=resolved_id, - ) + if invocation_id is not None: + resolved_id = invocation_id + if resolved_id in self._entries: + raise InvocationError( + "Invocation id is already registered.", + invocation_id=resolved_id, + ) + else: + resolved_id = self._generate_invocation_id() + while resolved_id in self._entries: + resolved_id = self._generate_invocation_id() entry = _InvocationEntryAsync(resolved_id) self._entries[resolved_id] = entry return resolved_id, entry diff --git a/sdk/webpubsub/azure-messaging-webpubsubclient/tests/test_invocation_manager.py b/sdk/webpubsub/azure-messaging-webpubsubclient/tests/test_invocation_manager.py index dde289ad9250..6f3240adaf99 100644 --- a/sdk/webpubsub/azure-messaging-webpubsubclient/tests/test_invocation_manager.py +++ b/sdk/webpubsub/azure-messaging-webpubsubclient/tests/test_invocation_manager.py @@ -54,6 +54,18 @@ def test_register_duplicate_raises(self): assert exc_info.value.invocation_id == "same-id" + def test_generated_id_skips_pending_custom_numeric_ids(self): + """Test generated IDs skip numeric IDs registered by callers.""" + inv_mgr = InvocationManager() + inv_mgr.register("1") + inv_mgr.register("3") + + first_id, _ = inv_mgr.register() + second_id, _ = inv_mgr.register() + + assert first_id == "2" + assert second_id == "4" + def test_reject(self): """Test rejecting an invocation""" inv_mgr = InvocationManager() diff --git a/sdk/webpubsub/azure-messaging-webpubsubclient/tests/test_invocation_manager_async.py b/sdk/webpubsub/azure-messaging-webpubsubclient/tests/test_invocation_manager_async.py index 98c3ff761484..6ca0bfd7e59b 100644 --- a/sdk/webpubsub/azure-messaging-webpubsubclient/tests/test_invocation_manager_async.py +++ b/sdk/webpubsub/azure-messaging-webpubsubclient/tests/test_invocation_manager_async.py @@ -51,6 +51,18 @@ def test_register_duplicate_raises(self): assert exc_info.value.invocation_id == "same-id" + def test_generated_id_skips_pending_custom_numeric_ids(self): + """Test generated IDs skip numeric IDs registered by callers.""" + inv_mgr = InvocationManagerAsync() + inv_mgr.register("1") + inv_mgr.register("3") + + first_id, _ = inv_mgr.register() + second_id, _ = inv_mgr.register() + + assert first_id == "2" + assert second_id == "4" + def test_reject_sets_event(self): """Test that rejecting sets the event""" inv_mgr = InvocationManagerAsync()