MCPcopy Index your code
hub / github.com/RustPython/RustPython / _tunnel

Method _tunnel

Lib/http/client.py:974–1004  ·  view source on GitHub ↗
(self)

Source from the content-addressed store, hash-verified

972 return ip
973
974 def _tunnel(self):
975 connect = b"CONNECT %s:%d %s\r\n" % (
976 self._wrap_ipv6(self._tunnel_host.encode("idna")),
977 self._tunnel_port,
978 self._http_vsn_str.encode("ascii"))
979 headers = [connect]
980 for header, value in self._tunnel_headers.items():
981 headers.append(f"{header}: {value}\r\n".encode("latin-1"))
982 headers.append(b"\r\n")
983 # Making a single send() call instead of one per line encourages
984 # the host OS to use a more optimal packet size instead of
985 # potentially emitting a series of small packets.
986 self.send(b"".join(headers))
987 del headers
988
989 response = self.response_class(self.sock, method=self._method)
990 try:
991 (version, code, message) = response._read_status()
992
993 self._raw_proxy_headers = _read_headers(response.fp)
994
995 if self.debuglevel > 0:
996 for header in self._raw_proxy_headers:
997 print('header:', header.decode())
998
999 if code != http.HTTPStatus.OK:
1000 self.close()
1001 raise OSError(f"Tunnel connection failed: {code} {message.strip()}")
1002
1003 finally:
1004 response.close()
1005
1006 def get_proxy_response_headers(self):
1007 """

Callers 1

connectMethod · 0.95

Calls 13

_wrap_ipv6Method · 0.95
sendMethod · 0.95
closeMethod · 0.95
_read_headersFunction · 0.85
_read_statusMethod · 0.80
printFunction · 0.50
encodeMethod · 0.45
itemsMethod · 0.45
appendMethod · 0.45
joinMethod · 0.45
decodeMethod · 0.45
stripMethod · 0.45

Tested by

no test coverage detected