Creates a listening unix socket. If a socket with the given name already exists, it will be deleted. If any other file with that name exists, an exception will be raised. Returns a socket object (not a list of socket objects like `bind_sockets`)
(
file: str, mode: int = 0o600, backlog: int = _DEFAULT_BACKLOG
)
| 190 | if hasattr(socket, "AF_UNIX"): |
| 191 | |
| 192 | def bind_unix_socket( |
| 193 | file: str, mode: int = 0o600, backlog: int = _DEFAULT_BACKLOG |
| 194 | ) -> socket.socket: |
| 195 | """Creates a listening unix socket. |
| 196 | |
| 197 | If a socket with the given name already exists, it will be deleted. |
| 198 | If any other file with that name exists, an exception will be |
| 199 | raised. |
| 200 | |
| 201 | Returns a socket object (not a list of socket objects like |
| 202 | `bind_sockets`) |
| 203 | """ |
| 204 | sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) |
| 205 | try: |
| 206 | sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) |
| 207 | except socket.error as e: |
| 208 | if errno_from_exception(e) != errno.ENOPROTOOPT: |
| 209 | # Hurd doesn't support SO_REUSEADDR |
| 210 | raise |
| 211 | sock.setblocking(False) |
| 212 | try: |
| 213 | st = os.stat(file) |
| 214 | except FileNotFoundError: |
| 215 | pass |
| 216 | else: |
| 217 | if stat.S_ISSOCK(st.st_mode): |
| 218 | os.remove(file) |
| 219 | else: |
| 220 | raise ValueError("File %s exists and is not a socket", file) |
| 221 | sock.bind(file) |
| 222 | os.chmod(file, mode) |
| 223 | sock.listen(backlog) |
| 224 | return sock |
| 225 | |
| 226 | |
| 227 | def add_accept_handler( |
nothing calls this directly
no test coverage detected