chore: directly use hash key to check messages

pkg/sinker/examples/redis-sink/main.go

@@ -26,14 +26,32 @@ func (rds *redisTestSink) Sink(ctx context.Context, datumStreamCh <-chan sinksdk
 		_ = d.Watermark()
 
 		// We use redis hashes to store messages.
-		// The name of a hash is pipelineName:sinkName.
 		// Each field of a hash is the content of a message and value of the field is the no. of occurrences of the message.
-		hkey := fmt.Sprintf("%s:%s", os.Getenv("NUMAFLOW_PIPELINE_NAME"), os.Getenv("NUMAFLOW_VERTEX_NAME"))
-		err := client.HIncrBy(ctx, hkey, string(d.Value()), 1).Err()
+		var hashKey string
+
+		monoVertexName := os.Getenv("NUMAFLOW_MONO_VERTEX_NAME")
+		pipelineName := os.Getenv("NUMAFLOW_PIPELINE_NAME")
+		vertexName := os.Getenv("NUMAFLOW_VERTEX_NAME")
+
+		if !(monoVertexName != "" || (pipelineName != "" && vertexName != "")) {
+			log.Println("Error: Either mono vertex name or pipeline name and vertex name must be set in the environment variables.")
+			return result.Append(sinksdk.ResponseFailure(d.ID(), fmt.Sprintf("either mono vertex name or pipeline name and vertex name must be set in the environment variables.")))
+		}
+
+		if monoVertexName != "" {
+			// if the mono vertex name environment variable is set, we are in a mono vertex
+			// in this case, we use the mono vertex name as the hash key
+			hashKey = monoVertexName
+		} else {
+			// if the mono vertex name environment variable is not set, we are in a pipeline
+			// in this case, the name of a hash is pipelineName:sinkName.
+			hashKey = fmt.Sprintf("%s:%s", pipelineName, vertexName)
+		}
+		err := client.HIncrBy(ctx, hashKey, string(d.Value()), 1).Err()
 		if err != nil {
 			log.Println("Set Error - ", err)
 		} else {
-			log.Printf("Incremented by 1 the no. of occurrences of %s under hash key %s\n", string(d.Value()), hkey)
+			log.Printf("Incremented by 1 the no. of occurrences of %s under hash key %s\n", string(d.Value()), hashKey)
 		}
 
 		id := d.ID()