diff --git a/deployments/deployment.yml b/deployments/deployment.yml index cf1c8bb2f91d6cebfa573a1ade6f1845644bb863..a2f5bf496ff9ab22573e6c5bc0f19fe45bee19c3 100644 --- a/deployments/deployment.yml +++ b/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" diff --git a/internal/usecases/consume/handlemessage.go b/internal/usecases/consume/handlemessage.go index 3a3bc119315f6b997910c7990027e8a23b9234b1..339e060150d5363852b9d64553e3a8d979f74d8b 100644 --- a/internal/usecases/consume/handlemessage.go +++ b/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) }