m17-gateway: hold reflector voice while the gateway speaks

cmd/m17-gateway/audio.go

@@ -132,6 +132,7 @@ func (g *Gateway) sayCallsign(callsign string) []byte {
 func (g *Gateway) playMessage(msg ...string) error {
 	log.Printf("[DEBUG] playMessage() setState(LocalCommand)")
 	defer g.setState(Idle)
+	defer g.holdVoice()()
 	lsf, err := m17.NewLSF(m17.DestinationAll, g.encodedCallsign.Callsign(), m17.LSFTypeStream, m17.LSFDataTypeVoice, 0)
 	if err != nil {
 		return fmt.Errorf("unable to make LSF: %w", err)
@@ -200,6 +201,7 @@ func (g *Gateway) echoStreamEnd() error {
 	defer g.setState(Idle)
 	var err error
 	if g.getState() == Echo {
+		defer g.holdVoice()()
 		log.Printf("[DEBUG] echoStreamEnd() %d frames", len(g.echoStream))
 		time.Sleep(250 * time.Millisecond)
 		sid := uint16(rand.Intn(0x10000))

cmd/m17-gateway/gateway.go

@@ -11,6 +11,7 @@ import (
 	"os/signal"
 	"path/filepath"
 	"sync"
+	"sync/atomic"
 	"syscall"
 	"time"
 
@@ -360,9 +361,10 @@ type Gateway struct {
 	lastFrameTimer *time.Timer
 	lastLSF        *m17.LSF // Workaround for reflectors that change the SRC during the stream
 	lastStreamID   uint16
-	netStreamHeld  bool // reflector voice is being held back for local RF
-	rfToLegacy     bool // the current RF stream is going to a legacy reflector
-	rfDirectLogged bool // and a warning that it will probably be dropped was logged
+	netStreamHeld  bool         // reflector voice is being held back
+	ownTX          atomic.Int32 // the gateway's own transmissions under way: announcements and echoes
+	rfToLegacy     bool         // the current RF stream is going to a legacy reflector
+	rfDirectLogged bool         // and a warning that it will probably be dropped was logged
 	echoStream     []m17.StreamDatagram
 	audioClips     map[string][]byte
 }
@@ -467,13 +469,21 @@ func (g *Gateway) TransmitVoiceStream(sd m17.StreamDatagram) error {
 	sd.LSF.SetECD(&sd.LSF.Src, g.inetClient.EncodedName)
 	sd.LSF.Src = g.encodedCallsign
 	sd.LSF.CalcCRC()
-	// Local RF takes precedence. While a radio is transmitting to us, hold
-	// the reflector stream back, then join it once the channel is clear: the
-	// modem starts a fresh transmission, preamble and LSF, with the next frame.
-	held := g.localRFActive()
+	// Local RF takes precedence, and so does the gateway's own voice. While
+	// a radio is transmitting to us, or the gateway is speaking, hold the
+	// reflector stream back, then join it once the channel is clear: the
+	// modem starts a fresh transmission, preamble and LSF, with the next
+	// frame. Two streams sent at once would take the transmitter from each
+	// other frame by frame, each with an EOT and a new LSF.
+	own := g.ownTX.Load() > 0
+	held := own || g.localRFActive()
 	if held != g.netStreamHeld {
 		if held {
-			log.Printf("[DEBUG] Holding Internet voice stream %04x for local RF", sd.StreamID)
+			reason := "local RF"
+			if own {
+				reason = "the gateway's own voice"
+			}
+			log.Printf("[DEBUG] Holding Internet voice stream %04x for %s", sd.StreamID, reason)
 		} else {
 			log.Printf("[DEBUG] Joining Internet voice stream %04x at frame %04x", sd.StreamID, sd.FrameNumber&0x7fff)
 		}
@@ -524,6 +534,13 @@ func (g *Gateway) TransmitVoiceStream(sd m17.StreamDatagram) error {
 	return err
 }
 
+// holdVoice holds reflector voice back while the gateway transmits
+// something of its own, until the returned function is called. Holds nest.
+func (g *Gateway) holdVoice() func() {
+	g.ownTX.Add(1)
+	return func() { g.ownTX.Add(-1) }
+}
+
 func (g *Gateway) getState() gatewayState {
 	g.stateMutex.Lock()
 	defer g.stateMutex.Unlock()

cmd/m17-gateway/packets_test.go

@@ -11,9 +11,10 @@ import (
 // recordingModem records when each packet is transmitted, and the frame
 // number of each voice frame.
 type recordingModem struct {
-	mu    sync.Mutex
-	sent  []time.Time
-	voice []uint16
+	mu      sync.Mutex
+	sent    []time.Time
+	voice   []uint16
+	onVoice func() // if set, called after each voice frame, outside mu
 }
 
 func (m *recordingModem) StartDecoding(func(uint16, []m17.SoftBit)) {}
@@ -22,8 +23,12 @@ func (m *recordingModem) Reset() error                              { return nil
 func (m *recordingModem) Close() error                              { return nil }
 func (m *recordingModem) TransmitVoiceStream(sd m17.StreamDatagram) error {
 	m.mu.Lock()
-	defer m.mu.Unlock()
 	m.voice = append(m.voice, sd.FrameNumber)
+	onVoice := m.onVoice
+	m.mu.Unlock()
+	if onVoice != nil {
+		onVoice()
+	}
 	return nil
 }
 func (m *recordingModem) TransmitPacket(m17.Packet) error {

cmd/m17-gateway/voice_test.go

@@ -51,6 +51,111 @@ func TestNetVoiceWaitsForRF(t *testing.T) {
 	}
 }
 
+// TestNetVoiceWaitsForSpeaking: reflector voice is held back while the
+// gateway speaks, through overlapping holds, and joined, with a fresh LSF,
+// once the last hold is released.
+func TestNetVoiceWaitsForSpeaking(t *testing.T) {
+	g, m := testGateway(t, 10*time.Millisecond, 10*time.Millisecond)
+	g.inetClient = &inet.Client{}
+	lsf, err := m17.NewLSF("@ALL", "W7GKB D", m17.LSFTypeStream, m17.LSFDataTypeVoice, 0)
+	if err != nil {
+		t.Fatal(err)
+	}
+	send := func(fn uint16) {
+		sd := m17.NewStreamDatagram(0x6425, fn, &lsf, make([]byte, 16))
+		if err := g.TransmitVoiceStream(sd); err != nil {
+			t.Fatal(err)
+		}
+	}
+	release := g.holdVoice()
+	inner := g.holdVoice() // overlaps the first, and ends before it
+	for fn := range uint16(3) {
+		send(fn)
+	}
+	inner()
+	for fn := uint16(3); fn < 5; fn++ {
+		send(fn)
+	}
+	if got := m.voiceFrames(); len(got) > 0 {
+		t.Fatalf("frames %v transmitted while the gateway spoke", got)
+	}
+	release()
+	send(5)
+	send(6 | 0x8000)
+	if got, want := m.voiceFrames(), []uint16{5, 6 | 0x8000}; !slices.Equal(got, want) {
+		t.Errorf("transmitted frames %v, want %v", got, want)
+	}
+}
+
+// TestOwnVoiceNotInterrupted: reflector voice that arrives during an
+// announcement or an echo waits for it to finish, instead of taking the
+// transmitter from it frame by frame.
+func TestOwnVoiceNotInterrupted(t *testing.T) {
+	for _, tc := range []struct {
+		name  string
+		speak func(g *Gateway) error
+		want  []uint16 // the gateway's own frames
+	}{
+		{"announcement", func(g *Gateway) error {
+			g.audioClips = map[string][]byte{"welcome": make([]byte, 2*16)}
+			return g.playMessage("welcome")
+		}, []uint16{0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10 | 0x8000}},
+		{"echo", func(g *Gateway) error {
+			lsf, err := m17.NewLSF("#ECHO", "N1ADJ", m17.LSFTypeStream, m17.LSFDataTypeVoice, 0)
+			if err != nil {
+				return err
+			}
+			g.setState(Echo)
+			g.echoStreamStart()
+			for _, fn := range []uint16{0, 1, 2 | 0x8000} {
+				l := lsf
+				g.echoStreamRecord(m17.NewStreamDatagram(0x2222, fn, &l, make([]byte, 16)))
+			}
+			return g.echoStreamEnd()
+		}, []uint16{0, 1, 2 | 0x8000}},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			g, m := testGateway(t, 10*time.Millisecond, 10*time.Millisecond)
+			g.inetClient = &inet.Client{}
+			cs, err := m17.EncodeCallsign("N1ADJ G")
+			if err != nil {
+				t.Fatal(err)
+			}
+			g.encodedCallsign = *cs
+			lsf, err := m17.NewLSF("@ALL", "W7GKB D", m17.LSFTypeStream, m17.LSFDataTypeVoice, 0)
+			if err != nil {
+				t.Fatal(err)
+			}
+			netFN := uint16(0x100)
+			send := func(fn uint16) {
+				sd := m17.NewStreamDatagram(0x6425, fn, &lsf, make([]byte, 16))
+				if err := g.TransmitVoiceStream(sd); err != nil {
+					t.Fatal(err)
+				}
+			}
+			// A reflector frame arrives with each frame the gateway sends.
+			arriving := true
+			m.onVoice = func() {
+				if arriving {
+					arriving = false
+					send(netFN)
+					netFN++
+					arriving = true
+				}
+			}
+			if err := tc.speak(g); err != nil {
+				t.Fatal(err)
+			}
+			arriving = false
+			send(netFN | 0x8000)
+			want := append(tc.want, netFN|0x8000)
+			if got := m.voiceFrames(); !slices.Equal(got, want) {
+				t.Errorf("transmitted frames %v, want %v", got, want)
+			}
+		})
+	}
+}
+
 // TestNetVoiceAfterRFStreamEnd: once a local stream has ended, reflector
 // voice waits only rfTurnaround, not the full rxHoldoff.
 func TestNetVoiceAfterRFStreamEnd(t *testing.T) {