funknetz/station/session.go

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 {
	defer s.Close()
	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.noteDownLocked(s.clock.Now(), true)
			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()
		s.mic = nil
		s.play = nil
		if s.fatal == "" {
			s.noteDownLocked(s.clock.Now(), false)
		} else {
			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.endFatal(msg, true)
		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.endFatal(noOpus, true)
		return errStopped
	}
	enc, err := opus.NewEncoder()
	if err != nil {
		return err
	}
	dec, err := opus.NewDecoder()
	if err != nil {
		enc.Close()
		return err
	}
	info, err := softwareIdentityBody()
	if err != nil {
		enc.Close()
		dec.Close()
		return err
	}
	if err := writeFrame(conn, tcpNodeInfo, info); err != nil {
		enc.Close()
		dec.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.up = true
	s.outage = false
	s.noteTransmitLocked()
	s.writeLog(s.clock.Now(), "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.endFatal(msg, false) }

// 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()
	now := s.clock.Now()
	if start {
		if call == s.cfg.callsign {
			return
		}
		s.closeTalkerLocked(now)
		was := s.transmitting()
		s.remoteTalker = call
		s.roomBusy = true
		s.talkCall = call
		s.talkStart = now
		if was && !s.transmitting() {
			s.endTransmit()
		}
		s.publish()
		return
	}
	s.closeTalkerLocked(now)
	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.noteTransmitLocked()
	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
		}
		if s.state.Talker == "" && s.logNotice != "" {
			s.state.Detail = s.logNotice
		}
	}
	select {
	case s.wake <- struct{}{}:
	default:
	}
}