diff --git a/cmd/m17-gateway/audio.go b/cmd/m17-gateway/audio.go index e564c4c..8aa1497 100644 --- a/cmd/m17-gateway/audio.go +++ b/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)) diff --git a/cmd/m17-gateway/gateway.go b/cmd/m17-gateway/gateway.go index 98c70ba..5405798 100644 --- a/cmd/m17-gateway/gateway.go +++ b/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() diff --git a/cmd/m17-gateway/packets_test.go b/cmd/m17-gateway/packets_test.go index 5ffeece..9e6242f 100644 --- a/cmd/m17-gateway/packets_test.go +++ b/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 { diff --git a/cmd/m17-gateway/voice_test.go b/cmd/m17-gateway/voice_test.go index 0008026..1ccdbd4 100644 --- a/cmd/m17-gateway/voice_test.go +++ b/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) {