diff --git a/agents/janusagent.go b/agents/janusagent.go index 7e748dc8c..ef60e100b 100644 --- a/agents/janusagent.go +++ b/agents/janusagent.go @@ -25,26 +25,38 @@ import ( "github.com/cgrates/cgrates/config" "github.com/cgrates/cgrates/engine" "github.com/cgrates/cgrates/utils" + janus "github.com/cgrates/janusgo" ) // NewJanusAgent will construct a JanusAgent -func NewJanusAgent(connMgr *engine.ConnManager, - sessionConns []string, filterS *engine.FilterS, - reqProcessors []*config.RequestProcessor) *JanusAgent { +func NewJanusAgent(cgrCfg *config.CGRConfig, + connMgr *engine.ConnManager, + filterS *engine.FilterS) *JanusAgent { return &JanusAgent{ - connMgr: connMgr, - filterS: filterS, - reqProcessors: reqProcessors, - sessionConns: sessionConns, + cgrCfg: cgrCfg, + connMgr: connMgr, + filterS: filterS, } } // JanusAgent is a gateway between HTTP and Janus Server over Websocket type JanusAgent struct { - connMgr *engine.ConnManager - filterS *engine.FilterS - reqProcessors []*config.RequestProcessor - sessionConns []string + cgrCfg *config.CGRConfig + connMgr *engine.ConnManager + filterS *engine.FilterS + jnsConn *janus.Gateway +} + +// Connect will create the connection to the Janus Server +func (ja *JanusAgent) Connect() (err error) { + ja.jnsConn, err = janus.Connect( + fmt.Sprintf("ws://%s", ja.cgrCfg.JanusAgentCfg().JanusConns[0].Address)) + return +} + +// Shutdown will close the connection to the Janus Server +func (ja *JanusAgent) Shutdown() error { + return ja.jnsConn.Close() } // ServeHTTP implements http.Handler interface diff --git a/config/config_defaults.go b/config/config_defaults.go index 50ca538bb..59628cb4d 100644 --- a/config/config_defaults.go +++ b/config/config_defaults.go @@ -369,7 +369,7 @@ const CGRATES_CFG_JSON = ` "enabled": false, // starts the EventReader service: "sessions_conns": ["*internal"], // RPC Connections IDs "ees_conns": [], // connection for routing processed and invalid messages through EEs - "concurrent_events":1, // number of events to generate concurrently on CGRateS side + "concurrent_events": 1, // number of events to generate concurrently on CGRateS side "partial_cache_ttl": "1s", // the duration to cache partial records when not pairing "readers": [ { @@ -395,7 +395,7 @@ const CGRATES_CFG_JSON = ` // "partialcsvFieldSeparator": "," // separator used when dumping the fields // FileCSV - "csvRowLength": 0, // Number of fields from csv file + "csvRowLength": 0, // Number of fields from csv file, -1 to disable checking, 0 to inherit the lenght of first record "csvFieldSeparator": ",", // separator used when reading the fields "csvHeaderDefineChar": ":", // the starting character for header definition used in case of CSV files // "csvLazyQuotes": false, // if a quote may appear in an unquoted field and a non-doubled quote may appear in a quoted field diff --git a/go.mod b/go.mod index 81a570481..7c744f176 100644 --- a/go.mod +++ b/go.mod @@ -80,6 +80,7 @@ require ( github.com/blevesearch/zapx/v15 v15.3.13 // indirect github.com/blevesearch/zapx/v16 v16.0.12 // indirect github.com/cenkalti/hub v1.0.1 // indirect + github.com/cgrates/janusgo v0.0.0-20240411150649-85fbf3213b5e // indirect github.com/couchbase/ghistogram v0.1.0 // indirect github.com/couchbase/moss v0.2.0 // indirect github.com/felixge/httpsnoop v1.0.4 // indirect @@ -93,6 +94,7 @@ require ( github.com/google/uuid v1.6.0 // indirect github.com/googleapis/enterprise-certificate-proxy v0.3.2 // indirect github.com/googleapis/gax-go/v2 v2.12.2 // indirect + github.com/gorilla/websocket v1.5.1 // indirect github.com/ishidawataru/sctp v0.0.0-20190922091402-408ec287e38c // indirect github.com/jackc/pgpassfile v1.0.0 // indirect github.com/jackc/pgservicefile v0.0.0-20221227161230-091c0ba34f0a // indirect @@ -108,6 +110,7 @@ require ( github.com/nats-io/nkeys v0.4.5 // indirect github.com/nats-io/nuid v1.0.1 // indirect github.com/pierrec/lz4/v4 v4.1.15 // indirect + github.com/rs/xid v1.5.0 // indirect github.com/xdg-go/pbkdf2 v1.0.0 // indirect github.com/xdg-go/scram v1.1.2 // indirect github.com/xdg-go/stringprep v1.0.4 // indirect diff --git a/go.sum b/go.sum index 7823b065c..f6506bf4b 100644 --- a/go.sum +++ b/go.sum @@ -68,6 +68,8 @@ github.com/cgrates/birpc v1.3.1-0.20211117095917-5b0ff29f3084 h1:YIEepjEOjeHaFre github.com/cgrates/birpc v1.3.1-0.20211117095917-5b0ff29f3084/go.mod h1:z/PmNnDPqSQALedKJv5T8+eXIq6XHa9J0St1YsvAVns= github.com/cgrates/fsock v0.0.0-20240322171959-35309017b3e0 h1:rnSM0tG6Cl8GXjyBQXw78WEaJolKfZqHTXkOCZh+w/k= github.com/cgrates/fsock v0.0.0-20240322171959-35309017b3e0/go.mod h1:bKByLko2HF33K+PbiiToAgevrrbr96C+7Pp3HGS6oag= +github.com/cgrates/janusgo v0.0.0-20240411150649-85fbf3213b5e h1:a0LmV+CiWKoLCyCMwOGSJl5rvs35+mdPbE/3QSh38LI= +github.com/cgrates/janusgo v0.0.0-20240411150649-85fbf3213b5e/go.mod h1:T/qWqdGS2VNOTrXN5PGXymbzPxvEJSZgVlFeSg3B3DY= github.com/cgrates/kamevapi v0.0.0-20240307160311-26273f03eedf h1:GbMJzvtwdX1OCEmsqSts/cRCIcIMvo8AYtC2dQExWlg= github.com/cgrates/kamevapi v0.0.0-20240307160311-26273f03eedf/go.mod h1:oEq/JbubkOD2pXHvDy4r7519NkxriONisrnVpkCaNJw= github.com/cgrates/ltcache v0.0.0-20240411152156-e673692056db h1:JRgzMS5kJ1WxaveoZ1YG/FowUDxFQXD3GjCHR7rH0Gk= @@ -163,6 +165,8 @@ github.com/googleapis/gax-go/v2 v2.12.2 h1:mhN09QQW1jEWeMF74zGR81R30z4VJzjZsfkUh github.com/googleapis/gax-go/v2 v2.12.2/go.mod h1:61M8vcyyXR2kqKFxKrfA22jaA8JGF7Dc8App1U3H6jc= github.com/gorhill/cronexpr v0.0.0-20180427100037-88b0669f7d75 h1:f0n1xnMSmBLzVfsMMvriDyA75NB/oBgILX2GcHXIQzY= github.com/gorhill/cronexpr v0.0.0-20180427100037-88b0669f7d75/go.mod h1:g2644b03hfBX9Ov0ZBDgXXens4rxSxmqFBbhvKv2yVA= +github.com/gorilla/websocket v1.5.1 h1:gmztn0JnHVt9JZquRuzLw3g4wouNVzKL15iLr/zn/QY= +github.com/gorilla/websocket v1.5.1/go.mod h1:x3kM2JMyaluk02fnUJpQuwD2dCS5NDG2ZHL0uE0tcaY= github.com/hpcloud/tail v1.0.0 h1:nfCOvKYfkgYP8hkirhJocXT2+zOD8yUNjXaWfTlyFKI= github.com/hpcloud/tail v1.0.0/go.mod h1:ab1qPbhIpdTxEkNHXyeSf5vhxWSCs/tWer42PpOxQnU= github.com/ishidawataru/sctp v0.0.0-20190922091402-408ec287e38c h1:PwVcPU2rqkJIG0Lz/UGbGcbfi/HhEbOIId+w4xkbGHQ= @@ -224,6 +228,8 @@ github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZN github.com/prometheus/client_model v0.0.0-20190812154241-14fe0d1b01d4/go.mod h1:xMI15A0UPsDsEKsMN9yxemIoYk6Tm2C1GtYGdfGttqA= github.com/rabbitmq/amqp091-go v1.9.0 h1:qrQtyzB4H8BQgEuJwhmVQqVHB9O4+MNDJCCAcpc3Aoo= github.com/rabbitmq/amqp091-go v1.9.0/go.mod h1:+jPrT9iY2eLjRaMSRHUhc3z14E/l85kv/f+6luSD3pc= +github.com/rs/xid v1.5.0 h1:mKX4bl4iPYJtEIxp6CYiUuLQ/8DYMoz0PUdtGgMFRVc= +github.com/rs/xid v1.5.0/go.mod h1:trrq9SKmegXys3aeAKXMUTdJsYXVwGY3RLcfgqegfbg= github.com/segmentio/kafka-go v0.4.44 h1:Vjjksniy0WSTZ7CuVJrz1k04UoZeTc77UV6Yyk6tLY4= github.com/segmentio/kafka-go v0.4.44/go.mod h1:HjF6XbOKh0Pjlkr5GVZxt6CsjjwnmhVOfURM5KMd8qg= github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= diff --git a/services/janusagent.go b/services/janusagent.go index 9e895b04b..8d644d0e2 100644 --- a/services/janusagent.go +++ b/services/janusagent.go @@ -49,6 +49,7 @@ type JanusAgent struct { cfg *config.CGRConfig filterSChan chan *engine.FilterS server *cores.Server + jA *agents.JanusAgent // we can realy stop the JanusAgent so keep a flag // if we registerd the jandlers @@ -59,20 +60,22 @@ type JanusAgent struct { // Start should jandle the sercive start func (ja *JanusAgent) Start() (err error) { - if ja.IsRunning() { - return utils.ErrServiceAlreadyRunning - } - filterS := <-ja.filterSChan ja.filterSChan <- filterS ja.Lock() + if ja.started { + ja.Unlock() + return utils.ErrServiceAlreadyRunning + } + ja.jA = agents.NewJanusAgent(ja.cfg, ja.connMgr, filterS) + if err = ja.jA.Connect(); err != nil { + return + } + ja.server.RegisterHttpHandler(ja.cfg.JanusAgentCfg().URL, ja.jA) ja.started = true - utils.Logger.Info(fmt.Sprintf("<%s> successfully started.", utils.JanusAgent)) - ja.server.RegisterHttpHandler(ja.cfg.JanusAgentCfg().URL, - agents.NewJanusAgent(ja.connMgr, ja.cfg.JanusAgentCfg().SessionSConns, filterS, - ja.cfg.JanusAgentCfg().RequestProcessors)) ja.Unlock() + utils.Logger.Info(fmt.Sprintf("<%s> successfully started.", utils.JanusAgent)) return } @@ -84,6 +87,7 @@ func (ja *JanusAgent) Reload() (err error) { // Shutdown stops the service func (ja *JanusAgent) Shutdown() (err error) { ja.Lock() + err = ja.jA.Shutdown() ja.started = false ja.Unlock() return // no shutdown for the momment