package jobs
import (
- "encoding/json"
"fmt"
- "os"
"sync"
log "github.com/sirupsen/logrus"
type TypeData struct {
TypeId string `json:"id"`
DMaaPTopicURL string `json:"dmaapTopicUrl"`
- Jobs map[string]JobInfo
+ jobsHandler *jobsHandler
}
type JobInfo struct {
InfoTypeIdentity string `json:"info_type_identity"`
}
-type JobTypeHandler interface {
- GetTypes() ([]config.TypeDefinition, error)
+type JobTypesManager interface {
+ LoadTypesFromConfiguration(types []config.TypeDefinition) []config.TypeDefinition
GetSupportedTypes() []string
}
-type JobHandler interface {
- AddJob(JobInfo) error
- DeleteJob(jobId string)
+type JobsManager interface {
+ AddJobFromRESTCall(JobInfo) error
+ DeleteJobFromRESTCall(jobId string)
}
-type JobHandlerImpl struct {
- mu sync.Mutex
- configFile string
+type JobsManagerImpl struct {
allTypes map[string]TypeData
pollClient restclient.HTTPClient
+ mrAddress string
distributeClient restclient.HTTPClient
}
-func NewJobHandlerImpl(typeConfigFilePath string, pollClient restclient.HTTPClient, distributeClient restclient.HTTPClient) *JobHandlerImpl {
- return &JobHandlerImpl{
- configFile: typeConfigFilePath,
+func NewJobsManagerImpl(pollClient restclient.HTTPClient, mrAddr string, distributeClient restclient.HTTPClient) *JobsManagerImpl {
+ return &JobsManagerImpl{
allTypes: make(map[string]TypeData),
pollClient: pollClient,
+ mrAddress: mrAddr,
distributeClient: distributeClient,
}
}
-func (jh *JobHandlerImpl) AddJob(ji JobInfo) error {
- jh.mu.Lock()
- defer jh.mu.Unlock()
- if err := jh.validateJobInfo(ji); err == nil {
- jobs := jh.allTypes[ji.InfoTypeIdentity].Jobs
- jobs[ji.InfoJobIdentity] = ji
+func (jm *JobsManagerImpl) AddJobFromRESTCall(ji JobInfo) error {
+ if err := jm.validateJobInfo(ji); err == nil {
+ typeData := jm.allTypes[ji.InfoTypeIdentity]
+ typeData.jobsHandler.addJobCh <- ji
log.Debug("Added job: ", ji)
return nil
} else {
}
}
-func (jh *JobHandlerImpl) DeleteJob(jobId string) {
- jh.mu.Lock()
- defer jh.mu.Unlock()
- for _, typeData := range jh.allTypes {
- delete(typeData.Jobs, jobId)
+func (jm *JobsManagerImpl) DeleteJobFromRESTCall(jobId string) {
+ for _, typeData := range jm.allTypes {
+ log.Debugf("Deleting job %v from type %v", jobId, typeData.TypeId)
+ typeData.jobsHandler.deleteJobCh <- jobId
}
log.Debug("Deleted job: ", jobId)
}
-func (jh *JobHandlerImpl) validateJobInfo(ji JobInfo) error {
- if _, ok := jh.allTypes[ji.InfoTypeIdentity]; !ok {
+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 (jh *JobHandlerImpl) GetTypes() ([]config.TypeDefinition, error) {
- jh.mu.Lock()
- defer jh.mu.Unlock()
- typeDefsByte, err := os.ReadFile(jh.configFile)
- if err != nil {
- return nil, err
- }
- typeDefs := struct {
- Types []config.TypeDefinition `json:"types"`
- }{}
- err = json.Unmarshal(typeDefsByte, &typeDefs)
- if err != nil {
- return nil, err
- }
- for _, typeDef := range typeDefs.Types {
- jh.allTypes[typeDef.Id] = TypeData{
+func (jm *JobsManagerImpl) LoadTypesFromConfiguration(types []config.TypeDefinition) []config.TypeDefinition {
+ for _, typeDef := range types {
+ jm.allTypes[typeDef.Id] = TypeData{
TypeId: typeDef.Id,
DMaaPTopicURL: typeDef.DmaapTopicURL,
- Jobs: make(map[string]JobInfo),
+ jobsHandler: newJobsHandler(typeDef.Id, typeDef.DmaapTopicURL, jm.pollClient, jm.distributeClient),
}
}
- return typeDefs.Types, nil
+ return types
}
-func (jh *JobHandlerImpl) GetSupportedTypes() []string {
- jh.mu.Lock()
- defer jh.mu.Unlock()
+func (jm *JobsManagerImpl) GetSupportedTypes() []string {
supportedTypes := []string{}
- for k := range jh.allTypes {
+ for k := range jm.allTypes {
supportedTypes = append(supportedTypes, k)
}
return supportedTypes
}
-func (jh *JobHandlerImpl) RunJobs(mRAddress string) {
- for {
- jh.pollAndDistributeMessages(mRAddress)
+func (jm *JobsManagerImpl) StartJobsForAllTypes() {
+ for _, jobType := range jm.allTypes {
+
+ go jobType.jobsHandler.startPollingAndDistribution(jm.mrAddress)
+
}
}
-func (jh *JobHandlerImpl) pollAndDistributeMessages(mRAddress string) {
- jh.mu.Lock()
- defer jh.mu.Unlock()
- for typeId, typeInfo := range jh.allTypes {
- log.Debugf("Processing jobs for type: %v", typeId)
- messagesBody, error := restclient.Get(mRAddress+typeInfo.DMaaPTopicURL, jh.pollClient)
- if error != nil {
- log.Warnf("Error getting data from MR. Cause: %v", error)
- continue
+type jobsHandler struct {
+ mu sync.Mutex
+ typeId string
+ topicUrl string
+ jobs map[string]job
+ addJobCh chan JobInfo
+ deleteJobCh chan string
+ pollClient restclient.HTTPClient
+ distributeClient restclient.HTTPClient
+}
+
+func newJobsHandler(typeId string, topicURL string, pollClient restclient.HTTPClient, distributeClient restclient.HTTPClient) *jobsHandler {
+ return &jobsHandler{
+ typeId: typeId,
+ topicUrl: topicURL,
+ jobs: make(map[string]job),
+ addJobCh: make(chan JobInfo),
+ deleteJobCh: make(chan string),
+ pollClient: pollClient,
+ distributeClient: distributeClient,
+ }
+}
+
+func (jh *jobsHandler) startPollingAndDistribution(mRAddress string) {
+ go func() {
+ for {
+ jh.pollAndDistributeMessages(mRAddress)
+ }
+ }()
+
+ go func() {
+ for {
+ jh.monitorManagementChannels()
}
- log.Debugf("Received messages: %v", string(messagesBody))
- jh.distributeMessages(messagesBody, typeInfo)
+ }()
+}
+
+func (jh *jobsHandler) 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.Warn("Error getting data from MR. Cause: ", error)
}
+ log.Debug("Received messages: ", string(messagesBody))
+ jh.distributeMessages(messagesBody)
}
-func (jh *JobHandlerImpl) distributeMessages(messages []byte, typeInfo TypeData) {
+func (jh *jobsHandler) distributeMessages(messages []byte) {
if len(messages) > 2 {
- for _, jobInfo := range typeInfo.Jobs {
- go jh.sendMessagesToConsumer(messages, jobInfo)
+ jh.mu.Lock()
+ defer jh.mu.Unlock()
+ for _, job := range jh.jobs {
+ if len(job.messagesChannel) < cap(job.messagesChannel) {
+ job.messagesChannel <- messages
+ } else {
+ jh.emptyMessagesBuffer(job)
+ }
}
}
}
-func (jh *JobHandlerImpl) sendMessagesToConsumer(messages []byte, jobInfo JobInfo) {
- log.Debugf("Processing job: %v", jobInfo.InfoJobIdentity)
- if postErr := restclient.Post(jobInfo.TargetUri, messages, jh.distributeClient); postErr != nil {
- log.Warnf("Error posting data for job: %v. Cause: %v", jobInfo, postErr)
+func (jh *jobsHandler) emptyMessagesBuffer(job job) {
+ log.Debug("Emptying message queue for job: ", job.jobInfo.InfoJobIdentity)
+out:
+ for {
+ select {
+ case <-job.messagesChannel:
+ default:
+ break out
+ }
+ }
+}
+
+func (jh *jobsHandler) monitorManagementChannels() {
+ select {
+ case addedJob := <-jh.addJobCh:
+ jh.addJob(addedJob)
+ case deletedJob := <-jh.deleteJobCh:
+ jh.deleteJob(deletedJob)
+ }
+}
+
+func (jh *jobsHandler) addJob(addedJob JobInfo) {
+ jh.mu.Lock()
+ log.Debug("Add job: ", addedJob)
+ newJob := newJob(addedJob, jh.distributeClient)
+ go newJob.start()
+ jh.jobs[addedJob.InfoJobIdentity] = newJob
+ jh.mu.Unlock()
+}
+
+func (jh *jobsHandler) deleteJob(deletedJob string) {
+ jh.mu.Lock()
+ log.Debug("Delete job: ", deletedJob)
+ j, exist := jh.jobs[deletedJob]
+ if exist {
+ j.controlChannel <- struct{}{}
+ delete(jh.jobs, deletedJob)
}
- log.Debugf("Messages distributed to consumer: %v.", jobInfo.Owner)
+ jh.mu.Unlock()
}
-func (jh *JobHandlerImpl) clearAll() {
- jh.allTypes = make(map[string]TypeData)
+type job struct {
+ jobInfo JobInfo
+ client restclient.HTTPClient
+ messagesChannel chan []byte
+ controlChannel chan struct{}
+}
+
+func newJob(j JobInfo, c restclient.HTTPClient) job {
+ return job{
+ jobInfo: j,
+ client: c,
+ messagesChannel: make(chan []byte, 10),
+ controlChannel: make(chan struct{}),
+ }
+}
+
+func (j *job) start() {
+out:
+ for {
+ select {
+ case <-j.controlChannel:
+ log.Debug("Stop distribution for job: ", j.jobInfo.InfoJobIdentity)
+ break out
+ case msg := <-j.messagesChannel:
+ j.sendMessagesToConsumer(msg)
+ }
+ }
+}
+
+func (j *job) sendMessagesToConsumer(messages []byte) {
+ log.Debug("Processing job: ", j.jobInfo.InfoJobIdentity)
+ if postErr := restclient.Post(j.jobInfo.TargetUri, messages, j.client); postErr != nil {
+ log.Warnf("Error posting data for job: %v. Cause: %v", j.jobInfo, postErr)
+ }
+ log.Debugf("Messages for job: %v distributed to consumer: %v", j.jobInfo.InfoJobIdentity, j.jobInfo.Owner)
}
Code Review / nonrtric.git / blobdiff