From 243d8fafe94f71489a428629c6c678cbbc3e3eca Mon Sep 17 00:00:00 2001
From: DanB <danb@cgrates.org>
Date: Mon, 24 Sep 2018 16:18:19 +0200
Subject: [PATCH] Diameter improved filter selector, writeOnConn method

---
 agents/dmtagent.go |  8 ++++++--
 agents/libdmt.go   | 20 ++++++++++++++------
 2 files changed, 20 insertions(+), 8 deletions(-)

diff --git a/agents/dmtagent.go b/agents/dmtagent.go
index 5d43a0e74..0422ed3a3 100644
--- a/agents/dmtagent.go
+++ b/agents/dmtagent.go
@@ -86,14 +86,16 @@ func (da *DiameterAgent) handleMessage(c diam.Conn, m *diam.Message) {
 	if err != nil {
 		utils.Logger.Err(fmt.Sprintf("<%s> decoding app: %d, err: %s",
 			utils.DiameterAgent, m.Header.ApplicationID, err.Error()))
+		writeOnConn(c, m.Answer(diam.NoCommonApplication))
 		return
 	}
 	dCmd, err := m.Dictionary().FindCommand(
 		m.Header.ApplicationID,
 		m.Header.CommandCode)
 	if err != nil {
-		utils.Logger.Err(fmt.Sprintf("<%s> decoding app: %d, command %d, err: %s",
+		utils.Logger.Warning(fmt.Sprintf("<%s> decoding app: %d, command %d, err: %s",
 			utils.DiameterAgent, m.Header.ApplicationID, m.Header.CommandCode, err.Error()))
+		writeOnConn(c, m.Answer(diam.CommandUnsupported))
 		return
 	}
 	reqVars := map[string]interface{}{
@@ -122,11 +124,13 @@ func (da *DiameterAgent) handleMessage(c diam.Conn, m *diam.Message) {
 		utils.Logger.Warning(
 			fmt.Sprintf("<%s> error: %s processing message: %s",
 				utils.DiameterAgent, err.Error(), m))
-		return // FixMe with returning some error on HTTP level
+		writeOnConn(c, m.Answer(diam.UnableToDeliver))
+		return
 	} else if !processed {
 		utils.Logger.Warning(
 			fmt.Sprintf("<%s> no request processor enabled, ignoring message %s from %s",
 				utils.DiameterAgent, m, c.RemoteAddr()))
+		writeOnConn(c, m.Answer(diam.UnableToDeliver))
 		return
 	}
 }
diff --git a/agents/libdmt.go b/agents/libdmt.go
index 1bfd9baeb..df17f61b6 100644
--- a/agents/libdmt.go
+++ b/agents/libdmt.go
@@ -109,6 +109,14 @@ func diamAVPAsIface(dAVP *diam.AVP) (val interface{}, err error) {
 	}
 }
 
+// writeOnConn writes the message on connection, logs failures
+func writeOnConn(c diam.Conn, m *diam.Message) {
+	if _, err := m.WriteTo(c); err != nil {
+		utils.Logger.Warning(fmt.Sprintf("<%s> failed writing message to %s, err: %s, msg: %s",
+			utils.DiameterAgent, c.RemoteAddr(), err.Error(), m))
+	}
+}
+
 // newDADataProvider constructs a DataProvider for a diameter message
 func newDADataProvider(m *diam.Message) config.DataProvider {
 	return &diameterDP{m: m, cache: config.NewNavigableMap(nil)}
@@ -176,8 +184,8 @@ func (dP *diameterDP) FieldAsInterface(fldPath []string) (data interface{}, err
 	slectedIdx := 0 // by default we select AVP[0]
 	if slctrStr != "" {
 		if slectedIdx, err = strconv.Atoi(slctrStr); err != nil { // not int, compile it as RSRParser
-			selIndxs := make(map[int][]struct{}) // use it to find intersection of all matched filters
-			slctrStrs := strings.Split(slctrStr, "|")
+			selIndxs := make(map[int]int) // use it to find intersection of all matched filters
+			slctrStrs := strings.Split(slctrStr, utils.PipeSep)
 			for _, slctrStr := range slctrStrs {
 				slctr, err := config.NewRSRParser(slctrStr, true)
 				if err != nil {
@@ -199,15 +207,15 @@ func (dP *diameterDP) FieldAsInterface(fldPath []string) (data interface{}, err
 						}
 						continue // filter not passing, not really error
 					} else {
-						selIndxs[k] = append(selIndxs[k], struct{}{}) // filter passing, index it
+						selIndxs[k+1] += 1 // filter passing, index it with one higher to cover 0
 					}
 				}
 			}
 			var oneMatches bool
-			for idx, matches := range selIndxs {
-				if len(matches) == len(slctrStrs) { // all filters in selection matching
+			for k, matches := range selIndxs {
+				if matches == len(slctrStrs) { // all filters in selection matching
 					oneMatches = true
-					slectedIdx = idx
+					slectedIdx = k - 1 // decrease it to reflect real index
 					break
 				}
 			}