- registrator := config.NewRegistratorImpl(configuration.InfoCoordinatorAddress)
- if types, err := jobtypes.GetTypes(); err == nil {
- if regErr := registrator.RegisterTypes(types); regErr != nil {
- log.Fatalf("Unable to register all types due to: %v", regErr)
- }
- } else {
- log.Fatalf("Unable to get types to register due to: %v", err)
+ log.Debug("Starting 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")
+ }
+ if configuration.DMaaPMRAddress == "" && configuration.KafkaBootstrapServers == "" {
+ return fmt.Errorf("at least one of DMAAP_MR_ADDR or KAFKA_BOOTSRAP_SERVERS must be provided")
+ }
+ return nil
+}
+func registerTypesAndProducer(jobTypesManager jobs.JobTypesManager, infoCoordinatorAddress string, callbackAddress string, client restclient.HTTPClient) error {
+ registrator := config.NewRegistratorImpl(infoCoordinatorAddress, client)
+ configTypes, err := config.GetJobTypesFromConfiguration("configs")
+ if err != nil {
+ return fmt.Errorf("unable to register all types due to: %v", err)
+ }
+ regErr := registrator.RegisterTypes(jobTypesManager.LoadTypesFromConfiguration(configTypes))
+ if regErr != nil {
+ return fmt.Errorf("unable to register all types due to: %v", regErr)