new-server/internal/drivers/lifx/client.go

package lifx

import (
	"context"
	"encoding/binary"
	"errors"
	"git.aiterp.net/lucifer/new-server/internal/lerrors"
	"log"
	"math/rand"
	"net"
	"sync"
	"sync/atomic"
	"time"
)

type Client struct {
	mu    sync.Mutex
	seq   uint8
	buf   []byte
	lastN int

	conn   *net.UDPConn
	source uint32
	debug  bool

	isHorrible uint32

	addrMap map[string]*net.UDPAddr
}

// Send sends the payload to the hardware addr specified. If it cannot find a mapped
// IP address from a past message, it will be broadcast and may be slow to use. If the
// address is left blank, the package is marked as tagged and broadcast.
//
// ErrInvalidAddress can be returned by this, and indicates a badly configured device.
// It should probably be logged.
func (c *Client) Send(addr string, payload Payload) (seq uint8, err error) {
	c.mu.Lock()
	seq = c.seq
	c.seq += 1
	c.mu.Unlock()

	packet := createPacket()
	packet.SetSource(c.source)
	packet.SetSequence(seq)

	sendAddr := &net.UDPAddr{IP: net.IPv4(255, 255, 255, 255), Port: 56700}
	if addr == "" {
		packet.SetTagged(true)

		if c.debug {
			log.Println("Broadcasting", payload, "seq", seq)
		}
	} else {
		err = packet.SetTarget(addr)
		if err != nil {
			return
		}

		c.mu.Lock()
		if udpAddr, ok := c.addrMap[addr]; ok {
			sendAddr = udpAddr
		}
		c.mu.Unlock()

		if c.debug {
			log.Println("Sending", payload, "to", addr, "seq", seq)
		}
	}

	packet.SetPayload(payload)

	_, err = c.conn.WriteToUDP(packet, sendAddr)
	if err != nil {
		return
	}

	return
}

// HorribleBroadcast "broadcasts" by blasting every IP address in the space with a tagged packet, because LIFX is LIFX.
func (c *Client) HorribleBroadcast(ctx context.Context, payload Payload) (seq uint8, err error) {
	c.mu.Lock()
	seq = c.seq
	c.seq += 1
	c.mu.Unlock()

	defer atomic.StoreUint32(&c.isHorrible, 0)

	packet := createPacket()
	packet.SetSource(c.source)
	packet.SetSequence(seq)
	packet.SetPayload(payload)
	packet.SetTagged(true)

	ifaces, err := net.Interfaces()
	if err != nil {
		return 0, err
	}

	connIP := c.conn.LocalAddr().(*net.UDPAddr).IP

	var minIPNum uint32
	var maxIPNum uint32
	for _, iface := range ifaces {
		addrs, err := iface.Addrs()
		if err != nil {
			continue
		}

		for _, addr := range addrs {
			naddr, ok := addr.(*net.IPNet)
			if !ok || !naddr.Contains(connIP) {
				continue
			}

			networkAddress := naddr.IP.Mask(naddr.Mask)
			cidr, bits := naddr.Mask.Size()
			minIPNum = binary.BigEndian.Uint32(networkAddress) + 1
			maxIPNum = minIPNum + (1 << (bits - cidr)) - 2

			break
		}
	}
	if minIPNum == maxIPNum {
		return
	}
	myIPNum := binary.BigEndian.Uint32(connIP)

	if c.debug {
		ipBuf := make(net.IP, 8)
		binary.BigEndian.PutUint32(ipBuf, minIPNum)
		binary.BigEndian.PutUint32(ipBuf[4:], maxIPNum-1)

		log.Println("Sending tagged", payload, "to range", ipBuf[:4], "-", ipBuf[4:])

		defer func() {
			log.Println("Completed sending tagged", payload, "to range", ipBuf[:4], "-", ipBuf[4:])
		}()
	}

	for ipNum := minIPNum; ipNum < maxIPNum; ipNum++ {
		if ipNum == myIPNum {
			continue
		}

		atomic.StoreUint32(&c.isHorrible, 1)

		ipBuf := make(net.IP, 4)
		binary.BigEndian.PutUint32(ipBuf, ipNum)
		addr := &net.UDPAddr{IP: ipBuf, Port: 56700}

		_, err = c.conn.WriteToUDP(packet, addr)
		if err != nil {
			if c.debug {
				log.Println("HorribleBroadcast™ aborted by error.")
			}

			return
		}

		if ipNum%256 == 0 {
			if c.debug {
				log.Println("Progress:", (ipNum-minIPNum)+1, "of", maxIPNum-minIPNum)
			}

			time.Sleep(time.Millisecond * 100)
		}

		if ctx.Err() != nil {
			if c.debug {
				log.Println("HorribleBroadcast™ aborted at call site.")
			}

			return
		}
	}

	return
}

// LastPacket gets the last read packet. This data is valid until the next call to Recv. The Packet may
// not be valid, however!
func (c *Client) LastPacket() Packet {
	if c.lastN < 36 {
		return nil
	}

	return c.buf[:c.lastN]
}

// Recv reads a message from the UDP socket. The data returned is decoded and will always be valid.
//
// However, these should be handled specifically:
// - ErrPayloadTooShort, ErrInvalidPacketSize: Garbage was received, please ignore.
// - ErrUnrecognizedPacketType: Log these and see what's up
// - ErrReadTimeout: The connection gets a 50ms read deadline. It should be used to do other things than wait
func (c *Client) Recv(timeout time.Duration) (target string, seq uint8, payload Payload, err error) {
	if c.buf == nil {
		c.buf = make([]byte, 2048)
	}

	if timeout > 0 {
		err = c.conn.SetReadDeadline(time.Now().Add(timeout))
	} else {
		err = c.conn.SetReadDeadline(time.Time{})
	}
	if err != nil {
		return
	}

	n, addr, err := c.conn.ReadFromUDP(c.buf)
	if n > 0 {
		c.lastN = n
	}
	if err != nil {
		if netErr, ok := err.(*net.OpError); ok && netErr.Timeout() {
			err = lerrors.ErrReadTimeout
			return
		}

		return
	}

	packet := Packet(c.buf[:n])
	if n < 2 || packet.Size() != n && packet.Protocol() != 1024 {
		err = lerrors.ErrInvalidAddress
		return
	}

	seq = packet.Sequence()
	target = packet.Target().String()

	payload, err = packet.Payload()
	if err != nil {
		return
	}

	// Learn the IP address from state service or ack messages.
	if service, ok := payload.(*StateService); ok && service.Service == 1 {
		c.mu.Lock()
		if c.addrMap == nil {
			c.addrMap = make(map[string]*net.UDPAddr)
		}
		c.addrMap[packet.Target().String()] = addr
		c.mu.Unlock()
	}

	if c.debug {
		log.Println("Received", payload, "from", target, "seq", seq)
	}

	return
}

// createClient creates a client that will last as long as the context.
func createClient(ctx context.Context, bindAddr string, debug bool) (*Client, error) {
	addr := net.ParseIP(bindAddr)
	if addr == nil {
		return nil, errors.New("invalid addr")
	}

	source := uint32(rand.Uint64())
	if source < 2 {
		source = 2
	}

	conn, err := net.ListenUDP("udp", &net.UDPAddr{
		IP:   addr,
		Port: 0,
		Zone: "",
	})
	if err != nil {
		return nil, err
	}

	_ = conn.SetWriteBuffer(2048 * 1024)

	go func() {
		<-ctx.Done()
		_ = conn.Close()
	}()

	return &Client{conn: conn, source: source, debug: debug}, nil
}