varnitgoyal95/janus
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

DE-121 - added env variable

Browse Source
This commit is contained in:
Nitin
2021-07-27 13:21:26 +05:30
parent ada5858be0
commit 3d3200d381
2 changed files with 14 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
main.go
View File

@@ -60,7 +60,7 @@ func init() {
producer_module.GetSchemaVersions()
// initialize producers
producer_module.InitializeProducers(configuration)
producer_module.InitializeProducers(configuration.Kafka, configuration.Env)
for i := 0; i < configuration.Server.Goroutines; i++ {
go lib.ProcessProtobufRequestChannel(configuration.Kafka.Kafka_Topic_Protobuf)

26
producer/producer_config.go
View File

@@ -16,12 +16,12 @@ var (
asyncProducer sarama.AsyncProducer
)
func GetSyncProducerConfig(configuration config.Configurations) *sarama.Config {
func GetSyncProducerConfig(kafkaConfiguration config.KafkaConfigurations, env string) *sarama.Config {
config := sarama.NewConfig()
kafkaConfiguration configuration.KafkaConfigurations
//kafkaConfiguration configuration.KafkaConfigurations
// security configs
// config.Net.TLS.Enable = true
if config.Env == "PROD" {
if env == "PROD" {
config.Net.SASL.Enable = true
config.Net.SASL.Handshake = true
config.Net.SASL.Mechanism = sarama.SASLTypeSCRAMSHA512
@@ -44,12 +44,12 @@ func GetSyncProducerConfig(configuration config.Configurations) *sarama.Config {
}
func GetAsyncProducerConfig(configuration config.Configurations) *sarama.Config {
func GetAsyncProducerConfig(kafkaConfiguration config.KafkaConfigurations, env string) *sarama.Config {
config := sarama.NewConfig()
kafkaConfiguration configuration.KafkaConfigurations
//kafkaConfiguration configuration.KafkaConfigurations
// security configs
// config.Net.TLS.Enable = true
if config.Env != "PROD" {
if env != "PROD" {
config.Net.SASL.Enable = true
config.Net.SASL.Handshake = true
config.Net.SASL.Mechanism = sarama.SASLTypeSCRAMSHA512
@@ -71,8 +71,8 @@ func GetAsyncProducerConfig(configuration config.Configurations) *sarama.Config
}
func GetSyncProducer(kafkaConfiguration config.KafkaConfigurations) sarama.SyncProducer {
config := GetSyncProducerConfig(kafkaConfiguration)
func GetSyncProducer(kafkaConfiguration config.KafkaConfigurations, env string) sarama.SyncProducer {
config := GetSyncProducerConfig(kafkaConfiguration, env)
brokerList := strings.Split(kafkaConfiguration.Bootstrap_Servers, ",")
producer, err := sarama.NewSyncProducer(brokerList, config)
if err != nil {
@@ -82,8 +82,8 @@ func GetSyncProducer(kafkaConfiguration config.KafkaConfigurations) sarama.SyncP
}
func GetAsyncProducer(kafkaConfiguration config.KafkaConfigurations) sarama.AsyncProducer {
config := GetAsyncProducerConfig(kafkaConfiguration)
func GetAsyncProducer(kafkaConfiguration config.KafkaConfigurations, env string) sarama.AsyncProducer {
config := GetAsyncProducerConfig(kafkaConfiguration, env)
brokerList := strings.Split(kafkaConfiguration.Bootstrap_Servers, ",")
producer, err := sarama.NewAsyncProducer(brokerList, config)
if err != nil {
@@ -93,10 +93,10 @@ func GetAsyncProducer(kafkaConfiguration config.KafkaConfigurations) sarama.Asyn
}
func InitializeProducers(configuration config) {
func InitializeProducers(kafkaConfiguration config.KafkaConfigurations, env string) {
sarama.Logger = log.New(os.Stdout, "[sarama] ", log.LstdFlags)
syncProducer = GetSyncProducer(configuration)
asyncProducer = GetAsyncProducer(configuration)
syncProducer = GetSyncProducer(kafkaConfiguration, env)
asyncProducer = GetAsyncProducer(kafkaConfiguration, env)
}
// using confluent-kafka-go