feat: idsse-897 rabbitmq_utils Rpc #87
Conversation
Co-authored-by: Geary Layne <[email protected]>
… _blocking_publish methods out of Publisher so Rpc can use them
|
Update: unit tests exist now. |
I found a bug that I thought I'd fix in your PR. Setting prefetch_count must be done before calling basic_consume.
Update rabbitmq_utils.py
| @@ -98,6 +98,12 @@ class PublishMessageParams(NamedTuple): | |||
| route_key: str | None = None | |||
|
|
|||
|
|
|||
| class RpcResponse(NamedTuple): | |||
There was a problem hiding this comment.
Do we need this class in addition to PublishMessageParams?
There was a problem hiding this comment.
Technically no, but PublishMessageParams is used for outbound RMQ messages and RpcResponse was originally written for inbound responses. I just don't know what to name the class if it's used for both.
There was a problem hiding this comment.
Maybe "RabbitMqMessage"?
There was a problem hiding this comment.
That works, or RmqMessage if you want shorter.
| @@ -117,7 +117,6 @@ def exec_cmd(commands: Sequence[str], timeout: int | None = None) -> Sequence[st | |||
|
|
|||
| def to_iso(date_time: datetime) -> str: | |||
| """Format a datetime instance to an ISO string""" | |||
| logger.debug('Datetime (%s) to iso', datetime) | |||
There was a problem hiding this comment.
I've wondered if these need to be here a few times.
| RpcResponse | None: The response message (body and properties), or None on request | ||
| timeout or error handling response. | ||
| """ | ||
| if not self.is_open: |
There was a problem hiding this comment.
I like that this gets handled
Co-authored-by: Geary Layne <[email protected]>
NOTE: depends on PR #88<-- mergedLinear Issue
IDSSE-897
Changes
Rpcclass in rabbitmq_utilsConsumerPublisherlogic to blocking_publish messages to an exchange, but reuses the Consumer's thread which is a constraint of Direct Reply-toConsumerthen listens to responses coming back over RabbitMQ's built-in Direct Reply-to queue and returns the response message to the caller when it is receivedCallers can simple instantiate an Rpc class (providing the params of the exchange to publish to), call
start(), then publish messages withsend_request(). This method will block until the receiving RMQ service sends a response (or times out), formatted as anRpcResponse(containing the body and properties of the message).Explanation
This type of logic existed in Risk Processor (called "RpcClient"), but moving it here and correctly using these newer rabbitmq_utils classes helps standardize behavior across IDSSe services.