adding configs

config/config.go

@@ -10,6 +10,8 @@ type Configurations struct {
 
 type ServerConfigurations struct {
 	Port                         string
+	Goroutines                   int
+	Kafka_Topic                  string
 }
 
 

config/config.yml

@@ -1,5 +1,7 @@
 server:
     port: 8000
+    goroutines: 1000
+    kafka_topic: KAFKA_TOPIC
 
 kafka:
     bootstrap_servers: BOOTSTRAP_SERVERS  # read from environment variable
@@ -13,4 +15,4 @@ kafka:
 
 schemaRegistry:
     endpoint: SCHEMA_REGISTRY_ENDPOINT
-    topics: TOPICS
+    topics: SCHEMA_TOPICS

lib/RequestHandler.go

@@ -19,10 +19,10 @@ type RequestObject struct {
 	Header   http.Header
 }
 
-func ProcessRequestChannel() {
+func ProcessRequestChannel(topic string) {
     for {
         request := <- RequestChannel
-        ClickstreamProtobufEventHandler(*request, "testgo")
+        ClickstreamProtobufEventHandler(*request, topic)
     }
 }
 
@@ -54,7 +54,7 @@ func ClickstreamProtobufEventHandler(request RequestObject, topic string) {
     }
 
     if message != nil {
-        log.Printf("WRITTEN TO KAFKA")
+        log.Printf("WRITING TO KAFKA")
     }
 
     producer_module.WriteMessageToKafkaAsync(message)

main.go

@@ -40,6 +40,7 @@ func init() {
 	configuration.Kafka.Sasl_Password = viper.GetString(configuration.Kafka.Sasl_Password)
 	configuration.SchemaRegistry.Endpoint = viper.GetString(configuration.SchemaRegistry.Endpoint)
 	configuration.SchemaRegistry.Topics = viper.GetString(configuration.SchemaRegistry.Topics)
+	configuration.Server.Kafka_Topic = viper.GetString(configuration.Server.Kafka_Topic)
 
 	port = configuration.Server.Port
 	log.Printf("PORT IS: ", port)
@@ -52,8 +53,8 @@ func init() {
 	producer_module.InitializeProducers(configuration.Kafka)
 
 	// TO DO: read number of goroutines from config
-	for i := 0; i < 10; i++ {
-		go lib.ProcessRequestChannel()
+	for i := 0; i < configuration.Server.Goroutines; i++ {
+		go lib.ProcessRequestChannel(configuration.Server.Kafka_Topic)
 	}
 
 	// sync producer using sarama