WIP local discovery

clientlibrary/service/lib_android.go

@@ -4,7 +4,7 @@ package service
 import "C"
 
 import (
-	ma "github.com/multiformats/go-multiaddr"
+	"github.com/anytypeio/go-anytype-middleware/net/addrs"
 )
 
 type InterfaceAddr interface {
@@ -26,12 +26,12 @@ type interfaceGetterAdapter struct {
 	interfaceGetter InterfaceAddrsGetter
 }
 
-func (i *interfaceGetterAdapter) InterfaceAddrs() []ma.InterfaceAddr {
+func (i *interfaceGetterAdapter) InterfaceAddrs() []addrs.InterfaceAddr {
 	iter := i.interfaceGetter.InterfaceAddrs()
 	addr := iter.Next()
-	var res []ma.InterfaceAddr
+	var res []addrs.InterfaceAddr
 	for addr != nil {
-		res = append(res, ma.InterfaceAddr{
+		res = append(res, addrs.InterfaceAddr{
 			Ip:     addr.Ip(),
 			Prefix: addr.Prefix(),
 		})
@@ -41,7 +41,7 @@ func (i *interfaceGetterAdapter) InterfaceAddrs() []ma.InterfaceAddr {
 }
 
 func SetInterfaceAddrsGetter(getter InterfaceAddrsGetter) {
-	ma.SetInterfaceAddrsGetter(&interfaceGetterAdapter{
+	addrs.SetInterfaceAddrsGetter(&interfaceGetterAdapter{
 		interfaceGetter: getter,
 	})
 }

net/addrs/interface.go

@@ -0,0 +1,21 @@
+//go:build !android
+// +build !android
+
+package addrs
+
+import "net"
+
+func SetInterfaceAddrsGetter(getter InterfaceAddrsGetter) {}
+
+type InterfaceAddr struct {
+	Ip     []byte
+	Prefix int
+}
+
+type InterfaceAddrsGetter interface {
+	InterfaceAddrs() []InterfaceAddr
+}
+
+func InterfaceAddrs() (addrs []net.Addr, err error) {
+	return net.InterfaceAddrs()
+}

net/addrs/interface_android.go

@@ -0,0 +1,67 @@
+package addrs
+
+import (
+	"fmt"
+	"net"
+	"sync"
+)
+
+var lock = sync.Mutex{}
+var interfaceGetter InterfaceAddrsGetter
+
+func SetInterfaceAddrsGetter(getter InterfaceAddrsGetter) {
+	lock.Lock()
+	defer lock.Unlock()
+	interfaceGetter = getter
+}
+
+type InterfaceAddr struct {
+	Ip     []byte
+	Prefix int
+}
+
+type InterfaceAddrsGetter interface {
+	InterfaceAddrs() []InterfaceAddr
+}
+
+func maskFromPrefix(prefix, base int) net.IPMask {
+	buf := make([]byte, base/8, base/8)
+	for i := 0; i < prefix/8; i++ {
+		buf[i] = 0xff
+	}
+	if prefix != base {
+		buf[prefix/8] = ((1 << prefix % 8) - 1) << (8 - prefix%8)
+	}
+	return buf
+}
+
+func ipV6MaskFromPrefix(prefix int) net.IPMask {
+	return maskFromPrefix(prefix, 128)
+}
+
+func ipV4MaskFromPrefix(prefix int) net.IPMask {
+	return maskFromPrefix(prefix, 32)
+}
+
+func InterfaceAddrs() (addrs []net.Addr, err error) {
+	lock.Lock()
+	if interfaceGetter == nil {
+		lock.Unlock()
+		return nil, fmt.Errorf("interface getter not set for Android")
+	}
+	lock.Unlock()
+	unmaskedAddrs := interfaceGetter.InterfaceAddrs()
+	for _, addr := range unmaskedAddrs {
+		var mask []byte
+		if len(addr.Ip) == 4 {
+			mask = ipV4MaskFromPrefix(addr.Prefix)
+		} else {
+			mask = ipV6MaskFromPrefix(addr.Prefix)
+		}
+		addrs = append(addrs, &net.IPNet{
+			IP:   addr.Ip,
+			Mask: mask,
+		})
+	}
+	return addrs, nil
+}

space/localdiscovery/localdiscovery.go

@@ -0,0 +1,135 @@
+package localdiscovery
+
+import (
+	"context"
+	"github.com/anytypeio/any-sync/accountservice"
+	"github.com/anytypeio/any-sync/app"
+	"github.com/anytypeio/any-sync/app/logger"
+	"github.com/anytypeio/go-anytype-middleware/net/addrs"
+	"github.com/libp2p/zeroconf/v2"
+	"go.uber.org/zap"
+	"strings"
+	"sync"
+)
+
+const (
+	CName = "client.space.localdiscovery"
+
+	serviceName   = "_p2p._localdiscovery"
+	mdnsDomain    = "local"
+	anytypePrefix = "anytype="
+)
+
+var log = logger.NewNamed(CName)
+
+type DiscoveredPeer struct {
+	Addr   string
+	PeerId string
+}
+
+type Notifier interface {
+	PeerDiscovered(peer DiscoveredPeer)
+}
+
+type LocalDiscovery interface {
+	SetNotifier(Notifier)
+	app.ComponentRunnable
+}
+
+type localDiscovery struct {
+	server *zeroconf.Server
+	peerId string
+
+	ctx       context.Context
+	cancel    context.CancelFunc
+	closeWait sync.WaitGroup
+
+	notifier Notifier
+}
+
+func (l *localDiscovery) SetNotifier(notifier Notifier) {
+	l.notifier = notifier
+}
+
+func (l *localDiscovery) Init(a *app.App) (err error) {
+	l.peerId = a.MustComponent(accountservice.CName).(accountservice.Service).Account().PeerId
+	return
+}
+
+func (l *localDiscovery) Run(ctx context.Context) (err error) {
+	l.ctx, l.cancel = context.WithCancel(ctx)
+	if err = l.startServer(); err != nil {
+		return
+	}
+	l.startListener(l.ctx)
+	return
+}
+
+func (l *localDiscovery) Name() (name string) {
+	return CName
+}
+
+func (l *localDiscovery) Close(ctx context.Context) (err error) {
+	l.cancel()
+	if l.server != nil {
+		l.server.Shutdown()
+	}
+	l.closeWait.Wait()
+	return nil
+}
+
+func (l *localDiscovery) startServer() (err error) {
+	interfaceAddrs, err := addrs.InterfaceAddrs()
+	if err != nil {
+		return
+	}
+	var (
+		ips  []string
+		txts []string
+	)
+	for _, addr := range interfaceAddrs {
+		ips = append(ips, addr.String())
+		txts = append(txts, anytypePrefix+addr.String())
+	}
+	l.server, err = zeroconf.RegisterProxy(
+		l.peerId,
+		serviceName,
+		mdnsDomain,
+		4001,
+		l.peerId,
+		ips,
+		txts,
+		nil,
+	)
+	return
+}
+
+func (l *localDiscovery) startListener(ctx context.Context) {
+	l.closeWait.Add(2)
+	listenCh := make(chan *zeroconf.ServiceEntry, 10)
+	go l.readChannel(listenCh)
+	go l.writeChannel(ctx, listenCh)
+}
+
+func (l *localDiscovery) readChannel(ch chan *zeroconf.ServiceEntry) {
+	defer l.closeWait.Done()
+	for entry := range ch {
+		for _, text := range entry.Text {
+			if !strings.HasPrefix(text, anytypePrefix) {
+				log.Debug("incorrect prefix, text", zap.String("text", text))
+				continue
+			}
+			l.notifier.PeerDiscovered(DiscoveredPeer{
+				Addr:   text[len(anytypePrefix):],
+				PeerId: entry.Service,
+			})
+		}
+	}
+}
+
+func (l *localDiscovery) writeChannel(ctx context.Context, ch chan *zeroconf.ServiceEntry) {
+	defer l.closeWait.Done()
+	if err := zeroconf.Browse(ctx, l.peerId, mdnsDomain, ch); err != nil {
+		log.Debug("browsing failed", zap.Error(err))
+	}
+}