RPC (Request/Response)

RPCClient (sync)

RPCClient

Synchronous RPC client over RabbitMQ.

Usage::

client = RPCClient(transport)
response = client.call("rpc.orders", b'{"id": 1}', timeout=5.0)
print(response.body)
client.close()

Source code in src/rabbitkit/rpc.py

class RPCClient:
    """Synchronous RPC client over RabbitMQ.

    Usage::

        client = RPCClient(transport)
        response = client.call("rpc.orders", b'{"id": 1}', timeout=5.0)
        print(response.body)
        client.close()
    """

    def __init__(
        self,
        transport: Any,
        *,
        serializer: Any | None = None,
        max_pending: int = 100,
        reply_connection: Any | None = None,
        close_reply_connection: bool = False,
        max_reply_bytes: int | None = None,
    ) -> None:
        self._transport = transport
        self._serializer = serializer
        self._reply_queue = DIRECT_REPLY_TO_QUEUE

        # Shared reply router holds max_reply_bytes / max_pending / _pending.
        self._router = _ReplyRouter(
            max_reply_bytes=max_reply_bytes,
            max_pending=max_pending,
        )

        # Dedicated reply connection. When provided, ``call()`` pumps it via
        # ``process_data_events`` while waiting so the broker's I/O thread is
        # never blocked. When ``None`` (e.g. in tests using a transport mock),
        # ``call()`` falls back to blocking on the future — which works when
        # replies are delivered out-of-band (the existing test harness).
        #
        # Ownership: by default the caller owns *reply_connection* and is
        # responsible for closing it (it may be shared). Set
        # ``close_reply_connection=True`` to have close() close it too.
        self._connection: Any | None = reply_connection
        self._close_reply_connection = bool(close_reply_connection)

        self._lock = threading.Lock()
        self._consuming = False
        self._consumer_tag: str | None = None
        self._starting = False
        # L-5: guards call()/_ensure_consuming() after close().
        self._closed = False

    def call(
        self,
        routing_key: str,
        body: bytes,
        *,
        timeout: float = 5.0,
        exchange: str = "",
        headers: dict[str, Any] | None = None,
    ) -> RabbitMessage:
        """Send an RPC request and wait for a response.

        Args:
            routing_key: The routing key (queue name) to send the request to.
            body: The request body.
            timeout: Maximum time to wait for a response (seconds).
            exchange: The exchange to publish to (default: "" for direct).
            headers: Optional headers to include in the request.

        Returns:
            RabbitMessage: The response message.

        Raises:
            RPCTimeoutError: If the response is not received within the timeout.
            RPCClientClosed: If the client was closed while waiting.
            RuntimeError: If max_pending calls is exceeded.

        Note:
            Sync RPC requires its own dedicated reply connection (passed via
            ``reply_connection``) so the I/O loop can be pumped while waiting.
            Calling sync RPC from inside a ``worker_count=1`` sync handler with
            a shared broker connection would otherwise deadlock.
        """
        # L-5: refuse to operate after close() instead of silently re-registering
        # a consumer on a half-torn-down client.
        if self._closed:
            raise RPCClientClosed("RPCClient is closed")
        self._ensure_consuming()

        correlation_id = str(uuid.uuid4())

        with self._lock:
            if self._router.is_full():
                raise RuntimeError(
                    f"Max pending RPC calls ({self._router.max_pending}) exceeded. "
                    "Consider increasing max_pending or reducing call rate."
                )
            fut: Future[RabbitMessage] = Future()
            self._router.register(correlation_id, _FutureSink(fut))

        # Publish request
        envelope = MessageEnvelope(
            routing_key=routing_key,
            body=body,
            exchange=exchange,
            reply_to=self._reply_queue,
            correlation_id=correlation_id,
            headers=headers or {},
        )
        self._transport.publish(envelope)

        # Wait for the response. If we own a dedicated reply connection, pump
        # its I/O loop ourselves; otherwise block directly on the future
        # (replies are delivered out-of-band, e.g. by a test harness or another
        # thread).
        # L-4: wrap the pump in try/finally so a `process_data_events` that
        # raises still pops the _pending entry — otherwise it leaks and
        # exhausts `max_pending` over time.
        deadline = time.monotonic() + timeout
        try:
            if self._connection is not None:
                while not fut.done():
                    remaining = deadline - time.monotonic()
                    if remaining <= 0:
                        break
                    self._connection.process_data_events(time_limit=min(0.01, remaining))
            else:
                remaining = deadline - time.monotonic()
                if remaining > 0:
                    try:
                        fut.result(timeout=remaining)
                    except TimeoutError:
                        pass  # handled as timeout below
        finally:
            if not fut.done():
                with self._lock:
                    self._router.pop(correlation_id)

        if not fut.done():
            raise RPCTimeoutError(correlation_id, timeout)

        with self._lock:
            self._router.pop(correlation_id)

        # Raises the stored exception (ReplyTooLargeError / RPCClientClosed)
        # or returns the resolved result.
        return fut.result()

    def close(self) -> None:
        """Close the RPC client and clean up.

        Cancels the reply consumer and resolves all pending waiters with an
        ``RPCClientClosed`` error so callers fail cleanly instead of hitting
        ``AttributeError`` on a ``None`` result.

        When constructed with ``close_reply_connection=True`` AND a
        ``reply_connection``, also closes the dedicated reply connection
        (ownership transferred to the client). Otherwise the caller retains
        ownership of *reply_connection* and must close it themselves.
        """
        if self._consumer_tag and self._transport:
            try:
                self._transport.cancel_consumer(self._consumer_tag)
            except Exception as e:
                logger.warning("Failed to cancel RPC reply consumer: %s", e)

        # Clean up pending calls — set an exception so waiters raise cleanly.
        # L-5: mark the client closed so subsequent call()/_ensure_consuming()
        # refuse to re-register a consumer on the torn-down client.
        with self._lock:
            self._closed = True
            self._router.close_all(RPCClientClosed("RPCClient was closed"))

        self._consuming = False
        self._consumer_tag = None

        # Low: optionally close the dedicated reply connection when the client
        # owns it. Default keeps the prior behaviour (caller-owned).
        if self._close_reply_connection and self._connection is not None:
            try:
                self._connection.close()
            except Exception as e:
                logger.warning("Failed to close RPC reply connection: %s", e)

    def _ensure_consuming(self) -> None:
        """Ensure the reply consumer is running.

        The network ``consume()`` call is made *outside* the lock to avoid
        self-deadlock and holding the lock across I/O. A ``_starting`` flag
        guards against concurrent first-callers each registering a consumer.
        """
        # L-5: refuse to (re-)register a consumer on a closed client.
        if self._closed:
            raise RPCClientClosed("RPCClient is closed")
        with self._lock:
            if self._consuming or self._starting:
                return
            self._starting = True

        def on_reply(message: RabbitMessage) -> None:
            """Handle reply messages — delegate correlation match to the router."""
            with self._lock:
                self._router.resolve(message)

        try:
            # amq.rabbitmq.reply-to is a broker pseudo-queue: it rejects any
            # Queue.Declare (declare=False) and requires a no-ack consumer
            # (no_ack=True) — the broker auto-acks each reply on delivery.
            consumer_tag = self._transport.consume(
                queue=self._reply_queue,
                callback=on_reply,
                no_ack=True,
                declare=False,
            )
        except Exception:
            with self._lock:
                self._starting = False
            raise

        with self._lock:
            self._consumer_tag = consumer_tag
            self._consuming = True
            self._starting = False

