def __init__()

in python/rocketmq/rpc_client.py [0:0]

• 47 lines of code
• 14 McCabe index (conditional complexity)

    def __init__(self, endpoints):
        self.Addresses = []

        self.scheme = AddressScheme.Unspecified
        self._hash = None

        if type(endpoints) == str:
            if endpoints.startswith(self.HttpPrefix):
                endpoints = endpoints[len(self.HttpPrefix):]
            if endpoints.startswith(self.HttpsPrefix):
                endpoints = endpoints[len(self.HttpsPrefix):]

            index = endpoints.find(self.EndpointSeparator)
            port = int(endpoints[index + 1:]) if index > 0 else 80
            host = endpoints[:index] if index > 0 else endpoints
            address = Address(host, port)
            self.Addresses.append(address)
            try:
                socket.inet_pton(socket.AF_INET, host)
                self.scheme = AddressScheme.IPv4
            except socket.error:
                try:
                    socket.inet_pton(socket.AF_INET6, host)
                    self.scheme = AddressScheme.IPv6
                except socket.error:
                    self.scheme = AddressScheme.DomainName
            self.Addresses.append(address)

            # Assuming AddressListEqualityComparer exists
            self._hash = 17
            self._hash = (self._hash * 31) + reduce(
                operator.xor, (hash(address) for address in self.Addresses)
            )
            self._hash = (self._hash * 31) + hash(self.scheme)
        else:
            self.Addresses = [
                Address(addr.host, addr.port) for addr in endpoints.addresses
            ]
            if not self.Addresses:
                raise Exception("No available address")

            if endpoints.scheme == "Ipv4":
                self.scheme = AddressScheme.Ipv4
            elif endpoints.scheme == "Ipv6":
                self.scheme = AddressScheme.Ipv6
            else:
                self.scheme = AddressScheme.DomainName
                if len(self.Addresses) > 1:
                    raise Exception(
                        "Multiple addresses are\
                                    not allowed in domain scheme"
                    )

            self._hash = self._calculate_hash()