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

Add Post support for HTTP and other types (amqp;sqs;etc...)

Browse Source
This commit is contained in:
TeoV
2020-06-04 16:16:08 +03:00
committed by Dan Christian Bogos
parent d20aecc7d9
commit 1302ffb8fe
9 changed files with 333 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
engine/action.go
View File

@@ -388,7 +388,7 @@ func sendAMQP(ub *Account, a *Action, acs Actions, extraData interface{}) error
}
err = PostersCache.PostAMQP(a.ExtraParameters, config.CgrConfig().GeneralCfg().PosterAttempts, body)
if err != nil && config.CgrConfig().GeneralCfg().FailedPostsDir != utils.META_NONE {
addFailedPost(a.ExtraParameters, utils.MetaAMQPjsonMap, utils.ActionsPoster+utils.HIERARCHY_SEP+a.ActionType, body)
AddFailedPost(a.ExtraParameters, utils.MetaAMQPjsonMap, utils.ActionsPoster+utils.HIERARCHY_SEP+a.ActionType, body)
err = nil
}
return err
@@ -401,7 +401,7 @@ func sendAWS(ub *Account, a *Action, acs Actions, extraData interface{}) error {
}
err = PostersCache.PostAMQPv1(a.ExtraParameters, config.CgrConfig().GeneralCfg().PosterAttempts, body)
if err != nil && config.CgrConfig().GeneralCfg().FailedPostsDir != utils.META_NONE {
addFailedPost(a.ExtraParameters, utils.MetaAMQPV1jsonMap, utils.ActionsPoster+utils.HIERARCHY_SEP+a.ActionType, body)
AddFailedPost(a.ExtraParameters, utils.MetaAMQPV1jsonMap, utils.ActionsPoster+utils.HIERARCHY_SEP+a.ActionType, body)
err = nil
}
return err
@@ -414,7 +414,7 @@ func sendSQS(ub *Account, a *Action, acs Actions, extraData interface{}) error {
}
err = PostersCache.PostSQS(a.ExtraParameters, config.CgrConfig().GeneralCfg().PosterAttempts, body)
if err != nil && config.CgrConfig().GeneralCfg().FailedPostsDir != utils.META_NONE {
addFailedPost(a.ExtraParameters, utils.MetaSQSjsonMap, utils.ActionsPoster+utils.HIERARCHY_SEP+a.ActionType, body)
AddFailedPost(a.ExtraParameters, utils.MetaSQSjsonMap, utils.ActionsPoster+utils.HIERARCHY_SEP+a.ActionType, body)
err = nil
}
return err
@@ -427,7 +427,7 @@ func sendKafka(ub *Account, a *Action, acs Actions, extraData interface{}) error
}
err = PostersCache.PostKafka(a.ExtraParameters, config.CgrConfig().GeneralCfg().PosterAttempts, body, utils.UUIDSha1Prefix())
if err != nil && config.CgrConfig().GeneralCfg().FailedPostsDir != utils.META_NONE {
addFailedPost(a.ExtraParameters, utils.MetaKafkajsonMap, utils.ActionsPoster+utils.HIERARCHY_SEP+a.ActionType, body)
AddFailedPost(a.ExtraParameters, utils.MetaKafkajsonMap, utils.ActionsPoster+utils.HIERARCHY_SEP+a.ActionType, body)
err = nil
}
return err
@@ -440,7 +440,7 @@ func sendS3(ub *Account, a *Action, acs Actions, extraData interface{}) error {
}
err = PostersCache.PostS3(a.ExtraParameters, config.CgrConfig().GeneralCfg().PosterAttempts, body, utils.UUIDSha1Prefix())
if err != nil && config.CgrConfig().GeneralCfg().FailedPostsDir != utils.META_NONE {
addFailedPost(a.ExtraParameters, utils.MetaS3jsonMap, utils.ActionsPoster+utils.HIERARCHY_SEP+a.ActionType, body)
AddFailedPost(a.ExtraParameters, utils.MetaS3jsonMap, utils.ActionsPoster+utils.HIERARCHY_SEP+a.ActionType, body)
err = nil
}
return err
@@ -459,7 +459,7 @@ func callURL(ub *Account, a *Action, acs Actions, extraData interface{}) error {
}
err = pstr.Post(body, utils.EmptyString)
if err != nil && config.CgrConfig().GeneralCfg().FailedPostsDir != utils.META_NONE {
addFailedPost(a.ExtraParameters, utils.MetaHTTPjson, utils.ActionsPoster+utils.HIERARCHY_SEP+a.ActionType, body)
AddFailedPost(a.ExtraParameters, utils.MetaHTTPjson, utils.ActionsPoster+utils.HIERARCHY_SEP+a.ActionType, body)
err = nil
}
return err
@@ -480,7 +480,7 @@ func callURLAsync(ub *Account, a *Action, acs Actions, extraData interface{}) er
go func() {
err := pstr.Post(body, utils.EmptyString)
if err != nil && config.CgrConfig().GeneralCfg().FailedPostsDir != utils.META_NONE {
addFailedPost(a.ExtraParameters, utils.MetaHTTPjson, utils.ActionsPoster+utils.HIERARCHY_SEP+a.ActionType, body)
AddFailedPost(a.ExtraParameters, utils.MetaHTTPjson, utils.ActionsPoster+utils.HIERARCHY_SEP+a.ActionType, body)
}
}()
return nil
@@ -1045,7 +1045,7 @@ func postEvent(ub *Account, a *Action, acs Actions, extraData interface{}) error
}
err = pstr.Post(body, utils.EmptyString)
if err != nil && config.CgrConfig().GeneralCfg().FailedPostsDir != utils.META_NONE {
addFailedPost(a.ExtraParameters, utils.MetaHTTPjson, utils.ActionsPoster+utils.HIERARCHY_SEP+a.ActionType, body)
AddFailedPost(a.ExtraParameters, utils.MetaHTTPjson, utils.ActionsPoster+utils.HIERARCHY_SEP+a.ActionType, body)
err = nil
}
return err

