From 430010eb68ab7bd027a9c0b5a3ae941468ddc8a5 Mon Sep 17 00:00:00 2001 From: Tripon Alexandru-Ionut Date: Wed, 5 Jun 2019 17:11:22 +0300 Subject: [PATCH] Updated Freeswitch agent connection indexing --- agents/fsagent.go | 240 ++++++++++++++++++++++++--------------------- engine/mapevent.go | 8 ++ engine/safevent.go | 7 ++ glide.lock | 2 +- 4 files changed, 142 insertions(+), 115 deletions(-) diff --git a/agents/fsagent.go b/agents/fsagent.go index 28b57a890..b05d269d0 100644 --- a/agents/fsagent.go +++ b/agents/fsagent.go @@ -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"]}, ) } diff --git a/engine/mapevent.go b/engine/mapevent.go index df6cf4f14..8ac141452 100644 --- a/engine/mapevent.go +++ b/engine/mapevent.go @@ -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 diff --git a/engine/safevent.go b/engine/safevent.go index a253d60a0..7214cd0bb 100644 --- a/engine/safevent.go +++ b/engine/safevent.go @@ -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 diff --git a/glide.lock b/glide.lock index 5f446a817..717baaa08 100644 --- a/glide.lock +++ b/glide.lock @@ -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