Methods:

call(routing_key: str, body: bytes, *, timeout: float = 5.0, exchange: str = '', headers: dict[str, Any] | None = None) -> RabbitMessage

Send an RPC request and wait for a response.

Parameters:

Name	Type	Description	Default
routing_key	str	The routing key (queue name) to send the request to.	required
body	bytes	The request body.	required
timeout	float	Maximum time to wait for a response (seconds).	5.0
exchange	str	The exchange to publish to (default: "" for direct).	''
headers	dict[str, Any] | None	Optional headers to include in the request.	None

Returns:

Name	Type	Description
RabbitMessage	RabbitMessage	The response message.

Raises:

Type	Description
RPCTimeoutError	If the response is not received within the timeout.
RPCClientClosed	If the client was closed while waiting.
RuntimeError	If max_pending calls is exceeded.

Note

Sync RPC requires its own dedicated reply connection (passed via reply_connection) so the I/O loop can be pumped while waiting. Calling sync RPC from inside a worker_count=1 sync handler with a shared broker connection would otherwise deadlock.

Source code in src/rabbitkit/rpc.py

def call(
    self,
    routing_key: str,
    body: bytes,
    *,
    timeout: float = 5.0,
    exchange: str = "",
    headers: dict[str, Any] | None = None,
) -> RabbitMessage:
    """Send an RPC request and wait for a response.

    Args:
        routing_key: The routing key (queue name) to send the request to.
        body: The request body.
        timeout: Maximum time to wait for a response (seconds).
        exchange: The exchange to publish to (default: "" for direct).
        headers: Optional headers to include in the request.

    Returns:
        RabbitMessage: The response message.

    Raises:
        RPCTimeoutError: If the response is not received within the timeout.
        RPCClientClosed: If the client was closed while waiting.
        RuntimeError: If max_pending calls is exceeded.

    Note:
        Sync RPC requires its own dedicated reply connection (passed via
        ``reply_connection``) so the I/O loop can be pumped while waiting.
        Calling sync RPC from inside a ``worker_count=1`` sync handler with
        a shared broker connection would otherwise deadlock.
    """
    # L-5: refuse to operate after close() instead of silently re-registering
    # a consumer on a half-torn-down client.
    if self._closed:
        raise RPCClientClosed("RPCClient is closed")
    self._ensure_consuming()

    correlation_id = str(uuid.uuid4())

    with self._lock:
        if self._router.is_full():
            raise RuntimeError(
                f"Max pending RPC calls ({self._router.max_pending}) exceeded. "
                "Consider increasing max_pending or reducing call rate."
            )
        fut: Future[RabbitMessage] = Future()
        self._router.register(correlation_id, _FutureSink(fut))

    # Publish request
    envelope = MessageEnvelope(
        routing_key=routing_key,
        body=body,
        exchange=exchange,
        reply_to=self._reply_queue,
        correlation_id=correlation_id,
        headers=headers or {},
    )
    self._transport.publish(envelope)

    # Wait for the response. If we own a dedicated reply connection, pump
    # its I/O loop ourselves; otherwise block directly on the future
    # (replies are delivered out-of-band, e.g. by a test harness or another
    # thread).
    # L-4: wrap the pump in try/finally so a `process_data_events` that
    # raises still pops the _pending entry — otherwise it leaks and
    # exhausts `max_pending` over time.
    deadline = time.monotonic() + timeout
    try:
        if self._connection is not None:
            while not fut.done():
                remaining = deadline - time.monotonic()
                if remaining <= 0:
                    break
                self._connection.process_data_events(time_limit=min(0.01, remaining))
        else:
            remaining = deadline - time.monotonic()
            if remaining > 0:
                try:
                    fut.result(timeout=remaining)
                except TimeoutError:
                    pass  # handled as timeout below
    finally:
        if not fut.done():
            with self._lock:
                self._router.pop(correlation_id)

    if not fut.done():
        raise RPCTimeoutError(correlation_id, timeout)

    with self._lock:
        self._router.pop(correlation_id)

    # Raises the stored exception (ReplyTooLargeError / RPCClientClosed)
    # or returns the resolved result.
    return fut.result()

