uppal/cgrates
1
0
Fork 1
mirror of https://github.com/cgrates/cgrates.git synced 2026-02-11 10:06:24 +05:00
Code Issues Packages Projects Releases Wiki Activity

modified astagent call methods over http

Browse Source
This commit is contained in:
gezimbll
2025-11-17 11:15:19 +01:00
committed by Dan Christian Bogos
parent 20a8c98999
commit 4bd459b486
14 changed files with 80 additions and 57 deletions
Download Patch File Download Diff File Expand all files Collapse all files

56
agents/astagent.go
View File

@@ -21,7 +21,6 @@ package agents
import (
"encoding/json"
"fmt"
"net/url"
"strconv"
"strings"
"sync"
@@ -74,12 +73,17 @@ func NewAsteriskAgent(cgrCfg *config.CGRConfig, astConnIdx int,
return sma, nil
}
// ARIConnector abstracts the transport layer (HTTP or WebSocket) for sending ARI commands.
type ARIConnector interface {
Call(method, uri string, queryStr map[string]string, bodyParams map[string]string) (aringo.RESTResponse, error)
}
// AsteriskAgent used to cominicate with asterisk
type AsteriskAgent struct {
cgrCfg *config.CGRConfig // Separate from smCfg since there can be multiple
connMgr *engine.ConnManager
astConnIdx int
astConn *aringo.ARInGO
astConn ARIConnector
astEvChan chan map[string]any
astErrChan chan error
eventsCache map[string]*utils.CGREvent // used to gather information about events during various phases
@@ -91,11 +95,19 @@ func (sma *AsteriskAgent) connectAsterisk(stopChan <-chan struct{}) (err error)
connCfg := sma.cgrCfg.AsteriskAgentCfg().AsteriskConns[sma.astConnIdx]
sma.astEvChan = make(chan map[string]any)
sma.astErrChan = make(chan error)
sma.astConn, err = aringo.NewARInGO(fmt.Sprintf("ws://%s/ari/events?api_key=%s:%s&app=%s",
connCfg.Address, connCfg.User, connCfg.Password, CGRAuthAPP), "http://cgrates.org",
connCfg.User, connCfg.Password, fmt.Sprintf("%s@%s", utils.CGRateS, utils.Version),
sma.astEvChan, sma.astErrChan, stopChan, connCfg.ConnectAttempts, connCfg.Reconnects,
connCfg.MaxReconnectInterval, utils.FibDuration)
if connCfg.AriWebSocket {
sma.astConn, err = aringo.NewARInGO(fmt.Sprintf("ws://%s/ari/events?api_key=%s:%s&app=%s",
connCfg.Address, connCfg.User, connCfg.Password, CGRAuthAPP), "http://cgrates.org",
connCfg.User, connCfg.Password, fmt.Sprintf("%s@%s", utils.CGRateS, utils.Version),
sma.astEvChan, sma.astErrChan, stopChan, connCfg.ConnectAttempts, connCfg.Reconnects,
connCfg.MaxReconnectInterval, utils.FibDuration)
} else {
sma.astConn, err = aringo.NewARInGOV1(fmt.Sprintf("ws://%s/ari/events?api_key=%s:%s&app=%s",
connCfg.Address, connCfg.User, connCfg.Password, CGRAuthAPP), "http://cgrates.org",
connCfg.User, connCfg.Password, connCfg.Address, fmt.Sprintf("%s@%s", utils.CGRateS, utils.Version),
sma.astEvChan, sma.astErrChan, stopChan, connCfg.ConnectAttempts, connCfg.Reconnects,
connCfg.MaxReconnectInterval, utils.FibDuration)
}
return
}
@@ -136,10 +148,8 @@ func (sma *AsteriskAgent) ListenAndServe(stopChan <-chan struct{}) (err error) {
// setChannelVar will set the value of a variable
func (sma *AsteriskAgent) setChannelVar(chanID string, vrblName, vrblVal string) (success bool) {
if _, err := sma.astConn.Call(aringo.HTTP_POST,
fmt.Sprintf("http://%s/ari/channels/%s/variable?variable=%s&value=%s", // Asterisk having issue with variable terminating empty so harcoding param in url
sma.cgrCfg.AsteriskAgentCfg().AsteriskConns[sma.astConnIdx].Address,
chanID, vrblName, vrblVal),
nil); err != nil {
fmt.Sprintf("channels/%s/variable", chanID), // Asterisk having issue with variable terminating empty so harcoding param in url
map[string]string{"variable": vrblName, "value": vrblVal}, nil); err != nil {
// Since we got error, disconnect channel
sma.hangupChannel(chanID,
fmt.Sprintf("<%s> error: <%s> setting <%s> for channelID: <%s>",
@@ -154,9 +164,8 @@ func (sma *AsteriskAgent) hangupChannel(channelID, warnMsg string) {
if warnMsg != "" {
utils.Logger.Warning(warnMsg)
}
if _, err := sma.astConn.Call(aringo.HTTP_DELETE, fmt.Sprintf("http://%s/ari/channels/%s",
sma.cgrCfg.AsteriskAgentCfg().AsteriskConns[sma.astConnIdx].Address, channelID),
url.Values{"reason": {"congestion"}}); err != nil {
if _, err := sma.astConn.Call(aringo.HTTP_DELETE, fmt.Sprintf("channels/%s", channelID), nil,
map[string]string{"reason": "congestion"}); err != nil {
utils.Logger.Warning(
fmt.Sprintf("<%s> failed disconnecting channel <%s>, err: %s",
utils.AsteriskAgent, channelID, err.Error()))
@@ -166,9 +175,7 @@ func (sma *AsteriskAgent) hangupChannel(channelID, warnMsg string) {
func (sma *AsteriskAgent) handleStasisStart(ev *SMAsteriskEvent) {
// Subscribe for channel updates even after we leave Stasis
if _, err := sma.astConn.Call(aringo.HTTP_POST,
fmt.Sprintf("http://%s/ari/applications/%s/subscription?eventSource=channel:%s",
sma.cgrCfg.AsteriskAgentCfg().AsteriskConns[sma.astConnIdx].Address,
CGRAuthAPP, ev.ChannelID()), nil); err != nil {
fmt.Sprintf("applications/%s/subscription", CGRAuthAPP), map[string]string{"eventSource": fmt.Sprintf("channel:%s", ev.ChannelID())}, nil); err != nil {
// Since we got error, disconnect channel
sma.hangupChannel(ev.ChannelID(),
fmt.Sprintf("<%s> error: %s subscribing for channelID: %s",
@@ -251,9 +258,8 @@ func (sma *AsteriskAgent) handleStasisStart(ev *SMAsteriskEvent) {
// Exit channel from stasis
if _, err := sma.astConn.Call(
aringo.HTTP_POST,
fmt.Sprintf("http://%s/ari/channels/%s/continue",
sma.cgrCfg.AsteriskAgentCfg().AsteriskConns[sma.astConnIdx].Address,
ev.ChannelID()), nil); err != nil {
fmt.Sprintf("channels/%s/continue",
ev.ChannelID()), nil, nil); err != nil {
}
// Done with processing event, cache it for later use
sma.evCacheMux.Lock()
@@ -396,13 +402,13 @@ func (sma *AsteriskAgent) V1DisconnectSession(ctx *context.Context, cgrEv utils.
func (sma *AsteriskAgent) V1GetActiveSessionIDs(ctx *context.Context, ignParam string,
sessionIDs *[]*sessions.SessionID) error {
var slMpIface []map[string]any // decode the result from ari into a slice of map[string]any
if byts, err := sma.astConn.Call(
restResp, err := sma.astConn.Call(
aringo.HTTP_GET,
fmt.Sprintf("http://%s/ari/channels",
sma.cgrCfg.AsteriskAgentCfg().AsteriskConns[sma.astConnIdx].Address),
nil); err != nil {
"channels", nil, nil)
if err != nil {
return err
} else if err := json.Unmarshal(byts, &slMpIface); err != nil {
}
if err := json.Unmarshal([]byte(restResp.MessageBody), &slMpIface); err != nil {
return err
}
var sIDs []*sessions.SessionID