(self, url, message_hash, message=None, subscribe_hash=None, subscription=None)
| 523 | return future |
| 524 | |
| 525 | def watch(self, url, message_hash, message=None, subscribe_hash=None, subscription=None): |
| 526 | # base exchange self.open starts the aiohttp Session in an async context |
| 527 | self.open() |
| 528 | backoff_delay = 0 |
| 529 | client = self.client(url) |
| 530 | if subscribe_hash is None and message_hash in client.futures: |
| 531 | return client.futures[message_hash] |
| 532 | future = client.future(message_hash) |
| 533 | |
| 534 | subscribed = client.subscriptions.get(subscribe_hash) |
| 535 | |
| 536 | if not subscribed: |
| 537 | client.subscriptions[subscribe_hash] = subscription or True |
| 538 | |
| 539 | selected_session = self.session |
| 540 | # http/s proxy is being set in other places |
| 541 | httpProxy, httpsProxy, socksProxy = self.check_ws_proxy_settings() |
| 542 | if (socksProxy): |
| 543 | selected_session = self.get_socks_proxy_session(socksProxy) |
| 544 | connected = client.connected if client.connected.done() \ |
| 545 | else asyncio.ensure_future(client.connect(selected_session, backoff_delay)) |
| 546 | |
| 547 | def after(fut): |
| 548 | # todo: decouple signing from subscriptions |
| 549 | options = self.safe_value(self.options, 'ws') |
| 550 | cost = self.safe_value(options, 'cost', 1) |
| 551 | if message: |
| 552 | async def send_message(): |
| 553 | if self.enableRateLimit: |
| 554 | await client.throttle(cost) |
| 555 | try: |
| 556 | await client.send(message) |
| 557 | except ConnectionError as e: |
| 558 | client.on_error(e) |
| 559 | except Exception as e: |
| 560 | client.on_error(e) |
| 561 | asyncio.ensure_future(send_message()) |
| 562 | |
| 563 | if not subscribed: |
| 564 | connected.add_done_callback(after) |
| 565 | |
| 566 | return future |
| 567 | |
| 568 | def on_connected(self, client, message=None): |
| 569 | # for user hooks |
no test coverage detected