(self, host, port, search_limit=100, skew=+0)
| 114 | stdin=self._handle, stdout=self._handle) |
| 115 | |
| 116 | def get_avail_port(self, host, port, search_limit=100, skew=+0): |
| 117 | try: |
| 118 | _, skew = current_process().name.split('-') |
| 119 | skew = int(skew) |
| 120 | except ValueError: |
| 121 | pass |
| 122 | this_port = None |
| 123 | for i in range(search_limit): |
| 124 | _sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) |
| 125 | _sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) |
| 126 | this_port = port + skew + i |
| 127 | try: |
| 128 | _sock.bind((host, this_port)) |
| 129 | except OSError as exc: |
| 130 | if exc.errno in [errno.EADDRINUSE, errno.EINVAL]: |
| 131 | continue |
| 132 | raise |
| 133 | else: |
| 134 | return _sock, this_port |
| 135 | raise Exception(NO_AVAILABLE_PORT.format(self=self)) |
| 136 | |
| 137 | def say(self, m): |
| 138 | print(m, file=self.out) |
no test coverage detected