package jingle import ( "fmt" "sync" log "github.com/sirupsen/logrus" "gosrc.io/xmpp/stanza" ) type IncomingProposal struct { SID string From string To string Media []string } // runs when JMI arrives for an unknown SID. // - (s, postReady, nil): accept; Manager registers s and calls postReady // (if non-nil) AFTER HandleJMI(propose) has reached StateRinging. Useful // for tg-side goroutines that could race back in via the bridge. // - (nil, _, nil): silently ignore. // - (nil, _, err): logged, propose dropped. type ProposalHandler func(IncomingProposal) (*Session, func(), error) // dispatches inbound Jingle IQ and JMI message stanzas to matching Sessions; // one Manager per gateway-side endpoint type Manager struct { LocalJID string // `from=` on IQ result/error replies Sender Sender // IQ replies only; Sessions own outbound traffic OnProposal ProposalHandler sessions sync.Map // sid -> *Session // stops two concurrent s for the same sid from both // invoking OnProposal pendingPropose sync.Map } func (m *Manager) Register(s *Session) { if s == nil { return } m.sessions.Store(s.SID(), s) } func (m *Manager) Unregister(sid string) { m.sessions.Delete(sid) } func (m *Manager) session(sid string) *Session { v, ok := m.sessions.Load(sid) if !ok { return nil } return v.(*Session) } // entrypoint for gosrc.io/xmpp's router (message and iq routes) func (m *Manager) HandlePacket(_ interface{}, p stanza.Packet) { // gosrc.io delivers *stanza.IQ but stanza.Message as value switch pkt := p.(type) { case *stanza.IQ: m.handleIQ(pkt) case stanza.Message: m.handleMessage(&pkt) case *stanza.Message: m.handleMessage(pkt) } } func (m *Manager) HandleIQ(iq *stanza.IQ) { m.handleIQ(iq) } func (m *Manager) HandleMessage(msg *stanza.Message) { m.handleMessage(msg) } func (m *Manager) handleIQ(iq *stanza.IQ) { j, ok := iq.Payload.(*JingleIQ) if !ok || j == nil { return } sess := m.session(j.SID) if sess == nil { m.replyError(iq, "item-not-found") return } _, errCondition := sess.HandleJingleIQ(iq, j) if errCondition != "" { m.replyError(iq, errCondition) return } if iq.Type == stanza.IQTypeSet || iq.Type == stanza.IQTypeGet { m.replyResult(iq) } if j.Action == ActionSessionTerminate { m.Unregister(j.SID) } } func (m *Manager) handleMessage(msg *stanza.Message) { for _, ext := range msg.Extensions { switch x := ext.(type) { case *JMIPropose: m.dispatchPropose(msg, x) case *JMIProceed: m.dispatchJMI(x.ID, msg.From, x) case *JMIRinging: m.dispatchJMI(x.ID, msg.From, x) case *JMIReject: m.dispatchJMI(x.ID, msg.From, x) m.Unregister(x.ID) case *JMIRetract: m.dispatchJMI(x.ID, msg.From, x) m.Unregister(x.ID) case *JMIFinish: // terminal, ignore } } } func (m *Manager) dispatchPropose(msg *stanza.Message, p *JMIPropose) { if existing := m.session(p.ID); existing != nil { return } // concurrent s for the same sid: only one dispatcher proceeds if _, loaded := m.pendingPropose.LoadOrStore(p.ID, struct{}{}); loaded { return } defer m.pendingPropose.Delete(p.ID) if m.OnProposal == nil { return } media := make([]string, 0, len(p.Descriptions)) for _, d := range p.Descriptions { media = append(media, d.Media) } sess, postReady, err := m.OnProposal(IncomingProposal{SID: p.ID, From: msg.From, To: msg.To, Media: media}) if err != nil || sess == nil { if err != nil { log.WithFields(log.Fields{"sid": p.ID, "from": msg.From}).WithError(err). Warn("jingle.Manager.dispatchPropose: OnProposal returned err") } return } m.Register(sess) sess.HandleJMI(msg.From, p) if postReady != nil { postReady() } } func (m *Manager) dispatchJMI(sid, from string, ext stanza.MsgExtension) { sess := m.session(sid) if sess == nil { log.WithFields(log.Fields{ "sid": sid, "from": from, "ext_type": fmt.Sprintf("%T", ext), }).Warn("jingle.Manager.dispatchJMI: no session for sid; dropping (stray / from SID mismatch?)") return } sess.HandleJMI(from, ext) } func (m *Manager) replyResult(iq *stanza.IQ) { if m.Sender == nil { return } out, err := stanza.NewIQ(stanza.Attrs{ Type: stanza.IQTypeResult, From: iq.To, To: iq.From, Id: iq.Id, }) if err != nil { return } _ = m.Sender.Send(out) } // ; condition from urn:ietf:params:xml:ns:xmpp-stanzas func (m *Manager) replyError(iq *stanza.IQ, condition string) { if m.Sender == nil { return } out, err := stanza.NewIQ(stanza.Attrs{ Type: stanza.IQTypeError, From: iq.To, To: iq.From, Id: iq.Id, }) if err != nil { return } out.Error = &stanza.Err{ Type: "cancel", Reason: condition, } _ = m.Sender.Send(out) }