"encoding/json"
"fmt"
"os"
- "path/filepath"
- "strings"
"sync"
log "github.com/sirupsen/logrus"
+ "oransc.org/nonrtric/dmaapmediatorproducer/internal/config"
"oransc.org/nonrtric/dmaapmediatorproducer/internal/restclient"
)
-type Type struct {
- TypeId string `json:"id"`
- DMaaPTopic string `json:"dmaapTopic"`
- Schema string `json:"schema"`
- Jobs map[string]JobInfo
+type TypeData struct {
+ TypeId string `json:"id"`
+ DMaaPTopicURL string `json:"dmaapTopicUrl"`
+ jobHandler *jobHandler
}
type JobInfo struct {
- Owner string `json:"owner"`
- LastUpdated string `json:"last_updated"`
- InfoJobIdentity string `json:"info_job_identity"`
- TargetUri string `json:"target_uri"`
- InfoJobData string `json:"info_job_data"`
- InfoTypeIdentity string `json:"info_type_identity"`
+ Owner string `json:"owner"`
+ LastUpdated string `json:"last_updated"`
+ InfoJobIdentity string `json:"info_job_identity"`
+ TargetUri string `json:"target_uri"`
+ InfoJobData interface{} `json:"info_job_data"`
+ InfoTypeIdentity string `json:"info_type_identity"`
}
-type JobHandler interface {
- AddJob(JobInfo) error
-}
-
-var (
- mu sync.Mutex
- typeDir = "configs"
- Handler JobHandler
- allJobs = make(map[string]Type)
-)
-
-func init() {
- Handler = newJobHandlerImpl()
+type JobTypesManager interface {
+ LoadTypesFromConfiguration() ([]config.TypeDefinition, error)
+ GetSupportedTypes() []string
}
-type jobHandlerImpl struct{}
-
-func newJobHandlerImpl() *jobHandlerImpl {
- return &jobHandlerImpl{}
+type JobsManager interface {
+ AddJob(JobInfo) error
+ DeleteJob(jobId string)
+}
+
+type JobsManagerImpl struct {
+ configFile string
+ allTypes map[string]TypeData
+ pollClient restclient.HTTPClient
+ mrAddress string
+ distributeClient restclient.HTTPClient
+}
+
+type jobHandler struct {
+ mu sync.Mutex
+ typeId string
+ topicUrl string
+ jobs map[string]JobInfo
+ addJobCh chan JobInfo
+ deleteJobCh chan string
+ pollClient restclient.HTTPClient
+ distributeClient restclient.HTTPClient
+}
+
+func NewJobsManagerImpl(typeConfigFilePath string, pollClient restclient.HTTPClient, mrAddr string, distributeClient restclient.HTTPClient) *JobsManagerImpl {
+ return &JobsManagerImpl{
+ configFile: typeConfigFilePath,
+ allTypes: make(map[string]TypeData),
+ pollClient: pollClient,
+ mrAddress: mrAddr,
+ distributeClient: distributeClient,
+ }
}
-func (jh *jobHandlerImpl) AddJob(ji JobInfo) error {
- mu.Lock()
- defer mu.Unlock()
- if err := validateJobInfo(ji); err == nil {
- jobs := allJobs[ji.InfoTypeIdentity].Jobs
- jobs[ji.InfoJobIdentity] = ji
+func (jm *JobsManagerImpl) AddJob(ji JobInfo) error {
+ if err := jm.validateJobInfo(ji); err == nil {
+ typeData := jm.allTypes[ji.InfoTypeIdentity]
+ typeData.jobHandler.addJobCh <- ji
+ log.Debug("Added job: ", ji)
return nil
} else {
return err
}
}
-func validateJobInfo(ji JobInfo) error {
- if _, ok := allJobs[ji.InfoTypeIdentity]; !ok {
+func (jm *JobsManagerImpl) DeleteJob(jobId string) {
+ for _, typeData := range jm.allTypes {
+ log.Debugf("Deleting job %v from type %v", jobId, typeData.TypeId)
+ typeData.jobHandler.deleteJobCh <- jobId
+ }
+ log.Debug("Deleted job: ", jobId)
+}
+
+func (jm *JobsManagerImpl) validateJobInfo(ji JobInfo) error {
+ if _, ok := jm.allTypes[ji.InfoTypeIdentity]; !ok {
return fmt.Errorf("type not supported: %v", ji.InfoTypeIdentity)
}
if ji.InfoJobIdentity == "" {
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 {
- if err != nil {
- return err
- }
- if strings.Contains(path, ".json") {
- if jobType, err := getType(path); err == nil {
- types = append(types, jobType)
- }
- }
- return nil
- })
+func (jm *JobsManagerImpl) LoadTypesFromConfiguration() ([]config.TypeDefinition, error) {
+ typeDefsByte, err := os.ReadFile(jm.configFile)
if err != nil {
return nil, err
}
- return types, nil
+ typeDefs := struct {
+ Types []config.TypeDefinition `json:"types"`
+ }{}
+ err = json.Unmarshal(typeDefsByte, &typeDefs)
+ if err != nil {
+ return nil, err
+ }
+ for _, typeDef := range typeDefs.Types {
+ addCh := make(chan JobInfo)
+ deleteCh := make(chan string)
+ jh := jobHandler{
+ typeId: typeDef.Id,
+ topicUrl: typeDef.DmaapTopicURL,
+ jobs: make(map[string]JobInfo),
+ addJobCh: addCh,
+ deleteJobCh: deleteCh,
+ pollClient: jm.pollClient,
+ distributeClient: jm.distributeClient,
+ }
+ jm.allTypes[typeDef.Id] = TypeData{
+ TypeId: typeDef.Id,
+ DMaaPTopicURL: typeDef.DmaapTopicURL,
+ jobHandler: &jh,
+ }
+ }
+ return typeDefs.Types, nil
}
-func GetSupportedTypes() []string {
- mu.Lock()
- defer mu.Unlock()
+func (jm *JobsManagerImpl) GetSupportedTypes() []string {
supportedTypes := []string{}
- for k := range allJobs {
+ for k := range jm.allTypes {
supportedTypes = append(supportedTypes, k)
}
return supportedTypes
}
-func AddJob(job JobInfo) error {
- return Handler.AddJob(job)
-}
-
-func getType(path string) (*Type, error) {
- 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
- }
- } else {
- return nil, err
- }
-}
+func (jm *JobsManagerImpl) StartJobs() {
+ for _, jobType := range jm.allTypes {
+
+ go jobType.jobHandler.start(jm.mrAddress)
-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
+func (jh *jobHandler) start(mRAddress string) {
+ go func() {
+ for {
+ jh.pollAndDistributeMessages(mRAddress)
}
- distributeMessages(messagesBody, typeInfo)
+ }()
+
+ go func() {
+ for {
+ jh.monitorManagementChannels()
+ }
+ }()
+}
+
+func (jh *jobHandler) pollAndDistributeMessages(mRAddress string) {
+ log.Debugf("Processing jobs for type: %v", jh.typeId)
+ messagesBody, error := restclient.Get(mRAddress+jh.topicUrl, jh.pollClient)
+ if error != nil {
+ log.Warnf("Error getting data from MR. Cause: %v", error)
}
+ log.Debugf("Received messages: %v", string(messagesBody))
+ jh.distributeMessages(messagesBody)
}
-func distributeMessages(messages []byte, typeInfo Type) {
+func (jh *jobHandler) distributeMessages(messages []byte) {
if len(messages) > 2 {
- mu.Lock()
- for _, jobInfo := range typeInfo.Jobs {
- go sendMessagesToConsumer(messages, jobInfo)
+ jh.mu.Lock()
+ defer jh.mu.Unlock()
+ for _, jobInfo := range jh.jobs {
+ go jh.sendMessagesToConsumer(messages, jobInfo)
}
- mu.Unlock()
}
}
-func sendMessagesToConsumer(messages []byte, jobInfo JobInfo) {
+func (jh *jobHandler) sendMessagesToConsumer(messages []byte, jobInfo JobInfo) {
log.Debugf("Processing job: %v", jobInfo.InfoJobIdentity)
- if postErr := restclient.Post(jobInfo.TargetUri, messages); postErr != nil {
+ if postErr := restclient.Post(jobInfo.TargetUri, messages, jh.distributeClient); postErr != nil {
log.Warnf("Error posting data for job: %v. Cause: %v", jobInfo, postErr)
}
-}
-
-func clearAll() {
- allJobs = make(map[string]Type)
+ log.Debugf("Messages distributed to consumer: %v.", jobInfo.Owner)
+}
+
+func (jh *jobHandler) monitorManagementChannels() {
+ select {
+ case addedJob := <-jh.addJobCh:
+ jh.mu.Lock()
+ log.Debugf("received %v from addJobCh\n", addedJob)
+ jh.jobs[addedJob.InfoJobIdentity] = addedJob
+ jh.mu.Unlock()
+ case deletedJob := <-jh.deleteJobCh:
+ jh.mu.Lock()
+ log.Debugf("received %v from deleteJobCh\n", deletedJob)
+ delete(jh.jobs, deletedJob)
+ jh.mu.Unlock()
+ }
}
Code Review / nonrtric.git / blobdiff