package main import ( config "com.navi.medici.janus/config" lib "com.navi.medici.janus/lib" producer_module "com.navi.medici.janus/producer" server "com.navi.medici.janus/server" "github.com/spf13/viper" "log" "strings" ) var ( port string cors string ) func init() { viper.SetConfigName("config") viper.AddConfigPath("./config") viper.SetConfigType("yml") viper.AutomaticEnv() var configuration config.Configurations if err := viper.ReadInConfig(); err != nil { log.Fatalln("Error reading config file, %s", err) } err := viper.Unmarshal(&configuration) if err != nil { log.Fatalln("Unable to decode into struct, %v", err) } //log.Print(configuration.Kafka.Bootstrap_Servers) // Following coinfigurations read from environment variables configuration.Env = viper.GetString(configuration.Env) configuration.Kafka.Bootstrap_Servers = viper.GetString(configuration.Kafka.Bootstrap_Servers) configuration.SchemaRegistry.Endpoint = viper.GetString(configuration.SchemaRegistry.Endpoint) configuration.SchemaRegistry.Topics = viper.GetString(configuration.SchemaRegistry.Topics) configuration.Kafka.Kafka_Topic_Json = viper.GetString(configuration.Kafka.Kafka_Topic_Json) configuration.Kafka.Kafka_Topic_Protobuf = viper.GetString(configuration.Kafka.Kafka_Topic_Protobuf) //log.Printf("Env is: ", configuration.Env) //if configuration.Env == "PROD" { configuration.Kafka.Sasl_User = viper.GetString(configuration.Kafka.Sasl_User) configuration.Kafka.Sasl_Password = viper.GetString(configuration.Kafka.Sasl_Password) //} port = configuration.Server.Port cors = viper.GetString(configuration.Server.Cors) log.Printf("PORT IS: %v", port) log.Printf(configuration.Kafka.Bootstrap_Servers) log.Printf(configuration.SchemaRegistry.Endpoint) producer_module.SchemaRegistryEndpoint = configuration.SchemaRegistry.Endpoint producer_module.TopicList = strings.Split(configuration.SchemaRegistry.Topics, ",") // initialize schema version map which contains latest schema version for topic(s) producer_module.GetSchemaVersions() // initialize producers producer_module.InitializeProducers(configuration.Kafka, configuration.Env) for i := 0; i < 2; i++ { go lib.ProcessProtobufRequestChannel(configuration.Kafka.Kafka_Topic_Protobuf) } for i := 0; i < configuration.Server.Goroutines; i++ { go lib.ProcessJsonRequestChannel(configuration.Kafka.Kafka_Topic_Json) } } func main() { log.Printf("Serving on http://0.0.0.0:", port) httpServer, err1 := server.NewServer(port, cors) metricsServer, err2 := server.MetricServer("4000") if err1 != nil { log.Fatalln("Unable to start server, ", err1) } if err2 != nil { log.Fatalln("Unable to start Metric server, ", err2) } go httpServer.HttpServer.Serve(httpServer.Listener) go metricsServer.HttpServer.Serve(metricsServer.Listener) select {} }