How to use the zeroconf.DNSService function in zeroconf

To help you get started, we’ve selected a few zeroconf examples, based on popular ways it is used in public projects.

Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.

github wmcbrine / tivoremote / zeroconf.py View on Github external
if info.type in self.servicetypes:
            self.servicetypes[info.type]+=1
        else:
            self.servicetypes[info.type]=1
        now = currentTimeMillis()
        nextTime = now
        i = 0
        while i < 3:
            if now < nextTime:
                self.wait(nextTime - now)
                now = currentTimeMillis()
                continue
            out = DNSOutgoing(_FLAGS_QR_RESPONSE | _FLAGS_AA)
            out.addAnswerAtTime(DNSPointer(info.type, _TYPE_PTR, 
                _CLASS_IN, ttl, info.name), 0)
            out.addAnswerAtTime(DNSService(info.name, _TYPE_SRV, 
                _CLASS_IN, ttl, info.priority, info.weight, info.port, 
                info.server), 0)
            out.addAnswerAtTime(DNSText(info.name, _TYPE_TXT, _CLASS_IN, 
                ttl, info.text), 0)
            if info.address:
                out.addAnswerAtTime(DNSAddress(info.server, _TYPE_A, 
                    _CLASS_IN, ttl, info.address), 0)
            self.send(out)
            i += 1
            nextTime += _REGISTER_TIME
github wmcbrine / hmeforpython / zeroconf.py View on Github external
def unregisterAllServices(self):
        """Unregister all registered services."""
        if len(self.services) > 0:
            now = currentTimeMillis()
            nextTime = now
            i = 0
            while i < 3:
                if now < nextTime:
                    self.wait(nextTime - now)
                    now = currentTimeMillis()
                    continue
                out = DNSOutgoing(_FLAGS_QR_RESPONSE | _FLAGS_AA)
                for info in self.services.values():
                    out.addAnswerAtTime(DNSPointer(info.type, _TYPE_PTR,
                        _CLASS_IN, 0, info.name), 0)
                    out.addAnswerAtTime(DNSService(info.name, _TYPE_SRV,
                        _CLASS_IN, 0, info.priority, info.weight,
                        info.port, info.server), 0)
                    out.addAnswerAtTime(DNSText(info.name, _TYPE_TXT,
                        _CLASS_IN, 0, info.text), 0)
                    if info.address:
                        out.addAnswerAtTime(DNSAddress(info.server,
                            _TYPE_A, _CLASS_IN, 0, info.address), 0)
                self.send(out)
                i += 1
                nextTime += _UNREGISTER_TIME
github wmcbrine / hmeforpython / zeroconf.py View on Github external
if info.type in self.servicetypes:
            self.servicetypes[info.type] += 1
        else:
            self.servicetypes[info.type] = 1
        now = currentTimeMillis()
        nextTime = now
        i = 0
        while i < 3:
            if now < nextTime:
                self.wait(nextTime - now)
                now = currentTimeMillis()
                continue
            out = DNSOutgoing(_FLAGS_QR_RESPONSE | _FLAGS_AA)
            out.addAnswerAtTime(DNSPointer(info.type, _TYPE_PTR,
                _CLASS_IN, ttl, info.name), 0)
            out.addAnswerAtTime(DNSService(info.name, _TYPE_SRV,
                _CLASS_IN, ttl, info.priority, info.weight, info.port,
                info.server), 0)
            out.addAnswerAtTime(DNSText(info.name, _TYPE_TXT, _CLASS_IN,
                ttl, info.text), 0)
            if info.address:
                out.addAnswerAtTime(DNSAddress(info.server, _TYPE_A,
                    _CLASS_IN, ttl, info.address), 0)
            self.send(out)
            i += 1
            nextTime += _REGISTER_TIME
github wmcbrine / hmeforpython / zeroconf.py View on Github external
else:
                del self.servicetypes[info.type]
        except:
            pass
        now = currentTimeMillis()
        nextTime = now
        i = 0
        while i < 3:
            if now < nextTime:
                self.wait(nextTime - now)
                now = currentTimeMillis()
                continue
            out = DNSOutgoing(_FLAGS_QR_RESPONSE | _FLAGS_AA)
            out.addAnswerAtTime(DNSPointer(info.type, _TYPE_PTR,
                _CLASS_IN, 0, info.name), 0)
            out.addAnswerAtTime(DNSService(info.name, _TYPE_SRV,
                _CLASS_IN, 0, info.priority, info.weight, info.port,
                info.name), 0)
            out.addAnswerAtTime(DNSText(info.name, _TYPE_TXT, _CLASS_IN,
                0, info.text), 0)
            if info.address:
                out.addAnswerAtTime(DNSAddress(info.server, _TYPE_A,
                    _CLASS_IN, 0, info.address), 0)
            self.send(out)
            i += 1
            nextTime += _UNREGISTER_TIME
