Added queryID as optional header key for backwards compatibility

Also added the Minio environment variables to the Kubernetes deployment.

deployments/deployment.yml

@@ -38,6 +38,12 @@ spec:
           value: redis.redis.svc.cluster.local:6379
         - name: LOG_MESSAGES
           value: "true"
+        - name: MINIO_ADDRESS
+          value: minio:9000
+        - name: MINIO_ACCESSKEYID
+          value: root
+        - name: MINIO_ACCESSKEY
+          value: DikkeDraak
         resources:
             requests:
               memory: "100Mi"

internal/usecases/consume/handlemessage.go

@@ -27,9 +27,10 @@ func (s *Service) HandleMessage(msg *broker.Message) {
 		return
 	}
 
-	queryID, ok := msg.Headers["queryID"].(string)
+	var queryID string
+	queryID, ok = msg.Headers["queryID"].(string)
 	if !ok {
-		return
+		queryID = ""
 	}
 
 	s.sendStatus("Received", &sessionID)
@@ -117,10 +118,9 @@ func (s *Service) HandleMessage(msg *broker.Message) {
 		return
 	}
 
-	s.sendStatus("Done", &sessionID)
+	s.sendStatus("Completed", &sessionID)
 
 	// Add type indicator to result from database
-	// TODO: Change key 'values' to key 'value'
 	var res interface{}
 	json.Unmarshal(*result, &res)
 	resultMsg := entity.MessageStruct{
@@ -134,9 +134,11 @@ func (s *Service) HandleMessage(msg *broker.Message) {
 	}
 
 	// Caching message in object store
-	cacheQueryResultContext, cancelCacheQueryResult := context.WithTimeout(context.Background(), time.Second*20)
-	defer cancelCacheQueryResult()
-	s.objectStore.Put(cacheQueryResultContext, "cached-queries", fmt.Sprintf("%s-%s", sessionID, queryID), bytes.NewReader(resultMsgBytes))
+	if queryID != "" {
+		cacheQueryResultContext, cancelCacheQueryResult := context.WithTimeout(context.Background(), time.Second*20)
+		defer cancelCacheQueryResult()
+		s.objectStore.Put(cacheQueryResultContext, "cached-queries", fmt.Sprintf("%s-%s", sessionID, queryID), bytes.NewReader(resultMsgBytes))
+	}
 
 	s.producer.PublishMessage(&resultMsgBytes, &sessionID)
 }