1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
|
from typing import Iterable
import warnings
from ._proxy import AnyioProxy
class ProxyChain:
def __init__(self, proxies: Iterable[AnyioProxy]):
warnings.warn(
'This implementation of ProxyChain is deprecated and will be removed in the future',
DeprecationWarning,
stacklevel=2,
)
self._proxies = proxies
async def connect(
self,
dest_host,
dest_port,
dest_ssl=None,
timeout=None,
):
_stream = None
proxies = list(self._proxies)
length = len(proxies) - 1
for i in range(length):
_stream = await proxies[i].connect(
dest_host=proxies[i + 1].proxy_host,
dest_port=proxies[i + 1].proxy_port,
timeout=timeout,
_stream=_stream,
)
_stream = await proxies[length].connect(
dest_host=dest_host,
dest_port=dest_port,
dest_ssl=dest_ssl,
timeout=timeout,
_stream=_stream,
)
return _stream
|
