ACIL FM

Nama	Ukuran	Aksi
__pycache__	-	Open
base_events.py	72.45 MB	View DL Edit
base_futures.py	2.51 MB	View DL Edit
base_subprocess.py	8.64 MB	View DL Edit
base_tasks.py	2.41 MB	View DL Edit
constants.py	888 B	View DL Edit
coroutines.py	8.59 MB	View DL Edit
events.py	25.83 MB	View DL Edit
exceptions.py	1.59 MB	View DL Edit
format_helpers.py	2.35 MB	View DL Edit
futures.py	13.71 MB	View DL Edit
locks.py	14.83 MB	View DL Edit
log.py	124 B	View DL Edit
proactor_events.py	31.39 MB	View DL Edit
protocols.py	6.79 MB	View DL Edit
queues.py	8.11 MB	View DL Edit
runners.py	2.08 MB	View DL Edit
selector_events.py	38.57 MB	View DL Edit
sslproto.py	26.82 MB	View DL Edit
staggered.py	5.85 MB	View DL Edit
streams.py	26.03 MB	View DL Edit
subprocess.py	7.88 MB	View DL Edit
tasks.py	33.61 MB	View DL Edit
threads.py	790 B	View DL Edit
transports.py	10.47 MB	View DL Edit
trsock.py	5.74 MB	View DL Edit
unix_events.py	50.54 MB	View DL Edit
windows_events.py	32.28 MB	View DL Edit
windows_utils.py	4.94 MB	View DL Edit
__init__.py	1.25 MB	View DL Edit
__main__.py	3.3 MB	View DL Edit

Edit file: /opt/alt/python39/lib64/python3.9/asyncio/trsock.py

import socket
import warnings

class TransportSocket:

"""A socket-like wrapper for exposing real transport sockets.

These objects can be safely returned by APIs like
    `transport.get_extra_info('socket')`.  All potentially disruptive
    operations (like "socket.close()") are banned.
    """

__slots__ = ('_sock',)

def __init__(self, sock: socket.socket):
        self._sock = sock

def _na(self, what):
        warnings.warn(
            f"Using {what} on sockets returned from get_extra_info('socket') "
            f"will be prohibited in asyncio 3.9. Please report your use case "
            f"to bugs.python.org.",
            DeprecationWarning, source=self)

@property
    def family(self):
        return self._sock.family

@property
    def type(self):
        return self._sock.type

@property
    def proto(self):
        return self._sock.proto

def __repr__(self):
        s = (
            f"<asyncio.TransportSocket fd={self.fileno()}, "
            f"family={self.family!s}, type={self.type!s}, "
            f"proto={self.proto}"
        )

if self.fileno() != -1:
            try:
                laddr = self.getsockname()
                if laddr:
                    s = f"{s}, laddr={laddr}"
            except socket.error:
                pass
            try:
                raddr = self.getpeername()
                if raddr:
                    s = f"{s}, raddr={raddr}"
            except socket.error:
                pass

return f"{s}>"

def __getstate__(self):
        raise TypeError("Cannot serialize asyncio.TransportSocket object")

def fileno(self):
        return self._sock.fileno()

def dup(self):
        return self._sock.dup()

def get_inheritable(self):
        return self._sock.get_inheritable()

def shutdown(self, how):
        # asyncio doesn't currently provide a high-level transport API
        # to shutdown the connection.
        self._sock.shutdown(how)

def getsockopt(self, *args, **kwargs):
        return self._sock.getsockopt(*args, **kwargs)

def setsockopt(self, *args, **kwargs):
        self._sock.setsockopt(*args, **kwargs)

def getpeername(self):
        return self._sock.getpeername()

def getsockname(self):
        return self._sock.getsockname()

def getsockbyname(self):
        return self._sock.getsockbyname()

def accept(self):
        self._na('accept() method')
        return self._sock.accept()

def connect(self, *args, **kwargs):
        self._na('connect() method')
        return self._sock.connect(*args, **kwargs)

def connect_ex(self, *args, **kwargs):
        self._na('connect_ex() method')
        return self._sock.connect_ex(*args, **kwargs)

def bind(self, *args, **kwargs):
        self._na('bind() method')
        return self._sock.bind(*args, **kwargs)

def ioctl(self, *args, **kwargs):
        self._na('ioctl() method')
        return self._sock.ioctl(*args, **kwargs)

def listen(self, *args, **kwargs):
        self._na('listen() method')
        return self._sock.listen(*args, **kwargs)

def makefile(self):
        self._na('makefile() method')
        return self._sock.makefile()

def sendfile(self, *args, **kwargs):
        self._na('sendfile() method')
        return self._sock.sendfile(*args, **kwargs)

def close(self):
        self._na('close() method')
        return self._sock.close()

def detach(self):
        self._na('detach() method')
        return self._sock.detach()

def sendmsg_afalg(self, *args, **kwargs):
        self._na('sendmsg_afalg() method')
        return self._sock.sendmsg_afalg(*args, **kwargs)

def sendmsg(self, *args, **kwargs):
        self._na('sendmsg() method')
        return self._sock.sendmsg(*args, **kwargs)

def sendto(self, *args, **kwargs):
        self._na('sendto() method')
        return self._sock.sendto(*args, **kwargs)

def send(self, *args, **kwargs):
        self._na('send() method')
        return self._sock.send(*args, **kwargs)

def sendall(self, *args, **kwargs):
        self._na('sendall() method')
        return self._sock.sendall(*args, **kwargs)

def set_inheritable(self, *args, **kwargs):
        self._na('set_inheritable() method')
        return self._sock.set_inheritable(*args, **kwargs)

def share(self, process_id):
        self._na('share() method')
        return self._sock.share(process_id)

def recv_into(self, *args, **kwargs):
        self._na('recv_into() method')
        return self._sock.recv_into(*args, **kwargs)

def recvfrom_into(self, *args, **kwargs):
        self._na('recvfrom_into() method')
        return self._sock.recvfrom_into(*args, **kwargs)

def recvmsg_into(self, *args, **kwargs):
        self._na('recvmsg_into() method')
        return self._sock.recvmsg_into(*args, **kwargs)

def recvmsg(self, *args, **kwargs):
        self._na('recvmsg() method')
        return self._sock.recvmsg(*args, **kwargs)

def recvfrom(self, *args, **kwargs):
        self._na('recvfrom() method')
        return self._sock.recvfrom(*args, **kwargs)

def recv(self, *args, **kwargs):
        self._na('recv() method')
        return self._sock.recv(*args, **kwargs)

def settimeout(self, value):
        if value == 0:
            return
        raise ValueError(
            'settimeout(): only 0 timeout is allowed on transport sockets')

def gettimeout(self):
        return 0

def setblocking(self, flag):
        if not flag:
            return
        raise ValueError(
            'setblocking(): transport sockets cannot be blocking')

def __enter__(self):
        self._na('context manager protocol')
        return self._sock.__enter__()

def __exit__(self, *err):
        self._na('context manager protocol')
        return self._sock.__exit__(*err)

Batal