Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
# tricky: get a pair of connected sockets
a.bind (self.address)
a.listen (1)
w.setblocking (0)
try:
w.connect (self.address)
except:
pass
r, addr = a.accept()
a.close()
w.setblocking (1)
self.trigger = w
asyncore.dispatcher.__init__ (self, r)
self.lock = _thread.allocate_lock()
self.thunks = []
self._trigger_connected = 0
self._cv = threading.Condition ()
self.__sendlock = None
self.__no_more_request = False
self.set_event_time ()
self.handler = None
self.auth = None
self.proxy = False
self.initialize_connection ()
self._closed = False
self.backend = False
self.ac_in_buffer = b''
self.incoming = []
self.producer_fifo = self.fifo_class ()
asyncore.dispatcher.__init__(self)
def __init__ (self, handler, host, port):
asyncore.dispatcher.__init__ (self)
self.handler = handler
self.create_socket(socket.AF_INET, socket.SOCK_STREAM)
self.connect((host, port))
def __init__ (self, family=socket.AF_INET, address=('', 9003)):
asyncore.dispatcher.__init__ (self)
self.create_socket (family, socket.SOCK_STREAM)
self.set_reuse_addr()
self.bind (address)
self.listen (5)
def __init__(self, host, port):
asyncore.dispatcher.__init__(self)
self.create_socket(socket.AF_INET, socket.SOCK_STREAM)
self.connect((host, port))
def __init__ (self, address, params = None, lock = None, logger = None):
dbconnect.AsynDBConnect.__init__ (self, address, params, lock, logger)
self.cur = None
self.retries = 0
asyncore.dispatcher.__init__ (self)