antplusbridge/cmd/bridge/ant.go

package main

import (
	"bufio"
	"context"
	"fmt"
	"io"
	"time"

	"github.com/google/gousb"
	"github.com/half2me/antgo/ant"
	"github.com/half2me/antgo/device"
)

type antUSB struct {
	in  *bufio.Reader
	out *gousb.OutEndpoint
}

func (d *antUSB) Read(p []byte) (int, error)  { return d.in.Read(p) }
func (d *antUSB) Write(p []byte) (int, error) { return d.out.Write(p) }

const (
	antDataPageMask         = 0x7F // Mask for data page number (7 bits, clears toggle bit)
	antDataPageCyclingPower = 0x10 // Data page 16 = Cycling Power Feature page
)

func resetAndWait(drv io.ReadWriter) error {
	if _, err := drv.Write(ant.SystemResetMessage()); err != nil {
		return err
	}
	// drain until we see the startup message (0x6F)
	for {
		msg, err := ant.ReadMsg(drv)
		if err != nil {
			return err
		}
		if msg.Class() == ant.MESSAGE_STARTUP {
			return nil
		}
	}
}

func ListenForTrainer(vendorId uint16, productId uint16, deviceNumber uint32,
	events chan<- Event) error {

	// Use gousb directly instead of antgo/driver/usb to manage context
	// lifecycle explicitly. The old code created a new gousb.Context on
	// every call via libusb_init and leaked it on error paths (device not
	// found, interface claim failure, etc.), which triggered a libusb
	// assertion after repeated ListenForTrainer calls.
	gctx := gousb.NewContext()
	defer gctx.Close()

	dev, err := gctx.OpenDeviceWithVIDPID(gousb.ID(vendorId), gousb.ID(productId))
	if err != nil {
		return fmt.Errorf("ant: open device: %w", err)
	}
	if dev == nil {
		return fmt.Errorf("ant: device not found (0x%04x:0x%04x)", vendorId, productId)
	}
	defer dev.Close()

	intf, done, err := dev.DefaultInterface()
	if err != nil {
		return fmt.Errorf("ant: claim interface: %w", err)
	}
	// Cleanup is ordered by defer stack: done() releases interface,
	// dev.Close() closes device, gctx.Close() closes context, cancel()
	// cancels the ANT message goroutine.
	defer done()

	out, err := intf.OutEndpoint(1)
	if err != nil {
		return fmt.Errorf("ant: open out endpoint: %w", err)
	}
	inp, err := intf.InEndpoint(1)
	if err != nil {
		return fmt.Errorf("ant: open in endpoint: %w", err)
	}

	drv := &antUSB{in: bufio.NewReader(inp), out: out}

	logMsg("ant: device opened")

	// Errors are returned (not panicked) so the caller can retry
	// gracefully instead of requiring a full process restart.
	if err := resetAndWait(drv); err != nil {
		return err
	}
	if err := device.StartRxScanModeWait(drv); err != nil {
		return err
	}
	logMsg("ant: scan mode started")

	msgs := make(chan ant.BroadcastMessage, 100)
	ctx, cancel := context.WithCancel(context.Background())
	// cancel is the final defer in the stack, ensuring the ANT message
	// goroutine is torn down after all USB resources are released.
	defer cancel()
	go func() {
		device.DumpBroadcastMessages(ctx, drv, msgs)
		logMsg("ant: dump broadcast messages ended")
	}()

	logMsg("ant: listening...")
	for {
		select {
		case msg := <-msgs:
			if deviceNumber != 0 && msg.DeviceNumber() != deviceNumber {
				logPrintf("ant: received msg from ignored device %v\n", msg.DeviceNumber())
				continue
			}
			//fmt.Printf("raw msg: device type=0x%02x device number=%d content=%x\n",
			//	msg.DeviceType(), msg.DeviceNumber(), msg.Content())

			switch msg.DeviceType() {
			case ant.DEVICE_TYPE_POWER:
				// Check data page first
				page := msg[4] & antDataPageMask
				if page != antDataPageCyclingPower {
					continue
				}
				cur := ant.PowerMessage(msg)
				logPrintf("ant: %v: power: %dW  cadence: %d rpm\n",
					msg.DeviceNumber(), cur.InstantaneousPower(), cur.InstantaneousCadence())
				select {
				case events <- Event{
					Power:   cur.InstantaneousPower(),
					Cadence: cur.InstantaneousCadence(),
					Now:     time.Now(),
				}:
				default:
					logPrintf("ant: dropped event, buffer full\n")
				}
			}
		case <-time.After(10 * time.Second):
			logMsg("ant: no more messages arrived, restarting")
			// Return without error so the caller can retry gracefully
			// rather than panicking and requiring a full restart.
			return nil
		}
	}
}