feat(kafka-clusters): implement plain, table, json, json-raw printers…

… for kafka-clusters

pkg/cmd/egress.go

@@ -9,7 +9,7 @@ import (
 
 // SimCmd represents the create command
 var EgressCmd = &cobra.Command{
-	Use:   "egress [stream-name]",
+	Use:   "egress (stream-name)",
 	Short: "Read from egress",
 	Run: func(cmd *cobra.Command, args []string) {
 		egress.Run(cmd, &args[0])

pkg/entity/kafka_cluster/kafka_cluster.go

@@ -32,15 +32,11 @@ func list() {
 }
 
 func get(name *string) {
-	cluster := GetCluster(name)
-	printer.Print(cluster)
-}
-
-func GetCluster(name *string) *entities.KafkaCluster {
 	req := &kafka_clusters.GetKafkaClusterRequest{Ref: ref(name)}
 	cluster, err := client.GetKafkaCluster(apiContext, req)
 	common.CliExit(err)
-	return cluster.KafkaCluster
+
+	printer.Print(cluster)
 }
 
 func NamesCompletion(cmd *cobra.Command, args []string, complete string) ([]string, cobra.ShellCompDirective) {

pkg/entity/kafka_cluster/printers.go

@@ -1,21 +1,114 @@
 package kafka_cluster
 
 import (
+	"fmt"
+	"github.com/jedib0t/go-pretty/v6/table"
 	"github.com/spf13/cobra"
+	"github.com/streammachineio/api-definitions-go/api/entities/v1"
+	"github.com/streammachineio/api-definitions-go/api/kafka_clusters/v1"
+	"google.golang.org/protobuf/proto"
+	"streammachine.io/strm/pkg/constants"
 	"streammachine.io/strm/pkg/util"
 )
 
 var printer util.Printer
 
-func configurePrinter(cmd *cobra.Command) util.Printer {
-	outputFormat := util.GetStringAndErr(cmd.Flags(), util.OutputFormatFlag)
+func configurePrinter(command *cobra.Command) util.Printer {
+	outputFormat := util.GetStringAndErr(command.Flags(), util.OutputFormatFlag)
 
 	switch outputFormat {
 	case "json":
+		return util.GenericPrettyJsonPrinter{}
+	case "json-raw":
 		return util.GenericRawJsonPrinter{}
 	case "table":
-		return util.GenericRawJsonPrinter{}
+		switch command.Parent().Name() {
+		case constants.ListCommandName:
+			return ListStreamsTablePrinter{}
+		case constants.GetCommandName:
+			return GetStreamTablePrinter{}
+		}
+
+		return util.GenericPrettyJsonPrinter{}
+	case "plain":
+		switch command.Parent().Name() {
+		case constants.ListCommandName:
+			return ListStreamsPlainPrinter{}
+		case constants.GetCommandName:
+			return GetStreamPlainPrinter{}
+		}
+
+		return util.GenericPrettyJsonPrinter{}
 	default:
-		return util.GenericRawJsonPrinter{}
+		return util.GenericPrettyJsonPrinter{}
+	}
+}
+
+type ListStreamsPlainPrinter struct{}
+type GetStreamPlainPrinter struct{}
+
+type ListStreamsTablePrinter struct{}
+type GetStreamTablePrinter struct{}
+
+func (p ListStreamsTablePrinter) Print(data proto.Message) {
+	listResponse, _ := (data).(*kafka_clusters.ListKafkaClustersResponse)
+	printTable(listResponse.KafkaClusters)
+}
+
+func (p GetStreamTablePrinter) Print(data proto.Message) {
+	getResponse, _ := (data).(*kafka_clusters.GetKafkaClusterResponse)
+	printTable([]*entities.KafkaCluster{getResponse.KafkaCluster})
+}
+
+func (p ListStreamsPlainPrinter) Print(data proto.Message) {
+	listResponse, _ := (data).(*kafka_clusters.ListKafkaClustersResponse)
+	printPlain(listResponse.KafkaClusters)
+}
+
+func (p GetStreamPlainPrinter) Print(data proto.Message) {
+	getResponse, _ := (data).(*kafka_clusters.GetKafkaClusterResponse)
+	printPlain([]*entities.KafkaCluster{getResponse.KafkaCluster})
+}
+
+func printTable(kafkaClusters []*entities.KafkaCluster) {
+	rows := make([]table.Row, 0, len(kafkaClusters))
+
+	for _, cluster := range kafkaClusters {
+
+		rows = append(rows, table.Row{
+			refToString(cluster.Ref),
+			cluster.BootstrapServers,
+			cluster.AuthMechanism,
+			cluster.TokenUri,
+		})
+	}
+
+	util.RenderTable(
+		table.Row{
+			"Name",
+			"Bootstrap Servers",
+			"Auth Mechanism",
+			"Token URI",
+		},
+		rows,
+	)
+}
+
+func printPlain(kafkaClusters []*entities.KafkaCluster) {
+	var names string
+	lastIndex := len(kafkaClusters) - 1
+
+	for index, cluster := range kafkaClusters {
+		names = names + refToString(cluster.Ref)
+
+		if index != lastIndex {
+			names = names + "\n"
+		}
 	}
+
+	fmt.Println(names)
+}
+
+func refToString(clusterRef *entities.KafkaClusterRef) string {
+	return fmt.Sprintf("%v/%v", clusterRef.BillingId, clusterRef.Name)
 }