feat(agent): originate one native call with bounded media and cleanup
This commit is contained in:
@@ -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
|
||||
}
|
||||
|
||||
@@ -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)}
|
||||
|
||||
@@ -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
|
||||
}
|
||||
@@ -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")
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user