119 lines
2.9 KiB
Go
119 lines
2.9 KiB
Go
package main
|
|
|
|
import (
|
|
_ "embed"
|
|
"encoding/gob"
|
|
"encoding/json"
|
|
"flag"
|
|
"io"
|
|
"log"
|
|
"math"
|
|
"net"
|
|
"net/http"
|
|
"sync"
|
|
"time"
|
|
|
|
"macker/datapoint"
|
|
)
|
|
|
|
//go:embed index.html
|
|
var page []byte
|
|
|
|
type location struct {
|
|
Point datapoint.DataPoint `json:"point"`
|
|
UpdatedAt time.Time `json:"updated_at"`
|
|
ReceivedAt time.Time `json:"received_at"`
|
|
}
|
|
|
|
type tracker struct {
|
|
mu sync.RWMutex
|
|
latest *location
|
|
}
|
|
|
|
func (t *tracker) accept(points []datapoint.DataPoint, now time.Time) {
|
|
t.mu.Lock()
|
|
defer t.mu.Unlock()
|
|
for _, p := range points {
|
|
if math.IsNaN(p.Lat) || math.IsInf(p.Lat, 0) || math.IsNaN(p.Long) || math.IsInf(p.Long, 0) || p.Lat < -90 || p.Lat > 90 || p.Long < -180 || p.Long > 180 {
|
|
continue
|
|
}
|
|
if p.Speed != nil && (math.IsNaN(*p.Speed) || math.IsInf(*p.Speed, 0) || *p.Speed < 0) {
|
|
p.Speed = nil
|
|
}
|
|
updated := now
|
|
if p.Timestamp != 0 {
|
|
if p.Timestamp > uint64(now.Add(time.Minute).UnixMilli()) {
|
|
continue
|
|
}
|
|
updated = time.UnixMilli(int64(p.Timestamp))
|
|
}
|
|
if t.latest != nil && !updated.After(t.latest.UpdatedAt) {
|
|
continue
|
|
}
|
|
t.latest = &location{Point: p, UpdatedAt: updated, ReceivedAt: now}
|
|
}
|
|
}
|
|
|
|
func (t *tracker) receive(c net.Conn) {
|
|
defer c.Close()
|
|
_ = c.SetReadDeadline(time.Now().Add(10 * time.Second))
|
|
var raw []byte
|
|
if err := gob.NewDecoder(io.LimitReader(c, datapoint.MaxBatchBytes+1024)).Decode(&raw); err != nil {
|
|
log.Printf("decode batch: %v", err)
|
|
return
|
|
}
|
|
points, err := datapoint.UnzipBatch(raw)
|
|
if err != nil {
|
|
log.Printf("decode points: %v", err)
|
|
return
|
|
}
|
|
t.accept(points, time.Now().UTC())
|
|
}
|
|
|
|
func (t *tracker) handler() http.Handler {
|
|
mux := http.NewServeMux()
|
|
mux.HandleFunc("GET /api/location", func(w http.ResponseWriter, r *http.Request) {
|
|
w.Header().Set("Content-Type", "application/json")
|
|
w.Header().Set("Cache-Control", "no-store")
|
|
t.mu.RLock()
|
|
defer t.mu.RUnlock()
|
|
_ = json.NewEncoder(w).Encode(t.latest)
|
|
})
|
|
mux.HandleFunc("GET /{$}", func(w http.ResponseWriter, r *http.Request) {
|
|
w.Header().Set("Content-Type", "text/html; charset=utf-8")
|
|
_, _ = w.Write(page)
|
|
})
|
|
return mux
|
|
}
|
|
|
|
func main() {
|
|
ingest := flag.String("ingest", ":2948", "TCP address for car GPS batches")
|
|
web := flag.String("http", ":8080", "HTTP address for the map")
|
|
flag.Parse()
|
|
ln, err := net.Listen("tcp", *ingest)
|
|
if err != nil {
|
|
log.Fatal(err)
|
|
}
|
|
defer ln.Close()
|
|
t := &tracker{}
|
|
go func() {
|
|
// Bound concurrent decoders and their memory usage.
|
|
slots := make(chan struct{}, 32)
|
|
for {
|
|
c, err := ln.Accept()
|
|
if err != nil {
|
|
return
|
|
}
|
|
select {
|
|
case slots <- struct{}{}:
|
|
go func() { defer func() { <-slots }(); t.receive(c) }()
|
|
default:
|
|
c.Close()
|
|
}
|
|
}
|
|
}()
|
|
log.Printf("GPS receiver on %s; map on %s", *ingest, *web)
|
|
srv := &http.Server{Addr: *web, Handler: t.handler(), ReadHeaderTimeout: 5 * time.Second, WriteTimeout: 15 * time.Second, IdleTimeout: 60 * time.Second}
|
|
log.Fatal(srv.ListenAndServe())
|
|
}
|