- if configuration.InfoProducerSupervisionCallbackUrl == "" {
- log.Fatal("Missing INFO_PRODUCER_SUPERVISION_CALLBACK_URL")
+ callbackAddress := fmt.Sprintf("%v:%v", configuration.InfoProducerHost, configuration.InfoProducerPort)
+
+ var cert tls.Certificate
+ if c, err := restclient.CreateClientCertificate(configuration.ProducerCertPath, configuration.ProducerKeyPath); err == nil {
+ cert = c
+ } else {
+ log.Fatalf("Stopping producer due to error: %v", err)
+ }
+ retryClient := restclient.CreateRetryClient(cert)
+
+ jobsManager := jobs.NewJobsManagerImpl("configs/type_config.json", retryClient, configuration.DMaaPMRAddress, restclient.CreateClientWithoutRetry(cert, 5*time.Second))
+ if err := registerTypesAndProducer(jobsManager, configuration.InfoCoordinatorAddress, callbackAddress, retryClient); err != nil {
+ log.Fatalf("Stopping producer due to: %v", err)