Merge "Add schemas for input events of ODU slice assurance usecase"
[nonrtric.git] / dmaap-mediator-producer / main.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 main
22
23 import (
24         "crypto/tls"
25         "fmt"
26         "net/http"
27         "time"
28
29         log "github.com/sirupsen/logrus"
30         "oransc.org/nonrtric/dmaapmediatorproducer/internal/config"
31         "oransc.org/nonrtric/dmaapmediatorproducer/internal/jobs"
32         "oransc.org/nonrtric/dmaapmediatorproducer/internal/restclient"
33         "oransc.org/nonrtric/dmaapmediatorproducer/internal/server"
34 )
35
36 var configuration *config.Config
37
38 func init() {
39         configuration = config.New()
40 }
41
42 func main() {
43         log.SetLevel(configuration.LogLevel)
44         log.Debug("Initializing DMaaP Mediator Producer")
45         log.Debug("Using configuration: ", configuration)
46         if err := validateConfiguration(configuration); err != nil {
47                 log.Fatalf("Stopping producer due to error: %v", err)
48         }
49         callbackAddress := fmt.Sprintf("%v:%v", configuration.InfoProducerHost, configuration.InfoProducerPort)
50
51         var cert tls.Certificate
52         if c, err := restclient.CreateClientCertificate(configuration.ProducerCertPath, configuration.ProducerKeyPath); err == nil {
53                 cert = c
54         } else {
55                 log.Fatalf("Stopping producer due to error: %v", err)
56         }
57         retryClient := restclient.CreateRetryClient(cert)
58
59         jobsManager := jobs.NewJobsManagerImpl("configs/type_config.json", retryClient, configuration.DMaaPMRAddress, restclient.CreateClientWithoutRetry(cert, 5*time.Second))
60         if err := registerTypesAndProducer(jobsManager, configuration.InfoCoordinatorAddress, callbackAddress, retryClient); err != nil {
61                 log.Fatalf("Stopping producer due to: %v", err)
62         }
63         jobsManager.StartJobsForAllTypes()
64
65         log.Debug("Starting DMaaP Mediator Producer")
66         go func() {
67                 log.Debugf("Starting callback server at port %v", configuration.InfoProducerPort)
68                 r := server.NewRouter(jobsManager)
69                 if restclient.IsUrlSecure(callbackAddress) {
70                         log.Fatalf("Server stopped: %v", http.ListenAndServeTLS(fmt.Sprintf(":%v", configuration.InfoProducerPort), configuration.ProducerCertPath, configuration.ProducerKeyPath, r))
71                 } else {
72                         log.Fatalf("Server stopped: %v", http.ListenAndServe(fmt.Sprintf(":%v", configuration.InfoProducerPort), r))
73                 }
74         }()
75
76         keepProducerAlive()
77 }
78
79 func validateConfiguration(configuration *config.Config) error {
80         if configuration.InfoProducerHost == "" {
81                 return fmt.Errorf("missing INFO_PRODUCER_HOST")
82         }
83         if configuration.ProducerCertPath == "" || configuration.ProducerKeyPath == "" {
84                 return fmt.Errorf("missing PRODUCER_CERT and/or PRODUCER_KEY")
85         }
86         return nil
87 }
88 func registerTypesAndProducer(jobTypesHandler jobs.JobTypesManager, infoCoordinatorAddress string, callbackAddress string, client restclient.HTTPClient) error {
89         registrator := config.NewRegistratorImpl(infoCoordinatorAddress, client)
90         if types, err := jobTypesHandler.LoadTypesFromConfiguration(); err == nil {
91                 if regErr := registrator.RegisterTypes(types); regErr != nil {
92                         return fmt.Errorf("unable to register all types due to: %v", regErr)
93                 }
94         } else {
95                 return fmt.Errorf("unable to get types to register due to: %v", err)
96         }
97         producer := config.ProducerRegistrationInfo{
98                 InfoProducerSupervisionCallbackUrl: callbackAddress + server.StatusPath,
99                 SupportedInfoTypes:                 jobTypesHandler.GetSupportedTypes(),
100                 InfoJobCallbackUrl:                 callbackAddress + server.AddJobPath,
101         }
102         if err := registrator.RegisterProducer("DMaaP_Mediator_Producer", &producer); err != nil {
103                 return fmt.Errorf("unable to register producer due to: %v", err)
104         }
105         return nil
106 }
107
108 func keepProducerAlive() {
109         forever := make(chan int)
110         <-forever
111 }