comparison hgext/zeroconf/Zeroconf.py @ 41519:a36f462cf533

zeroconf: Python 3 porting of vendored library

A quick glance through this module reveals that most of it "just works"
on Python 3 with the source transformer active. There are a few
places where we need to ensure we're using str.

Differential Revision: https://phab.mercurial-scm.org/D5804

41518:530d211ae9a8

import errno
import itertools
import select
import socket
import string
import struct
import threading
import time
import traceback

_LISTENER_TIME = 200
_BROWSER_TIME = 500

# Some DNS constants

_MDNS_ADDR = '224.0.0.251'
_MDNS_PORT = 5353
_DNS_PORT = 53
_DNS_TTL = 60 * 60 # one hour default TTL

_MAX_MSG_TYPICAL = 1460 # unused

class DNSEntry(object):
    """A DNS entry"""

    def __init__(self, name, type, clazz):
        self.key = string.lower(name)
        self.name = name
        self.type = type
        self.clazz = clazz & _CLASS_MASK
        self.unique = (clazz & _CLASS_UNIQUE) != 0

        off = self.offset
        next = -1
        first = off

        while True:
            len = ord(self.data[off])
            off += 1
            if len == 0:
                break
            t = len & 0xC0
            if t == 0x00:
                result = ''.join((result, self.readUTF(off, len) + '.'))
                off += len
            elif t == 0xC0:
                if next < 0:
                    next = off + 1
                off = ((len & 0x3F) << 8) | ord(self.data[off])
                if off >= first:
                    raise BadDomainNameCircular(off)
                first = off
            else:
                raise BadDomainName(off)

        except Exception:
            # Some versions of linux raise an exception even though
            # SO_REUSEADDR and SO_REUSEPORT have been set, so ignore it
            pass
        self.socket.setsockopt(socket.SOL_IP, socket.IP_ADD_MEMBERSHIP,
            socket.inet_aton(_MDNS_ADDR) + socket.inet_aton('0.0.0.0'))

        self.listeners = []
        self.browsers = []
        self.services = {}
        self.servicetypes = {}

            globals()['_GLOBAL_DONE'] = 1
            self.notifyAll()
            self.engine.notify()
            self.unregisterAllServices()
            self.socket.setsockopt(socket.SOL_IP, socket.IP_DROP_MEMBERSHIP,
                socket.inet_aton(_MDNS_ADDR) + socket.inet_aton('0.0.0.0'))
            self.socket.close()

# Test a few module features, including service registration, service
# query (for Zoe), and service unregistration.

41519:a36f462cf533

import errno
import itertools
import select
import socket
import struct
import threading
import time
import traceback

_LISTENER_TIME = 200
_BROWSER_TIME = 500

# Some DNS constants

_MDNS_ADDR = r'224.0.0.251'
_MDNS_PORT = 5353
_DNS_PORT = 53
_DNS_TTL = 60 * 60 # one hour default TTL

_MAX_MSG_TYPICAL = 1460 # unused

class DNSEntry(object):
    """A DNS entry"""

    def __init__(self, name, type, clazz):
        self.key = name.lower()
        self.name = name
        self.type = type
        self.clazz = clazz & _CLASS_MASK
        self.unique = (clazz & _CLASS_UNIQUE) != 0

        off = self.offset
        next = -1
        first = off

        while True:
            len = ord(self.data[off:off + 1])
            off += 1
            if len == 0:
                break
            t = len & 0xC0
            if t == 0x00:
                result = ''.join((result, self.readUTF(off, len) + '.'))
                off += len
            elif t == 0xC0:
                if next < 0:
                    next = off + 1
                off = ((len & 0x3F) << 8) | ord(self.data[off:off + 1])
                if off >= first:
                    raise BadDomainNameCircular(off)
                first = off
            else:
                raise BadDomainName(off)

        except Exception:
            # Some versions of linux raise an exception even though
            # SO_REUSEADDR and SO_REUSEPORT have been set, so ignore it
            pass
        self.socket.setsockopt(socket.SOL_IP, socket.IP_ADD_MEMBERSHIP,
            socket.inet_aton(_MDNS_ADDR) + socket.inet_aton(r'0.0.0.0'))

        self.listeners = []
        self.browsers = []
        self.services = {}
        self.servicetypes = {}

            globals()['_GLOBAL_DONE'] = 1
            self.notifyAll()
            self.engine.notify()
            self.unregisterAllServices()
            self.socket.setsockopt(socket.SOL_IP, socket.IP_DROP_MEMBERSHIP,
                socket.inet_aton(_MDNS_ADDR) + socket.inet_aton(r'0.0.0.0'))
            self.socket.close()

# Test a few module features, including service registration, service
# query (for Zoe), and service unregistration.