我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用socket.settimeout()。
def handle_request(self): """Handle one request, possibly blocking. Respects self.timeout. """ # Support people who used socket.settimeout() to escape # handle_request before self.timeout was available. timeout = self.socket.gettimeout() if timeout is None: timeout = self.timeout elif self.timeout is not None: timeout = min(timeout, self.timeout) fd_sets = select.select([self], [], [], timeout) if not fd_sets[0]: self.handle_timeout() return self._handle_request_noblock()
def handle_request(self): """Handle one request, possibly blocking. Respects self.timeout. """ # Support people who used socket.settimeout() to escape # handle_request before self.timeout was available. timeout = self.socket.gettimeout() if timeout is None: timeout = self.timeout elif self.timeout is not None: timeout = min(timeout, self.timeout) fd_sets = _eintr_retry(select.select, [self], [], [], timeout) if not fd_sets[0]: self.handle_timeout() return self._handle_request_noblock()
def _send_socket(self, cmd, rtnCmd, ip, port): socket = self._socket try: _LOGGER.debug('Sending to GW {0}'.format(cmd)) self._read_unwanted_data() socket.settimeout(30.0) socket.sendto(cmd.encode(), (ip, port)) socket.settimeout(30.0) data, addr = socket.recvfrom(1024) if len(data) is not None: resp = json.loads(data.decode()) _LOGGER.debug('Recieved from GW {0}'.format(resp)) if resp["cmd"] == rtnCmd: return resp else: _LOGGER.error("Response from {0} does not match return cmd".format(ip)) _LOGGER.error(data) else: _LOGGER.error("No response from Gateway") except socket.timeout: _LOGGER.error("Cannot connect to Gateway") socket.close()
def __init__(self, microWebSrv, socket, addr) : socket.settimeout(2) self._microWebSrv = microWebSrv self._socket = socket self._addr = addr self._method = None self._path = None self._httpVer = None self._resPath = "/" self._queryString = "" self._queryParams = { } self._headers = { } self._contentType = None self._contentLength = 0 self._processRequest() # ------------------------------------------------------------------------
def _sock_operation(self, count, timeout, method, *args): """ Test the specified socket method. The method is run at most `count` times and must raise a socket.timeout within `timeout` + self.fuzz seconds. """ self.sock.settimeout(timeout) method = getattr(self.sock, method) for i in range(count): t1 = time.time() try: method(*args) except socket.timeout as e: delta = time.time() - t1 break else: self.fail('socket.timeout was not raised') # These checks should account for timing unprecision self.assertLess(delta, timeout + self.fuzz) self.assertGreater(delta, timeout - 1.0)
def testConnectTimeout(self): # Choose a private address that is unlikely to exist to prevent # failures due to the connect succeeding before the timeout. # Use a dotted IP address to avoid including the DNS lookup time # with the connect time. This avoids failing the assertion that # the timeout occurred fast enough. addr = ('10.0.0.0', 12345) # Test connect() timeout _timeout = 0.001 self.sock.settimeout(_timeout) _t1 = time.time() self.assertRaises(socket.error, self.sock.connect, addr) _t2 = time.time() _delta = abs(_t1 - _t2) self.assertTrue(_delta < _timeout + self.fuzz, "timeout (%g) is more than %g seconds more than expected (%g)" %(_delta, self.fuzz, _timeout))
def testRecvTimeout(self): # Test recv() timeout _timeout = 0.02 with test_support.transient_internet(self.addr_remote[0]): self.sock.connect(self.addr_remote) self.sock.settimeout(_timeout) _t1 = time.time() self.assertRaises(socket.timeout, self.sock.recv, 1024) _t2 = time.time() _delta = abs(_t1 - _t2) self.assertTrue(_delta < _timeout + self.fuzz, "timeout (%g) is %g seconds more than expected (%g)" %(_delta, self.fuzz, _timeout))
def testAcceptTimeout(self): # Test accept() timeout _timeout = 2 self.sock.settimeout(_timeout) # Prevent "Address already in use" socket exceptions test_support.bind_port(self.sock, self.localhost) self.sock.listen(5) _t1 = time.time() self.assertRaises(socket.error, self.sock.accept) _t2 = time.time() _delta = abs(_t1 - _t2) self.assertTrue(_delta < _timeout + self.fuzz, "timeout (%g) is %g seconds more than expected (%g)" %(_delta, self.fuzz, _timeout))
def setup(self): self.connection = self.request if self.timeout is not None: self.connection.settimeout(self.timeout) if self.disable_nagle_algorithm: self.connection.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, True) self.rfile = self.connection.makefile('rb', self.rbufsize) self.wfile = self.connection.makefile('wb', self.wbufsize)
