diff --git a/internal/asterisk/answer.go b/internal/asterisk/answer.go index 1e31481..d44806d 100644 --- a/internal/asterisk/answer.go +++ b/internal/asterisk/answer.go @@ -32,14 +32,7 @@ func awaitStasisStart(ctx context.Context, subscription ari.Subscription, channe recent = recent[1:] } recent = append(recent, typ) - matched := false - for _, key := range event.Keys() { - if key != nil && key.Kind == "channel" && key.ID == channelID { - matched = true - break - } - } - if !matched { + if !eventBelongsToChannel(event, channelID) { continue } switch typ { @@ -63,3 +56,12 @@ func awaitStasisStart(ctx context.Context, subscription ari.Subscription, channe } } } + +func eventBelongsToChannel(event ari.Event, channelID string) bool { + for _, key := range event.Keys() { + if key != nil && key.Kind == "channel" && key.ID == channelID { + return true + } + } + return false +} diff --git a/internal/asterisk/answer_test.go b/internal/asterisk/answer_test.go index 7c638c3..8498ada 100644 --- a/internal/asterisk/answer_test.go +++ b/internal/asterisk/answer_test.go @@ -15,6 +15,7 @@ type answerEvents struct { } func (f answerEvents) Events() <-chan ari.Event { return f.events } +func (f answerEvents) Cancel() {} func TestAwaitNativeAnswerRequiresMatchingStasisStart(t *testing.T) { sub := answerEvents{events: make(chan ari.Event, 3)} diff --git a/internal/asterisk/call.go b/internal/asterisk/call.go new file mode 100644 index 0000000..7d8337e --- /dev/null +++ b/internal/asterisk/call.go @@ -0,0 +1,200 @@ +package asterisk + +import ( + "context" + "errors" + "fmt" + "net" + "net/netip" + "strconv" + "sync" + "sync/atomic" + "time" + + "git.ipao.vip/rogee/go-sip/internal/media" + "github.com/CyCoreSystems/ari/v5" +) + +// NativeDial contains only the Dispatcher-selected SIP identity and the +// Agent's explicitly configured SLIN16 RTP payload profile. +type NativeDial struct { + ExecutionID, TrunkID, DialedCallee, CallerID string + AnswerTimeout time.Duration + MediaPayloadType uint8 +} + +// NativeCall owns one real ARI channel, bridge, ExternalMedia channel and RTP +// socket; closing it cannot originate or replay a second call. +type NativeCall struct { + Media *media.RTPStream + Context context.Context + cancel context.CancelCauseFunc + monitorDone chan struct{} + destroyed atomic.Bool + client ari.Client + subscription ari.Subscription + outbound *ari.ChannelHandle + external *ari.ChannelHandle + bridge *ari.BridgeHandle + closeOnce sync.Once + closeErr error +} + +// Originate uses the issued Cell-local ARI credential. The caller must have +// already persisted the signed instruction and passed the host capture gate. +func (l Loader) Originate(ctx context.Context, request NativeDial) (*NativeCall, error) { + if _, err := approvedOriginateRequest(request.ExecutionID, request.TrunkID, request.DialedCallee, request.CallerID, request.AnswerTimeout); err != nil { + return nil, err + } + client, err := l.OpenARI() + if err != nil { + return nil, err + } + return dialWithClient(ctx, client, request) +} + +func dialWithClient(ctx context.Context, client ari.Client, request NativeDial) (_ *NativeCall, err error) { + call := &NativeCall{client: client} + defer func() { + if err != nil { + err = errors.Join(err, call.Close()) + } + }() + if ctx == nil || client == nil { + return nil, errors.New("native ARI call context and client required") + } + call.Context, call.cancel = context.WithCancelCause(ctx) + originate, err := approvedOriginateRequest(request.ExecutionID, request.TrunkID, request.DialedCallee, request.CallerID, request.AnswerTimeout) + if err != nil { + return nil, err + } + call.Media, err = media.ListenRTPWithFormat("127.0.0.1:0", request.MediaPayloadType, media.FormatSLIN16, 16000) + if err != nil { + return nil, fmt.Errorf("prepare Agent RTP capture: %w", err) + } + key := ari.NewKey(ari.ChannelKey, request.ExecutionID) + call.subscription = client.Bus().Subscribe(key, "StasisStart", "StasisEnd", "ChannelHangupRequest", "ChannelDestroyed") + if call.subscription == nil { + return nil, errors.New("native ARI channel subscription unavailable before origination") + } + call.outbound, err = client.Channel().Originate(key, originate) + if err != nil { + // The HTTP response can be lost after Asterisk has created the + // channel. Do not originate again: try to stop the same identity. + call.outbound = client.Channel().Get(key) + return nil, fmt.Errorf("native SIP origination outcome unknown: %T", err) + } + answerCtx, cancel := context.WithTimeout(ctx, request.AnswerTimeout) + defer cancel() + if err := awaitStasisStart(answerCtx, call.subscription, request.ExecutionID); err != nil { + return nil, err + } + bridgeKey := ari.NewKey(ari.BridgeKey, request.ExecutionID+"-bridge") + call.bridge, err = client.Bridge().Create(bridgeKey, "mixing", "go-sip-agent") + if err != nil { + // A lost reply does not prove that the bridge was not created. + call.bridge = client.Bridge().Get(bridgeKey) + return nil, fmt.Errorf("create Agent ARI mixing bridge outcome unknown: %T", err) + } + if err := call.bridge.AddChannel(call.outbound.ID()); err != nil { + return nil, fmt.Errorf("add answered channel to ARI bridge: %T", err) + } + mediaKey := ari.NewKey(ari.ChannelKey, request.ExecutionID+"-media") + call.external, err = client.Channel().ExternalMedia(mediaKey, ari.ExternalMediaOptions{ + App: "go-sip-agent", ExternalHost: net.JoinHostPort("127.0.0.1", strconv.Itoa(call.Media.LocalAddr().(*net.UDPAddr).Port)), + Format: "slin16", Encapsulation: "rtp", Transport: "udp", ConnectionType: "client", Direction: "both", + }) + if err != nil { + // Use the same identity to stop an ExternalMedia channel created + // before the HTTP outcome became unknown; never create another. + call.external = client.Channel().Get(mediaKey) + return nil, fmt.Errorf("create Agent ARI external media outcome unknown: %T", err) + } + address, err := call.external.GetVariable("UNICASTRTP_LOCAL_ADDRESS") + if err != nil { + return nil, fmt.Errorf("read Agent ARI RTP peer address: %T", err) + } + port, err := call.external.GetVariable("UNICASTRTP_LOCAL_PORT") + if err != nil { + return nil, fmt.Errorf("read Agent ARI RTP peer port: %T", err) + } + peer, err := netip.ParseAddr(address) + if err != nil || !peer.IsLoopback() { + return nil, errors.New("Agent ARI RTP peer is not a local Cell address") + } + if err := call.Media.SetPeer(net.JoinHostPort(address, port)); err != nil { + return nil, fmt.Errorf("configure Agent RTP peer: %w", err) + } + if err := call.bridge.AddChannel(call.external.ID()); err != nil { + return nil, fmt.Errorf("add Agent RTP channel to ARI bridge: %T", err) + } + call.monitorDone = make(chan struct{}) + go call.watch(request.ExecutionID) + return call, nil +} + +func (c *NativeCall) watch(channelID string) { + defer close(c.monitorDone) + for { + select { + case <-c.Context.Done(): + return + case event, open := <-c.subscription.Events(): + if !open || event == nil { + c.cancel(errors.New("native ARI event subscription lost during call")) + return + } + if !eventBelongsToChannel(event, channelID) { + continue + } + switch event.GetType() { + case "ChannelDestroyed": + if ended, ok := event.(*ari.ChannelDestroyed); ok { + c.destroyed.Store(true) + c.cancel(fmt.Errorf("native channel ended: cause=%d", ended.Cause)) + } else { + c.cancel(errors.New("native channel ended: invalid destroy event")) + } + return + case "ChannelHangupRequest", "StasisEnd": + c.cancel(fmt.Errorf("native channel ended: event=%s", event.GetType())) + return + } + } + } +} + +// Close releases each resource once. Errors remain visible to the caller; a +// failed Hangup never silently turns an uncertain call into a success. +func (c *NativeCall) Close() error { + if c == nil { + return nil + } + c.closeOnce.Do(func() { + if c.cancel != nil { + c.cancel(context.Canceled) + } + if c.subscription != nil { + c.subscription.Cancel() + } + if c.external != nil { + c.closeErr = errors.Join(c.closeErr, c.external.Hangup()) + } + if c.outbound != nil && !c.destroyed.Load() { + c.closeErr = errors.Join(c.closeErr, c.outbound.Hangup()) + } + if c.bridge != nil { + c.closeErr = errors.Join(c.closeErr, c.bridge.Delete()) + } + if c.Media != nil { + c.closeErr = errors.Join(c.closeErr, c.Media.Close()) + } + if c.client != nil { + c.client.Close() + } + if c.monitorDone != nil { + <-c.monitorDone + } + }) + return c.closeErr +} diff --git a/internal/asterisk/call_test.go b/internal/asterisk/call_test.go new file mode 100644 index 0000000..5dd3714 --- /dev/null +++ b/internal/asterisk/call_test.go @@ -0,0 +1,217 @@ +package asterisk + +import ( + "context" + "errors" + "net" + "strings" + "testing" + "time" + + "github.com/CyCoreSystems/ari/v5" +) + +type testARIClient struct { + ari.Client + channels *testChannels + bridges *testBridges + bus *testBus + closed int +} + +func (c *testARIClient) Channel() ari.Channel { return c.channels } +func (c *testARIClient) Bridge() ari.Bridge { return c.bridges } +func (c *testARIClient) Bus() ari.Bus { return c.bus } +func (c *testARIClient) Close() { c.closed++ } + +type testBus struct { + ari.Bus + sub ari.Subscription +} + +func (b *testBus) Subscribe(_ *ari.Key, _ ...string) ari.Subscription { return b.sub } + +type testChannels struct { + ari.Channel + events chan ari.Event + originate ari.OriginateRequest + media ari.ExternalMediaOptions + issued int + originateErr error + externalErr error + hungup []string + mediaPeerIP string +} + +func (c *testChannels) Originate(_ *ari.Key, request ari.OriginateRequest) (*ari.ChannelHandle, error) { + c.issued++ + c.originate = request + if c.originateErr != nil { + return nil, c.originateErr + } + if c.events != nil { + c.events <- &ari.StasisStart{EventData: ari.EventData{Type: "StasisStart"}, Channel: ari.ChannelData{ID: request.ChannelID}} + } + return ari.NewChannelHandle(ari.NewKey(ari.ChannelKey, request.ChannelID), c, nil), nil +} +func (c *testChannels) Get(key *ari.Key) *ari.ChannelHandle { return ari.NewChannelHandle(key, c, nil) } +func (c *testChannels) ExternalMedia(key *ari.Key, options ari.ExternalMediaOptions) (*ari.ChannelHandle, error) { + c.media = options + if c.externalErr != nil { + return nil, c.externalErr + } + return ari.NewChannelHandle(key, c, nil), nil +} +func (c *testChannels) GetVariable(_ *ari.Key, name string) (string, error) { + switch name { + case "UNICASTRTP_LOCAL_ADDRESS": + return c.mediaPeerIP, nil + case "UNICASTRTP_LOCAL_PORT": + return "17000", nil + } + return "", errors.New("unexpected ARI variable") +} +func (c *testChannels) Hangup(key *ari.Key, _ string) error { + c.hungup = append(c.hungup, key.ID) + return nil +} + +type testBridges struct { + ari.Bridge + attached []string + deleted int + createErr error +} + +func (b *testBridges) Create(key *ari.Key, _, _ string) (*ari.BridgeHandle, error) { + if b.createErr != nil { + return nil, b.createErr + } + return ari.NewBridgeHandle(key, b, nil), nil +} +func (b *testBridges) Get(key *ari.Key) *ari.BridgeHandle { return ari.NewBridgeHandle(key, b, nil) } +func (b *testBridges) AddChannel(_ *ari.Key, channelID string) error { + b.attached = append(b.attached, channelID) + return nil +} +func (b *testBridges) Delete(_ *ari.Key) error { b.deleted++; return nil } + +func TestNativeCallUsesOneARIOriginateAndActualRTPBridge(t *testing.T) { + events := make(chan ari.Event, 2) + client := &testARIClient{ + channels: &testChannels{events: events, mediaPeerIP: "127.0.0.1"}, + bridges: &testBridges{}, + bus: &testBus{sub: answerEvents{events: events}}, + } + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + call, err := dialWithClient(ctx, client, NativeDial{ + ExecutionID: "exec-1", TrunkID: "shuqi", DialedCallee: "708915000000001", CallerID: "BD1234", + AnswerTimeout: time.Second, MediaPayloadType: 118, + }) + if err != nil { + t.Fatal(err) + } + if call.Media == nil || client.channels.issued != 1 || client.channels.originate.Endpoint != "PJSIP/708915000000001@shuqi" || + client.channels.originate.CallerID != "BD1234" || client.channels.originate.ChannelID != "exec-1" || + client.channels.media.Format != "slin16" || client.channels.media.App != "go-sip-agent" || + len(client.bridges.attached) != 2 || client.bridges.attached[0] != "exec-1" { + t.Fatal("native channel/ExternalMedia was not attached exactly once to the approved bridge") + } + if host, _, err := net.SplitHostPort(client.channels.media.ExternalHost); err != nil || host != "127.0.0.1" { + t.Fatal("native RTP capture must bind the local Cell only") + } + if err := call.Close(); err != nil || len(client.channels.hungup) != 2 || client.bridges.deleted != 1 || client.closed != 1 { + t.Fatalf("native call resources must be closed exactly once: err=%v", err) + } +} + +func TestNativeCallStopsMediaOnMatchingCarrierHangup(t *testing.T) { + events := make(chan ari.Event, 3) + client := &testARIClient{channels: &testChannels{events: events, mediaPeerIP: "127.0.0.1"}, bridges: &testBridges{}, bus: &testBus{sub: answerEvents{events: events}}} + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + call, err := dialWithClient(ctx, client, NativeDial{ExecutionID: "exec-1", TrunkID: "shuqi", DialedCallee: "708915000000001", CallerID: "BD1234", AnswerTimeout: time.Second, MediaPayloadType: 118}) + if err != nil { + t.Fatal(err) + } + defer call.Close() + events <- &ari.ChannelDestroyed{EventData: ari.EventData{Type: "ChannelDestroyed"}, Channel: ari.ChannelData{ID: "someone-else"}, Cause: 16} + select { + case <-call.Context.Done(): + t.Fatal("another caller cannot cancel this approved call") + case <-time.After(10 * time.Millisecond): + } + events <- &ari.ChannelDestroyed{EventData: ari.EventData{Type: "ChannelDestroyed"}, Channel: ari.ChannelData{ID: "exec-1"}, Cause: 16} + select { + case <-call.Context.Done(): + if cause := context.Cause(call.Context); cause == nil || !strings.Contains(cause.Error(), "channel ended") { + t.Fatalf("real call must expose carrier hangup cause, got %v", cause) + } + case <-time.After(100 * time.Millisecond): + t.Fatal("carrier hangup left real AI/media running") + } + if err := call.Close(); err != nil || len(client.channels.hungup) != 1 || client.channels.hungup[0] != "exec-1-media" { + t.Fatalf("destroyed carrier channel must not be hung up again: hungup=%v err=%v", client.channels.hungup, err) + } +} + +func TestNativeCallUnknownOriginationMustAttemptHangupWithoutRetry(t *testing.T) { + client := &testARIClient{ + channels: &testChannels{originateErr: errors.New("connection lost during originate")}, + bridges: &testBridges{}, + bus: &testBus{sub: answerEvents{events: make(chan ari.Event)}}, + } + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + if _, err := dialWithClient(ctx, client, NativeDial{ExecutionID: "exec-1", TrunkID: "shuqi", DialedCallee: "708915000000001", CallerID: "BD1234", AnswerTimeout: time.Second, MediaPayloadType: 118}); err == nil || !strings.Contains(err.Error(), "outcome unknown") { + t.Fatalf("unknown originate outcome must never be reported as a completed call: %v", err) + } + if client.channels.issued != 1 || len(client.channels.hungup) != 1 || client.channels.hungup[0] != "exec-1" || client.closed != 1 { + t.Fatal("unknown originate outcome must not dial twice and must attempt to stop the issued channel") + } +} + +func TestNativeCallUnknownBridgeOrMediaCreationCleansKnownIdentities(t *testing.T) { + for _, step := range []string{"bridge", "external-media"} { + t.Run(step, func(t *testing.T) { + events := make(chan ari.Event, 1) + channels := &testChannels{events: events, mediaPeerIP: "127.0.0.1"} + bridges := &testBridges{} + if step == "bridge" { + bridges.createErr = errors.New("lost bridge response") + } else { + channels.externalErr = errors.New("lost external-media response") + } + client := &testARIClient{channels: channels, bridges: bridges, bus: &testBus{sub: answerEvents{events: events}}} + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + if _, err := dialWithClient(ctx, client, NativeDial{ExecutionID: "exec-1", TrunkID: "shuqi", DialedCallee: "708915000000001", CallerID: "BD1234", AnswerTimeout: time.Second, MediaPayloadType: 118}); err == nil { + t.Fatal("unknown bridge/media outcome cannot be treated as a live audio session") + } + wantHungup := 1 + if step == "external-media" { + wantHungup = 2 + } + if channels.issued != 1 || len(channels.hungup) != wantHungup || bridges.deleted != 1 || client.closed != 1 { + t.Fatalf("known ARI identities must be cleaned without a second dial: hungup=%v deleted=%d closed=%d", channels.hungup, bridges.deleted, client.closed) + } + }) + } +} + +func TestNativeCallAnswerTimeoutNeverCreatesFakeMedia(t *testing.T) { + client := &testARIClient{ + channels: &testChannels{mediaPeerIP: "127.0.0.1"}, + bridges: &testBridges{}, + bus: &testBus{sub: answerEvents{events: make(chan ari.Event)}}, + } + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Millisecond) + defer cancel() + if _, err := dialWithClient(ctx, client, NativeDial{ExecutionID: "exec-1", TrunkID: "shuqi", DialedCallee: "708915000000001", CallerID: "BD1234", AnswerTimeout: time.Second, MediaPayloadType: 118}); err == nil || !strings.Contains(err.Error(), "deadline") { + t.Fatalf("unanswered channel must fail rather than synthesize media: %v", err) + } + if client.channels.issued != 1 || len(client.channels.hungup) == 0 || client.closed != 1 { + t.Fatal("timed-out real channel cannot leak or be originated a second time") + } +}