Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def connect (self):
self.create_socket (socket.AF_INET, socket.SOCK_STREAM)
try:
asynchat.async_chat.connect (self, self.address)
except:
self.handle_error ()
def _initiate_send_ts (self):
with self.__sendlock:
return asynchat.async_chat.initiate_send (self)
from . import adns
from ..protocols.http2 import H2_PROTOCOLS
from ..athreads.fifo import await_fifo
from rs4.ssl_ import resolve_cert_reqs, resolve_ssl_version, create_urllib3_context
from collections import deque
from ..protocols.http import respcodes
import random
DEBUG = 0
DEFAULT_ZOMBIE_TIMEOUT = 60
DEFAULT_KEEP_ALIVE = 2
class SocketPanic (Exception): pass
class TimeOut (Exception): pass
class AsynConnect (asynchat.async_chat):
ac_in_buffer_size = 65535
ac_out_buffer_size = 65535
zombie_timeout = DEFAULT_ZOMBIE_TIMEOUT
keep_alive = DEFAULT_KEEP_ALIVE
request_count = 0
active = 0
fifo_class = deque
keep_connect = True
is_ssl = False
def __init__ (self, address, lock = None, logger = None):
self.address = address
self.lock = lock
self.logger = logger
self._cv = threading.Condition ()
self.__sendlock = None