Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def emit(self, event, *args, **kw):
socketIO_packet_type = 2
path = kw.get("path", "")
callback, args = find_callback(args, kw)
ack_id = self._set_ack_callback(callback) if callback else None
args = [event] + list(args)
socketIO_packet_data, binary_packets = format_socketIO_packet_data(
path, ack_id, args
)
if binary_packets:
socketIO_packet_type += 3
self._message(str(socketIO_packet_type) + socketIO_packet_data)
for packet in binary_packets:
self._message(packet)
def disconnect(self, path=""):
if not path or not self._opened:
self._close()
elif path:
socketIO_packet_type = 1
socketIO_packet_data, _ = format_socketIO_packet_data(path)
try:
self._message(str(socketIO_packet_type) + socketIO_packet_data)
except (TimeoutError, ConnectionError):
pass
try:
namespace = self._namespace_by_path.pop(path)
namespace.on_disconnect()
except KeyError:
pass
def connect(self, path, with_transport_instance=False):
socketIO_packet_type = 0
socketIO_packet_data, _ = format_socketIO_packet_data(path)
self._message(
str(socketIO_packet_type) + socketIO_packet_data, with_transport_instance
)
def _ack(self, path, ack_id, *args):
socketIO_packet_type = 3
socketIO_packet_data, binary_packets = format_socketIO_packet_data(
path, ack_id, args
)
if binary_packets:
socketIO_packet_type += 3
self._message(str(socketIO_packet_type) + socketIO_packet_data)
for packet in binary_packets:
self._message(packet)