Implementation of the synchronous request-response (aka RPC) pattern. A single instance can send many requests to the same or different addresses. :param connection: Connection for requests and responses. :param address: Address for all requests. If not specified, each request
| 578 | |
| 579 | |
| 580 | class SyncRequestResponse(IncomingMessageHandler): |
| 581 | """ |
| 582 | Implementation of the synchronous request-response (aka RPC) pattern. |
| 583 | A single instance can send many requests to the same or different |
| 584 | addresses. |
| 585 | |
| 586 | :param connection: Connection for requests and responses. |
| 587 | :param address: Address for all requests. If not specified, each request |
| 588 | must have the address property set. Successive messages may have |
| 589 | different addresses. |
| 590 | """ |
| 591 | |
| 592 | correlation_id = AtomicCount() |
| 593 | |
| 594 | def __init__(self, connection: BlockingConnection, address: Optional[str] = None) -> None: |
| 595 | super(SyncRequestResponse, self).__init__() |
| 596 | self.connection = connection |
| 597 | self.address = address |
| 598 | self.sender = self.connection.create_sender(self.address) |
| 599 | # dynamic=true generates a unique address dynamically for this receiver. |
| 600 | # credit=1 because we want to receive 1 response message initially. |
| 601 | self.receiver = self.connection.create_receiver(None, dynamic=True, credit=1, handler=self) |
| 602 | self.response = None |
| 603 | |
| 604 | def call(self, request: 'Message') -> 'Message': |
| 605 | """ |
| 606 | Send a request message, wait for and return the response message. |
| 607 | |
| 608 | :param request: Request message. If ``self.address`` is not set the |
| 609 | request message address must be set and will be used. |
| 610 | """ |
| 611 | if not self.address and not request.address: |
| 612 | raise ValueError("Request message has no address: %s" % request) |
| 613 | request.reply_to = self.reply_to |
| 614 | request.correlation_id = correlation_id = str(self.correlation_id.next()) |
| 615 | self.sender.send(request) |
| 616 | |
| 617 | def wakeup(): |
| 618 | return self.response and (self.response.correlation_id == correlation_id) |
| 619 | |
| 620 | self.connection.wait(wakeup, msg="Waiting for response") |
| 621 | response = self.response |
| 622 | self.response = None # Ready for next response. |
| 623 | self.receiver.flow(1) # Set up credit for the next response. |
| 624 | return response |
| 625 | |
| 626 | @property |
| 627 | def reply_to(self) -> str: |
| 628 | """ |
| 629 | The dynamic address of our receiver. |
| 630 | """ |
| 631 | return self.receiver.remote_source.address |
| 632 | |
| 633 | def on_message(self, event: 'Event') -> None: |
| 634 | """ |
| 635 | Called when we receive a message for our receiver. |
| 636 | |
| 637 | :param event: The event which occurs when a message is received. |