macker/client/main.go

package main

//NOTE: I use emoji in comments sometimes to keep the functions from
//blurring together. I know there are better dyslexia coping strategies,
//but this is where I'm at right now. For some reason, this apparently
//offends people. Welcome to the internet.

import (
	"encoding/gob"
	"encoding/json"
	"flag"
	"fmt"
	"math"
	"net"
	"os"
	"strings"
	"time"

	"github.com/amenzhinsky/go-gpsd"

	"macker/datapoint"
)

var LOG_LEVEL int

func solidifyData(v gpsd.Report) (datapoint.DataPoint, error) {
	tpv, ok := v.(*gpsd.TPV)
	if !ok {
		return datapoint.DataPoint{}, fmt.Errorf("not a TPV report: %T", v)
	}

	dp := datapoint.DataPoint{
		// Epy is the north-south (latitude) error, Epx the east-west (longitude) one.
		Lat:     tpv.Lat,
		LatErr:  tpv.Epy,
		Long:    tpv.Lon,
		LongErr: tpv.Epx,
		Alt:     tpv.Alt,
		AltErr:  tpv.Epv,
	}

	if tpv.Mode < 2 {
		return datapoint.DataPoint{}, fmt.Errorf("GPS has no position fix")
	}
	if !math.IsNaN(tpv.Speed) && !math.IsInf(tpv.Speed, 0) && tpv.Speed >= 0 {
		speed := tpv.Speed
		dp.Speed = &speed
	}

	// gpsd omits "time" on some reports; only stamp what it actually gave us.
	if !tpv.Time.IsZero() {
		dp.Timestamp = uint64(tpv.Time.UnixMilli())
	}

	return dp, nil
}

// Monitor owns the gpsd connection and pumps position fixes onto PipeOut.
// Use the pointer form: gpsd.Client embeds a sync.Mutex and must not be copied.
type Monitor struct {
	PipeOut chan datapoint.DataPoint
	C       *gpsd.Client
}

// enableWatch requests a JSON stream from gpsd.
//
// NOTE: we deliberately do not use gpsd.Client.Stream here. It sends
// "?WATCH={...}" with no terminating semicolon (see Stream() in the library's
// client.go, which ends with `return c.Send([]byte(s))` after only appending
// "}"). The gpsd JSON protocol requires every command to be terminated by a
// semicolon; the trailing CR/LF is optional but the semicolon is not. gpsd
// 3.27 is lenient and honours the malformed request anyway, which hides the
// bug, but a stricter or non-gpsd server would silently ignore the WATCH and we
// would then block forever waiting on g.C(). Building the command ourselves
// keeps the protocol correct without forking the dependency.
func enableWatch(g *gpsd.Client, device string) error {
	var cmd strings.Builder
	cmd.WriteString(`?WATCH={"enable":true,"json":true`)
	if device != "" {
		fmt.Fprintf(&cmd, `,"device":%q`, device)
	}
	cmd.WriteString("};")

	if err := g.Send([]byte(cmd.String())); err != nil {
		return fmt.Errorf("send WATCH: %w", err)
	}
	return nil
}

func (m *Monitor) StartRun(device string) error {
	g, err := gpsd.Dial(gpsd.DefaultAddress)
	if err != nil {
		return fmt.Errorf("dial gpsd at %s: %w", gpsd.DefaultAddress, err)
	}

	if err := enableWatch(g, device); err != nil {
		g.Close()
		return err
	}

	// Publish the client only once the connection is known good, so a failed
	// StartRun never leaves a half-built Monitor behind.
	m.C = g

	// The connection must stay open for the lifetime of the reader, so Close
	// belongs here and not in a defer on StartRun -- StartRun returns long
	// before this goroutine does.
	go func() {
		defer g.Close()
		for v := range g.C() {
			if LOG_LEVEL > 2 {
				fmt.Println("Startrun goroutine loop")
			}

			dp, err := solidifyData(v)
			if err != nil {
				// VERSION, DEVICES, WATCH and SKY all land here.
				if LOG_LEVEL > 2 {
					fmt.Printf("skipping report: %v\n", err)
				}
				continue
			}

			m.PipeOut <- dp
		}

		// Only reached when gpsd drops the connection; main reports the cause.
		if LOG_LEVEL > 0 {
			fmt.Println("gpsd reader loop ended")
		}
	}()

	return nil
}

