X-Git-Url: https://gerrit.o-ran-sc.org/r/gitweb?a=blobdiff_plain;f=dmaap-mediator-producer%2Finternal%2Fjobs%2Fjobs.go;h=eec59c39046175ed569bf1df2609a28b28c640c6;hb=280385634f160bb78a8944998e9106a2f6549eb0;hp=10eaf680ff3b4c1050c72d606d39cb07ee7c870d;hpb=f93e105a428fcf22795effb23ac17f1472ba5f7e;p=nonrtric.git

diff --git a/dmaap-mediator-producer/internal/jobs/jobs.go b/dmaap-mediator-producer/internal/jobs/jobs.go
index 10eaf680..eec59c39 100644
--- a/dmaap-mediator-producer/internal/jobs/jobs.go
+++ b/dmaap-mediator-producer/internal/jobs/jobs.go
@@ -21,14 +21,22 @@
 package jobs
 
 import (
+	"encoding/json"
+	"fmt"
 	"os"
 	"path/filepath"
 	"strings"
+	"sync"
+
+	log "github.com/sirupsen/logrus"
+	"oransc.org/nonrtric/dmaapmediatorproducer/internal/restclient"
 )
 
 type Type struct {
-	TypeId string
-	Schema string
+	TypeId     string `json:"id"`
+	DMaaPTopic string `json:"dmaapTopic"`
+	Schema     string `json:"schema"`
+	Jobs       map[string]JobInfo
 }
 
 type JobInfo struct {
@@ -45,9 +53,10 @@ type JobHandler interface {
 }
 
 var (
+	mu      sync.Mutex
 	typeDir = "configs"
 	Handler JobHandler
-	allJobs = make(map[string]map[string]JobInfo)
+	allJobs = make(map[string]Type)
 )
 
 func init() {
@@ -61,17 +70,34 @@ func newJobHandlerImpl() *jobHandlerImpl {
 }
 
 func (jh *jobHandlerImpl) AddJob(ji JobInfo) error {
-	if jobs, ok := allJobs[ji.InfoTypeIdentity]; ok {
-		if _, ok := jobs[ji.InfoJobIdentity]; ok {
-			// TODO: Update job
-		} else {
-			jobs[ji.InfoJobIdentity] = ji
-		}
+	mu.Lock()
+	defer mu.Unlock()
+	if err := validateJobInfo(ji); err == nil {
+		jobs := allJobs[ji.InfoTypeIdentity].Jobs
+		jobs[ji.InfoJobIdentity] = ji
+		return nil
+	} else {
+		return err
+	}
+}
+
+func validateJobInfo(ji JobInfo) error {
+	if _, ok := allJobs[ji.InfoTypeIdentity]; !ok {
+		return fmt.Errorf("type not supported: %v", ji.InfoTypeIdentity)
+	}
+	if ji.InfoJobIdentity == "" {
+		return fmt.Errorf("missing required job identity: %v", ji)
+	}
+	// Temporary for when there are only REST callbacks needed
+	if ji.TargetUri == "" {
+		return fmt.Errorf("missing required target URI: %v", ji)
 	}
 	return nil
 }
 
 func GetTypes() ([]*Type, error) {
+	mu.Lock()
+	defer mu.Unlock()
 	types := make([]*Type, 0, 1)
 	err := filepath.Walk(typeDir,
 		func(path string, info os.FileInfo, err error) error {
@@ -92,6 +118,8 @@ func GetTypes() ([]*Type, error) {
 }
 
 func GetSupportedTypes() []string {
+	mu.Lock()
+	defer mu.Unlock()
 	supportedTypes := []string{}
 	for k := range allJobs {
 		supportedTypes = append(supportedTypes, k)
@@ -104,19 +132,63 @@ func AddJob(job JobInfo) error {
 }
 
 func getType(path string) (*Type, error) {
-	fileName := filepath.Base(path)
-	typeName := strings.TrimSuffix(fileName, filepath.Ext(fileName))
-
-	if typeSchema, err := os.ReadFile(path); err == nil {
-		typeInfo := Type{
-			TypeId: typeName,
-			Schema: string(typeSchema),
-		}
-		if _, ok := allJobs[typeName]; !ok {
-			allJobs[typeName] = make(map[string]JobInfo)
+	if typeDefinition, err := os.ReadFile(path); err == nil {
+		var dat map[string]interface{}
+		if marshalError := json.Unmarshal(typeDefinition, &dat); marshalError == nil {
+			schema, _ := json.Marshal(dat["schema"])
+			typeInfo := Type{
+				TypeId:     dat["id"].(string),
+				DMaaPTopic: dat["dmaapTopic"].(string),
+				Schema:     string(schema),
+				Jobs:       make(map[string]JobInfo),
+			}
+			if _, ok := allJobs[typeInfo.TypeId]; !ok {
+				allJobs[typeInfo.TypeId] = typeInfo
+			}
+			return &typeInfo, nil
+		} else {
+			return nil, marshalError
 		}
-		return &typeInfo, nil
 	} else {
 		return nil, err
 	}
 }
+
+func RunJobs(mRAddress string) {
+	for {
+		pollAndDistributeMessages(mRAddress)
+	}
+}
+
+func pollAndDistributeMessages(mRAddress string) {
+	for typeId, typeInfo := range allJobs {
+		log.Debugf("Processing jobs for type: %v", typeId)
+		messagesBody, error := restclient.Get(fmt.Sprintf("%v/events/%v/users/dmaapmediatorproducer", mRAddress, typeInfo.DMaaPTopic))
+		if error != nil {
+			log.Warnf("Error getting data from MR. Cause: %v", error)
+			continue
+		}
+		distributeMessages(messagesBody, typeInfo)
+	}
+}
+
+func distributeMessages(messages []byte, typeInfo Type) {
+	if len(messages) > 2 {
+		mu.Lock()
+		for _, jobInfo := range typeInfo.Jobs {
+			go sendMessagesToConsumer(messages, jobInfo)
+		}
+		mu.Unlock()
+	}
+}
+
+func sendMessagesToConsumer(messages []byte, jobInfo JobInfo) {
+	log.Debugf("Processing job: %v", jobInfo.InfoJobIdentity)
+	if postErr := restclient.Post(jobInfo.TargetUri, messages); postErr != nil {
+		log.Warnf("Error posting data for job: %v. Cause: %v", jobInfo, postErr)
+	}
+}
+
+func clearAll() {
+	allJobs = make(map[string]Type)
+}