def __init__()

in fbtftp/base_handler.py [0:0]

• 42 lines of code
• 6 McCabe index (conditional complexity)

    def __init__(self, server_addr, peer, path, options, stats_callback):
        """
        Class that deals with talking to a single client. Being a subclass of
        `multiprocessing.Process` this will run in a separate process from the
        main process.

        Note:
            Do not use this class as is, inherit from it and override the
            `get_response_data` method which must return a subclass of
            `ResponseData`.

        Args:
            server_addr (tuple): (ip, port) of the server

            peer (tuple): (ip, port of) the peer

            path (string): requested file

            options (dict): a dictionary containing the options the client
                wants to negotiate.

            stats_callback (callable): a callable that will be executed at the
                end of the session. It gets passed an instance of the
                `SessionStats` class.
        """
        self._timeout = int(options["default_timeout"])
        self._server_addr = server_addr
        self._reset_timeout()
        self._retries = int(options["retries"])
        self._block_size = constants.DEFAULT_BLKSIZE
        self._last_block_sent = 0
        self._retransmits = 0
        self._global_retransmits = 0
        self._current_block = None
        self._should_stop = False
        self._waiting_last_ack = False
        self._path = path
        self._options = options
        self._stats_callback = stats_callback
        self._response_data = None
        self._listener = None

        self._peer = peer
        logging.info(
            "New connection from peer `%s` asking for path `%s`"
            % (str(peer), str(path))
        )
        self._family = socket.AF_INET6
        # the format of the peer tuple is different for v4 and v6
        if isinstance(ipaddress.ip_address(server_addr[0]), ipaddress.IPv4Address):
            self._family = socket.AF_INET
            # peer address format is different in v4 world
            self._peer = (self._peer[0].replace("::ffff:", ""), self._peer[1])

        self._stats = SessionStats(self._server_addr, self._peer, self._path)

        try:
            self._response_data = self.get_response_data()
        except FileNotFoundError as e:
            logging.warning(str(e))
            self._stats.error = {
                "error_code": constants.ERR_FILE_NOT_FOUND,
                "error_message": str(e),
            }
        except Exception as e:
            logging.exception("Caught exception: %s." % e)
            self._stats.error = {
                "error_code": constants.ERR_UNDEFINED,
                "error_message": str(e),
            }

        super().__init__()