feat(module,cast): enhance casting device discovery speed

pkg/module/cast/cast.go

@@ -3,10 +3,10 @@ package cast
 import (
 	"context"
 	"net"
+	"sync"
 	"time"
 
 	"github.com/barnybug/go-cast"
-	"github.com/barnybug/go-cast/discovery"
 	"github.com/pkg/errors"
 	"gitlab.com/wpetit/goweb/logger"
 )
@@ -18,6 +18,15 @@ type Device struct {
 	Name string `goja:"name" json:"name"`
 }
 
+type CachedDevice struct {
+	Device
+	UpdatedAt time.Time
+}
+
+func (d CachedDevice) Expired() bool {
+	return d.UpdatedAt.Add(30 * time.Minute).Before(time.Now())
+}
+
 type DeviceStatus struct {
 	CurrentApp DeviceStatusCurrentApp `goja:"currentApp" json:"currentApp"`
 	Volume     DeviceStatusVolume     `goja:"volume" json:"volume"`
@@ -35,9 +44,36 @@ type DeviceStatusVolume struct {
 }
 
 const (
-	serviceDiscoveryPollingInterval time.Duration = 2 * time.Second
+	serviceDiscoveryPollingInterval time.Duration = 500 * time.Millisecond
 )
 
+var cache sync.Map
+
+func getCachedDevice(uuid string) (Device, bool) {
+	value, exists := cache.Load(uuid)
+	if !exists {
+		return Device{}, false
+	}
+
+	cachedDevice, ok := value.(CachedDevice)
+	if !ok {
+		return Device{}, false
+	}
+
+	if cachedDevice.Expired() {
+		return Device{}, false
+	}
+
+	return cachedDevice.Device, true
+}
+
+func cacheDevice(dev Device) {
+	cache.Store(dev.UUID, CachedDevice{
+		Device:    dev,
+		UpdatedAt: time.Now(),
+	})
+}
+
 func getDeviceClientByUUID(ctx context.Context, uuid string) (*cast.Client, error) {
 	device, err := FindDeviceByUUID(ctx, uuid)
 	if err != nil {
@@ -49,82 +85,114 @@ func getDeviceClientByUUID(ctx context.Context, uuid string) (*cast.Client, erro
 	return client, nil
 }
 
-func FindDeviceByUUID(ctx context.Context, uuid string) (*Device, error) {
-	service := discovery.NewService(ctx)
-	defer service.Stop()
+func FindDeviceByUUID(ctx context.Context, uuid string) (Device, error) {
+	device, exists := getCachedDevice(uuid)
+	if exists {
+		return device, nil
+	}
 
-	go func() {
-		if err := service.Run(ctx, serviceDiscoveryPollingInterval); err != nil {
-			logger.Error(ctx, "error while running cast service discovery", logger.E(errors.WithStack(err)))
-		}
-	}()
+	ctx, cancel := context.WithCancel(ctx)
+	defer cancel()
 
-LOOP:
-	for {
-		select {
-		case c := <-service.Found():
-			if c.Uuid() == uuid {
-				return &Device{
-					Host: c.IP().To4(),
-					Port: c.Port(),
-					Name: c.Name(),
-					UUID: c.Uuid(),
-				}, nil
-			}
-		case <-ctx.Done():
-			break LOOP
+	devices, err := SearchDevices(ctx)
+	if err != nil {
+		return Device{}, nil
+	}
+
+	for dev := range devices {
+		if dev.UUID == uuid {
+			return dev, nil
 		}
 	}
 
-	if err := ctx.Err(); err != nil {
-		return nil, errors.WithStack(err)
-	}
-
-	return nil, errors.WithStack(ErrDeviceNotFound)
+	return Device{}, errors.Errorf("could not find device '%s'", uuid)
 }
 
-func FindDevices(ctx context.Context) ([]*Device, error) {
-	service := discovery.NewService(ctx)
-	defer service.Stop()
+func ListDevices(ctx context.Context, refresh bool) ([]Device, error) {
+	devices := make([]Device, 0)
 
-	go func() {
-		if err := service.Run(ctx, serviceDiscoveryPollingInterval); err != nil && !errors.Is(err, context.DeadlineExceeded) {
-			logger.Error(ctx, "error while running cast service discovery", logger.E(errors.WithStack(err)))
-		}
-	}()
-
-	devices := make([]*Device, 0)
-	found := make(map[string]struct{})
-
-LOOP:
-	for {
-		select {
-		case c := <-service.Found():
-			if _, exists := found[c.Uuid()]; exists {
-				continue
+	if !refresh {
+		cache.Range(func(key, value any) bool {
+			cached, ok := value.(CachedDevice)
+			if !ok || cached.Expired() {
+				return true
 			}
 
-			devices = append(devices, &Device{
-				Host: c.IP().To4(),
-				Port: c.Port(),
-				Name: c.Name(),
-				UUID: c.Uuid(),
-			})
-			found[c.Uuid()] = struct{}{}
+			devices = append(devices, cached.Device)
+			return true
+		})
 
-		case <-ctx.Done():
-			break LOOP
-		}
+		return devices, nil
 	}
 
-	if err := ctx.Err(); err != nil && !errors.Is(err, context.DeadlineExceeded) {
+	ch, err := SearchDevices(ctx)
+	if err != nil {
 		return nil, errors.WithStack(err)
 	}
 
+	for dev := range ch {
+		devices = append(devices, dev)
+	}
+
 	return devices, nil
 }
 
+var searchDevicesMutex sync.Mutex
+
+func SearchDevices(ctx context.Context) (chan Device, error) {
+	service := NewService(ctx)
+	defer service.Stop()
+
+	go func() {
+		searchDevicesMutex.Lock()
+		defer searchDevicesMutex.Unlock()
+
+		if err := service.Run(ctx, serviceDiscoveryPollingInterval); err != nil && !errors.Is(err, context.DeadlineExceeded) {
+			logger.Error(ctx, "error while running cast service discovery", logger.E(errors.WithStack(err)))
+		}
+	}()
+
+	devices := make(chan Device)
+
+	go func() {
+		defer close(devices)
+
+		found := make(map[string]struct{})
+
+	LOOP:
+		for {
+			select {
+			case c := <-service.Found():
+				dev := Device{
+					Host: c.IP().To4(),
+					Port: c.Port(),
+					Name: c.Name(),
+					UUID: c.Uuid(),
+				}
+
+				if _, exists := found[dev.UUID]; !exists {
+					devices <- dev
+
+					found[dev.UUID] = struct{}{}
+				}
+
+				cacheDevice(dev)
+
+			case <-ctx.Done():
+				break LOOP
+			}
+		}
+	}()
+
+	return devices, nil
+}
+
+var loadURLMutex sync.Mutex
+
 func LoadURL(ctx context.Context, deviceUUID string, url string) error {
+	loadURLMutex.Lock()
+	defer loadURLMutex.Unlock()
+
 	client, err := getDeviceClientByUUID(ctx, deviceUUID)
 	if err != nil {
 		return errors.WithStack(err)
@@ -153,7 +221,12 @@ func isLoadURLContextExceeded(err error) bool {
 	return err.Error() == "Failed to send load command: context deadline exceeded"
 }
 
+var stopCastMutex sync.Mutex
+
 func StopCast(ctx context.Context, deviceUUID string) error {
+	stopCastMutex.Lock()
+	defer stopCastMutex.Unlock()
+
 	client, err := getDeviceClientByUUID(ctx, deviceUUID)
 	if err != nil {
 		return errors.WithStack(err)
@@ -171,7 +244,12 @@ func StopCast(ctx context.Context, deviceUUID string) error {
 	return nil
 }
 
+var getStatusMutex sync.Mutex
+
 func getStatus(ctx context.Context, deviceUUID string) (*DeviceStatus, error) {
+	getStatusMutex.Lock()
+	defer getStatusMutex.Unlock()
+
 	client, err := getDeviceClientByUUID(ctx, deviceUUID)
 	if err != nil {
 		return nil, errors.WithStack(err)

pkg/module/cast/cast_test.go

@@ -26,11 +26,24 @@ func TestCastLoadURL(t *testing.T) {
 	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
 	defer cancel()
 
-	devices, err := FindDevices(ctx)
+	devices, err := ListDevices(ctx, true)
 	if err != nil {
 		t.Error(errors.WithStack(err))
 	}
 
+	t.Logf("DEVICES: %s", spew.Sdump(devices))
+
+	if e, g := 1, len(devices); e != g {
+		t.Fatalf("len(devices): expected '%v', got '%v'", e, g)
+	}
+
+	devices, err = ListDevices(ctx, false)
+	if err != nil {
+		t.Error(errors.WithStack(err))
+	}
+
+	t.Logf("CACHED DEVICES: %s", spew.Sdump(devices))
+
 	if e, g := 1, len(devices); e != g {
 		t.Fatalf("len(devices): expected '%v', got '%v'", e, g)
 	}
@@ -52,7 +65,7 @@ func TestCastLoadURL(t *testing.T) {
 		t.Error(errors.WithStack(err))
 	}
 
-	spew.Dump(status)
+	t.Logf("DEVICE STATUS: %s", spew.Sdump(status))
 
 	ctx, cancel4 := context.WithTimeout(context.Background(), 15*time.Second)
 	defer cancel4()

pkg/module/cast/discovery.go

@@ -0,0 +1,250 @@
+package cast
+
+import (
+	"context"
+	"net"
+	"regexp"
+	"strconv"
+	"strings"
+	"sync"
+	"time"
+
+	"gitlab.com/wpetit/goweb/logger"
+
+	"github.com/barnybug/go-cast"
+	"github.com/barnybug/go-cast/log"
+	"github.com/hashicorp/mdns"
+	"github.com/pkg/errors"
+)
+
+type Service struct {
+	found     chan *cast.Client
+	entriesCh chan *mdns.ServiceEntry
+
+	stopPeriodic chan struct{}
+}
+
+func NewService(ctx context.Context) *Service {
+	s := &Service{
+		found:     make(chan *cast.Client),
+		entriesCh: make(chan *mdns.ServiceEntry, 10),
+	}
+
+	go s.listener(ctx)
+	return s
+}
+
+func (d *Service) Run(ctx context.Context, interval time.Duration) error {
+	ifaces, err := findMulticastInterfaces(ctx)
+	if err != nil {
+		return errors.WithStack(err)
+	}
+
+	var wg sync.WaitGroup
+
+	for _, iface := range ifaces {
+		hasIPv4, hasIPv6, err := retrieveSupportedProtocols(iface)
+		if err != nil {
+			return errors.WithStack(err)
+		}
+
+		if !hasIPv4 && !hasIPv6 {
+			continue
+		}
+
+		if err := d.queryIface(iface, !hasIPv4, !hasIPv6); err != nil {
+			return errors.WithStack(err)
+		}
+
+		pollCtx, cancel := context.WithCancel(ctx)
+		defer cancel()
+
+		wg.Add(1)
+
+		go func(ctx context.Context, iface net.Interface) {
+			defer wg.Done()
+
+			if err := d.pollInterface(ctx, iface, interval, !hasIPv4, !hasIPv6); err != nil {
+				if errors.Is(err, context.DeadlineExceeded) || errors.Is(err, context.Canceled) {
+					return
+				}
+
+				logger.Error(
+					ctx, "could not poll interface",
+					logger.E(errors.WithStack(err)), logger.F("iface", iface.Name),
+				)
+			}
+		}(pollCtx, iface)
+	}
+
+	wg.Wait()
+
+	return nil
+}
+
+func (d *Service) queryIface(iface net.Interface, disableIPv4, disableIPv6 bool) error {
+	err := mdns.Query(&mdns.QueryParam{
+		Service:     "_googlecast._tcp",
+		Domain:      "local",
+		Timeout:     3 * time.Second,
+		Entries:     d.entriesCh,
+		Interface:   &iface,
+		DisableIPv6: disableIPv6,
+		DisableIPv4: disableIPv4,
+	})
+	if err != nil {
+		return errors.WithStack(err)
+	}
+
+	return nil
+}
+
+func (d *Service) pollInterface(ctx context.Context, iface net.Interface, interval time.Duration, disableIPv4, disableIPv6 bool) error {
+	ticker := time.NewTicker(interval)
+	for {
+		select {
+		case <-ticker.C:
+			if err := d.queryIface(iface, disableIPv4, disableIPv6); err != nil {
+				return errors.WithStack(err)
+			}
+
+		case <-ctx.Done():
+			if err := ctx.Err(); err != nil {
+				return errors.WithStack(err)
+			}
+
+			return nil
+		}
+	}
+}
+
+func (d *Service) Stop() {
+	if d.stopPeriodic != nil {
+		close(d.stopPeriodic)
+		d.stopPeriodic = nil
+	}
+}
+
+func (d *Service) Found() chan *cast.Client {
+	return d.found
+}
+
+func (d *Service) listener(ctx context.Context) {
+	for entry := range d.entriesCh {
+		name := strings.Split(entry.Name, "._googlecast")
+		// Skip everything that doesn't have googlecast in the fdqn
+		if len(name) < 2 {
+			continue
+		}
+
+		log.Printf("New entry: %#v\n", entry)
+		client := cast.NewClient(entry.AddrV4, entry.Port)
+		info := decodeTxtRecord(entry.Info)
+		client.SetName(info["fn"])
+		client.SetInfo(info)
+
+		select {
+		case d.found <- client:
+		case <-time.After(time.Second):
+		case <-ctx.Done():
+			break
+		}
+	}
+}
+
+func decodeDnsEntry(text string) string {
+	text = strings.Replace(text, `\.`, ".", -1)
+	text = strings.Replace(text, `\ `, " ", -1)
+
+	re := regexp.MustCompile(`([\\][0-9][0-9][0-9])`)
+	text = re.ReplaceAllStringFunc(text, func(source string) string {
+		i, err := strconv.Atoi(source[1:])
+		if err != nil {
+			return ""
+		}
+
+		return string([]byte{byte(i)})
+	})
+
+	return text
+}
+
+func decodeTxtRecord(txt string) map[string]string {
+	m := make(map[string]string)
+
+	s := strings.Split(txt, "|")
+	for _, v := range s {
+		s := strings.Split(v, "=")
+		if len(s) == 2 {
+			m[s[0]] = s[1]
+		}
+	}
+
+	return m
+}
+
+func isIPv4(ip net.IP) bool {
+	return strings.Count(ip.String(), ":") < 2
+}
+
+func isIPv6(ip net.IP) bool {
+	return strings.Count(ip.String(), ":") >= 2
+}
+
+func findMulticastInterfaces(ctx context.Context) ([]net.Interface, error) {
+	ifaces, err := net.Interfaces()
+	if err != nil {
+		return nil, nil
+	}
+
+	multicastIfaces := make([]net.Interface, 0)
+
+	for _, iface := range ifaces {
+		if iface.Flags&net.FlagLoopback == net.FlagLoopback {
+			continue
+		}
+
+		if iface.Flags&net.FlagRunning != net.FlagRunning {
+			continue
+		}
+
+		if iface.Flags&net.FlagMulticast != net.FlagMulticast {
+			continue
+		}
+
+		multicastIfaces = append(multicastIfaces, iface)
+	}
+
+	return multicastIfaces, nil
+}
+
+func retrieveSupportedProtocols(iface net.Interface) (bool, bool, error) {
+	adresses, err := iface.Addrs()
+	if err != nil {
+		return false, false, errors.WithStack(err)
+	}
+
+	hasIPv4 := false
+	hasIPv6 := false
+
+	for _, addr := range adresses {
+		ip, _, err := net.ParseCIDR(addr.String())
+		if err != nil {
+			return false, false, errors.WithStack(err)
+		}
+
+		if isIPv4(ip) {
+			hasIPv4 = true
+		}
+
+		if isIPv6(ip) {
+			hasIPv6 = true
+		}
+
+		if hasIPv4 && hasIPv6 {
+			return hasIPv4, hasIPv6, nil
+		}
+	}
+
+	return hasIPv4, hasIPv6, nil
+}

pkg/module/cast/module.go

@@ -2,7 +2,6 @@ package cast
 
 import (
 	"context"
-	"sync"
 	"time"
 
 	"forge.cadoles.com/arcad/edge/pkg/app"
@@ -19,14 +18,6 @@ const (
 type Module struct {
 	ctx    context.Context
 	server *app.Server
-	mutex  struct {
-		devices        sync.RWMutex
-		refreshDevices sync.Mutex
-		loadURL        sync.Mutex
-		quitApp        sync.Mutex
-		getStatus      sync.Mutex
-	}
-	devices []*Device
 }
 
 func (m *Module) Name() string {
@@ -66,14 +57,11 @@ func (m *Module) refreshDevices(call goja.FunctionCall, rt *goja.Runtime) goja.V
 	promise := m.server.NewPromise()
 
 	go func() {
-		m.mutex.refreshDevices.Lock()
-		defer m.mutex.refreshDevices.Unlock()
-
 		ctx, cancel := context.WithTimeout(context.Background(), timeout)
 		defer cancel()
 
-		devices, err := FindDevices(ctx)
-		if err != nil && !errors.Is(err, context.DeadlineExceeded) {
+		devices, err := ListDevices(ctx, true)
+		if err != nil {
 			err = errors.WithStack(err)
 			logger.Error(ctx, "error refreshing casting devices list", logger.E(errors.WithStack(err)))
 
@@ -82,24 +70,19 @@ func (m *Module) refreshDevices(call goja.FunctionCall, rt *goja.Runtime) goja.V
 			return
 		}
 
-		if err == nil {
-			m.mutex.devices.Lock()
-			m.devices = devices
-			m.mutex.devices.Unlock()
-		}
-
-		devicesCopy := m.getDevicesCopy(devices)
-		promise.Resolve(devicesCopy)
+		promise.Resolve(devices)
 	}()
 
 	return rt.ToValue(promise)
 }
 
 func (m *Module) getDevices(call goja.FunctionCall, rt *goja.Runtime) goja.Value {
-	m.mutex.devices.RLock()
-	defer m.mutex.devices.RUnlock()
+	ctx := context.Background()
 
-	devices := m.getDevicesCopy(m.devices)
+	devices, err := ListDevices(ctx, false)
+	if err != nil {
+		panic(rt.ToValue(errors.WithStack(err)))
+	}
 
 	return rt.ToValue(devices)
 }
@@ -122,9 +105,6 @@ func (m *Module) loadUrl(call goja.FunctionCall, rt *goja.Runtime) goja.Value {
 	promise := m.server.NewPromise()
 
 	go func() {
-		m.mutex.loadURL.Lock()
-		defer m.mutex.loadURL.Unlock()
-
 		ctx, cancel := context.WithTimeout(context.Background(), timeout)
 		defer cancel()
 
@@ -160,9 +140,6 @@ func (m *Module) stopCast(call goja.FunctionCall, rt *goja.Runtime) goja.Value {
 	promise := m.server.NewPromise()
 
 	go func() {
-		m.mutex.quitApp.Lock()
-		defer m.mutex.quitApp.Unlock()
-
 		ctx, cancel := context.WithTimeout(context.Background(), timeout)
 		defer cancel()
 
@@ -198,9 +175,6 @@ func (m *Module) getStatus(call goja.FunctionCall, rt *goja.Runtime) goja.Value
 	promise := m.server.NewPromise()
 
 	go func() {
-		m.mutex.getStatus.Lock()
-		defer m.mutex.getStatus.Unlock()
-
 		ctx, cancel := context.WithTimeout(context.Background(), timeout)
 		defer cancel()
 
@@ -220,21 +194,6 @@ func (m *Module) getStatus(call goja.FunctionCall, rt *goja.Runtime) goja.Value
 	return m.server.ToValue(promise)
 }
 
-func (m *Module) getDevicesCopy(devices []*Device) []Device {
-	devicesCopy := make([]Device, 0, len(m.devices))
-
-	for _, d := range devices {
-		devicesCopy = append(devicesCopy, Device{
-			UUID: d.UUID,
-			Name: d.Name,
-			Host: d.Host,
-			Port: d.Port,
-		})
-	}
-
-	return devicesCopy
-}
-
 func (m *Module) parseTimeout(rawTimeout string) (time.Duration, error) {
 	var (
 		timeout time.Duration
@@ -256,8 +215,7 @@ func (m *Module) parseTimeout(rawTimeout string) (time.Duration, error) {
 func CastModuleFactory() app.ServerModuleFactory {
 	return func(server *app.Server) app.ServerModule {
 		return &Module{
-			server:  server,
-			devices: make([]*Device, 0),
+			server: server,
 		}
 	}
 }