func genMonitor() (chan datapoint.DataPoint, *Monitor) {
	if LOG_LEVEL > 0 {
		fmt.Println("Starting genMonitor")
	}

	// Buffered so a momentary stall in the sender cannot back-pressure into
	// gpsd's read loop, which would make gpsd drop reports on the floor.
	retChan := make(chan datapoint.DataPoint, 16)

	retMon := &Monitor{
		PipeOut: retChan,
	}

	if LOG_LEVEL > 0 {
		fmt.Println("Ending genMonitor\n------------------------------------1")
	}

	return retChan, retMon
}

func printDp(input datapoint.DataPoint) {
	printString := fmt.Sprintf("\rTime: %d,Lat: %f Long: %f", input.Timestamp, input.Lat, input.Long)
	fmt.Print(printString)
}

// Retry pacing for sendToServer. The backoff doubles from the starter delay up
// to the cap, and after sendOffMaxRetries failures it wraps back around to the
// starter delay and starts doubling again.
const (
	sendOffStartBackoff = 5 * time.Second
	sendOffMaxBackoff   = 60 * time.Second
	sendOffMaxRetries   = 100
)

// backoffFor returns how long to sleep before retry number retry.
//
// The doubling is capped instead of left unbounded: 5s doubled repeatedly
// overflows time.Duration after roughly 58 steps, and a cycle runs 100 retries.
func backoffFor(retry int) time.Duration {
	if retry >= sendOffMaxRetries {
		retry = 0
	}

	wait := sendOffStartBackoff
	for i := 0; i < retry; i++ {
		wait *= 2
		if wait >= sendOffMaxBackoff {
			return sendOffMaxBackoff
		}
	}
	return wait
}

// FUNCTION TO SEND THE COLLECTED DATA POINTS TO THE SERVER 🖥🖥🖥🖥🖥🖥🖥🖥🖥🖥🖥🖥🖥🖥🖥🖥🖥🖥🖥🖥🖥🖥
//
// The payload is JSON, gob-encoded as an opaque []byte. sendBuffer is left
// untouched on failure so the caller can retry the same points; the caller
// clears them once we return without error.
func sendToServer(targAddr, targPort string, sendBuffer *[]datapoint.DataPoint, errorChan chan error, retry int) {

	if !strings.HasPrefix(targPort, ":") {
		targPort = ":" + targPort
	}

	con, err := net.DialTimeout("tcp", net.JoinHostPort(targAddr, strings.TrimPrefix(targPort, ":")), 10*time.Second)
	if err != nil {
		errorChan <- fmt.Errorf("dial %s%s: %w", targAddr, targPort, err)
		time.Sleep(backoffFor(retry))
		sendToServer(targAddr, targPort, sendBuffer, errorChan, retry+1)
		return
	}

	// Without this we leak a connection on every flush.
	defer con.Close()
	con.SetWriteDeadline(time.Now().Add(10 * time.Second))

	str, err := json.Marshal(sendBuffer)
	if err != nil {
		// Bad data, not a bad connection: retrying will not help.
		errorChan <- fmt.Errorf("marshal batch: %w", err)
		return
	}

	if encErr := gob.NewEncoder(con).Encode(str); encErr != nil {
		errorChan <- fmt.Errorf("send batch to %s%s: %w", targAddr, targPort, encErr)
		time.Sleep(backoffFor(retry))
		sendToServer(targAddr, targPort, sendBuffer, errorChan, retry+1)
	}
}