close() -> None

Close the RPC client and clean up.

Cancels the reply consumer and resolves all pending waiters with an RPCClientClosed error so callers fail cleanly instead of hitting AttributeError on a None result.

When constructed with close_reply_connection=True AND a reply_connection, also closes the dedicated reply connection (ownership transferred to the client). Otherwise the caller retains ownership of reply_connection and must close it themselves.

Source code in src/rabbitkit/rpc.py

def close(self) -> None:
    """Close the RPC client and clean up.

    Cancels the reply consumer and resolves all pending waiters with an
    ``RPCClientClosed`` error so callers fail cleanly instead of hitting
    ``AttributeError`` on a ``None`` result.

    When constructed with ``close_reply_connection=True`` AND a
    ``reply_connection``, also closes the dedicated reply connection
    (ownership transferred to the client). Otherwise the caller retains
    ownership of *reply_connection* and must close it themselves.
    """
    if self._consumer_tag and self._transport:
        try:
            self._transport.cancel_consumer(self._consumer_tag)
        except Exception as e:
            logger.warning("Failed to cancel RPC reply consumer: %s", e)

    # Clean up pending calls — set an exception so waiters raise cleanly.
    # L-5: mark the client closed so subsequent call()/_ensure_consuming()
    # refuse to re-register a consumer on the torn-down client.
    with self._lock:
        self._closed = True
        self._router.close_all(RPCClientClosed("RPCClient was closed"))

    self._consuming = False
    self._consumer_tag = None

    # Low: optionally close the dedicated reply connection when the client
    # owns it. Default keeps the prior behaviour (caller-owned).
    if self._close_reply_connection and self._connection is not None:
        try:
            self._connection.close()
        except Exception as e:
            logger.warning("Failed to close RPC reply connection: %s", e)

