package station import ( "context" "crypto/hmac" "crypto/sha1" "encoding/binary" "errors" "net" "strconv" "sync" "time" "gitea.neitzel.de/konrad/funknetz/internal/opus" ) const noOpus = "The reflector did not offer OPUS." var errStopped = errors.New("session stopped") // Run keeps the reflector session until ctx is canceled. func (s *Station) Run(ctx context.Context) error { for { if ctx.Err() != nil { return nil } s.mu.Lock() if s.fatal != "" { s.mu.Unlock() <-ctx.Done() return nil } s.state.Connection = "Connecting" s.publish() host, port := s.cfg.host, s.cfg.port s.mu.Unlock() conn, err := dialStation(ctx, host, port) if err != nil { if ctx.Err() != nil { return nil } s.mu.Lock() s.state.Connection = "Disconnected" s.publish() s.mu.Unlock() if !s.wait(ctx, 2*time.Second) { return nil } continue } err = s.session(ctx, conn) conn.Close() s.mu.Lock() s.closeUDP() if s.fatal == "" { s.state.Connection = "Disconnected" s.remoteTalker = "" s.roomBusy = false } s.mic = nil s.play = nil s.publish() fatal := s.fatal != "" s.mu.Unlock() if ctx.Err() != nil { return nil } if fatal { <-ctx.Done() return nil } if err != nil && errors.Is(err, errStopped) { <-ctx.Done() return nil } if !s.wait(ctx, 2*time.Second) { return nil } } } func dialStation(ctx context.Context, host string, port int) (net.Conn, error) { var d net.Dialer d.Timeout = 5 * time.Second return d.DialContext(ctx, "tcp", net.JoinHostPort(host, strconv.Itoa(port))) } func (s *Station) wait(ctx context.Context, d time.Duration) bool { timer := s.clock.NewTimer(d) defer timer.Stop() select { case <-ctx.Done(): return false case <-timer.C(): return true } } func (s *Station) session(ctx context.Context, conn net.Conn) error { done := make(chan struct{}) var closeOnce sync.Once closeConn := func() { closeOnce.Do(func() { conn.Close() }) } defer close(done) go func() { select { case <-ctx.Done(): closeConn() case <-done: } }() tcpCh := make(chan frame, 8) tcpErr := make(chan error, 1) go readTCP(conn, tcpCh, tcpErr, done) if err := writeFrame(conn, tcpProtoVer, []byte{0, 2, 0, 0}); err != nil { return err } lastTCP := s.clock.Now() tcpRx := lastTCP.Add(15 * time.Second) var udpCh chan []byte var udpErr chan error udpReady := false var udpRx time.Time for { s.mu.Lock() lastUDP := s.lastUDPSend s.mu.Unlock() next := lastTCP.Add(10 * time.Second) if tcpRx.Before(next) { next = tcpRx } if udpReady { if u := lastUDP.Add(15 * time.Second); u.Before(next) { next = u } if udpRx.Before(next) { next = udpRx } } wait := next.Sub(s.clock.Now()) if wait < 0 { wait = 0 } timer := s.clock.NewTimer(wait) var udpRecv <-chan []byte var udpFail <-chan error if udpCh != nil { udpRecv = udpCh udpFail = udpErr } var reason string var tcpf frame var udpp []byte select { case <-ctx.Done(): timer.Stop() return ctx.Err() case err := <-tcpErr: timer.Stop() return err case err := <-udpFail: timer.Stop() return err case tcpf = <-tcpCh: timer.Stop() reason = "tcp" case udpp = <-udpRecv: timer.Stop() reason = "udp" case <-timer.C(): select { case tcpf = <-tcpCh: reason = "tcp" default: select { case udpp = <-udpRecv: reason = "udp" default: reason = "timer" } } } now := s.clock.Now() switch reason { case "tcp": tcpRx = now.Add(15 * time.Second) if err := s.onTCP(conn, tcpf, done, &udpCh, &udpErr, &udpReady, &udpRx, &lastTCP); err != nil { return err } case "udp": s.mu.Lock() if clientIDOf(udpp) == s.clientID { udpRx = now.Add(60 * time.Second) s.onDatagram(udpp) } s.mu.Unlock() default: if !now.Before(tcpRx) { return errors.New("tcp receive timeout") } if udpReady && !now.Before(udpRx) { return errors.New("udp receive timeout") } if !now.Before(lastTCP.Add(10 * time.Second)) { if err := writeFrame(conn, tcpHeartbeat, nil); err != nil { return err } lastTCP = s.clock.Now() } s.mu.Lock() if udpReady && s.udp != nil && !s.clock.Now().Before(s.lastUDPSend.Add(15*time.Second)) { if err := s.sendUDP(udpHeartbeat, nil); err != nil { s.mu.Unlock() return err } } s.mu.Unlock() } } } func readTCP(conn net.Conn, out chan<- frame, errc chan<- error, done <-chan struct{}) { for { f, err := readFrame(conn) if err != nil { select { case errc <- err: case <-done: } return } select { case out <- f: case <-done: return } } } func readUDP(conn *net.UDPConn, out chan<- []byte, errc chan<- error, done <-chan struct{}) { buf := make([]byte, 2048) for { n, err := conn.Read(buf) if err != nil { select { case errc <- err: case <-done: } return } pkt := append([]byte(nil), buf[:n]...) select { case out <- pkt: case <-done: return } } } func (s *Station) onTCP(conn net.Conn, f frame, done <-chan struct{}, udpCh *chan []byte, udpErr *chan error, udpReady *bool, udpRx *time.Time, lastTCP *time.Time) error { switch f.typ { case tcpHeartbeat, tcpAuthOK: return nil case tcpAuthChallenge: chal, _, err := parseBlob(f.body) if err != nil || len(chal) != 20 { return errors.New("bad challenge") } mac := hmac.New(sha1.New, []byte(s.cfg.authKey)) mac.Write(chal) body := appendString(nil, s.cfg.callsign) body = appendBlob(body, mac.Sum(nil)) if err := writeFrame(conn, tcpAuthResponse, body); err != nil { return err } *lastTCP = s.clock.Now() return nil case tcpError: msg, _, err := parseString(f.body) if err != nil { msg = "could not be read" } s.stop(msg) return errStopped case tcpServerInfo: return s.onServerInfo(conn, f.body, done, udpCh, udpErr, udpReady, udpRx, lastTCP) case tcpTalkerStart: s.onTalker(f.body, true) return nil case tcpTalkerStop: s.onTalker(f.body, false) return nil default: return nil } } func (s *Station) onServerInfo(conn net.Conn, body []byte, done <-chan struct{}, udpCh *chan []byte, udpErr *chan error, udpReady *bool, udpRx *time.Time, lastTCP *time.Time) error { id, codecs, err := parseServerInfo(body) if err != nil { return err } if !hasOPUS(codecs) { s.stop(noOpus) return errStopped } enc, err := opus.NewEncoder() if err != nil { return err } dec, err := opus.NewDecoder() if err != nil { enc.Close() return err } var tg [4]byte binary.BigEndian.PutUint32(tg[:], uint32(s.cfg.talkGroup)) if err := writeFrame(conn, tcpSelectTG, tg[:]); err != nil { enc.Close() dec.Close() return err } *lastTCP = s.clock.Now() remote, ok := conn.RemoteAddr().(*net.TCPAddr) if !ok { enc.Close() dec.Close() return errors.New("tcp peer has no address") } udpConn, err := net.DialUDP("udp", nil, &net.UDPAddr{IP: remote.IP, Port: remote.Port}) if err != nil { enc.Close() dec.Close() return err } ch := make(chan []byte, 32) errc := make(chan error, 1) go readUDP(udpConn, ch, errc, done) s.mu.Lock() s.clientID = id s.seq = 0 s.udp = udpConn s.enc = enc s.dec = dec err = s.sendUDP(udpHeartbeat, nil) if err != nil { s.mu.Unlock() enc.Close() dec.Close() udpConn.Close() return err } s.state.Connection = "Connected" s.publish() s.mu.Unlock() *udpCh = ch *udpErr = errc *udpReady = true *udpRx = s.clock.Now().Add(60 * time.Second) return nil } func (s *Station) stop(msg string) { s.mu.Lock() defer s.mu.Unlock() s.fatal = msg s.publish() } // Fail stops retries and shows message until the operator quits. func (s *Station) Fail(message string) { s.stop(message) } func (s *Station) onTalker(body []byte, start bool) { tg, rest, err := parseU32(body) if err != nil || int(tg) != s.cfg.talkGroup { return } call, _, err := parseString(rest) if err != nil { return } s.mu.Lock() defer s.mu.Unlock() if start { if call == s.cfg.callsign { return } was := s.transmitting() s.remoteTalker = call s.roomBusy = true if was && !s.transmitting() { s.endTransmit() } s.publish() return } s.remoteTalker = "" s.roomBusy = false s.publish() } func (s *Station) onDatagram(pkt []byte) { if len(pkt) < 6 { return } typ := binary.BigEndian.Uint16(pkt[:2]) if typ != udpAudio || s.dec == nil { return } if s.transmitting() { return } packet, _, err := parseBlob(pkt[6:]) if err != nil { return } pcm, err := s.dec.Decode(packet) if err != nil { return } s.play = append(s.play, pcm...) s.roomBusy = true s.publish() } func clientIDOf(pkt []byte) uint16 { if len(pkt) < 6 { return ^uint16(0) } return binary.BigEndian.Uint16(pkt[2:4]) } func (s *Station) sendUDP(typ uint16, body []byte) error { if s.udp == nil { return errors.New("udp is down") } buf := make([]byte, 6+len(body)) binary.BigEndian.PutUint16(buf[0:2], typ) binary.BigEndian.PutUint16(buf[2:4], s.clientID) binary.BigEndian.PutUint16(buf[4:6], s.seq) s.seq++ copy(buf[6:], body) if _, err := s.udp.Write(buf); err != nil { return err } s.lastUDPSend = s.clock.Now() return nil } func (s *Station) endTransmit() { if s.enc != nil && s.udp != nil && len(s.mic) > 0 { frame := make([]int16, opus.FrameSamples) copy(frame, s.mic) s.mic = nil if pkt, err := s.enc.Encode(frame); err == nil { // A lost datagram ends the session when the receive timer expires. _ = s.sendUDP(udpAudio, appendBlob(nil, pkt)) } } else { s.mic = nil } if s.udp != nil { // A lost datagram ends the session when the receive timer expires. _ = s.sendUDP(udpFlush, nil) } } func (s *Station) closeUDP() { if s.udp != nil { s.udp.Close() s.udp = nil } if s.enc != nil { s.enc.Close() s.enc = nil } if s.dec != nil { s.dec.Close() s.dec = nil } } func (s *Station) transmitting() bool { if s.fatal != "" || s.state.Connection != "Connected" || s.roomBusy { return false } return s.buttonHeld || s.spaceHeld } func (s *Station) publish() { s.state.Callsign = s.cfg.callsign s.state.TalkGroup = s.cfg.talkGroup if s.fatal != "" { s.state.Connection = "Disconnected" s.state.Detail = s.fatal s.state.Talker = "" s.state.Transmitting = false } else { s.state.Detail = "Idle" tx := s.transmitting() s.state.Transmitting = tx if tx { s.state.Talker = s.cfg.callsign s.play = nil } else { s.state.Talker = s.remoteTalker } } select { case s.wake <- struct{}{}: default: } }