github wmcbrine / tivoremote / zeroconf.py View on Github external
else:
                del self.servicetypes[info.type]
        except:
            pass
        now = currentTimeMillis()
        nextTime = now
        i = 0
        while i < 3:
            if now < nextTime:
                self.wait(nextTime - now)
                now = currentTimeMillis()
                continue
            out = DNSOutgoing(_FLAGS_QR_RESPONSE | _FLAGS_AA)
            out.addAnswerAtTime(DNSPointer(info.type, _TYPE_PTR, 
                _CLASS_IN, 0, info.name), 0)
            out.addAnswerAtTime(DNSService(info.name, _TYPE_SRV, 
                _CLASS_IN, 0, info.priority, info.weight, info.port, 
                info.name), 0)
            out.addAnswerAtTime(DNSText(info.name, _TYPE_TXT, _CLASS_IN, 
                0, info.text), 0)
            if info.address:
                out.addAnswerAtTime(DNSAddress(info.server, _TYPE_A, 
                    _CLASS_IN, 0, info.address), 0)
            self.send(out)
            i += 1
            nextTime += _UNREGISTER_TIME
github wmcbrine / tivoremote / zeroconf.py View on Github external
"""Reads the answers, authorities and additionals section of the 
        packet"""
        n = self.numAnswers + self.numAuthorities + self.numAdditionals
        for i in xrange(n):
            domain = self.readName()
            type, clazz, ttl, length = self.unpack('!HHiH')

            rec = None
            if type == _TYPE_A:
                rec = DNSAddress(domain, type, clazz, ttl, self.readString(4))
            elif type == _TYPE_CNAME or type == _TYPE_PTR:
                rec = DNSPointer(domain, type, clazz, ttl, self.readName())
            elif type == _TYPE_TXT:
                rec = DNSText(domain, type, clazz, ttl, self.readString(length))
            elif type == _TYPE_SRV:
                rec = DNSService(domain, type, clazz, ttl,
                    self.readUnsignedShort(), self.readUnsignedShort(),
                    self.readUnsignedShort(), self.readName())
            elif type == _TYPE_HINFO:
                rec = DNSHinfo(domain, type, clazz, ttl,
                    self.readCharacterString(), self.readCharacterString())
            elif type == _TYPE_AAAA:
                rec = DNSAddress(domain, type, clazz, ttl, self.readString(16))
            else:
                # Try to ignore types we don't know about
                # Skip the payload for the resource record so the next
                # records can be parsed correctly
                self.offset += length

            if rec is not None:
                self.answers.append(rec)
github wmcbrine / hmeforpython / zeroconf.py View on Github external
if out is None:
                        out = DNSOutgoing(_FLAGS_QR_RESPONSE | _FLAGS_AA)

                    # Answer A record queries for any service addresses we know
                    if question.type in (_TYPE_A, _TYPE_ANY):
                        for service in self.services.values():
                            if service.server == question.name.lower():
                                out.addAnswer(msg, DNSAddress(question.name,
                                    _TYPE_A, _CLASS_IN | _CLASS_UNIQUE,
                                    _DNS_TTL, service.address))

                    service = self.services.get(question.name.lower(), None)
                    if not service: continue

                    if question.type in (_TYPE_SRV, _TYPE_ANY):
                        out.addAnswer(msg, DNSService(question.name,
                            _TYPE_SRV, _CLASS_IN | _CLASS_UNIQUE,
                            _DNS_TTL, service.priority, service.weight,
                            service.port, service.server))
                    if question.type in (_TYPE_TXT, _TYPE_ANY):
                        out.addAnswer(msg, DNSText(question.name,
                            _TYPE_TXT, _CLASS_IN | _CLASS_UNIQUE,
                            _DNS_TTL, service.text))
                    if question.type == _TYPE_SRV:
                        out.addAdditionalAnswer(DNSAddress(service.server,
                            _TYPE_A, _CLASS_IN | _CLASS_UNIQUE,
                            _DNS_TTL, service.address))
                except:
                    traceback.print_exc()

        if out is not None and out.answers:
            out.id = msg.id
github wmcbrine / hmeforpython / zeroconf.py View on Github external
"""Reads the answers, authorities and additionals section of the
        packet"""
        n = self.numAnswers + self.numAuthorities + self.numAdditionals
        for i in xrange(n):
            domain = self.readName()
            type, clazz, ttl, length = self.unpack('!HHiH')

            rec = None
            if type == _TYPE_A:
                rec = DNSAddress(domain, type, clazz, ttl, self.readString(4))
            elif type == _TYPE_CNAME or type == _TYPE_PTR:
                rec = DNSPointer(domain, type, clazz, ttl, self.readName())
            elif type == _TYPE_TXT:
                rec = DNSText(domain, type, clazz, ttl, self.readString(length))
            elif type == _TYPE_SRV:
                rec = DNSService(domain, type, clazz, ttl,
                    self.readUnsignedShort(), self.readUnsignedShort(),
                    self.readUnsignedShort(), self.readName())
            elif type == _TYPE_HINFO:
                rec = DNSHinfo(domain, type, clazz, ttl,
                    self.readCharacterString(), self.readCharacterString())
            elif type == _TYPE_AAAA:
                rec = DNSAddress(domain, type, clazz, ttl, self.readString(16))
            else:
                # Try to ignore types we don't know about
                # Skip the payload for the resource record so the next
                # records can be parsed correctly
                self.offset += length

            if rec is not None:
                self.answers.append(rec)
github wmcbrine / tivoremote / zeroconf.py View on Github external
def __eq__(self, other):
        """Tests equality on priority, weight, port and server"""
        return (isinstance(other, DNSService) and
                self.priority == other.priority and
                self.weight == other.weight and
                self.port == other.port and
                self.server == other.server)