AsyncRPCClient (async)

AsyncRPCClient

Asynchronous RPC client over RabbitMQ.

Usage::

client = AsyncRPCClient(transport)
response = await client.call("rpc.orders", b'{"id": 1}', timeout=5.0)
print(response.body)
await client.close()

Source code in src/rabbitkit/rpc.py

class AsyncRPCClient:
    """Asynchronous RPC client over RabbitMQ.

    Usage::

        client = AsyncRPCClient(transport)
        response = await client.call("rpc.orders", b'{"id": 1}', timeout=5.0)
        print(response.body)
        await client.close()
    """

    def __init__(
        self,
        transport: Any,
        *,
        serializer: Any | None = None,
        max_pending: int = 100,
        max_reply_bytes: int | None = None,
    ) -> None:
        self._transport = transport
        self._serializer = serializer
        self._reply_queue = DIRECT_REPLY_TO_QUEUE

        # Shared reply router holds max_reply_bytes / max_pending / _pending.
        self._router = _ReplyRouter(
            max_reply_bytes=max_reply_bytes,
            max_pending=max_pending,
        )

        self._lock = asyncio.Lock()
        self._consuming = False
        self._consumer_tag: str | None = None
        # L6: guards call()/_ensure_consuming() after close() — mirrors the
        # sync client's `_closed` guard (see its `_ensure_consuming` for why:
        # without it, a call() after close() would happily re-register a
        # consumer on a torn-down transport).
        self._closed = False

    async def call(
        self,
        routing_key: str,
        body: bytes,
        *,
        timeout: float = 5.0,
        exchange: str = "",
        headers: dict[str, Any] | None = None,
    ) -> RabbitMessage:
        """Send an RPC request and wait for a response.

        Args:
            routing_key: The routing key (queue name) to send the request to.
            body: The request body.
            timeout: Maximum time to wait for a response (seconds).
            exchange: The exchange to publish to (default: "" for direct).
            headers: Optional headers to include in the request.

        Returns:
            RabbitMessage: The response message.

        Raises:
            RPCTimeoutError: If the response is not received within the timeout.
            RPCClientClosed: If the client was closed (or is closed while waiting).
            RuntimeError: If max_pending calls is exceeded.
        """
        # L6: refuse to operate after close() instead of silently
        # re-registering a consumer on a half-torn-down client.
        if self._closed:
            raise RPCClientClosed("AsyncRPCClient is closed")
        await self._ensure_consuming()

        correlation_id = str(uuid.uuid4())

        async with self._lock:
            if self._router.is_full():
                raise RuntimeError(
                    f"Max pending RPC calls ({self._router.max_pending}) exceeded. "
                    "Consider increasing max_pending or reducing call rate."
                )
            loop = asyncio.get_running_loop()
            future: asyncio.Future[RabbitMessage] = loop.create_future()
            self._router.register(correlation_id, _AsyncFutureSink(future))

        # Publish request
        envelope = MessageEnvelope(
            routing_key=routing_key,
            body=body,
            exchange=exchange,
            reply_to=self._reply_queue,
            correlation_id=correlation_id,
            headers=headers or {},
        )
        await self._transport.publish(envelope)

        # Wait for response with timeout. R-timeout: asyncio.timeout (3.11+)
        # replaces asyncio.wait_for to avoid the wrapper-task overhead.
        try:
            async with asyncio.timeout(timeout):
                result = await future
        except TimeoutError:
            async with self._lock:
                self._router.pop(correlation_id)
            raise RPCTimeoutError(correlation_id, timeout) from None

        async with self._lock:
            self._router.pop(correlation_id)

        return result

    async def close(self) -> None:
        """Close the RPC client and clean up.

        Cancels the reply consumer and resolves all pending calls with a
        typed ``RPCClientClosed`` (L6) — matching the sync client's
        behavior — so an in-flight caller's ``await future`` raises cleanly
        instead of a bare ``asyncio.CancelledError`` that could be confused
        with the caller's own cancellation.
        """
        if self._consumer_tag and self._transport:
            try:
                await self._transport.cancel_consumer(self._consumer_tag)
            except Exception as e:
                logger.warning("Failed to cancel RPC reply consumer: %s", e)

        # Clean up pending calls. L6: mark closed so subsequent
        # call()/_ensure_consuming() refuse to re-register a consumer on the
        # torn-down client.
        async with self._lock:
            self._closed = True
            self._router.close_all(RPCClientClosed("AsyncRPCClient was closed"))

        self._consuming = False
        self._consumer_tag = None

    async def _ensure_consuming(self) -> None:
        """Ensure the reply consumer is running.

        Uses the existing lock to prevent concurrent first-calls from each
        registering a duplicate consumer on amq.rabbitmq.reply-to (which
        supports exactly one consumer per channel).
        """
        # L6: refuse to (re-)register a consumer on a closed client.
        if self._closed:
            raise RPCClientClosed("AsyncRPCClient is closed")
        async with self._lock:
            if self._consuming:
                return

            async def on_reply(message: RabbitMessage) -> None:
                """Handle reply messages — delegate correlation match to the router."""
                async with self._lock:
                    self._router.resolve(message)

            # amq.rabbitmq.reply-to is a broker pseudo-queue: it rejects any
            # Queue.Declare (declare=False) and requires a no-ack consumer
            # (no_ack=True) — the broker auto-acks each reply on delivery.
            self._consumer_tag = await self._transport.consume(
                queue=self._reply_queue,
                callback=on_reply,
                no_ack=True,
                declare=False,
            )
            self._consuming = True

