telegabber/calls/signaling/xmppsig/pion_loop_test.go

package xmppsig

import (
	"context"
	"sync/atomic"
	"testing"
	"time"

	"dev.narayana.im/narayana/telegabber/xmpp/jingle"

	"github.com/pion/webrtc/v4"
	"gosrc.io/xmpp/stanza"
)

// every Send dispatches into the other Manager's HandlePacket on a fresh
// goroutine, surfacing trickle/ordering bugs naturally
type linkedSender struct{ target *jingle.Manager }

func (ls *linkedSender) Send(p stanza.Packet) error {
	go ls.target.HandlePacket(nil, p)
	return nil
}
func (ls *linkedSender) SendIQ(_ context.Context, iq *stanza.IQ) (chan stanza.IQ, error) {
	return make(chan stanza.IQ, 1), ls.Send(iq)
}

type calleeBundle struct {
	callee      *Callee
	pcConnected <-chan struct{}
}

// full xmppsig path against itself: Caller on A, Callee on B, paired via
// linkedSenders. Both PCs must reach Connected and both ends must see the
// peer's track. Track-attachment timing matches setupAudio: attach after
// SetRemoteDescription (pre-negotiation ReplaceTrack doesn't wire pion's encoder).
func TestXmppsigPionLoop_TwoSidesConnectAndDeliverTrack(t *testing.T) {
	if testing.Short() {
		t.Skip("e2e ICE handshake")
	}

	mA := &jingle.Manager{LocalJID: "a@gw/telegabber"}
	mB := &jingle.Manager{LocalJID: "b@gw/telegabber"}
	mA.Sender = &linkedSender{target: mB}
	mB.Sender = &linkedSender{target: mA}

	adapterA := NewAdapter(AdapterConfig{
		Sender: mA.Sender, LocalJID: mA.LocalJID, Manager: mA,
		PCFactory: DefaultPCFactory,
	})
	adapterB := NewAdapter(AdapterConfig{
		Sender: mB.Sender, LocalJID: mB.LocalJID, Manager: mB,
		PCFactory: DefaultPCFactory,
	})

	calleeReady := make(chan calleeBundle, 1)
	mB.OnProposal = func(p jingle.IncomingProposal) (*jingle.Session, func(), error) {
		callee, sess, err := adapterB.NewCallee(p)
		if err != nil {
			return nil, nil, err
		}
		pc := callee.PeerConnection()
		// production pattern: on OnRemoteSDP, slot a track into the
		// sender DefaultPCFactory pre-created
		callee.SetOnRemoteSDP(func(string) {
			tr := newSilenceTrack(t, "b-stream")
			if err := replaceFirstEmptySender(pc, tr); err != nil {
				t.Errorf("B replaceFirstEmptySender: %v", err)
				return
			}
			startSilencePump(t, tr)
		})
		calleeReady <- calleeBundle{callee: callee, pcConnected: connectedSignal(pc)}
		go func() {
			deadline := time.Now().Add(2 * time.Second)
			for time.Now().Before(deadline) && sess.State() != jingle.StateRinging {
				time.Sleep(5 * time.Millisecond)
			}
			callee.Accept()
		}()
		return sess, nil, nil
	}

	caller, err := adapterA.NewCaller("b@gw/telegabber", 42)
	if err != nil {
		t.Fatalf("NewCaller: %v", err)
	}
	pcA := caller.PeerConnection()
	caller.SetOnRemoteSDP(func(string) {
		tr := newSilenceTrack(t, "a-stream")
		if err := replaceFirstEmptySender(pcA, tr); err != nil {
			t.Errorf("A replaceFirstEmptySender: %v", err)
			return
		}
		startSilencePump(t, tr)
	})
	connA := connectedSignal(pcA)

	caller.Start()

	var bundle calleeBundle
	select {
	case bundle = <-calleeReady:
	case <-time.After(5 * time.Second):
		t.Fatal("OnProposal never fired on B side")
	}

	// install handlers AFTER Start so tracks landing during
	// SetRemoteDescription go through hookTrack's buffered path
	var aGot, bGot atomic.Int64
	caller.SetTrackHandler(func(*webrtc.TrackRemote) { aGot.Add(1) })
	bundle.callee.SetTrackHandler(func(*webrtc.TrackRemote) { bGot.Add(1) })

	const deadline = 15 * time.Second
	select {
	case <-connA:
	case <-time.After(deadline):
		t.Fatalf("Caller PC never Connected (state=%s)", pcA.ConnectionState())
	}
	select {
	case <-bundle.pcConnected:
	case <-time.After(deadline):
		t.Fatalf("Callee PC never Connected (state=%s)",
			bundle.callee.PeerConnection().ConnectionState())
	}

	waitUntilTrue(t, "A received B's track", 10*time.Second, func() bool { return aGot.Load() > 0 })
	waitUntilTrue(t, "B received A's track", 10*time.Second, func() bool { return bGot.Load() > 0 })
}

// fresh opus track; caller attaches and starts the pump at the right
// point in the negotiation
func newSilenceTrack(t *testing.T, streamID string) *webrtc.TrackLocalStaticSample {
	t.Helper()
	tr, err := webrtc.NewTrackLocalStaticSample(
		webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeOpus, ClockRate: 48000, Channels: 2},
		"audio", streamID,
	)
	if err != nil {
		t.Fatalf("NewTrackLocalStaticSample: %v", err)
	}
	return tr
}

// mirrors orchestrator.attachOutgoingTrack; falls back to AddTrack for
// bare PCFactory tests
func replaceFirstEmptySender(pc *webrtc.PeerConnection, tr webrtc.TrackLocal) error {
	for _, s := range pc.GetSenders() {
		if s == nil {
			continue
		}
		if err := s.ReplaceTrack(tr); err == nil {
			return nil
		}
	}
	_, err := pc.AddTrack(tr)
	return err
}

// pushes 20ms opus silence into tr; stops at test cleanup
func startSilencePump(t *testing.T, tr *webrtc.TrackLocalStaticSample) {
	t.Helper()
	stop := make(chan struct{})
	t.Cleanup(func() {
		defer func() { _ = recover() }() // tolerate double-close in failure paths
		close(stop)
	})
	go pumpSilenceOpus(tr, stop)
}

// closes once pc reaches Connected; must be installed before pc could reach
// Connected because pion fires on state change, not on register
func connectedSignal(pc *webrtc.PeerConnection) <-chan struct{} {
	ch := make(chan struct{})
	var once atomic.Bool
	pc.OnConnectionStateChange(func(s webrtc.PeerConnectionState) {
		if s == webrtc.PeerConnectionStateConnected && once.CompareAndSwap(false, true) {
			close(ch)
		}
	})
	return ch
}