Files
OpsLog/internal/integrations/udp/server.go
T
rouggy 47992b5f03 fix(decodes): a decode's mode is a marker, not a mode name
Every station on an already-worked band was flagged NEW MODE.

A WSJT-X Decode does not carry the mode's name. It carries the
one-character marker from the decode line - "~" for FT8, "+" for FT4 - and
that character was passed straight through as though it were a mode. The
status resolver then compared "~" against the modes worked for the entity,
matched nothing, and concluded the mode had never been worked. Same cause
put "~ -07" in the comment of every decode spot pushed to the FlexRadio
panadapter, which nobody had traced back.

Resolved through a marker table, with the mode from the sender's last
Status as the fallback - Status is the message that carries the real name.
So an unlisted or future marker degrades to correct rather than to
nonsense, and a sender that puts the name in the field directly is believed
as-is. With neither available the mode is left empty, which makes the
resolver answer "worked": the safe side, since a wrong mode invents a
new-mode flag exactly as the marker did.
2026-08-18 11:40:13 +02:00

797 lines
28 KiB
Go

package udp
import (
"context"
"encoding/binary"
"fmt"
"net"
"regexp"
"strconv"
"strings"
"sync"
"syscall"
"time"
"golang.org/x/net/ipv4"
"hamlog/internal/applog"
)
// remoteTuneHz turns a <FREQ> value from a remote-call packet into Hz.
//
// The field has NO agreed unit, and the same sender uses two of them. DXHunter
// documents "<FREQ>10.136" — MHz — but a log from a working station showed it
// echoing "<FREQ>2107400" straight back: the frequency OpsLog had just
// published to it in the N1MM RadioInfo broadcast, whose <Freq> is in units of
// 10 Hz. Read as MHz, that asked the rig for 2 107 400 MHz, and every tune
// request failed with "out of the 11-digit CAT range" from the first second
// after launch.
//
// So the unit is inferred: try each one and keep the first that lands on an
// amateur band. Anything a station is asked to tune to is, by definition, in
// one. Hz is tried before 10 Hz because the one overlap between them — a 40 m
// frequency in Hz reads as a 4 m one in tens of Hz — is far more likely to be
// 40 m. Nothing plausible means nothing is tuned: a wrong band is worse than a
// request that visibly did nothing.
func remoteTuneHz(s string) int64 {
v, err := strconv.ParseFloat(s, 64)
if err != nil || v <= 0 {
return 0
}
for _, hz := range []int64{int64(v * 1e6), int64(v * 1e3), int64(v), int64(v * 10)} {
if bandFromHz(hz) != "" {
return hz
}
}
// Refusing in silence is how the previous version's failure looked from the
// outside: a spot clicked in another program, and nothing happening here.
applog.Printf("udp: remote_call <FREQ>%s is not a frequency in any amateur band in MHz, kHz or Hz — not tuning\n", s)
return 0
}
// remoteFreqRe / remoteModeRe pull the optional tune request out of a
// ServiceRemoteCall packet: "<FREQ>10.136" (MHz) and "<MODE>FT8". Both accept
// an optional closing tag for proper-XML senders.
var (
remoteFreqRe = regexp.MustCompile(`(?i)<FREQ>\s*([0-9]+(?:\.[0-9]+)?)`)
remoteModeRe = regexp.MustCompile(`(?i)<MODE>\s*([A-Z0-9-]+)`)
)
// reusingListenConfig builds a net.ListenConfig that sets SO_REUSEADDR
// (and SO_REUSEPORT on Unix) on the underlying socket before bind. This
// is the only way for two processes to share a UDP port on Windows — Go
// doesn't expose the option directly, but ListenConfig.Control hooks the
// raw socket and lets us call setsockopt.
func reusingListenConfig() net.ListenConfig {
return net.ListenConfig{
Control: func(network, address string, c syscall.RawConn) error {
var opErr error
err := c.Control(func(fd uintptr) {
opErr = setSocketReuse(fd)
})
if err != nil {
return err
}
return opErr
},
}
}
// decodeTime turns WSJT-X's milliseconds-since-midnight into a UTC instant.
//
// The sender gives a time of DAY with no date, so the date comes from our own
// clock — and the two can straddle midnight: a decode stamped 23:59:58 that
// reaches us at 00:00:01 would otherwise be dated a day late and sort to the top
// of the list for the rest of the session. More than half a day apart is read as
// the wrong side of midnight and moved.
func decodeTime(msSinceMidnight uint32) time.Time {
now := time.Now().UTC()
midnight := time.Date(now.Year(), now.Month(), now.Day(), 0, 0, 0, 0, time.UTC)
at := midnight.Add(time.Duration(msSinceMidnight) * time.Millisecond)
switch {
case at.Sub(now) > 12*time.Hour:
at = at.AddDate(0, 0, -1) // stamped late yesterday, arrived after midnight
case now.Sub(at) > 12*time.Hour:
at = at.AddDate(0, 0, 1) // stamped just after midnight, our clock still on the old day
}
return at
}
// Event is what a Server emits to its consumer for every parsed packet.
// At most one of the fields is populated per event.
type Event struct {
ConfigID int64
Service ServiceType
Source string // remote addr that sent the packet, for diagnostics
DXCall string // ServiceWSJT (Status) or ServiceRemoteCall
DXGrid string // ServiceWSJT (Status)
Mode string // ServiceWSJT (Status/Decode)
FreqHz int64 // ServiceWSJT (Status)
LoggedADIF string // ServiceWSJT (LoggedADIF), ServiceADIF or ServiceN1MM
// A WSJT-X Decode (heard station) to render on the panadapter.
DecodeCall string // transmitting (DE) callsign
DecodeGrid string // 4-char grid, CQ decodes only
DecodeFreqHz int64 // RF frequency (dial + audio offset)
DecodeSNR int // reported SNR (dB)
DecodeCQ bool // the decode was a CQ
DecodeMsg string // the decoded line as printed ("CQ K1ABC FN42")
// DecodeAt is the decode's own UTC timestamp, rebuilt from the sender's
// milliseconds-since-midnight. It is what groups decodes into T/R periods:
// a period's worth arrives in one burst, so arrival time would put them all
// in whichever slot the burst happened to land in.
DecodeAt time.Time
// DecodeTRPeriod is the transmit/receive period in seconds, from the last
// Status of the same program (15 = FT8). 0 when the sender never said.
DecodeTRPeriod int
DecodeDial int64 // dial frequency the decode was heard on, for the band
DecodeOffAir bool // decoded from a file rather than off the air
// The three fields below are shown in the panel AND replayed verbatim when
// answering the station — WSJT-X matches a Reply against its own decode list,
// so every one has to go back exactly as it came.
DecodeDT float64 // seconds into the slot the transmission started
DecodeAudioHz int64 // audio offset inside the passband
DecodeMs uint32 // the decode's raw ms-since-midnight, as sent
DecodeLowConf bool
// ProgramID is the sending application's own id ("WSJT-X", "MSHV", or
// "WSJT-X - 2" for a second instance started with --rig-name). It is what
// tells two receivers apart on one multicast group — and it is the address a
// Reply message would have to be sent back to, so it is carried even though
// nothing replies yet.
ProgramID string
// TxMessage is what the operator's digital app is sending, with Transmitting
// true while the carrier is actually up. From Status, so ~1 Hz.
TxMessage string
Transmitting bool
DECall string // the operator's own call, as the digital app knows it
// ClearCall is set when a WSJT/JTDX/MSHV Status message reports an EMPTY DX
// Call after previously reporting one — i.e. the operator cleared the call in
// the digital app. OpsLog clears its entry to match.
ClearCall bool
// TuneFreqHz / TuneMode carry an explicit "tune the radio" request embedded
// in a ServiceRemoteCall packet (DXHunter spot click sends
// "<CALLSIGN>VP5G<FREQ>10.136<MODE>FT8"). Zero/empty = no tune requested —
// the packet only fills the entry callsign, exactly as before.
TuneFreqHz int64
TuneMode string
}
// Server is a single inbound UDP listener.
type Server struct {
cfg Config
conn *net.UDPConn
out chan<- Event
stop chan struct{}
done chan struct{}
stopped bool
mu sync.Mutex
// WSJT: dial frequency from the last Status, added to Decode audio offsets —
// keyed by the WSJT-X instance id, NOT one value per listener. Two instances
// share one multicast group, so a single value was overwritten by whichever
// sent Status last: decodes from the 50.313 receiver were placed on the
// 50.400 panadapter. WSJT-X requires --rig-name for a second instance, so the
// id is distinct whenever there is more than one.
dialHz map[string]int64
// trPeriod is the T/R period (seconds) from each program's last Status —
// what tells a decode which slot it belongs to.
trPeriod map[string]int
// lastFrom is the address each program's packets arrive from — where a Reply
// has to be sent. See SendReply.
lastFrom map[string]*net.UDPAddr
// lastMode is the mode NAME from each program's last Status, used to resolve
// a Decode's one-character mode marker.
lastMode map[string]string
lastDX string // WSJT: last non-empty DX Call seen, to detect a clear
// badPkts counts datagrams this listener could not parse, so the diagnostic
// dump below stays bounded. A misconfigured port is not a one-off: the
// sender that produced "bad magic 0x3132372e" put out ~150 packets a second,
// which fills the whole rotating log with the same line and buries the
// evidence of anything else.
badPkts int
}
// maxBadPktDumps is how many unparseable datagrams a listener describes in full
// before going quiet. Enough to identify the sender and the payload; few enough
// that a permanently misconfigured port costs a handful of lines, not a log.
const maxBadPktDumps = 5
// describePacket renders a datagram for the log: its size, a printable preview
// and the first bytes in hex.
//
// Both forms, deliberately. "bad magic 0x3132372e" is already readable as ASCII
// "127." to someone who thinks to decode it — and that one fact (the sender is
// emitting text, not WSJT-X binary) is the whole diagnosis. The hex stays for
// the case where the payload really is binary and the preview shows nothing.
func describePacket(pkt []byte) string {
const maxShown = 96
head := pkt
if len(head) > maxShown {
head = head[:maxShown]
}
var text, hex strings.Builder
for _, b := range head {
if b >= 0x20 && b < 0x7f {
text.WriteByte(b)
} else {
text.WriteByte('.')
}
fmt.Fprintf(&hex, "%02x ", b)
}
more := ""
if len(pkt) > maxShown {
more = "…"
}
return fmt.Sprintf("%d bytes | text %q%s | hex %s%s", len(pkt), text.String(), more, strings.TrimSpace(hex.String()), more)
}
func newServer(cfg Config, out chan<- Event) *Server {
return &Server{
cfg: cfg,
out: out,
stop: make(chan struct{}),
done: make(chan struct{}),
}
}
func (s *Server) start() error {
var conn *net.UDPConn
if s.cfg.Multicast {
group := strings.TrimSpace(s.cfg.MulticastGroup)
if group == "" {
return fmt.Errorf("multicast enabled but group address is empty")
}
groupIP := net.ParseIP(group)
if groupIP == nil {
return fmt.Errorf("bad multicast group %q", group)
}
gaddr := &net.UDPAddr{IP: groupIP, Port: s.cfg.Port}
// Bind to INADDR_ANY:port so the kernel will forward packets
// addressed to the multicast group from any interface. Then
// JoinGroup() on every up & multicast-capable interface — Windows
// won't route multicast through interfaces we haven't explicitly
// joined, and the "default" interface picked by
// net.ListenMulticastUDP isn't always the one MSHV/WSJT sends on.
// ListenConfig with SO_REUSEADDR lets us share the port with
// Log4OM / other listeners already bound to 2237.
lc := reusingListenConfig()
pc, err := lc.ListenPacket(context.Background(), "udp4", fmt.Sprintf("0.0.0.0:%d", s.cfg.Port))
if err != nil {
return fmt.Errorf("listen :%d for multicast: %w", s.cfg.Port, err)
}
c, ok := pc.(*net.UDPConn)
if !ok {
_ = pc.Close()
return fmt.Errorf("internal: ListenPacket returned %T not *net.UDPConn", pc)
}
p := ipv4.NewPacketConn(c)
ifaces, _ := net.Interfaces()
joined := 0
for _, ifi := range ifaces {
if ifi.Flags&net.FlagUp == 0 || ifi.Flags&net.FlagMulticast == 0 {
continue
}
if err := p.JoinGroup(&ifi, gaddr); err != nil {
applog.Printf("udp: [%s] join %s on %s: %v\n", s.cfg.Name, gaddr.IP, ifi.Name, err)
continue
}
joined++
}
if joined == 0 {
_ = c.Close()
return fmt.Errorf("couldn't join multicast %s on any interface", gaddr.IP)
}
conn = c
applog.Printf("udp: [%s] listening on multicast %s on %d interface(s) (service=%s)\n",
s.cfg.Name, gaddr, joined, s.cfg.ServiceType)
} else {
lc := reusingListenConfig()
pc, err := lc.ListenPacket(context.Background(), "udp4", fmt.Sprintf("0.0.0.0:%d", s.cfg.Port))
if err != nil {
return fmt.Errorf("listen udp :%d: %w", s.cfg.Port, err)
}
c, ok := pc.(*net.UDPConn)
if !ok {
_ = pc.Close()
return fmt.Errorf("internal: ListenPacket returned %T not *net.UDPConn", pc)
}
conn = c
applog.Printf("udp: [%s] listening on unicast :%d (service=%s)\n", s.cfg.Name, s.cfg.Port, s.cfg.ServiceType)
}
s.conn = conn
go s.run()
return nil
}
func (s *Server) run() {
defer close(s.done)
buf := make([]byte, 64*1024)
for {
select {
case <-s.stop:
return
default:
}
_ = s.conn.SetReadDeadline(time.Now().Add(500 * time.Millisecond))
n, remote, err := s.conn.ReadFromUDP(buf)
if err != nil {
if ne, ok := err.(net.Error); ok && ne.Timeout() {
continue
}
// Closed by stop(): exit silently.
return
}
if n == 0 {
continue
}
pkt := make([]byte, n)
copy(pkt, buf[:n])
go s.handle(pkt, remote)
}
}
// logBadPacket reports a datagram this listener could not parse, with enough of
// it to identify the sender — then falls silent.
//
// The point is the SENDER: an unparseable packet on a WSJT port almost always
// means another program is broadcasting on it, or the service type is wrong for
// what is actually arriving. The remote address names the culprit, and the
// payload preview says what it really is. Neither was logged before, so the
// operator saw only a magic number repeated a few hundred times a second.
func (s *Server) logBadPacket(kind string, remote *net.UDPAddr, pkt []byte, err error) {
s.mu.Lock()
s.badPkts++
n := s.badPkts
s.mu.Unlock()
if n > maxBadPktDumps {
return
}
applog.Printf("udp: [%s] %s parse error from %s: %v — %s\n",
s.cfg.Name, kind, remote, err, describePacket(pkt))
if n == maxBadPktDumps {
applog.Printf("udp: [%s] further unparseable packets on port %d will not be logged — "+
"check that the sender belongs on this port and that the service type matches\n",
s.cfg.Name, s.cfg.Port)
}
}
// noteIgnoredADIF reports a payload the ADIF listener could not use, ONCE per
// cause and then quietly.
//
// This used to log a line per datagram. An operator's log arrived with four
// hundred identical "ADIF payload ignored" lines and nothing else legible: two
// inbound listeners were configured on the SAME multicast group and port
// (239.255.0.1:2237, one WSJT and one ADIF), so every WSJT-X decode packet was
// delivered to both, and the ADIF one rejected each of them in writing. An FT8
// cycle is dozens of decodes every fifteen seconds.
//
// So the WSJT-X magic number is checked here. When it matches, the payload is
// not "chatter" — it is a specific, fixable misconfiguration, and saying which
// is the difference between a log the operator can act on and one they cannot.
func (s *Server) noteIgnoredADIF(pkt []byte) {
s.mu.Lock()
s.badPkts++
n := s.badPkts
s.mu.Unlock()
if n > maxBadPktDumps {
return
}
if len(pkt) >= 4 && binary.BigEndian.Uint32(pkt[:4]) == wsjtMagic {
applog.Printf("udp: [%s] this is WSJT-X traffic, not ADIF — port %d is shared with a WSJT source; "+
"give the ADIF forwarder (JTAlert / GridTracker) its own port, or set this listener's service to WSJT\n",
s.cfg.Name, s.cfg.Port)
// One line is the whole diagnosis: nothing is gained by counting to five.
s.mu.Lock()
s.badPkts = maxBadPktDumps + 1
s.mu.Unlock()
return
}
applog.Printf("udp: [%s] ADIF payload ignored (no <call:>/<eor>) — %s\n", s.cfg.Name, describePacket(pkt))
if n == maxBadPktDumps {
applog.Printf("udp: [%s] further unusable payloads on port %d will not be logged\n",
s.cfg.Name, s.cfg.Port)
}
}
func (s *Server) handle(pkt []byte, remote *net.UDPAddr) {
ev := Event{ConfigID: s.cfg.ID, Service: s.cfg.ServiceType, Source: remote.String()}
switch s.cfg.ServiceType {
case ServiceWSJT:
w, ok, err := ParseWSJT(pkt)
if err != nil {
s.logBadPacket("WSJT", remote, pkt, err)
return
}
if !ok {
return
}
// Where this application's packets come from, so a Reply can be sent back
// to it. Per PROGRAM, not per listener: two receivers share one multicast
// group, and a reply must reach the one that heard the station — and it
// must go to the sender's own address, never to the group.
if w.ProgramID != "" && remote != nil {
s.mu.Lock()
if s.lastFrom == nil {
s.lastFrom = map[string]*net.UDPAddr{}
}
s.lastFrom[w.ProgramID] = remote
s.mu.Unlock()
}
// Status carries the current dial frequency; remember it so Decode audio
// offsets can be turned into RF frequencies for the panadapter.
if w.FreqHz > 0 && !w.IsDecode {
s.mu.Lock()
if s.dialHz == nil {
s.dialHz = map[string]int64{}
}
s.dialHz[w.ProgramID] = w.FreqHz
// The T/R period travels with Status, and a decode has to be told
// which slot it belongs to — so it is remembered per program the
// same way the dial is.
if w.TRPeriod > 0 {
if s.trPeriod == nil {
s.trPeriod = map[string]int{}
}
s.trPeriod[w.ProgramID] = w.TRPeriod
}
// The mode NAME, which only Status carries: a Decode gives the
// one-character marker instead. See DecodeModeName.
if w.Mode != "" {
if s.lastMode == nil {
s.lastMode = map[string]string{}
}
s.lastMode[w.ProgramID] = w.Mode
}
s.mu.Unlock()
}
if !w.IsDecode && (w.TxMessage != "" || w.DECall != "") {
// What the operator is sending, and to whom. Carried on every Status,
// so the consumer sees it change as the QSO progresses. The program id
// travels with it because a second receiver has a transmit state of
// its own.
ev.TxMessage = w.TxMessage
ev.Transmitting = w.Transmitting
ev.DECall = w.DECall
ev.ProgramID = w.ProgramID
}
if w.IsDecode {
s.mu.Lock()
dial := s.dialHz[w.ProgramID]
tr := s.trPeriod[w.ProgramID]
statusMode := s.lastMode[w.ProgramID]
s.mu.Unlock()
if dial <= 0 {
// No Status from THIS instance yet. Guessing with another
// instance's dial is what produced the wrong-panadapter spots,
// so drop the decode and wait — Status arrives every second.
return
}
ev.DecodeCall = w.DecodeCall
ev.DecodeGrid = w.DecodeGrid
ev.DecodeFreqHz = dial + w.DeltaFreqHz
ev.DecodeSNR = w.SNR
ev.DecodeCQ = w.IsCQ
ev.Mode = DecodeModeName(w.Mode, statusMode)
ev.DecodeMsg = w.DecodeMsg
ev.DecodeAt = decodeTime(w.DecodeMsSinceMidnight)
ev.DecodeTRPeriod = tr
ev.DecodeDial = dial
ev.DecodeOffAir = w.OffAir
ev.ProgramID = w.ProgramID
ev.DecodeDT = w.DeltaTime
ev.DecodeAudioHz = w.DeltaFreqHz
ev.DecodeMs = w.DecodeMsSinceMidnight
ev.DecodeLowConf = w.LowConfidence
break
}
// Only a logged QSO is worth a line — WSJT-X/MSHV send a Status packet
// every second, and logging each one buried the rest of the file.
if len(w.LoggedADIF) > 0 {
applog.Printf("udp: [%s] WSJT QSO logged: prog=%q dx_call=%q grid=%q mode=%q freq=%d adif_len=%d\n",
s.cfg.Name, w.ProgramID, w.DXCall, w.DXGrid, w.Mode, w.FreqHz, len(w.LoggedADIF))
}
ev.DXCall = w.DXCall
ev.DXGrid = w.DXGrid
ev.Mode = w.Mode
ev.FreqHz = w.FreqHz
ev.LoggedADIF = w.LoggedADIF
// A Status with an empty DX Call, right after one that had a call, means the
// operator cleared it in WSJT-X / JTDX / MSHV. Fire ONE clear (tracked per
// server) — an idle app sends empty Status every second, and we must not
// re-clear (which would fight a manual entry) on each of those.
s.mu.Lock()
prev := s.lastDX
s.lastDX = w.DXCall
s.mu.Unlock()
if w.DXCall == "" && prev != "" {
ev.ClearCall = true
}
case ServiceADIF:
// JTAlert / GridTracker forward a text ADIF record after a QSO is
// logged. Guard against keep-alive / non-ADIF chatter on the socket:
// only forward payloads that actually carry a callsign field and a
// record terminator.
text := string(pkt)
low := strings.ToLower(text)
if !strings.Contains(low, "<call:") || !strings.Contains(low, "<eor") {
s.noteIgnoredADIF(pkt)
return
}
ev.LoggedADIF = text
case ServiceRemoteCall:
// Common payload shapes seen in the wild:
// "F4XYZ" (bare callsign)
// "CALL F4XYZ" (text prefix)
// "<CALLSIGN>F4XYZ<CALLSIGN>" (DXHunter-style tags)
// "<CALLSIGN>F4XYZ</CALLSIGN>" (proper XML)
// "<CALLSIGN>VP5G<FREQ>10.136<MODE>FT8" (DXHunter with CAT tune)
// Strip every angle-bracket tag, normalise whitespace, take the
// last non-empty token. Upper-case for downstream consistency.
text := string(pkt)
// NEVER act on an N1MM RadioInfo datagram.
//
// It is not a spot click, it is a radio TELLING the world where it is —
// and on a station where an inbound remote-call row and an outbound
// RadioInfo row share a port, the one OpsLog just sent arrives straight
// back on the loopback. The tag-stripping below then reads its last token,
// <ActiveRadioNr>1</ActiveRadioNr>, as the callsign "1", and <Freq> as a
// tune request — for the frequency the rig is already on.
//
// That was harmless only for as long as <Freq> was misread: in tens of Hz
// it looks like a wild number, every tune failed "out of the CAT range",
// and the loop died there. Reading the unit correctly closed it. With
// WSJT-X/JTDX "Fake It", which shifts the dial for each over, every
// transmission then produced a burst of sets echoing between OpsLog and
// itself until the rig stopped answering IF; and the shared CAT link
// dropped. Reported as Fake It causing CAT disconnections.
if low := strings.ToLower(text); strings.Contains(low, "<radioinfo") {
return
}
// Optional tune request: <FREQ>MHz and <MODE>str ride along with the
// callsign so a DXHunter spot click can drive OpsLog's CAT. Extract
// (and cut) them BEFORE the generic tag-stripping below, which would
// otherwise leave their values as stray tokens and corrupt the
// "last token = callsign" heuristic.
if m := remoteFreqRe.FindStringSubmatch(text); m != nil {
ev.TuneFreqHz = remoteTuneHz(m[1])
text = strings.Replace(text, m[0], " ", 1)
}
if m := remoteModeRe.FindStringSubmatch(text); m != nil {
ev.TuneMode = strings.ToUpper(m[1])
text = strings.Replace(text, m[0], " ", 1)
}
// Drop every <...> tag (open or close) — works for both
// <CALLSIGN>...<CALLSIGN> and <CALLSIGN>...</CALLSIGN>.
for {
start := strings.IndexByte(text, '<')
if start < 0 {
break
}
end := strings.IndexByte(text[start:], '>')
if end < 0 {
break
}
text = text[:start] + " " + text[start+end+1:]
}
text = strings.TrimSpace(text)
parts := strings.Fields(text)
if len(parts) == 0 {
return
}
call := strings.ToUpper(parts[len(parts)-1])
// A callsign has a letter in it. Without this, any status XML that ends
// in a number is read as a station — the RadioInfo above was exactly
// that, and refusing it by shape as well as by name means the next
// program to broadcast its state on this port cannot drive the rig
// either.
if !strings.ContainsAny(call, "ABCDEFGHIJKLMNOPQRSTUVWXYZ") {
return
}
ev.DXCall = call
case ServiceN1MM:
adifText, ok, err := ParseN1MM(pkt)
if err != nil {
s.logBadPacket("N1MM", remote, pkt, err)
return
}
if !ok {
applog.Printf("udp: [%s] N1MM datagram ignored (not a loggable contact)\n", s.cfg.Name)
return
}
applog.Printf("udp: [%s] N1MM contact decoded (%d bytes ADIF)\n", s.cfg.Name, len(adifText))
ev.LoggedADIF = adifText
default:
return
}
// Empty events are useless; skip — EXCEPT a clear signal, which is meant to be
// empty (the DX Call was cleared in the digital app), and a tune-only
// request (freq with no callsign).
// TxMessage rides on Status, which also carries the DX call — but a Status
// with an empty DX call and a live transmit message (calling CQ) used to be
// dropped here, and that is exactly the message the decodes panel needs.
if ev.DXCall == "" && ev.LoggedADIF == "" && ev.DecodeCall == "" && ev.TxMessage == "" && !ev.ClearCall && ev.TuneFreqHz == 0 {
return
}
select {
case s.out <- ev:
default:
// Drop on backpressure rather than block the read loop.
}
}
func (s *Server) close() {
s.mu.Lock()
if s.stopped {
s.mu.Unlock()
return
}
s.stopped = true
stop, done, conn := s.stop, s.done, s.conn
s.mu.Unlock()
if conn != nil {
_ = conn.Close()
}
if stop != nil {
close(stop)
}
if done != nil {
<-done
}
}
// ── Outbound emitter ──────────────────────────────────────────────────
// SendUDP sends payload to dst (host:port). Unicast or directed broadcast.
// Returns the error from the write; the connection is closed before return.
func SendUDP(dst string, payload []byte) error {
conn, err := net.Dial("udp4", dst)
if err != nil {
return fmt.Errorf("dial %s: %w", dst, err)
}
defer conn.Close()
_ = conn.SetWriteDeadline(time.Now().Add(2 * time.Second))
_, err = conn.Write(payload)
return err
}
// ── Manager ───────────────────────────────────────────────────────────
// Manager owns every inbound Server and exposes a helper to emit on
// outbound connections at QSO-save time. It reloads from the Repo on
// demand (after a CRUD change in the Settings panel).
type Manager struct {
repo *Repo
out chan Event
// noADIFOnce keeps the "nothing to forward to" note to one line a session
// rather than one per QSO logged.
noADIFOnce sync.Once
// httpFailAt throttles the complaint from a custom URL row, per row: a
// switch that is off answers the same way on every band change.
httpFailMu sync.Mutex
httpFailAt map[int64]time.Time
mu sync.Mutex
inbound map[int64]*Server
outbound []Config
}
func NewManager(repo *Repo) *Manager {
return &Manager{
repo: repo,
out: make(chan Event, 64),
inbound: map[int64]*Server{},
}
}
// Events returns the channel inbound parsed events are delivered on.
// The app exposes these as Wails events.
func (m *Manager) Events() <-chan Event { return m.out }
// Reload restarts every server based on the current Repo contents.
// Existing servers are stopped, the snapshot is rebuilt from scratch.
// Errors on individual rows are logged via the returned slice; the
// caller can surface them in the UI.
func (m *Manager) Reload(ctx context.Context) []string {
applog.Printf("udp: Reload() called")
m.mu.Lock()
old := m.inbound
m.inbound = map[int64]*Server{}
m.outbound = nil
m.mu.Unlock()
for _, s := range old {
s.close()
}
cfgs, err := m.repo.List(ctx)
if err != nil {
applog.Printf("udp: Reload list failed: %v", err)
return []string{fmt.Sprintf("load udp configs: %v", err)}
}
applog.Printf("udp: Reload found %d config(s) in DB", len(cfgs))
var errs []string
for _, c := range cfgs {
applog.Printf("udp: cfg id=%d name=%q dir=%s service=%s port=%d mcast=%v group=%q enabled=%v",
c.ID, c.Name, c.Direction, c.ServiceType, c.Port, c.Multicast, c.MulticastGroup, c.Enabled)
if !c.Enabled {
continue
}
if c.Direction == Outbound {
m.mu.Lock()
m.outbound = append(m.outbound, c)
m.mu.Unlock()
continue
}
srv := newServer(c, m.out)
if err := srv.start(); err != nil {
applog.Printf("udp: start %q failed: %v", c.Name, err)
errs = append(errs, fmt.Sprintf("%s: %v", c.Name, err))
continue
}
m.mu.Lock()
m.inbound[c.ID] = srv
m.mu.Unlock()
}
warnSharedPorts(cfgs)
applog.Printf("udp: Reload done — %d server(s) running, %d error(s)", len(m.inbound), len(errs))
return errs
}
// warnSharedPorts names an inbound and an outbound row sitting on the same
// port, because that is a loop: what OpsLog sends there, OpsLog receives.
//
// It is how a station ended up re-tuning its own rig from its own RadioInfo
// broadcasts. The parser refuses that particular payload now, but the
// arrangement stays wrong for anything else that lands on the port, and it is
// invisible in a settings panel that shows one row at a time.
func warnSharedPorts(cfgs []Config) {
in := map[int]string{}
for _, c := range cfgs {
if c.Enabled && c.Direction != Outbound {
in[c.Port] = c.Name
}
}
for _, c := range cfgs {
if !c.Enabled || c.Direction != Outbound {
continue
}
if name, ok := in[c.Port]; ok {
applog.Printf("udp: %q sends on port %d and %q listens on it — OpsLog will receive its own messages there; give one of the two another port",
c.Name, c.Port, name)
}
}
}
// Outbound returns the active outbound configs matching a service type.
// Used by the QSO save path to push notifications to listeners.
func (m *Manager) Outbound(service ServiceType) []Config {
m.mu.Lock()
defer m.mu.Unlock()
var out []Config
for _, c := range m.outbound {
if c.ServiceType == service {
out = append(out, c)
}
}
return out
}
// StopAll closes every running server. Called at app shutdown.
func (m *Manager) StopAll() {
m.mu.Lock()
old := m.inbound
m.inbound = map[int64]*Server{}
m.outbound = nil
m.mu.Unlock()
for _, s := range old {
s.close()
}
}