2 // ========================LICENSE_START=================================
5 // Copyright (C) 2021: Nordix Foundation
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
11 // http://www.apache.org/licenses/LICENSE-2.0
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===================================
29 "github.com/hashicorp/go-retryablehttp"
30 log "github.com/sirupsen/logrus"
31 "oransc.org/nonrtric/dmaapmediatorproducer/internal/config"
32 "oransc.org/nonrtric/dmaapmediatorproducer/internal/jobs"
33 "oransc.org/nonrtric/dmaapmediatorproducer/internal/restclient"
34 "oransc.org/nonrtric/dmaapmediatorproducer/internal/server"
37 const timeoutDistributionClient = time.Second * 5
38 const retryWaitMax = time.Minute
39 const retryMax = int(^uint(0) >> 1)
41 var configuration *config.Config
42 var retryClient restclient.HTTPClient
43 var jobHandler *jobs.JobHandlerImpl
46 configuration = config.New()
50 log.SetLevel(configuration.LogLevel)
51 log.Debug("Initializing DMaaP Mediator Producer")
52 if err := validateConfiguration(configuration); err != nil {
53 log.Fatalf("Stopping producer due to error: %v", err)
55 callbackAddress := fmt.Sprintf("%v:%v", configuration.InfoProducerHost, configuration.InfoProducerPort)
57 distributionClient := &http.Client{
58 Timeout: timeoutDistributionClient,
61 rawRetryClient := retryablehttp.NewClient()
62 rawRetryClient.RetryWaitMax = retryWaitMax
63 rawRetryClient.RetryMax = retryMax
64 retryClient = rawRetryClient.StandardClient()
66 jobHandler = jobs.NewJobHandlerImpl("configs/type_config.json", retryClient, distributionClient)
67 if err := registerTypesAndProducer(jobHandler, configuration.InfoCoordinatorAddress, callbackAddress); err != nil {
68 log.Fatalf("Stopping producer due to: %v", err)
71 log.Debug("Starting DMaaP Mediator Producer")
72 wg := new(sync.WaitGroup)
74 // add two goroutines to `wg` WaitGroup, one for each running go routine
77 log.Debugf("Starting callback server at port %v", configuration.InfoProducerPort)
79 r := server.NewRouter(jobHandler)
80 log.Warn(http.ListenAndServe(fmt.Sprintf(":%v", configuration.InfoProducerPort), r))
85 jobHandler.RunJobs(fmt.Sprintf("%v:%v", configuration.MRHost, configuration.MRPort))
89 // wait until WaitGroup is done
91 log.Debug("Stopping DMaaP Mediator Producer")
94 func validateConfiguration(configuration *config.Config) error {
95 if configuration.InfoProducerHost == "" {
96 return fmt.Errorf("missing INFO_PRODUCER_HOST")
101 func registerTypesAndProducer(jobHandler jobs.JobTypeHandler, infoCoordinatorAddress string, callbackAddress string) error {
102 registrator := config.NewRegistratorImpl(infoCoordinatorAddress, retryClient)
103 if types, err := jobHandler.GetTypes(); err == nil {
104 if regErr := registrator.RegisterTypes(types); regErr != nil {
105 return fmt.Errorf("unable to register all types due to: %v", regErr)
108 return fmt.Errorf("unable to get types to register due to: %v", err)
110 producer := config.ProducerRegistrationInfo{
111 InfoProducerSupervisionCallbackUrl: callbackAddress + server.StatusPath,
112 SupportedInfoTypes: jobHandler.GetSupportedTypes(),
113 InfoJobCallbackUrl: callbackAddress + server.AddJobPath,
115 if err := registrator.RegisterProducer("DMaaP_Mediator_Producer", &producer); err != nil {
116 return fmt.Errorf("unable to register producer due to: %v", err)