Methods:

call(routing_key: str, body: bytes, *, timeout: float = 5.0, exchange: str = '', headers: dict[str, Any] | None = None) -> RabbitMessage async

Send an RPC request and wait for a response.

Parameters:

Name	Type	Description	Default
routing_key	str	The routing key (queue name) to send the request to.	required
body	bytes	The request body.	required
timeout	float	Maximum time to wait for a response (seconds).	5.0
exchange	str	The exchange to publish to (default: "" for direct).	''
headers	dict[str, Any] | None	Optional headers to include in the request.	None

Returns:

Name	Type	Description
RabbitMessage	RabbitMessage	The response message.

Raises:

Type	Description
RPCTimeoutError	If the response is not received within the timeout.
RPCClientClosed	If the client was closed (or is closed while waiting).
RuntimeError	If max_pending calls is exceeded.

Source code in src/rabbitkit/rpc.py

async def call(
    self,
    routing_key: str,
    body: bytes,
    *,
    timeout: float = 5.0,
    exchange: str = "",
    headers: dict[str, Any] | None = None,
) -> RabbitMessage:
    """Send an RPC request and wait for a response.

    Args:
        routing_key: The routing key (queue name) to send the request to.
        body: The request body.
        timeout: Maximum time to wait for a response (seconds).
        exchange: The exchange to publish to (default: "" for direct).
        headers: Optional headers to include in the request.

    Returns:
        RabbitMessage: The response message.

    Raises:
        RPCTimeoutError: If the response is not received within the timeout.
        RPCClientClosed: If the client was closed (or is closed while waiting).
        RuntimeError: If max_pending calls is exceeded.
    """
    # L6: refuse to operate after close() instead of silently
    # re-registering a consumer on a half-torn-down client.
    if self._closed:
        raise RPCClientClosed("AsyncRPCClient is closed")
    await self._ensure_consuming()

    correlation_id = str(uuid.uuid4())

    async with self._lock:
        if self._router.is_full():
            raise RuntimeError(
                f"Max pending RPC calls ({self._router.max_pending}) exceeded. "
                "Consider increasing max_pending or reducing call rate."
            )
        loop = asyncio.get_running_loop()
        future: asyncio.Future[RabbitMessage] = loop.create_future()
        self._router.register(correlation_id, _AsyncFutureSink(future))

    # Publish request
    envelope = MessageEnvelope(
        routing_key=routing_key,
        body=body,
        exchange=exchange,
        reply_to=self._reply_queue,
        correlation_id=correlation_id,
        headers=headers or {},
    )
    await self._transport.publish(envelope)

    # Wait for response with timeout. R-timeout: asyncio.timeout (3.11+)
    # replaces asyncio.wait_for to avoid the wrapper-task overhead.
    try:
        async with asyncio.timeout(timeout):
            result = await future
    except TimeoutError:
        async with self._lock:
            self._router.pop(correlation_id)
        raise RPCTimeoutError(correlation_id, timeout) from None

    async with self._lock:
        self._router.pop(correlation_id)

    return result

