dmxconnect/hardware/os2l/OS2LEndpoint.go

package os2l

import (
	"context"
	"dmxconnect/hardware"
	"encoding/json"
	"fmt"
	"net"
	"strings"
	"sync"
	"time"

	"github.com/rs/zerolog/log"
	"github.com/wailsapp/wails/v2/pkg/runtime"
)

// Message represents an OS2L message
type Message struct {
	Event string  `json:"evt"`
	Name  string  `json:"name"`
	State string  `json:"state"`
	ID    int64   `json:"id"`
	Param float64 `json:"param"`
}

// Endpoint contains the data of an OS2L endpoint
type Endpoint struct {
	wg sync.WaitGroup

	info           hardware.EndpointInfo // The basic info for this endpoint
	serverIP       string                // OS2L server IP
	serverPort     int                   // OS2L server port
	listener       net.Listener          // Net listener (TCP)
	listenerCancel context.CancelFunc    // Call this function to cancel the endpoint activation

	eventCallback        func(any)                                                           // This callback is called for returning events
	addDeviceCallback    func(context.Context, hardware.DeviceInfo, hardware.Endpoint) error // Add a device to the hardware
	removeDeviceCallback func(context.Context, string) error                                 // Remove a device from the hardware
}

// NewOS2LEndpoint creates a new OS2L endpoint
func NewOS2LEndpoint(endpointData hardware.EndpointInfo) (*Endpoint, error) {
	endpoint := &Endpoint{
		info:          endpointData,
		listener:      nil,
		eventCallback: nil,
	}
	log.Trace().Str("file", "OS2LEndpoint").Str("name", endpointData.Name).Str("s/n", endpointData.SerialNumber).Msg("OS2L endpoint created")
	return endpoint, endpoint.loadSettings(endpointData.Settings)
}

// SetEventCallback sets the callback for returning events
func (p *Endpoint) SetEventCallback(eventCallback func(any)) {
	p.eventCallback = eventCallback
}

// SetDeviceArrivalCallback is called when we need to add a new device to the hardware
func (p *Endpoint) SetDeviceArrivalCallback(adc func(context.Context, hardware.DeviceInfo, hardware.Endpoint) error) {
	p.addDeviceCallback = adc
}

// SetDeviceRemovalCallback is called when we need to remove a device from the hardware
func (p *Endpoint) SetDeviceRemovalCallback(rdc func(context.Context, string) error) {
	p.removeDeviceCallback = rdc
}

// Connect connects the OS2L endpoint
func (p *Endpoint) Connect(ctx context.Context) error {
	runtime.EventsEmit(ctx, string(hardware.EndpointStatusUpdated), p.GetInfo(), hardware.EndpointStatusConnecting)

	var err error
	addr := net.TCPAddr{Port: p.serverPort, IP: net.ParseIP(p.serverIP)}

	log.Debug().Any("addr", addr).Msg("parametres de connexion à la connexion")
	p.listener, err = net.ListenTCP("tcp", &addr)
	if err != nil {
		runtime.EventsEmit(ctx, string(hardware.EndpointStatusUpdated), p.GetInfo(), hardware.EndpointStatusDisconnected)
		return fmt.Errorf("unable to set the OS2L TCP listener")
	}

	runtime.EventsEmit(ctx, string(hardware.EndpointStatusUpdated), p.GetInfo(), hardware.EndpointStatusDeactivated)
	log.Info().Str("file", "OS2LEndpoint").Msg("OS2L endpoint connected")

	// TODO: To remove : simulate a device arrival/removal
	p.addDeviceCallback(ctx, hardware.DeviceInfo{
		SerialNumber: "0DE3FF",
		Name:         "OS2L test device",
		Manufacturer: "BlueSig",
		Version:      "0.1.0-dev",
	}, p)
	return nil
}

// handleMessage handles an OS2L message
func (p *Endpoint) handleMessage(raw []byte) error {
	message := Message{}
	err := json.Unmarshal(raw, &message)
	if err != nil {
		return fmt.Errorf("Unable to parse the OS2L message: %w", err)
	}
	log.Debug().Str("event", message.Event).Str("name", message.Name).Str("state", message.State).Int("ID", int(message.ID)).Float64("param", message.Param).Msg("OS2L event received")
	// Return the event to the provider
	if p.eventCallback != nil {
		go p.eventCallback(message)
	}
	return nil
}

// loadSettings check and load the settings in the endpoint
func (p *Endpoint) loadSettings(settings map[string]any) error {
	// Check if the IP exists
	serverIP, found := settings["os2lIp"]
	if !found {
		// Set default IP address
		serverIP = "127.0.0.1"
	}
	// Check if it is a string
	ipSetting, ok := serverIP.(string)
	if ok {
		p.serverIP = ipSetting
	} else {
		return fmt.Errorf("The specified IP is not a string")
	}
	// Check if the port exists
	serverPort, found := settings["os2lPort"]
	if !found {
		// Set default port
		serverPort = 9995
	}
	switch v := serverPort.(type) {
	case int:
		p.serverPort = v
	case float64:
		p.serverPort = int(v) // JSON numbers are float64
	default:
		return fmt.Errorf("The specified port is not a number, got %T", serverPort)
	}

	return nil
}