2
engine/cdre.go
View File

@@ -290,7 +290,7 @@ func (cdre *CDRExporter) postCdr(cdr *CDR) (err error) {
err = PostersCache.PostS3(cdre.exportPath, cdre.attempts, body.([]byte), utils.ConcatenatedKey(cdr.CGRID, cdr.RunID))
}
if err != nil && cdre.fallbackPath != utils.META_NONE {
addFailedPost(cdre.exportPath, cdre.exportFormat, utils.CDRPoster, body)
AddFailedPost(cdre.exportPath, cdre.exportFormat, utils.CDRPoster, body)
}
return
}

2
engine/libcdre.go
View File

@@ -58,7 +58,7 @@ func writeFailedPosts(itmID string, value interface{}) {
return
}
func addFailedPost(expPath, format, module string, ev interface{}) {
func AddFailedPost(expPath, format, module string, ev interface{}) {
key := utils.ConcatenatedKey(expPath, format, module)
var failedPost *ExportEvents
if x, ok := failedPostCache.Get(key); ok {

6
engine/libcdre_test.go
View File

@@ -38,7 +38,7 @@ func TestSetFldPostCacheTTL(t *testing.T) {
func TestAddFldPost(t *testing.T) {
SetFailedPostCacheTTL(time.Duration(5 * time.Second))
addFailedPost("path1", "format1", "module1", "1")
AddFailedPost("path1", "format1", "module1", "1")
x, ok := failedPostCache.Get(utils.ConcatenatedKey("path1", "format1", "module1"))
if !ok {
t.Error("Error reading from cache")
@@ -60,8 +60,8 @@ func TestAddFldPost(t *testing.T) {
if !reflect.DeepEqual(eOut, failedPost) {
t.Errorf("Expecting: %+v, received: %+v", utils.ToJSON(eOut), utils.ToJSON(failedPost))
}
addFailedPost("path1", "format1", "module1", "2")
addFailedPost("path2", "format2", "module2", "3")
AddFailedPost("path1", "format1", "module1", "2")
AddFailedPost("path2", "format2", "module2", "3")
x, ok = failedPostCache.Get(utils.ConcatenatedKey("path1", "format1", "module1"))
if !ok {
t.Error("Error reading from cache")

4
engine/poster_it_test.go
View File

@@ -75,7 +75,7 @@ func TestHttpJsonPoster(t *testing.T) {
if err = pstr.Post(jsn, utils.EmptyString); err == nil {
t.Error("Expected error")
}
addFailedPost("http://localhost:8080/invalid", utils.CONTENT_JSON, "test1", jsn)
AddFailedPost("http://localhost:8080/invalid", utils.CONTENT_JSON, "test1", jsn)
time.Sleep(2)
fs, err := filepath.Glob("/tmp/test1*")
if err != nil {
@@ -108,7 +108,7 @@ func TestHttpBytesPoster(t *testing.T) {
if err = pstr.Post(content, utils.EmptyString); err == nil {
t.Error("Expected error")
}
addFailedPost("http://localhost:8080/invalid", utils.CONTENT_JSON, "test2", content)
AddFailedPost("http://localhost:8080/invalid", utils.CONTENT_JSON, "test2", content)
time.Sleep(2)
fs, err := filepath.Glob("/tmp/test2*")
if err != nil {