如何开发 Avahi 客户端/服务器

Posted 2023-04-14

【中文标题】如何开发 Avahi 客户端/服务器

【英文标题】：How to develop an Avahi client/server

【发布时间】：2011-03-26 16:57:37

【问题描述】：

我正在尝试使用 python 开发客户端/服务器解决方案，服务器必须使用 Avahi 广播服务可用性。我正在使用以下代码发布服务：

import avahi
import dbus

__all__ = ["ZeroconfService"]

class ZeroconfService:
    """A simple class to publish a network service with zeroconf using
    avahi.

    """

    def __init__(self, name, port, stype="_http._tcp",
                 domain="", host="", text=""):
        self.name = name
        self.stype = stype
        self.domain = domain
        self.host = host
        self.port = port
        self.text = text

    def publish(self):
        bus = dbus.SystemBus()
        server = dbus.Interface(
                         bus.get_object(
                                 avahi.DBUS_NAME,
                                 avahi.DBUS_PATH_SERVER),
                        avahi.DBUS_INTERFACE_SERVER)

        g = dbus.Interface(
                    bus.get_object(avahi.DBUS_NAME,
                                   server.EntryGroupNew()),
                    avahi.DBUS_INTERFACE_ENTRY_GROUP)

        g.AddService(avahi.IF_UNSPEC, avahi.PROTO_UNSPEC,dbus.UInt32(0),
                     self.name, self.stype, self.domain, self.host,
                     dbus.UInt16(self.port), self.text)

        g.Commit()
        self.group = g

    def unpublish(self):
        self.group.Reset()


def test():
    service = ZeroconfService(name="TestService", port=3000)
    service.publish()
    raw_input("Press any key to unpublish the service ")
    service.unpublish()


if __name__ == "__main__":
    test()

至于客户，我正在尝试使用以下方式搜索服务：

# http://avahi.org/wiki/PythonBrowseExample
import dbus, gobject, avahi
from dbus import DBusException
from dbus.mainloop.glib import DBusGMainLoop

# Looks for iTunes shares

TYPE = "_http._tcp"

def service_resolved(*args):
    print 'service resolved'
    print 'name:', args[2]
    print 'address:', args[7]
    print 'port:', args[8]

def print_error(*args):
    print 'error_handler'
    print args[0]

def myhandler(interface, protocol, name, stype, domain, flags):
    print "Found service '%s' type '%s' domain '%s' " % (name, stype, domain)

    if flags & avahi.LOOKUP_RESULT_LOCAL:
            # local service, skip
            pass

    server.ResolveService(interface, protocol, name, stype, 
        domain, avahi.PROTO_UNSPEC, dbus.UInt32(0), 
        reply_handler=service_resolved, error_handler=print_error)

loop = DBusGMainLoop()

bus = dbus.SystemBus(mainloop=loop)

server = dbus.Interface( bus.get_object(avahi.DBUS_NAME, '/'),
        'org.freedesktop.Avahi.Server')

sbrowser = dbus.Interface(bus.get_object(avahi.DBUS_NAME,
        server.ServiceBrowserNew(avahi.IF_UNSPEC,
            avahi.PROTO_UNSPEC, TYPE, 'local', dbus.UInt32(0))),
        avahi.DBUS_INTERFACE_SERVICE_BROWSER)

sbrowser.connect_to_signal("ItemNew", myhandler)

gobject.MainLoop().run()

但是客户端没有检测到服务何时启动。关于我做错了什么的任何想法？

【参考方案1】：

我发现代码按预期工作。我的防火墙规则阻止了与 avahi 相关的发布。

【讨论】：

这曾经工作得很好，我重新使用了你的优秀代码。但是使用最新的 Fedora（也许还有其他），我得到：'dbus.exceptions.DBusException: org.freedesktop.DBus.Error.UnknownMethod: Method "AddService" with signature "iiussssqs" on interface "org.freedesktop.Avahi.EntryGroup" “不存在”

更多细节和修复可以在这里找到：xpra.org/trac/ticket/1153你现在必须在TXT记录上调用'avahi.string_array_to_txt_array'。