| 69 | |
| 70 | |
| 71 | class Broker(MessagingHandler): |
| 72 | def __init__(self, url): |
| 73 | super(Broker, self).__init__() |
| 74 | self.url = url |
| 75 | self.queues = {} |
| 76 | |
| 77 | def on_start(self, event): |
| 78 | self.acceptor = event.container.listen(self.url) |
| 79 | |
| 80 | def _queue(self, address): |
| 81 | if address not in self.queues: |
| 82 | self.queues[address] = Queue() |
| 83 | return self.queues[address] |
| 84 | |
| 85 | def on_connection_opening(self, event): |
| 86 | event.connection.offered_capabilities = 'ANONYMOUS-RELAY' |
| 87 | |
| 88 | def on_link_opening(self, event): |
| 89 | if event.link.is_sender: |
| 90 | if event.link.remote_source.dynamic: |
| 91 | address = str(uuid.uuid4()) |
| 92 | event.link.source.address = address |
| 93 | q = Queue(True) |
| 94 | self.queues[address] = q |
| 95 | q.subscribe(event.link) |
| 96 | elif event.link.remote_source.address: |
| 97 | event.link.source.address = event.link.remote_source.address |
| 98 | self._queue(event.link.source.address).subscribe(event.link) |
| 99 | elif event.link.remote_target.address: |
| 100 | event.link.target.address = event.link.remote_target.address |
| 101 | |
| 102 | def _unsubscribe(self, link): |
| 103 | if link.source.address in self.queues and self.queues[link.source.address].unsubscribe(link): |
| 104 | del self.queues[link.source.address] |
| 105 | |
| 106 | def on_link_closing(self, event): |
| 107 | if event.link.is_sender: |
| 108 | self._unsubscribe(event.link) |
| 109 | |
| 110 | def on_connection_closing(self, event): |
| 111 | self.remove_stale_consumers(event.connection) |
| 112 | |
| 113 | def on_disconnected(self, event): |
| 114 | self.remove_stale_consumers(event.connection) |
| 115 | |
| 116 | def remove_stale_consumers(self, connection): |
| 117 | link = connection.link_head(Endpoint.REMOTE_ACTIVE) |
| 118 | while link: |
| 119 | if link.is_sender: |
| 120 | self._unsubscribe(link) |
| 121 | link = link.next(Endpoint.REMOTE_ACTIVE) |
| 122 | |
| 123 | def on_sendable(self, event): |
| 124 | self._queue(event.link.source.address).dispatch(event.link) |
| 125 | |
| 126 | def on_message(self, event): |
| 127 | address = event.link.target.address |
| 128 | if address is None: |