(
self,
timeout: float = _INITIAL_CONNECT_TIMEOUT,
connect_timeout: Optional[Union[float, datetime.timedelta]] = None,
)
| 98 | return primary, secondary |
| 99 | |
| 100 | def start( |
| 101 | self, |
| 102 | timeout: float = _INITIAL_CONNECT_TIMEOUT, |
| 103 | connect_timeout: Optional[Union[float, datetime.timedelta]] = None, |
| 104 | ) -> "Future[Tuple[socket.AddressFamily, Any, IOStream]]": |
| 105 | self.try_connect(iter(self.primary_addrs)) |
| 106 | self.set_timeout(timeout) |
| 107 | if connect_timeout is not None: |
| 108 | self.set_connect_timeout(connect_timeout) |
| 109 | return self.future |
| 110 | |
| 111 | def try_connect(self, addrs: Iterator[Tuple[socket.AddressFamily, Tuple]]) -> None: |
| 112 | try: |
no test coverage detected