// sendLoop owns the pending points and hands them to sendToServer.
//
// It is the only goroutine touching the buffer, which is why no mutex is
// needed: main copies each batch in over batches, and this goroutine is the
// sole appender and clearer.
func sendLoop(targAddr, targPort string, batches <-chan []datapoint.DataPoint, interval time.Duration, batchSize int, errorChan chan error) {
	var buf []datapoint.DataPoint

	flush := func() {
		if len(buf) == 0 {
			return
		}

		// Returns only on success or an unmarshalable batch; either way these
		// points have left our hands.
		sendToServer(targAddr, targPort, &buf, errorChan, 0)
		buf = buf[:0]
	}

	ticker := time.NewTicker(interval)
	defer ticker.Stop()

	for {
		select {
		case b := <-batches:
			buf = append(buf, b...)
			if len(buf) >= batchSize {
				flush()
			}

		case <-ticker.C:
			flush()
		}
	}
}

func manageErrors(input chan error) {
	for er := range input {
		// For now, just print
		fmt.Println(er)
	}
}

// HERE'S MAIN ⬇️⬇️⬇️⬇️⬇️⬇️⬇️⬇️⬇️⬇️⬇️⬇️⬇️⬇️⬇️⬇️⬇️⬇️⬇️⬇️⬇️⬇️⬇️⬇️⬇️⬇️⬇️⬇️⬇️⬇️⬇️⬇️⬇️⬇️⬇️⬇️⬇️⬇️⬇️⬇️⬇️⬇️⬇️⬇️⬇️⬇️⬇️⬇️⬇️⬇️
func main() {
	targAddr := flag.String("addr", "localhost", "address of the server to send points to")
	targPort := flag.String("port", "2948", "port of the server to send points to")
	interval := flag.Duration("interval", 5*time.Second, "how often to flush buffered points to the server")
	batchSize := flag.Int("batch", 25, "flush as soon as this many points are buffered")
	device := flag.String("device", "", "gpsd device to watch (empty watches every device)")
	logLevel := flag.Int("log", 0, "log verbosity (0 quiet, 1 lifecycle, 2+ per-report)")
	flag.Parse()

	if *interval <= 0 || *batchSize <= 0 {
		fmt.Fprintln(os.Stderr, "interval and batch must be positive")
		os.Exit(2)
	}
	LOG_LEVEL = *logLevel

	errorChan := make(chan error)
	go manageErrors(errorChan)

	// Buffered so bursts of fixes do not stall the gpsd read loop. A full
	// channel means the sender is behind on a slow or dead server; we drop
	// the batch and say so rather than block gpsd's rx, which would make
	// gpsd discard reports entirely.
	batches := make(chan []datapoint.DataPoint, 64)
	go sendLoop(*targAddr, *targPort, batches, *interval, *batchSize, errorChan)

	monitorOutput, m := genMonitor()

	if err := m.StartRun(*device); err != nil {
		fmt.Println(err)
		os.Exit(1)
	}

	ticker := time.NewTicker(*interval)
	defer ticker.Stop()
	pending := make([]datapoint.DataPoint, 0, *batchSize)

	for {
		select {
		case dp := <-monitorOutput:
			printDp(dp)
			pending = append(pending, dp)

			if len(pending) >= *batchSize {
				select {
				case batches <- pending:
				default:
					fmt.Println("send queue full, dropping batch of", len(pending))
				}
				// Fresh backing array: sendLoop appends into its own buffer and
				// we must not have it alias ours.
				pending = make([]datapoint.DataPoint, 0, *batchSize)
			}

		case <-ticker.C:
			if len(pending) == 0 {
				continue
			}

			select {
			case batches <- pending:
			default:
				fmt.Println("send queue full, dropping batch of", len(pending))
			}
			pending = make([]datapoint.DataPoint, 0, *batchSize)

		case <-m.C.Done():
			// Without this the program would block on a channel that can
			// never receive again, and hang with no explanation.
			fmt.Println("gpsd connection closed:", m.C.Err())
			return
		}
	}
}