// telegram side of signaling.Bridge. // Adapter is per-session; Manager is keyed by (jid, callID). // Caller = gateway issued CreateCall, Callee = gateway received Pending. package tgsig import ( "sync" "time" "dev.narayana.im/narayana/telegabber/calls/audio" "dev.narayana.im/narayana/telegabber/calls/signaling" "gotgcalls/ntgcalls" log "github.com/sirupsen/logrus" "github.com/zelenin/go-tdlib/client" ) // past this point ntgcalls is probably stuck and the user hears silence - // audio sent before Connected is dropped const ntgConnectStallThreshold = 7 * time.Second // narrow upstream APIs to what we actually use, so tests can fake them type TdlibClient interface { CreateCall(req *client.CreateCallRequest) (*client.CallId, error) AcceptCall(req *client.AcceptCallRequest) (*client.Ok, error) DiscardCall(req *client.DiscardCallRequest) (*client.Ok, error) SendCallSignalingData(req *client.SendCallSignalingDataRequest) (*client.Ok, error) } type NtgCallsClient interface { OnSignal(callback ntgcalls.SignalCallback) OnConnectionChange(callback ntgcalls.ConnectionChangeCallback) CreateP2PCall(chatId int64) error SkipExchange(chatId int64, encryptionKey []byte, isOutgoing bool) error ConnectP2P(chatId int64, rtcServers []ntgcalls.RTCServer, versions []string, p2pAllowed bool) error SendSignalingData(chatId int64, data []byte) error Stop(chatId int64) error } type Adapter struct { tdlib TdlibClient ntg NtgCallsClient audioNtg audio.NtgClient // nil disables audio (signaling-only tests) jid string mgr *Manager mu sync.Mutex byUserID map[int64]*callBase // gates ntg/tdlib touches against Close()/Free() on the C++ object. // dispatchers take RLock via withCallContext; Close takes Lock. lifecycleLock sync.RWMutex closed bool // per-chatID stall warning while ntgcalls is in Connecting ntgConnectTimersMu sync.Mutex ntgConnectTimers map[int64]*time.Timer } // runs fn under the lifecycle RLock; returns false (and skips fn) if closed func (a *Adapter) withCallContext(fn func()) bool { a.lifecycleLock.RLock() defer a.lifecycleLock.RUnlock() if a.closed { return false } fn() return true } func New(tdlib TdlibClient, ntg NtgCallsClient, audioNtg audio.NtgClient, jid string, mgr *Manager) *Adapter { a := &Adapter{ tdlib: tdlib, ntg: ntg, audioNtg: audioNtg, jid: jid, mgr: mgr, byUserID: make(map[int64]*callBase), } a.attachNtgCallbacks() return a } func (a *Adapter) JID() string { return a.jid } // outbound call; callID is unknown until CreateCall returns, so // the entry registers with Manager from Caller.Start func (a *Adapter) NewCaller(userID int64, isVideo bool) *Caller { return &Caller{callBase: callBase{ adapter: a, userID: userID, isVideo: isVideo, side: signaling.CallerSide, }} } // inbound call; callID is known so the entry registers immediately func (a *Adapter) NewCallee(callID int32, userID int64, isVideo bool) *Callee { c := &Callee{callBase: callBase{ adapter: a, userID: userID, isVideo: isVideo, callID: callID, side: signaling.CalleeSide, }} a.registerSide(userID, &c.callBase) a.mgr.Register(a.jid, callID, &c.callBase) return c } func (a *Adapter) registerSide(userID int64, s *callBase) { a.mu.Lock() a.byUserID[userID] = s a.mu.Unlock() } func (a *Adapter) dropSide(userID int64) { a.mu.Lock() delete(a.byUserID, userID) a.mu.Unlock() } func (a *Adapter) lookupSideByUser(userID int64) (*callBase, bool) { a.mu.Lock() defer a.mu.Unlock() s, ok := a.byUserID[userID] return s, ok } // terminates every active call and marks the adapter closed; after this // returns no goroutine is mid-ntg/tdlib through here, safe to Free() the // C++ ntgcalls object. Bridge.Terminate runs outside the lock and is // synchronous so audio halves close (and ClearStreams runs) while C++ // is still alive func (a *Adapter) Close() { a.lifecycleLock.Lock() if a.closed { a.lifecycleLock.Unlock() return } a.mu.Lock() sides := make([]*callBase, 0, len(a.byUserID)) for _, s := range a.byUserID { sides = append(sides, s) } a.byUserID = make(map[int64]*callBase) a.mu.Unlock() a.closed = true a.lifecycleLock.Unlock() for _, s := range sides { if b := s.getBridge(); b != nil { b.Terminate(signaling.ReasonHangup) } if cid := s.getCallID(); cid != 0 { a.mgr.Drop(a.jid, cid) } } for _, s := range sides { if err := a.ntg.Stop(s.userID); err != nil { log.WithField("user_id", s.userID).WithError(err).Debug("ntgcalls Stop during teardown") } } } func (a *Adapter) attachNtgCallbacks() { a.ntg.OnSignal(func(chatID int64, data []byte) { // invoked from the ntgcalls C++ thread; guard against Close()/Free() a.withCallContext(func() { s, ok := a.lookupSideByUser(chatID) if !ok { log.WithFields(log.Fields{ "jid": a.jid, "user_id": chatID, }).Warn("ntgcalls OnSignal for unknown user") return } if _, err := a.tdlib.SendCallSignalingData(&client.SendCallSignalingDataRequest{ CallId: s.getCallID(), Data: data, }); err != nil { log.WithFields(log.Fields{ "call_id": s.getCallID(), "user_id": chatID, }).WithError(err).Error("Failed to send signaling data to TDLib") } }) }) a.ntg.OnConnectionChange(func(chatID int64, info ntgcalls.NetworkInfo) { entry := log.WithFields(log.Fields{ "jid": a.jid, "user_id": chatID, "state": info.State, "kind": info.Kind, }) entry.Info("ntgcalls connection state changed") // the TG peer can hear nothing while ntgcalls is still negotiating // with TG's TURN relays even though the XMPP side has RTP flowing switch info.State { case ntgcalls.Connecting: a.startNtgConnectStallTimer(chatID) case ntgcalls.Connected, ntgcalls.Failed, ntgcalls.Timeout, ntgcalls.Closed: a.stopNtgConnectStallTimer(chatID) } s, ok := a.lookupSideByUser(chatID) if !ok { return } b := s.getBridge() if b == nil { return } switch info.State { case ntgcalls.Connected: b.MediaConnected(s.side) case ntgcalls.Failed: b.MediaFailed(signaling.ReasonMediaFailed) case ntgcalls.Timeout: b.MediaFailed(signaling.ReasonConnectTimeout) } }) } // (re)arms the stall timer; Connecting can fire more than once func (a *Adapter) startNtgConnectStallTimer(chatID int64) { a.stopNtgConnectStallTimer(chatID) t := time.AfterFunc(ntgConnectStallThreshold, func() { log.WithFields(log.Fields{ "jid": a.jid, "user_id": chatID, "after": ntgConnectStallThreshold, }).Warn("ntgcalls still Connecting - TG peer is likely hearing silence") }) a.ntgConnectTimersMu.Lock() if a.ntgConnectTimers == nil { a.ntgConnectTimers = make(map[int64]*time.Timer) } a.ntgConnectTimers[chatID] = t a.ntgConnectTimersMu.Unlock() } func (a *Adapter) stopNtgConnectStallTimer(chatID int64) { a.ntgConnectTimersMu.Lock() t, ok := a.ntgConnectTimers[chatID] if ok { delete(a.ntgConnectTimers, chatID) } a.ntgConnectTimersMu.Unlock() if t != nil { t.Stop() } }