X-Git-Url: https://gerrit.o-ran-sc.org/r/gitweb?a=blobdiff_plain;f=dmaap-mediator-producer%2Fmain.go;h=380087f3c495e185d6daeab9f9126b900a4cb050;hb=b3896f4ad7912be9e12c05e7d4770fa39752d797;hp=beeb995d9c0b329ef1d057f864ffd6e442a5256a;hpb=0af0d6013c831bfea032c34ce85e1dcf28b10855;p=nonrtric.git diff --git a/dmaap-mediator-producer/main.go b/dmaap-mediator-producer/main.go index beeb995d..380087f3 100644 --- a/dmaap-mediator-producer/main.go +++ b/dmaap-mediator-producer/main.go @@ -21,11 +21,12 @@ package main import ( + "crypto/tls" "fmt" "net/http" - "sync" "time" + "github.com/hashicorp/go-retryablehttp" log "github.com/sirupsen/logrus" "oransc.org/nonrtric/dmaapmediatorproducer/internal/config" "oransc.org/nonrtric/dmaapmediatorproducer/internal/jobs" @@ -33,12 +34,7 @@ import ( "oransc.org/nonrtric/dmaapmediatorproducer/internal/server" ) -const timeoutHTTPClient = time.Second * 5 -const timeoutPollClient = time.Second * 15 - var configuration *config.Config -var httpClient restclient.HTTPClient -var jobHandler *jobs.JobHandlerImpl func init() { configuration = config.New() @@ -52,50 +48,68 @@ func main() { } callbackAddress := fmt.Sprintf("%v:%v", configuration.InfoProducerHost, configuration.InfoProducerPort) - httpClient = &http.Client{ - Timeout: timeoutHTTPClient, - } - pollClient := &http.Client{ - Timeout: timeoutPollClient, + var retryClient restclient.HTTPClient + if cert, err := createClientCertificate(); err == nil { + retryClient = createRetryClient(cert) + } else { + log.Fatalf("Stopping producer due to error: %v", err) } - jobHandler = jobs.NewJobHandlerImpl("configs/type_config.json", pollClient, httpClient) - if err := registerTypesAndProducer(jobHandler, configuration.InfoCoordinatorAddress, callbackAddress); err != nil { + + jobsManager := jobs.NewJobsManagerImpl("configs/type_config.json", retryClient, configuration.DMaaPMRAddress, &http.Client{ + Timeout: time.Second * 5, + }) + if err := registerTypesAndProducer(jobsManager, configuration.InfoCoordinatorAddress, callbackAddress, retryClient); err != nil { log.Fatalf("Stopping producer due to: %v", err) } + jobsManager.StartJobs() log.Debug("Starting DMaaP Mediator Producer") - wg := new(sync.WaitGroup) - - // add two goroutines to `wg` WaitGroup, one for each running go routine - wg.Add(2) - - log.Debugf("Starting callback server at port %v", configuration.InfoProducerPort) go func() { - r := server.NewRouter(jobHandler) - log.Warn(http.ListenAndServe(fmt.Sprintf(":%v", configuration.InfoProducerPort), r)) - wg.Done() + log.Debugf("Starting callback server at port %v", configuration.InfoProducerPort) + r := server.NewRouter(jobsManager) + log.Fatalf("Server stopped: %v", http.ListenAndServeTLS(fmt.Sprintf(":%v", configuration.InfoProducerPort), configuration.ProducerCertPath, configuration.ProducerKeyPath, r)) }() - go func() { - jobHandler.RunJobs(fmt.Sprintf("%v:%v", configuration.MRHost, configuration.MRPort)) - wg.Done() - }() - - // wait until WaitGroup is done - wg.Wait() - log.Debug("Stopping DMaaP Mediator Producer") + keepProducerAlive() } func validateConfiguration(configuration *config.Config) error { if configuration.InfoProducerHost == "" { return fmt.Errorf("missing INFO_PRODUCER_HOST") } + if configuration.ProducerCertPath == "" || configuration.ProducerKeyPath == "" { + return fmt.Errorf("missing PRODUCER_CERT and/or PRODUCER_KEY") + } return nil } -func registerTypesAndProducer(jobHandler jobs.JobTypeHandler, infoCoordinatorAddress string, callbackAddress string) error { - registrator := config.NewRegistratorImpl(infoCoordinatorAddress, httpClient) - if types, err := jobHandler.GetTypes(); err == nil { +func createClientCertificate() (*tls.Certificate, error) { + if cert, err := tls.LoadX509KeyPair(configuration.ProducerCertPath, configuration.ProducerKeyPath); err == nil { + return &cert, nil + } else { + return nil, fmt.Errorf("cannot create x509 keypair from cert file %s and key file %s due to: %v", configuration.ProducerCertPath, configuration.ProducerKeyPath, err) + } +} + +func createRetryClient(cert *tls.Certificate) *http.Client { + rawRetryClient := retryablehttp.NewClient() + rawRetryClient.RetryWaitMax = time.Minute + rawRetryClient.RetryMax = int(^uint(0) >> 1) + rawRetryClient.HTTPClient.Transport = &http.Transport{ + TLSClientConfig: &tls.Config{ + Certificates: []tls.Certificate{ + *cert, + }, + InsecureSkipVerify: true, + }, + } + + return rawRetryClient.StandardClient() +} + +func registerTypesAndProducer(jobHandler jobs.JobTypesManager, infoCoordinatorAddress string, callbackAddress string, client restclient.HTTPClient) error { + registrator := config.NewRegistratorImpl(infoCoordinatorAddress, client) + if types, err := jobHandler.LoadTypesFromConfiguration(); err == nil { if regErr := registrator.RegisterTypes(types); regErr != nil { return fmt.Errorf("unable to register all types due to: %v", regErr) } @@ -112,3 +126,8 @@ func registerTypesAndProducer(jobHandler jobs.JobTypeHandler, infoCoordinatorAdd } return nil } + +func keepProducerAlive() { + forever := make(chan int) + <-forever +}