// Disconnect disconnects the MIDI endpoint
func (p *Endpoint) Disconnect(ctx context.Context) error {

	// Close the TCP listener if not null
	if p.listener != nil {
		p.listener.Close()
	}

	runtime.EventsEmit(ctx, string(hardware.EndpointStatusUpdated), p.GetInfo(), hardware.EndpointStatusDisconnected)

	log.Info().Str("file", "OS2LEndpoint").Msg("OS2L endpoint disconnected")
	// TODO: To remove : simulate a device arrival/removal
	p.removeDeviceCallback(ctx, "0DE3FF")
	return nil
}

// Activate activates the OS2L endpoint
func (p *Endpoint) Activate(ctx context.Context) error {
	// Create a derived context to handle deactivation
	var listenerCtx context.Context
	listenerCtx, p.listenerCancel = context.WithCancel(ctx)

	if p.listener == nil {
		return fmt.Errorf("the listener isn't defined")
	}

	p.wg.Add(1)
	go func() {
		defer p.wg.Done()

		for {
			select {
			case <-listenerCtx.Done():
				return
			default:
				p.listener.(*net.TCPListener).SetDeadline(time.Now().Add(1 * time.Second))
				conn, err := p.listener.Accept()
				if err != nil {
					if ne, ok := err.(net.Error); ok && ne.Timeout() {
						continue
					}
					if strings.Contains(err.Error(), "use of closed network connection") || strings.Contains(err.Error(), "invalid argument") {
						return
					}
					log.Err(err).Str("file", "OS2LEndpoint").Msg("unable to accept the connection")
					continue
				}

				// Every client is handled in a dedicated goroutine
				p.wg.Add(1)
				go func(c net.Conn) {
					defer p.wg.Done()
					defer c.Close()

					buffer := make([]byte, 1024)
					for {
						select {
						case <-listenerCtx.Done():
							return
						default:
							c.SetReadDeadline(time.Now().Add(500 * time.Millisecond))
							n, err := c.Read(buffer)
							if err != nil {
								if ne, ok := err.(net.Error); ok && ne.Timeout() {
									// Lecture a expiré → vérifier si le contexte est annulé
									select {
									case <-listenerCtx.Done():
										return
									default:
										continue // pas annulé → relancer Read
									}
								}
								return // autre erreur ou EOF
							}

							p.handleMessage(buffer[:n])
						}
					}
				}(conn)
			}
		}
	}()
	runtime.EventsEmit(ctx, string(hardware.EndpointStatusUpdated), p.GetInfo(), hardware.EndpointStatusActivated)

	log.Info().Str("file", "OS2LEndpoint").Msg("OS2L endpoint activated")

	return nil
}

// Deactivate deactivates the OS2L endpoint
func (p *Endpoint) Deactivate(ctx context.Context) error {
	if p.listener == nil {
		return fmt.Errorf("the listener isn't defined")
	}

	// Cancel listener by context
	if p.listenerCancel != nil {
		p.listenerCancel()
	}

	runtime.EventsEmit(ctx, string(hardware.EndpointStatusUpdated), p.GetInfo(), hardware.EndpointStatusDeactivated)

	log.Info().Str("file", "OS2LEndpoint").Msg("OS2L endpoint deactivated")
	return nil
}

// SetSettings sets a specific setting for this endpoint
func (p *Endpoint) SetSettings(ctx context.Context, settings map[string]any) error {
	err := p.loadSettings(settings)
	if err != nil {
		return fmt.Errorf("unable to load settings: %w", err)
	}
	// Reconnect the endpoint
	p.wg.Add(1)
	go func() {
		defer p.wg.Done()
		err := p.Deactivate(ctx)
		if err != nil {
			log.Err(err).Str("sn", p.GetInfo().SerialNumber).Msg("unable to deactivate")
			return
		}
		err = p.Disconnect(ctx)
		if err != nil {
			log.Err(err).Str("sn", p.GetInfo().SerialNumber).Msg("unable to disconnect")
			return
		}
		// Add a sleep to view changes
		time.Sleep(500 * time.Millisecond)
		err = p.Connect(ctx)
		if err != nil {
			log.Err(err).Str("sn", p.GetInfo().SerialNumber).Msg("unable to connect")
			return
		}
		err = p.Activate(ctx)
		if err != nil {
			log.Err(err).Str("sn", p.GetInfo().SerialNumber).Msg("unable to activate")
			return
		}
	}()
	log.Info().Str("sn", p.GetInfo().SerialNumber).Msg("endpoint settings set")
	return nil
}

// SetDeviceProperty - not implemented for this kind of endpoint
func (p *Endpoint) SetDeviceProperty(context.Context, uint32, byte) error {
	return nil
}

// GetSettings gets the endpoint settings
func (p *Endpoint) GetSettings() map[string]any {
	return map[string]any{
		"os2lIp":   p.serverIP,
		"os2lPort": p.serverPort,
	}
}

// GetInfo gets the endpoint information
func (p *Endpoint) GetInfo() hardware.EndpointInfo {
	return p.info
}

// WaitStop stops the endpoint
func (p *Endpoint) WaitStop() error {
	log.Info().Str("file", "OS2LEndpoint").Str("s/n", p.info.SerialNumber).Msg("waiting for OS2L endpoint to close...")
	p.wg.Wait()
	log.Info().Str("file", "OS2LEndpoint").Str("s/n", p.info.SerialNumber).Msg("OS2L endpoint closed!")
	return nil
}