(self, event)
| 39 | self.conn = event.container.connect(self.url, desired_capabilities="ANONYMOUS-RELAY") |
| 40 | |
| 41 | def on_connection_opened(self, event): |
| 42 | capabilities = event.connection.remote_offered_capabilities |
| 43 | if capabilities and 'ANONYMOUS-RELAY' in capabilities: |
| 44 | self.receiver = event.container.create_receiver(self.conn, self.address) |
| 45 | self.server = self.container.create_sender(self.conn, None) |
| 46 | else: |
| 47 | global exit_status |
| 48 | print("Server needs a broker which supports ANONYMOUS-RELAY", file=sys.stderr) |
| 49 | exit_status = 1 |
| 50 | c = event.connection |
| 51 | c.condition = Condition('amqp:not-implemented', description="ANONYMOUS-RELAY required") |
| 52 | c.close() |
| 53 | |
| 54 | def on_message(self, event): |
| 55 | print("Received", event.message) |
nothing calls this directly
no test coverage detected