kaf consume add value-filter

cmd/kaf/consume.go

@@ -6,6 +6,7 @@ import (
 	"encoding/binary"
 	"encoding/json"
 	"fmt"
+	"regexp"
 	"sync"
 	"text/tabwriter"
 
@@ -37,6 +38,10 @@ var (
 
 	limitMessagesFlag int64
 
+	valueFilterFlag string
+
+	valReg *regexp.Regexp
+
 	reg *proto.DescriptorRegistry
 )
 
@@ -55,6 +60,7 @@ func init() {
 	consumeCmd.Flags().Int64VarP(&limitMessagesFlag, "limit-messages", "l", 0, "Limit messages per partition")
 	consumeCmd.Flags().StringVarP(&groupFlag, "group", "g", "", "Consumer Group to use for consume")
 	consumeCmd.Flags().BoolVar(&groupCommitFlag, "commit", false, "Commit Group offset after receiving messages. Works only if consuming as Consumer Group")
+	consumeCmd.Flags().StringVar(&valueFilterFlag, "value-filter", "", "msg will display if it contains --value-filter regex string, but all msgs will display if the value of value-filter is empty")
 
 	keyfmt = prettyjson.NewFormatter()
 	keyfmt.Newline = " " // Replace newline with space to avoid condensed output.
@@ -95,6 +101,10 @@ var consumeCmd = &cobra.Command{
 		topic := args[0]
 		client := getClientFromConfig(cfg)
 
+		if len(valueFilterFlag) > 0 {
+			valReg = regexp.MustCompile(valueFilterFlag)
+		}
+
 		switch offsetFlag {
 		case "oldest":
 			offset = sarama.OffsetOldest
@@ -306,10 +316,21 @@ func handleMessage(msg *sarama.ConsumerMessage, mu *sync.Mutex) {
 		w.Flush()
 	}
 
+	outputSign := true
+	if len(valueFilterFlag) > 0 {
+		isFound := len(valReg.Find(dataToDisplay))
+		if isFound <= 0 {
+			outputSign = false
+		}
+	}
+
 	mu.Lock()
 	stderr.WriteTo(errWriter)
-	_, _ = colorableOut.Write(dataToDisplay)
-	fmt.Fprintln(outWriter)
+
+	if outputSign {
+		_, _ = colorableOut.Write(dataToDisplay)
+		fmt.Fprintln(outWriter)
+	}
 	mu.Unlock()
 
 }