dmaap-mediator-producer/internal/jobs/jobs.go

   1 // -
   2 //   ========================LICENSE_START=================================
   3 //   O-RAN-SC
   4 //   %%
   5 //   Copyright (C) 2021: Nordix Foundation
   6 //   %%
   7 //   Licensed under the Apache License, Version 2.0 (the "License");
   8 //   you may not use this file except in compliance with the License.
   9 //   You may obtain a copy of the License at
  10 //
  11 //        http://www.apache.org/licenses/LICENSE-2.0
  12 //
  13 //   Unless required by applicable law or agreed to in writing, software
  14 //   distributed under the License is distributed on an "AS IS" BASIS,
  15 //   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  16 //   See the License for the specific language governing permissions and
  17 //   limitations under the License.
  18 //   ========================LICENSE_END===================================
  19 //
  20
  21 package jobs
  22
  23 import (
  24         "encoding/json"
  25         "fmt"
  26         "os"
  27         "sync"
  28
  29         log "github.com/sirupsen/logrus"
  30         "oransc.org/nonrtric/dmaapmediatorproducer/internal/config"
  31         "oransc.org/nonrtric/dmaapmediatorproducer/internal/restclient"
  32 )
  33
  34 type TypeData struct {
  35         TypeId        string `json:"id"`
  36         DMaaPTopicURL string `json:"dmaapTopicUrl"`
  37         Jobs          map[string]JobInfo
  38 }
  39
  40 type JobInfo struct {
  41         Owner            string      `json:"owner"`
  42         LastUpdated      string      `json:"last_updated"`
  43         InfoJobIdentity  string      `json:"info_job_identity"`
  44         TargetUri        string      `json:"target_uri"`
  45         InfoJobData      interface{} `json:"info_job_data"`
  46         InfoTypeIdentity string      `json:"info_type_identity"`
  47 }
  48
  49 type JobTypeHandler interface {
  50         GetTypes() ([]config.TypeDefinition, error)
  51         GetSupportedTypes() []string
  52 }
  53
  54 type JobHandler interface {
  55         AddJob(JobInfo) error
  56         DeleteJob(jobId string)
  57 }
  58
  59 type JobHandlerImpl struct {
  60         mu               sync.Mutex
  61         configFile       string
  62         allTypes         map[string]TypeData
  63         pollClient       restclient.HTTPClient
  64         distributeClient restclient.HTTPClient
  65 }
  66
  67 func NewJobHandlerImpl(typeConfigFilePath string, pollClient restclient.HTTPClient, distributeClient restclient.HTTPClient) *JobHandlerImpl {
  68         return &JobHandlerImpl{
  69                 configFile:       typeConfigFilePath,
  70                 allTypes:         make(map[string]TypeData),
  71                 pollClient:       pollClient,
  72                 distributeClient: distributeClient,
  73         }
  74 }
  75
  76 func (jh *JobHandlerImpl) AddJob(ji JobInfo) error {
  77         jh.mu.Lock()
  78         defer jh.mu.Unlock()
  79         if err := jh.validateJobInfo(ji); err == nil {
  80                 jobs := jh.allTypes[ji.InfoTypeIdentity].Jobs
  81                 jobs[ji.InfoJobIdentity] = ji
  82                 log.Debug("Added job: ", ji)
  83                 return nil
  84         } else {
  85                 return err
  86         }
  87 }
  88
  89 func (jh *JobHandlerImpl) DeleteJob(jobId string) {
  90         jh.mu.Lock()
  91         defer jh.mu.Unlock()
  92         for _, typeData := range jh.allTypes {
  93                 delete(typeData.Jobs, jobId)
  94         }
  95         log.Debug("Deleted job: ", jobId)
  96 }
  97
  98 func (jh *JobHandlerImpl) validateJobInfo(ji JobInfo) error {
  99         if _, ok := jh.allTypes[ji.InfoTypeIdentity]; !ok {
 100                 return fmt.Errorf("type not supported: %v", ji.InfoTypeIdentity)
 101         }
 102         if ji.InfoJobIdentity == "" {
 103                 return fmt.Errorf("missing required job identity: %v", ji)
 104         }
 105         // Temporary for when there are only REST callbacks needed
 106         if ji.TargetUri == "" {
 107                 return fmt.Errorf("missing required target URI: %v", ji)
 108         }
 109         return nil
 110 }
 111
 112 func (jh *JobHandlerImpl) GetTypes() ([]config.TypeDefinition, error) {
 113         jh.mu.Lock()
 114         defer jh.mu.Unlock()
 115         typeDefsByte, err := os.ReadFile(jh.configFile)
 116         if err != nil {
 117                 return nil, err
 118         }
 119         typeDefs := struct {
 120                 Types []config.TypeDefinition `json:"types"`
 121         }{}
 122         err = json.Unmarshal(typeDefsByte, &typeDefs)
 123         if err != nil {
 124                 return nil, err
 125         }
 126         for _, typeDef := range typeDefs.Types {
 127                 jh.allTypes[typeDef.Id] = TypeData{
 128                         TypeId:        typeDef.Id,
 129                         DMaaPTopicURL: typeDef.DmaapTopicURL,
 130                         Jobs:          make(map[string]JobInfo),
 131                 }
 132         }
 133         return typeDefs.Types, nil
 134 }
 135
 136 func (jh *JobHandlerImpl) GetSupportedTypes() []string {
 137         jh.mu.Lock()
 138         defer jh.mu.Unlock()
 139         supportedTypes := []string{}
 140         for k := range jh.allTypes {
 141                 supportedTypes = append(supportedTypes, k)
 142         }
 143         return supportedTypes
 144 }
 145
 146 func (jh *JobHandlerImpl) RunJobs(mRAddress string) {
 147         for {
 148                 jh.pollAndDistributeMessages(mRAddress)
 149         }
 150 }
 151
 152 func (jh *JobHandlerImpl) pollAndDistributeMessages(mRAddress string) {
 153         jh.mu.Lock()
 154         defer jh.mu.Unlock()
 155         for typeId, typeInfo := range jh.allTypes {
 156                 log.Debugf("Processing jobs for type: %v", typeId)
 157                 messagesBody, error := restclient.Get(fmt.Sprintf("%v/%v", mRAddress, typeInfo.DMaaPTopicURL), jh.pollClient)
 158                 if error != nil {
 159                         log.Warnf("Error getting data from MR. Cause: %v", error)
 160                         continue
 161                 }
 162                 jh.distributeMessages(messagesBody, typeInfo)
 163         }
 164 }
 165
 166 func (jh *JobHandlerImpl) distributeMessages(messages []byte, typeInfo TypeData) {
 167         if len(messages) > 2 {
 168                 for _, jobInfo := range typeInfo.Jobs {
 169                         go jh.sendMessagesToConsumer(messages, jobInfo)
 170                 }
 171         }
 172 }
 173
 174 func (jh *JobHandlerImpl) sendMessagesToConsumer(messages []byte, jobInfo JobInfo) {
 175         log.Debugf("Processing job: %v", jobInfo.InfoJobIdentity)
 176         if postErr := restclient.Post(jobInfo.TargetUri, messages, jh.distributeClient); postErr != nil {
 177                 log.Warnf("Error posting data for job: %v. Cause: %v", jobInfo, postErr)
 178         }
 179 }
 180
 181 func (jh *JobHandlerImpl) clearAll() {
 182         jh.allTypes = make(map[string]TypeData)
 183 }