parameters: (
              self, protocol_factory, host=None, port=None,
              *,
              family=socket.AF_UNSPEC,
              flags=socket.AI_PASSIVE,
              sock=None,
              backlog=100,
              ssl=None,
              reuse_address=None,
              reuse_port=None,
              ssl_handshake_timeout=None,
              start_serving=True)
return_statement: return server
docstring: Create a TCP server.
  
          The host parameter can be a string, in that case the TCP server is
          bound to host and port.
  
          The host parameter can also be a sequence of strings and in that case
          the TCP server is bound to all hosts of the sequence. If a host
          appears multiple times (possibly indirectly e.g. when hostnames
          resolve to the same IP address), the server is only bound once to that
          host.
  
          Return a Server object which can be used to stop the service.
  
          This method is a coroutine.
docstring_summary: Create a TCP server.
docstring_tokens: Create a TCP server .
function: async def create_server(
              self, protocol_factory, host=None, port=None,
              *,
              family=socket.AF_UNSPEC,
              flags=socket.AI_PASSIVE,
              sock=None,
              backlog=100,
              ssl=None,
              reuse_address=None,
              reuse_port=None,
              ssl_handshake_timeout=None,
              start_serving=True):
          """Create a TCP server.
  
          The host parameter can be a string, in that case the TCP server is
          bound to host and port.
  
          The host parameter can also be a sequence of strings and in that case
          the TCP server is bound to all hosts of the sequence. If a host
          appears multiple times (possibly indirectly e.g. when hostnames
          resolve to the same IP address), the server is only bound once to that
          host.
  
          Return a Server object which can be used to stop the service.
  
          This method is a coroutine.
          """
          if isinstance(ssl, bool):
              raise TypeError('ssl argument must be an SSLContext or None')
  
          if ssl_handshake_timeout is not None and ssl is None:
              raise ValueError(
                  'ssl_handshake_timeout is only meaningful with ssl')
  
          if host is not None or port is not None:
              if sock is not None:
                  raise ValueError(
                      'host/port and sock can not be specified at the same time')
  
              if reuse_address is None:
                  reuse_address = os.name == 'posix' and sys.platform != 'cygwin'
              sockets = []
              if host == '':
                  hosts = [None]
              elif (isinstance(host, str) or
                    not isinstance(host, collections.abc.Iterable)):
                  hosts = [host]
              else:
                  hosts = host
  
              fs = [self._create_server_getaddrinfo(host, port, family=family,
                                                    flags=flags)
                    for host in hosts]
              infos = await tasks.gather(*fs, loop=self)
              infos = set(itertools.chain.from_iterable(infos))
  
              completed = False
              try:
                  for res in infos:
                      af, socktype, proto, canonname, sa = res
                      try:
                          sock = socket.socket(af, socktype, proto)
                      except socket.error:
                          # Assume it's a bad family/type/protocol combination.
                          if self._debug:
                              logger.warning('create_server() failed to create '
                                             'socket.socket(%r, %r, %r)',
                                             af, socktype, proto, exc_info=True)
                          continue
                      sockets.append(sock)
                      if reuse_address:
                          sock.setsockopt(
                              socket.SOL_SOCKET, socket.SO_REUSEADDR, True)
                      if reuse_port:
                          _set_reuseport(sock)
                      # Disable IPv4/IPv6 dual stack support (enabled by
                      # default on Linux) which makes a single socket
                      # listen on both address families.
                      if (_HAS_IPv6 and
                              af == socket.AF_INET6 and
                              hasattr(socket, 'IPPROTO_IPV6')):
                          sock.setsockopt(socket.IPPROTO_IPV6,
                                          socket.IPV6_V6ONLY,
                                          True)
                      try:
                          sock.bind(sa)
                      except OSError as err:
                          raise OSError(err.errno, 'error while attempting '
                                        'to bind on address %r: %s'
                                        % (sa, err.strerror.lower())) from None
                  completed = True
              finally:
                  if not completed:
                      for sock in sockets:
                          sock.close()
          else:
              if sock is None:
                  raise ValueError('Neither host/port nor sock were specified')
              if sock.type != socket.SOCK_STREAM:
                  raise ValueError(f'A Stream Socket was expected, got {sock!r}')
              sockets = [sock]
  
          for sock in sockets:
              sock.setblocking(False)
  
          server = Server(self, sockets, protocol_factory,
                          ssl, backlog, ssl_handshake_timeout)
          if start_serving:
              server._start_serving()
              # Skip one loop iteration so that all 'loop.add_reader'
              # go through.
              await tasks.sleep(0, loop=self)
  
          if self._debug:
              logger.info("%r is serving", server)
          return server
function_tokens: async def create_server ( self , protocol_factory , host = None , port = None , * , family = socket . AF_UNSPEC , flags = socket . AI_PASSIVE , sock = None , backlog = 100 , ssl = None , reuse_address = None , reuse_port = None , ssl_handshake_timeout = None , start_serving = True ) : if isinstance ( ssl , bool ) : raise TypeError ( 'ssl argument must be an SSLContext or None' ) if ssl_handshake_timeout is not None and ssl is None : raise ValueError ( 'ssl_handshake_timeout is only meaningful with ssl' ) if host is not None or port is not None : if sock is not None : raise ValueError ( 'host/port and sock can not be specified at the same time' ) if reuse_address is None : reuse_address = os . name == 'posix' and sys . platform != 'cygwin' sockets = [ ] if host == '' : hosts = [ None ] elif ( isinstance ( host , str ) or not isinstance ( host , collections . abc . Iterable ) ) : hosts = [ host ] else : hosts = host fs = [ self . _create_server_getaddrinfo ( host , port , family = family , flags = flags ) for host in hosts ] infos = await tasks . gather ( * fs , loop = self ) infos = set ( itertools . chain . from_iterable ( infos ) ) completed = False try : for res in infos : af , socktype , proto , canonname , sa = res try : sock = socket . socket ( af , socktype , proto ) except socket . error : # Assume it's a bad family/type/protocol combination. if self . _debug : logger . warning ( 'create_server() failed to create ' 'socket.socket(%r, %r, %r)' , af , socktype , proto , exc_info = True ) continue sockets . append ( sock ) if reuse_address : sock . setsockopt ( socket . SOL_SOCKET , socket . SO_REUSEADDR , True ) if reuse_port : _set_reuseport ( sock ) # Disable IPv4/IPv6 dual stack support (enabled by # default on Linux) which makes a single socket # listen on both address families. if ( _HAS_IPv6 and af == socket . AF_INET6 and hasattr ( socket , 'IPPROTO_IPV6' ) ) : sock . setsockopt ( socket . IPPROTO_IPV6 , socket . IPV6_V6ONLY , True ) try : sock . bind ( sa ) except OSError as err : raise OSError ( err . errno , 'error while attempting ' 'to bind on address %r: %s' % ( sa , err . strerror . lower ( ) ) ) from None completed = True finally : if not completed : for sock in sockets : sock . close ( ) else : if sock is None : raise ValueError ( 'Neither host/port nor sock were specified' ) if sock . type != socket . SOCK_STREAM : raise ValueError ( f'A Stream Socket was expected, got {sock!r}' ) sockets = [ sock ] for sock in sockets : sock . setblocking ( False ) server = Server ( self , sockets , protocol_factory , ssl , backlog , ssl_handshake_timeout ) if start_serving : server . _start_serving ( ) # Skip one loop iteration so that all 'loop.add_reader' # go through. await tasks . sleep ( 0 , loop = self ) if self . _debug : logger . info ( "%r is serving" , server ) return server