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

Updated Freeswitch agent connection indexing

Browse Source
This commit is contained in:
Tripon Alexandru-Ionut
2019-06-05 17:11:22 +03:00
committed by Dan Christian Bogos
parent f89b55dad8
commit 430010eb68
4 changed files with 142 additions and 115 deletions
Download Patch File Download Diff File Expand all files Collapse all files

240
agents/fsagent.go
View File

@@ -33,11 +33,6 @@ import (
"github.com/cgrates/rpcclient"
)
type fsSockWithConfig struct {
fsSock *fsock.FSock
cfg *config.FsConnCfg
}
func NewFSsessions(fsAgentConfig *config.FsAgentCfg,
sS rpcclient.RpcClientConnection, timezone string) (fsa *FSsessions) {
if sS != nil && reflect.ValueOf(sS).IsNil() {
@@ -45,8 +40,8 @@ func NewFSsessions(fsAgentConfig *config.FsAgentCfg,
}
fsa = &FSsessions{
cfg: fsAgentConfig,
conns: make(map[string]*fsSockWithConfig),
senderPools: make(map[string]*fsock.FSockPool),
conns: make([]*fsock.FSock, len(fsAgentConfig.EventSocketConns)),
senderPools: make([]*fsock.FSockPool, len(fsAgentConfig.EventSocketConns)),
sS: sS,
timezone: timezone,
}
@@ -57,106 +52,110 @@ func NewFSsessions(fsAgentConfig *config.FsAgentCfg,
// and the active sessions
type FSsessions struct {
cfg *config.FsAgentCfg
conns map[string]*fsSockWithConfig // Keep the list here for connection management purposes
senderPools map[string]*fsock.FSockPool // Keep sender pools here
conns []*fsock.FSock // Keep the list here for connection management purposes
senderPools []*fsock.FSockPool // Keep sender pools here
sS rpcclient.RpcClientConnection // Connection towards CGR-SessionS component
timezone string
}
func (sm *FSsessions) createHandlers() map[string][]func(string, string) {
ca := func(body, connId string) {
func (sm *FSsessions) createHandlers() map[string][]func(string, int) {
ca := func(body string, connIdx int) {
sm.onChannelAnswer(
NewFSEvent(body), connId)
NewFSEvent(body), connIdx)
}
ch := func(body, connId string) {
ch := func(body string, connIdx int) {
sm.onChannelHangupComplete(
NewFSEvent(body), connId)
NewFSEvent(body), connIdx)
}
handlers := map[string][]func(string, string){
handlers := map[string][]func(string, int){
"CHANNEL_ANSWER": {ca},
"CHANNEL_HANGUP_COMPLETE": {ch},
}
if sm.cfg.SubscribePark {
cp := func(body, connId string) {
cp := func(body string, connIdx int) {
sm.onChannelPark(
NewFSEvent(body), connId)
NewFSEvent(body), connIdx)
}
handlers["CHANNEL_PARK"] = []func(string, string){cp}
handlers["CHANNEL_PARK"] = []func(string, int){cp}
}
return handlers
}
// Sets the call timeout valid of starting of the call
func (sm *FSsessions) setMaxCallDuration(uuid, connId string,
func (sm *FSsessions) setMaxCallDuration(uuid string, connIdx int,
maxDur time.Duration, destNr string) error {
if len(sm.cfg.EmptyBalanceContext) != 0 {
_, err := sm.conns[connId].fsSock.SendApiCmd(
_, err := sm.conns[connIdx].SendApiCmd(
fmt.Sprintf("uuid_setvar %s execute_on_answer sched_transfer +%d %s XML %s\n\n",
uuid, int(maxDur.Seconds()), destNr, sm.cfg.EmptyBalanceContext))
if err != nil {
utils.Logger.Err(
fmt.Sprintf("<%s> Could not transfer the call to empty balance context, error: <%s>, connId: %s",
utils.FreeSWITCHAgent, err.Error(), connId))
fmt.Sprintf("<%s> Could not transfer the call to empty balance context, error: <%s>, connIdx: %v",
utils.FreeSWITCHAgent, err.Error(), connIdx))
return err
}
return nil
} else if len(sm.cfg.EmptyBalanceAnnFile) != 0 {
if _, err := sm.conns[connId].fsSock.SendApiCmd(
}
if len(sm.cfg.EmptyBalanceAnnFile) != 0 {
if _, err := sm.conns[connIdx].SendApiCmd(
fmt.Sprintf("sched_broadcast +%d %s playback!manager_request::%s aleg\n\n",
int(maxDur.Seconds()), uuid, sm.cfg.EmptyBalanceAnnFile)); err != nil {
utils.Logger.Err(
fmt.Sprintf("<%s> Could not send uuid_broadcast to freeswitch, error: <%s>, connId: %s",
utils.FreeSWITCHAgent, err.Error(), connId))
return err
}
return nil
} else {
_, err := sm.conns[connId].fsSock.SendApiCmd(
fmt.Sprintf("uuid_setvar %s execute_on_answer sched_hangup +%d alloted_timeout\n\n",
uuid, int(maxDur.Seconds())))
if err != nil {
utils.Logger.Err(
fmt.Sprintf("<%s> Could not send sched_hangup command to freeswitch, error: <%s>, connId: %s",
utils.FreeSWITCHAgent, err.Error(), connId))
fmt.Sprintf("<%s> Could not send uuid_broadcast to freeswitch, error: <%s>, connIdx: %v",
utils.FreeSWITCHAgent, err.Error(), connIdx))
return err
}
return nil
}
_, err := sm.conns[connIdx].SendApiCmd(
fmt.Sprintf("uuid_setvar %s execute_on_answer sched_hangup +%d alloted_timeout\n\n",
uuid, int(maxDur.Seconds())))
if err != nil {
utils.Logger.Err(
fmt.Sprintf("<%s> Could not send sched_hangup command to freeswitch, error: <%s>, connIdx: %v",
utils.FreeSWITCHAgent, err.Error(), connIdx))
return err
}
return nil
}
// Sends the transfer command to unpark the call to freeswitch
func (sm *FSsessions) unparkCall(uuid, connId, call_dest_nb, notify string) (err error) {
_, err = sm.conns[connId].fsSock.SendApiCmd(
func (sm *FSsessions) unparkCall(uuid string, connIdx int, call_dest_nb, notify string) (err error) {
_, err = sm.conns[connIdx].SendApiCmd(
fmt.Sprintf("uuid_setvar %s cgr_notify %s\n\n", uuid, notify))
if err != nil {
utils.Logger.Err(
fmt.Sprintf("<%s> Could not send unpark api notification to freeswitch, error: <%s>, connId: %s",
utils.FreeSWITCHAgent, err.Error(), connId))
fmt.Sprintf("<%s> Could not send unpark api notification to freeswitch, error: <%s>, connIdx: %v",
utils.FreeSWITCHAgent, err.Error(), connIdx))
return
}
if _, err = sm.conns[connId].fsSock.SendApiCmd(
if _, err = sm.conns[connIdx].SendApiCmd(
fmt.Sprintf("uuid_transfer %s %s\n\n", uuid, call_dest_nb)); err != nil {
utils.Logger.Err(
fmt.Sprintf("<%s> Could not send unpark api call to freeswitch, error: <%s>, connId: %s",
utils.FreeSWITCHAgent, err.Error(), connId))
fmt.Sprintf("<%s> Could not send unpark api call to freeswitch, error: <%s>, connIdx: %v",
utils.FreeSWITCHAgent, err.Error(), connIdx))
}
return
}
func (sm *FSsessions) onChannelPark(fsev FSEvent, connId string) {
func (sm *FSsessions) onChannelPark(fsev FSEvent, connIdx int) {
if fsev.GetReqType(utils.META_DEFAULT) == utils.META_NONE { // Not for us
return
}
fsev[VarCGROriginHost] = utils.FirstNonEmpty(fsev[VarCGROriginHost], sm.conns[connId].cfg.Alias) // rewrite the OriginHost variable if it is empty
if connIdx >= len(sm.conns) { // protection against index out of range panic
err := fmt.Errorf("Index out of range[0,%v): %v ", len(sm.conns), connIdx)
utils.Logger.Err(fmt.Sprintf("<%s> %s", utils.FreeSWITCHAgent, err.Error()))
return
}
fsev[VarCGROriginHost] = utils.FirstNonEmpty(fsev[VarCGROriginHost], sm.cfg.EventSocketConns[connIdx].Alias) // rewrite the OriginHost variable if it is empty
authArgs := fsev.V1AuthorizeArgs()
authArgs.CGREvent.Event[FsConnID] = connId // Attach the connection ID
authArgs.CGREvent.Event[FsConnID] = connIdx // Attach the connection ID
var authReply sessions.V1AuthorizeReply
if err := sm.sS.Call(utils.SessionSv1AuthorizeEvent, authArgs, &authReply); err != nil {
utils.Logger.Err(
fmt.Sprintf("<%s> Could not authorize event %s, error: %s",
utils.FreeSWITCHAgent, fsev.GetUUID(), err.Error()))
sm.unparkCall(fsev.GetUUID(), connId,
sm.unparkCall(fsev.GetUUID(), connIdx,
fsev.GetCallDestNr(utils.META_DEFAULT), err.Error())
return
}
@@ -165,13 +164,13 @@ func (sm *FSsessions) onChannelPark(fsev FSEvent, connId string) {
if _, has := authReply.Attributes.CGREvent.Event[fldName]; !has {
continue //maybe removed
}
if _, err := sm.conns[connId].fsSock.SendApiCmd(
if _, err := sm.conns[connIdx].SendApiCmd(
fmt.Sprintf("uuid_setvar %s %s %s\n\n", fsev.GetUUID(), fldName,
authReply.Attributes.CGREvent.Event[fldName])); err != nil {
utils.Logger.Info(
fmt.Sprintf("<%s> error %s setting channel variabile: %s",
utils.FreeSWITCHAgent, err.Error(), fldName))
sm.unparkCall(fsev.GetUUID(), connId,
sm.unparkCall(fsev.GetUUID(), connIdx,
fsev.GetCallDestNr(utils.META_DEFAULT), err.Error())
return
}
@@ -180,83 +179,93 @@ func (sm *FSsessions) onChannelPark(fsev FSEvent, connId string) {
if authReply.MaxUsage != nil {
if *authReply.MaxUsage != -1 { // For calls different than unlimited, set limits
if *authReply.MaxUsage == 0 {
sm.unparkCall(fsev.GetUUID(), connId,
sm.unparkCall(fsev.GetUUID(), connIdx,
fsev.GetCallDestNr(utils.META_DEFAULT), utils.ErrInsufficientCredit.Error())
return
}
sm.setMaxCallDuration(fsev.GetUUID(), connId,
sm.setMaxCallDuration(fsev.GetUUID(), connIdx,
*authReply.MaxUsage, fsev.GetCallDestNr(utils.META_DEFAULT))
}
}
if authReply.ResourceAllocation != nil {
if _, err := sm.conns[connId].fsSock.SendApiCmd(fmt.Sprintf("uuid_setvar %s %s %s\n\n",
if _, err := sm.conns[connIdx].SendApiCmd(fmt.Sprintf("uuid_setvar %s %s %s\n\n",
fsev.GetUUID(), CGRResourceAllocation, *authReply.ResourceAllocation)); err != nil {
utils.Logger.Info(
fmt.Sprintf("<%s> error %s setting channel variabile: %s",
utils.FreeSWITCHAgent, err.Error(), CGRResourceAllocation))
sm.unparkCall(fsev.GetUUID(), connId,
sm.unparkCall(fsev.GetUUID(), connIdx,
fsev.GetCallDestNr(utils.META_DEFAULT), err.Error())
return
}
}
if authReply.Suppliers != nil {
fsArray := SliceAsFsArray(authReply.Suppliers.SuppliersWithParams())
if _, err := sm.conns[connId].fsSock.SendApiCmd(fmt.Sprintf("uuid_setvar %s %s %s\n\n",
if _, err := sm.conns[connIdx].SendApiCmd(fmt.Sprintf("uuid_setvar %s %s %s\n\n",
fsev.GetUUID(), utils.CGR_SUPPLIERS, fsArray)); err != nil {
utils.Logger.Info(fmt.Sprintf("<%s> error setting suppliers: %s",
utils.FreeSWITCHAgent, err.Error()))
sm.unparkCall(fsev.GetUUID(), connId, fsev.GetCallDestNr(utils.META_DEFAULT), err.Error())
sm.unparkCall(fsev.GetUUID(), connIdx, fsev.GetCallDestNr(utils.META_DEFAULT), err.Error())
return
}
}
sm.unparkCall(fsev.GetUUID(), connId,
sm.unparkCall(fsev.GetUUID(), connIdx,
fsev.GetCallDestNr(utils.META_DEFAULT), AUTH_OK)
}
func (sm *FSsessions) onChannelAnswer(fsev FSEvent, connId string) {
func (sm *FSsessions) onChannelAnswer(fsev FSEvent, connIdx int) {
if fsev.GetReqType(utils.META_DEFAULT) == utils.META_NONE { // Do not process this request
return
}
_, err := sm.conns[connId].fsSock.SendApiCmd(
if connIdx >= len(sm.conns) { // protection against index out of range panic
err := fmt.Errorf("Index out of range[0,%v): %v ", len(sm.conns), connIdx)
utils.Logger.Err(fmt.Sprintf("<%s> %s", utils.FreeSWITCHAgent, err.Error()))
return
}
_, err := sm.conns[connIdx].SendApiCmd(
fmt.Sprintf("uuid_setvar %s %s %s\n\n", fsev.GetUUID(),
utils.CGROriginHost, utils.FirstNonEmpty(sm.conns[connId].cfg.Alias,
sm.conns[connId].cfg.Address)))
utils.CGROriginHost, utils.FirstNonEmpty(sm.cfg.EventSocketConns[connIdx].Alias,
sm.cfg.EventSocketConns[connIdx].Address)))
if err != nil {
utils.Logger.Err(
fmt.Sprintf("<%s> error %s setting channel variabile: %s",
utils.FreeSWITCHAgent, err.Error(), VarCGROriginHost))
return
}
fsev[VarCGROriginHost] = utils.FirstNonEmpty(fsev[VarCGROriginHost], sm.conns[connId].cfg.Alias) // rewrite the OriginHost variable if it is empty
fsev[VarCGROriginHost] = utils.FirstNonEmpty(fsev[VarCGROriginHost], sm.cfg.EventSocketConns[connIdx].Alias) // rewrite the OriginHost variable if it is empty
chanUUID := fsev.GetUUID()
if missing := fsev.MissingParameter(sm.timezone); missing != "" {
sm.disconnectSession(connId, chanUUID, "",
sm.disconnectSession(connIdx, chanUUID, "",
utils.NewErrMandatoryIeMissing(missing).Error())
return
}
initSessionArgs := fsev.V1InitSessionArgs()
initSessionArgs.CGREvent.Event[FsConnID] = connId // Attach the connection ID so we can properly disconnect later
initSessionArgs.CGREvent.Event[FsConnID] = connIdx // Attach the connection ID so we can properly disconnect later
var initReply sessions.V1InitSessionReply
if err := sm.sS.Call(utils.SessionSv1InitiateSession,
initSessionArgs, &initReply); err != nil {
utils.Logger.Err(
fmt.Sprintf("<%s> could not process answer for event %s, error: %s",
utils.FreeSWITCHAgent, chanUUID, err.Error()))
sm.disconnectSession(connId, chanUUID, "", err.Error())
sm.disconnectSession(connIdx, chanUUID, "", err.Error())
return
}
}
func (sm *FSsessions) onChannelHangupComplete(fsev FSEvent, connId string) {
func (sm *FSsessions) onChannelHangupComplete(fsev FSEvent, connIdx int) {
if fsev.GetReqType(utils.META_DEFAULT) == utils.META_NONE { // Do not process this request
return
}
if connIdx >= len(sm.conns) { // protection against index out of range panic
err := fmt.Errorf("Index out of range[0,%v): %v ", len(sm.conns), connIdx)
utils.Logger.Err(fmt.Sprintf("<%s> %s", utils.FreeSWITCHAgent, err.Error()))
return
}
var reply string
fsev[VarCGROriginHost] = utils.FirstNonEmpty(fsev[VarCGROriginHost], sm.conns[connId].cfg.Alias) // rewrite the OriginHost variable if it is empty
if fsev[VarAnswerEpoch] != "0" { // call was answered
fsev[VarCGROriginHost] = utils.FirstNonEmpty(fsev[VarCGROriginHost], sm.cfg.EventSocketConns[connIdx].Alias) // rewrite the OriginHost variable if it is empty
if fsev[VarAnswerEpoch] != "0" { // call was answered
terminateSessionArgs := fsev.V1TerminateSessionArgs()
terminateSessionArgs.CGREvent.Event[FsConnID] = connId // Attach the connection ID in case we need to create a session and disconnect it
terminateSessionArgs.CGREvent.Event[FsConnID] = connIdx // Attach the connection ID in case we need to create a session and disconnect it
if err := sm.sS.Call(utils.SessionSv1TerminateSession,
terminateSessionArgs, &reply); err != nil {
utils.Logger.Err(
@@ -283,34 +292,31 @@ func (sm *FSsessions) onChannelHangupComplete(fsev FSEvent, connId string) {
func (sm *FSsessions) Connect() error {
eventFilters := map[string][]string{"Call-Direction": {"inbound"}}
errChan := make(chan error)
for _, connCfg := range sm.cfg.EventSocketConns {
connId := utils.GenUUID()
for connIdx, connCfg := range sm.cfg.EventSocketConns {
fSock, err := fsock.NewFSock(connCfg.Address, connCfg.Password, connCfg.Reconnects,
sm.createHandlers(), eventFilters, utils.Logger.GetSyslog(), connId)
sm.createHandlers(), eventFilters, utils.Logger.GetSyslog(), connIdx)
if err != nil {
return err
} else if !fSock.Connected() {
return errors.New("Could not connect to FreeSWITCH")
} else {
sm.conns[connId] = &fsSockWithConfig{
fsSock: fSock,
cfg: connCfg,
}
}
if !fSock.Connected() {
return errors.New("Could not connect to FreeSWITCH")
}
sm.conns[connIdx] = fSock
utils.Logger.Info(fmt.Sprintf("<%s> successfully connected to FreeSWITCH at: <%s>", utils.FreeSWITCHAgent, connCfg.Address))
go func() { // Start reading in own goroutine, return on error
if err := sm.conns[connId].fsSock.ReadEvents(); err != nil {
if err := sm.conns[connIdx].ReadEvents(); err != nil {
errChan <- err
}
}()
if fsSenderPool, err := fsock.NewFSockPool(5, connCfg.Address, connCfg.Password, 1, sm.cfg.MaxWaitConnection,
make(map[string][]func(string, string)), make(map[string][]string), utils.Logger.GetSyslog(), connId); err != nil {
fsSenderPool, err := fsock.NewFSockPool(5, connCfg.Address, connCfg.Password, 1, sm.cfg.MaxWaitConnection,
make(map[string][]func(string, int)), make(map[string][]string), utils.Logger.GetSyslog(), connIdx)
if err != nil {
return fmt.Errorf("Cannot connect FreeSWITCH senders pool, error: %s", err.Error())
} else if fsSenderPool == nil {
return errors.New("Cannot connect FreeSWITCH senders pool.")
} else {
sm.senderPools[connId] = fsSenderPool
}
if fsSenderPool == nil {
return errors.New("Cannot connect FreeSWITCH senders pool.")
}
sm.senderPools[connIdx] = fsSenderPool
}
err := <-errChan // Will keep the Connect locked until the first error in one of the connections
return err
@@ -318,52 +324,53 @@ func (sm *FSsessions) Connect() error {
// fsev.GetCallDestNr(utils.META_DEFAULT)
// Disconnects a session by sending hangup command to freeswitch
func (sm *FSsessions) disconnectSession(connId, uuid, redirectNr, notify string) error {
if _, err := sm.conns[connId].fsSock.SendApiCmd(
func (sm *FSsessions) disconnectSession(connIdx int, uuid, redirectNr, notify string) error {
if _, err := sm.conns[connIdx].SendApiCmd(
fmt.Sprintf("uuid_setvar %s cgr_notify %s\n\n", uuid, notify)); err != nil {
utils.Logger.Err(
fmt.Sprintf("<%s> error: %s when attempting to disconnect channelID: %s over connID: %s",
utils.FreeSWITCHAgent, err.Error(), uuid, connId))
fmt.Sprintf("<%s> error: %s when attempting to disconnect channelID: %s over connIdx: %v",
utils.FreeSWITCHAgent, err.Error(), uuid, connIdx))
return err
}
if notify == utils.ErrInsufficientCredit.Error() {
if len(sm.cfg.EmptyBalanceContext) != 0 {
if _, err := sm.conns[connId].fsSock.SendApiCmd(fmt.Sprintf("uuid_transfer %s %s XML %s\n\n",
if _, err := sm.conns[connIdx].SendApiCmd(fmt.Sprintf("uuid_transfer %s %s XML %s\n\n",
uuid, redirectNr, sm.cfg.EmptyBalanceContext)); err != nil {
utils.Logger.Err(fmt.Sprintf("<%s> Could not transfer the call to empty balance context, error: <%s>, connId: %s",
utils.FreeSWITCHAgent, err.Error(), connId))
utils.Logger.Err(fmt.Sprintf("<%s> Could not transfer the call to empty balance context, error: <%s>, connIdx: %v",
utils.FreeSWITCHAgent, err.Error(), connIdx))
return err
}
return nil
} else if len(sm.cfg.EmptyBalanceAnnFile) != 0 {
if _, err := sm.conns[connId].fsSock.SendApiCmd(fmt.Sprintf("uuid_broadcast %s playback!manager_request::%s aleg\n\n",
}
if len(sm.cfg.EmptyBalanceAnnFile) != 0 {
if _, err := sm.conns[connIdx].SendApiCmd(fmt.Sprintf("uuid_broadcast %s playback!manager_request::%s aleg\n\n",
uuid, sm.cfg.EmptyBalanceAnnFile)); err != nil {
utils.Logger.Err(fmt.Sprintf("<%s> Could not send uuid_broadcast to freeswitch, error: <%s>, connId: %s",
utils.FreeSWITCHAgent, err.Error(), connId))
utils.Logger.Err(fmt.Sprintf("<%s> Could not send uuid_broadcast to freeswitch, error: <%s>, connIdx: %v",
utils.FreeSWITCHAgent, err.Error(), connIdx))
return err
}
return nil
}
}
if err := sm.conns[connId].fsSock.SendMsgCmd(uuid,
if err := sm.conns[connIdx].SendMsgCmd(uuid,
map[string]string{"call-command": "hangup", "hangup-cause": "MANAGER_REQUEST"}); err != nil {
utils.Logger.Err(
fmt.Sprintf("<%s> Could not send disconect msg to freeswitch, error: <%s>, connId: %s",
utils.FreeSWITCHAgent, err.Error(), connId))
fmt.Sprintf("<%s> Could not send disconect msg to freeswitch, error: <%s>, connIdx: %v",
utils.FreeSWITCHAgent, err.Error(), connIdx))
return err
}
return nil
}
func (sm *FSsessions) Shutdown() (err error) {
for connId, fSockWithCfg := range sm.conns {
if !fSockWithCfg.fsSock.Connected() {
utils.Logger.Err(fmt.Sprintf("<%s> Cannot shutdown sessions, fsock not connected for connection id: %s", utils.FreeSWITCHAgent, connId))
for connIdx, fSock := range sm.conns {
if !fSock.Connected() {
utils.Logger.Err(fmt.Sprintf("<%s> Cannot shutdown sessions, fsock not connected for connection index: %v", utils.FreeSWITCHAgent, connIdx))
continue
}
utils.Logger.Info(fmt.Sprintf("<%s> Shutting down all sessions on connection id: %s", utils.FreeSWITCHAgent, connId))
if _, err = fSockWithCfg.fsSock.SendApiCmd("hupall MANAGER_REQUEST cgr_reqtype *prepaid"); err != nil {
utils.Logger.Err(fmt.Sprintf("<%s> Error on calls shutdown: %s, connection id: %s", utils.FreeSWITCHAgent, err.Error(), connId))
utils.Logger.Info(fmt.Sprintf("<%s> Shutting down all sessions on connection index: %v", utils.FreeSWITCHAgent, connIdx))
if _, err = fSock.SendApiCmd("hupall MANAGER_REQUEST cgr_reqtype *prepaid"); err != nil {
utils.Logger.Err(fmt.Sprintf("<%s> Error on calls shutdown: %s, connection index: %v", utils.FreeSWITCHAgent, err.Error(), connIdx))
}
}
return
@@ -378,14 +385,19 @@ func (sm *FSsessions) Call(serviceMethod string, args interface{}, reply interfa
func (fsa *FSsessions) V1DisconnectSession(args utils.AttrDisconnectSession, reply *string) (err error) {
ev := engine.NewMapEvent(args.EventStart)
channelID := ev.GetStringIgnoreErrors(utils.OriginID)
connID, err := ev.GetString(FsConnID)
connIdx, err := ev.GetInt64(FsConnID)
if err != nil {
utils.Logger.Err(
fmt.Sprintf("<%s> error: <%s:%s> when attempting to disconnect channelID: <%s>",
utils.FreeSWITCHAgent, err.Error(), FsConnID, channelID))
return
}
if err = fsa.disconnectSession(connID, channelID,
if int(connIdx) >= len(fsa.conns) { // protection against index out of range panic
err := fmt.Errorf("Index out of range[0,%v): %v ", len(fsa.conns), connIdx)
utils.Logger.Err(fmt.Sprintf("<%s> %s", utils.FreeSWITCHAgent, err.Error()))
return err
}
if err = fsa.disconnectSession(int(connIdx), channelID,
utils.FirstNonEmpty(ev.GetStringIgnoreErrors(CALL_DEST_NR), ev.GetStringIgnoreErrors(SIP_REQ_USER)),
utils.ErrInsufficientCredit.Error()); err != nil {
return
@@ -397,24 +409,24 @@ func (fsa *FSsessions) V1DisconnectSession(args utils.AttrDisconnectSession, rep
func (fsa *FSsessions) V1GetActiveSessionIDs(ignParam string,
sessionIDs *[]*sessions.SessionID) (err error) {
var sIDs []*sessions.SessionID
for connId, senderPool := range fsa.senderPools {
for connIdx, senderPool := range fsa.senderPools {
fsConn, err := senderPool.PopFSock()
if err != nil {
utils.Logger.Err(fmt.Sprintf("<%s> Error on pop FSock: %s, connection id: %s",
utils.FreeSWITCHAgent, err.Error(), connId))
utils.Logger.Err(fmt.Sprintf("<%s> Error on pop FSock: %s, connection index: %v",
utils.FreeSWITCHAgent, err.Error(), connIdx))
continue
}
activeChanStr, err := fsConn.SendApiCmd("show channels")
senderPool.PushFSock(fsConn)
if err != nil {
utils.Logger.Err(fmt.Sprintf("<%s> Error on push FSock: %s, connection id: %s",
utils.FreeSWITCHAgent, err.Error(), connId))
utils.Logger.Err(fmt.Sprintf("<%s> Error on push FSock: %s, connection index: %v",
utils.FreeSWITCHAgent, err.Error(), connIdx))
continue
}
aChans := fsock.MapChanData(activeChanStr)
for _, fsAChan := range aChans {
sIDs = append(sIDs, &sessions.SessionID{
OriginHost: fsa.conns[connId].cfg.Alias,
OriginHost: fsa.cfg.EventSocketConns[connIdx].Alias,
OriginID: fsAChan["uuid"]},
)
}

8
engine/mapevent.go
View File

@@ -81,6 +81,14 @@ func (me MapEvent) GetString(fldName string) (out string, err error) {
return utils.IfaceAsString(fldIface)
}
func (me MapEvent) GetInt64(fldName string) (out int64, err error) {
fldIface, has := me[fldName]
if !has {
return 0, utils.ErrNotFound
}
return utils.IfaceAsInt64(fldIface)
}
func (me MapEvent) GetStringIgnoreErrors(fldName string) (out string) {
out, _ = me.GetString(fldName)
return

7
engine/safevent.go
View File

@@ -126,6 +126,13 @@ func (se *SafEvent) GetString(fldName string) (out string, err error) {
return
}
func (se SafEvent) GetInt64(fldName string) (out int64, err error) {
se.RLock()
out, err = se.Me.GetInt64(fldName)
se.RUnlock()
return
}
func (se *SafEvent) GetStringIgnoreErrors(fldName string) (out string) {
out, _ = se.GetString(fldName)
return

2
glide.lock generated
View File

@@ -10,7 +10,7 @@ imports:
- name: github.com/cgrates/aringo
version: f996da7890eaec95ba13240253744446e17e6598
- name: github.com/cgrates/fsock
version: bcbd5e75c07dddb12ac86f1f861f2bdddc1d4596
version: 4759d9e84c74981872c5c2bbffe6f23ecba2ea3c
- name: github.com/cgrates/kamevapi
version: 0e0d0379606fd8f12b53c6da6aeb28544f7bfa37
- name: github.com/cgrates/ltcache