close() -> None async

Close the RPC client and clean up.

Cancels the reply consumer and resolves all pending calls with a typed RPCClientClosed (L6) — matching the sync client's behavior — so an in-flight caller's await future raises cleanly instead of a bare asyncio.CancelledError that could be confused with the caller's own cancellation.

Source code in src/rabbitkit/rpc.py

async def close(self) -> None:
    """Close the RPC client and clean up.

    Cancels the reply consumer and resolves all pending calls with a
    typed ``RPCClientClosed`` (L6) — matching the sync client's
    behavior — so an in-flight caller's ``await future`` raises cleanly
    instead of a bare ``asyncio.CancelledError`` that could be confused
    with the caller's own cancellation.
    """
    if self._consumer_tag and self._transport:
        try:
            await self._transport.cancel_consumer(self._consumer_tag)
        except Exception as e:
            logger.warning("Failed to cancel RPC reply consumer: %s", e)

    # Clean up pending calls. L6: mark closed so subsequent
    # call()/_ensure_consuming() refuse to re-register a consumer on the
    # torn-down client.
    async with self._lock:
        self._closed = True
        self._router.close_all(RPCClientClosed("AsyncRPCClient was closed"))

    self._consuming = False
    self._consumer_tag = None

Exceptions

RPCTimeoutError

Bases: TimeoutError

Raised when an RPC call times out waiting for a response.

Source code in src/rabbitkit/rpc.py

class RPCTimeoutError(TimeoutError):
    """Raised when an RPC call times out waiting for a response."""

    def __init__(self, correlation_id: str, timeout: float) -> None:
        self.correlation_id = correlation_id
        self.timeout = timeout
        super().__init__(f"RPC call timed out after {timeout}s (correlation_id={correlation_id})")

RPCClientClosed

Bases: RuntimeError

Raised by :meth:RPCClient.call when the client has been closed.

close() resolves all outstanding waiters with this error instead of leaving result=None (which previously caused AttributeError).

Source code in src/rabbitkit/rpc.py

class RPCClientClosed(RuntimeError):
    """Raised by :meth:`RPCClient.call` when the client has been closed.

    ``close()`` resolves all outstanding waiters with this error instead of
    leaving ``result=None`` (which previously caused ``AttributeError``).
    """

ReplyTooLargeError

Bases: Exception

Raised when an RPC reply body exceeds max_reply_bytes (L-8).

Source code in src/rabbitkit/rpc.py

class ReplyTooLargeError(Exception):
    """Raised when an RPC reply body exceeds ``max_reply_bytes`` (L-8)."""

    def __init__(self, correlation_id: str, size: int, limit: int) -> None:
        self.correlation_id = correlation_id
        self.size = size
        self.limit = limit
        super().__init__(
            f"RPC reply (correlation_id={correlation_id}) body size {size} exceeds max_reply_bytes limit {limit}"
        )