diff --git a/Doc/library/socket.rst b/Doc/library/socket.rst --- a/Doc/library/socket.rst +++ b/Doc/library/socket.rst @@ -325,6 +325,24 @@ .. versionchanged:: 3.2 support for the :keyword:`with` statement was added. +.. function:: create_server_sock(address[, reuse_addr[, queue_size]]) + + Convenience function which creates a TCP server bound to *address* (a 2-tuple + ``(host, port)``) and return the socket object. Internally it takes care of + choosing the right address family (IPv4 or IPv6) depending on the host + specified in *address*. The returned socket can then be used to accept() + new connections as in: + + >>> server = socket.create_server_sock(('localhost', 8000)) + >>> while True: + ... sock, addr = server.accept() + ... # handle new sock connection + + *reuse_addr* defaults to True on POSIX and tells the kernel to reuse a local + socket in ``TIME_WAIT`` state. *queue_size* is the maximum number of queued + connections passed to :meth:`socket.listen`. + + .. versionadded:: 3.4 .. function:: getaddrinfo(host, port, family=0, type=0, proto=0, flags=0) @@ -1427,4 +1445,3 @@ details of socket semantics. For Unix, refer to the manual pages; for Windows, see the WinSock (or Winsock 2) specification. For IPv6-ready APIs, readers may want to refer to :rfc:`3493` titled Basic Socket Interface Extensions for IPv6. - diff --git a/Lib/socket.py b/Lib/socket.py --- a/Lib/socket.py +++ b/Lib/socket.py @@ -435,3 +435,44 @@ raise err else: raise error("getaddrinfo returns an empty list") + +def create_server_sock(address, + reuse_addr=os.name == 'posix' and sys.platform != 'cygwin', + queue_size=5): + """Convenience function which creates a TCP server bound to + *address* and return the socket object. + Internally it takes care of choosing the right address family + (IPv4 or IPv6) depending on the host specified in *address*. + The returned socket can then be used to accept() new connections + as in: + + >>> server = create_server_sock(('localhost', 8000)) + >>> while True: + ... sock, addr = server.accept() + ... # handle new sock connection + """ + host, port = address + if host == "": + # http://mail.python.org/pipermail/python-ideas/2013-March/019937.html + host = None + err = None + info = getaddrinfo(host, port, AF_UNSPEC, SOCK_STREAM, 0, AI_PASSIVE) + for res in getaddrinfo(host, port, 0, SOCK_STREAM): + af, socktype, proto, canonname, sa = res + sock = None + try: + sock = socket(af, socktype, proto) + if reuse_addr: + sock.setsockopt(SOL_SOCKET, SO_REUSEADDR, 1) + sock.bind(sa) + sock.listen(queue_size) + return sock + except error as _: + err = _ + if sock is not None: + sock.close() + + if err is not None: + raise err + else: + raise error("getaddrinfo returns an empty list") diff --git a/Lib/test/test_socket.py b/Lib/test/test_socket.py --- a/Lib/test/test_socket.py +++ b/Lib/test/test_socket.py @@ -4249,6 +4249,14 @@ with self.assertRaises(socket.timeout): socket.create_connection((HOST, 1234)) + def test_create_server_sock(self): + port = support.find_unused_port() + with socket.create_server_sock(('localhost', port)) as sock: + self.assertEqual(sock.getsockname()[1], port) + self.assertEqual(sock.type, socket.SOCK_STREAM) + self.assertIn(sock.family, (socket.AF_INET, + getattr(socket, 'AF_INET6', 0))) + @unittest.skipUnless(thread, 'Threading required for this test.') class NetworkConnectionAttributesTest(SocketTCPTest, ThreadableTest):