Add schemas commands for pulsarctl

main.go

@@ -7,10 +7,11 @@ import (
 	"github.com/streamnative/pulsarctl/pkg/cmdutils"
 	"github.com/streamnative/pulsarctl/pkg/ctl/cluster"
 	"github.com/streamnative/pulsarctl/pkg/ctl/completion"
-	"github.com/streamnative/pulsarctl/pkg/ctl/topic"
 	"github.com/streamnative/pulsarctl/pkg/ctl/functions"
-    "github.com/streamnative/pulsarctl/pkg/ctl/sources"
-    "github.com/streamnative/pulsarctl/pkg/ctl/tenant"
+	"github.com/streamnative/pulsarctl/pkg/ctl/schemas"
+	"github.com/streamnative/pulsarctl/pkg/ctl/sources"
+	"github.com/streamnative/pulsarctl/pkg/ctl/tenant"
+	"github.com/streamnative/pulsarctl/pkg/ctl/topic"
 	"os"
 )
 
@@ -69,6 +70,7 @@ func addCommands(flagGrouping *cmdutils.FlagGrouping) {
 	rootCmd.AddCommand(functions.Command(flagGrouping))
 	rootCmd.AddCommand(sources.Command(flagGrouping))
 	rootCmd.AddCommand(topic.Command(flagGrouping))
+	rootCmd.AddCommand(schemas.Command(flagGrouping))
 }
 
 func main() {

pkg/ctl/functions/trigger_test.go

@@ -18,9 +18,9 @@
 package functions
 
 import (
-	`bytes`
-	`encoding/json`
-	`github.com/streamnative/pulsarctl/pkg/pulsar`
+	"bytes"
+	"encoding/json"
+	"github.com/streamnative/pulsarctl/pkg/pulsar"
 	"github.com/stretchr/testify/assert"
 	"testing"
 	"time"

pkg/ctl/schemas/delete.go

@@ -0,0 +1,71 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package schemas
+
+import (
+	"github.com/streamnative/pulsarctl/pkg/cmdutils"
+	"github.com/streamnative/pulsarctl/pkg/pulsar"
+)
+
+func deleteSchema(vc *cmdutils.VerbCmd) {
+	desc := pulsar.LongDescription{}
+	desc.CommandUsedFor = "Delete the latest schema for a topic"
+	desc.CommandPermission = "This command requires namespace admin permissions."
+
+	var examples []pulsar.Example
+	del := pulsar.Example{
+		Desc:    "Delete the latest schema for a topic",
+		Command: "pulsarctl schemas delete <topic name>",
+	}
+	examples = append(examples, del)
+	desc.CommandExamples = examples
+
+	var out []pulsar.Output
+	successOut := pulsar.Output{
+		Desc: "normal output",
+		Out:  "Deleted <topic name> successfully",
+	}
+
+	notTopicName := pulsar.Output{
+		Desc: "you must specify a topic name, please check if the topic name is provided",
+		Out:  "[✖]  only one argument is allowed to be used as a name",
+	}
+	out = append(out, successOut, notTopicName)
+	desc.CommandOutput = out
+
+	vc.SetDescription(
+		"delete",
+		"Delete the latest schema for a topic",
+		desc.ToString(),
+		"delete",
+	)
+
+	vc.SetRunFuncWithNameArg(func() error {
+		return doDeleteSchema(vc)
+	})
+}
+
+func doDeleteSchema(vc *cmdutils.VerbCmd) error {
+	topic := vc.NameArg
+	admin := cmdutils.NewPulsarClient()
+	err := admin.Schemas().DeleteSchema(topic)
+	if err == nil {
+		vc.Command.Printf("Deleted %s successfully", topic)
+	}
+	return err
+}

pkg/ctl/schemas/get.go

@@ -0,0 +1,128 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package schemas
+
+import (
+	"github.com/spf13/pflag"
+	"github.com/streamnative/pulsarctl/pkg/cmdutils"
+	"github.com/streamnative/pulsarctl/pkg/pulsar"
+)
+
+func getSchema(vc *cmdutils.VerbCmd) {
+	desc := pulsar.LongDescription{}
+	desc.CommandUsedFor = "Get the schema for a topic."
+	desc.CommandPermission = "This command requires namespace admin permissions."
+
+	vc.SetDescription(
+		"get",
+		"Get the schema for a topic",
+		desc.ToString(),
+		"get",
+	)
+
+	var examples []pulsar.Example
+	del := pulsar.Example{
+		Desc:    "Get the schema for a topic",
+		Command: "pulsarctl schemas get <topic name>",
+	}
+
+	delWithVersion := pulsar.Example{
+		Desc: "Get the schema for a topic with version",
+		Command: "pulsarctl schemas get <topic name> \n" +
+			"\t--version 2",
+	}
+
+	examples = append(examples, del, delWithVersion)
+	desc.CommandExamples = examples
+
+	var out []pulsar.Output
+	successOut := pulsar.Output{
+		Desc: "normal output",
+		Out: "{\n" +
+			"  \"name\": \"test-schema\",\n" +
+			"  \"schema\": {\n" +
+			"    \"type\": \"record\",\n" +
+			"    \"name\": \"Test\",\n" +
+			"    \"fields\": [\n" +
+			"      {\n" +
+			"        \"name\": \"id\",\n" +
+			"        \"type\": [\n" +
+			"          \"null\",\n" +
+			"          \"int\"\n" +
+			"        ]\n" +
+			"      },\n" +
+			"      {\n" +
+			"        \"name\": \"name\",\n" +
+			"        \"type\": [\n" +
+			"          \"null\",\n" +
+			"          \"string\"\n" +
+			"        ]\n" +
+			"      }\n" +
+			"    ]\n" +
+			"  },\n" +
+			"  \"type\": \"AVRO\",\n" +
+			"  \"properties\": {}\n" +
+			"}",
+	}
+
+	failOut := pulsar.Output{
+		Desc: "HTTP 404 Not Found, please check if the topic name you entered is correct",
+		Out:  "[✖]  code: 404 reason: Not Found",
+	}
+
+	notTopicName := pulsar.Output{
+		Desc: "you must specify a topic name, please check if the topic name is provided",
+		Out:  "[✖]  only one argument is allowed to be used as a name",
+	}
+
+	out = append(out, successOut, failOut, notTopicName)
+	desc.CommandOutput = out
+
+	schemaData := &pulsar.SchemaData{}
+
+	vc.SetRunFuncWithNameArg(func() error {
+		return doGetSchema(vc, schemaData)
+	})
+
+	vc.FlagSetGroup.InFlagSet("SchemaConfig", func(flagSet *pflag.FlagSet) {
+		flagSet.Int64Var(
+			&schemaData.Version,
+			"version",
+			0,
+			"the schema version info")
+	})
+}
+
+func doGetSchema(vc *cmdutils.VerbCmd, schemaData *pulsar.SchemaData) error {
+	topic := vc.NameArg
+
+	admin := cmdutils.NewPulsarClient()
+	if schemaData.Version == 0 {
+		schemaInfoWithVersion, err := admin.Schemas().GetSchemaInfoWithVersion(topic)
+		if err == nil {
+			PrintSchema(vc.Command.OutOrStdout(), schemaInfoWithVersion)
+		}
+		return err
+	}
+	info, err := admin.Schemas().GetSchemaInfoByVersion(topic, schemaData.Version)
+	if err == nil {
+		cmdutils.PrintJson(vc.Command.OutOrStdout(), info)
+	}
+
+	return err
+}

pkg/ctl/schemas/schema_test.go

@@ -0,0 +1,69 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package schemas
+
+import (
+	"github.com/stretchr/testify/assert"
+	"os"
+	"strings"
+	"testing"
+)
+
+func TestSchema(t *testing.T) {
+	fileName := "avro-schema"
+	f, err := os.Create(fileName)
+	assert.Nil(t, err)
+	defer os.Remove(fileName)
+
+	_, err = f.WriteString("{\n" +
+		"   \"type\": \"AVRO\",\n" +
+		"   \"schema\": \"{\\\"type\\\":\\\"record\\\",\\\"name\\\":\\\"Test\\\",\\\"fields\\\":[{\\\"name\\\":\\\"id\\\",\\\"type\\\":[\\\"null\\\",\\\"int\\\"]},{\\\"name\\\":\\\"name\\\",\\\"type\\\":[\\\"null\\\",\\\"string\\\"]}]}\",\n" +
+		"   \"properties\": {}\n" +
+		"}\n")
+	assert.Nil(t, err)
+
+	args := []string{"upload", "test-schema", "-f", fileName}
+	out, _, err := TestSchemasCommands(uploadSchema, args)
+	assert.Nil(t, err)
+	assert.Equal(t, "Upload test-schema successfully", out.String())
+
+	getArgs := []string{"get", "test-schema"}
+	getOut, _, err := TestSchemasCommands(getSchema, getArgs)
+
+	t.Log(getOut.String())
+	assert.Nil(t, err)
+	assert.True(t, strings.Contains(getOut.String(), "AVRO"))
+	assert.True(t, strings.Contains(getOut.String(), "test-schema"))
+
+	delArgs := []string{"delete", "test-schema"}
+	delOut, _, err := TestSchemasCommands(deleteSchema, delArgs)
+	assert.Nil(t, err)
+	assert.Equal(t, delOut.String(), "Deleted test-schema successfully")
+}
+
+func TestFailSchema(t *testing.T) {
+	getArgs := []string{"get", "fail-schema"}
+	_, execErr, _ := TestSchemasCommands(getSchema, getArgs)
+	assert.NotNil(t, execErr)
+
+	uploadErr := "open avro-schema: no such file or directory"
+	args := []string{"upload", "test-schema", "-f", "avro-schema"}
+	_, execErr, _ = TestSchemasCommands(uploadSchema, args)
+	assert.NotNil(t, execErr)
+	assert.Equal(t, execErr.Error(), uploadErr)
+}

pkg/ctl/schemas/schemas.go

@@ -0,0 +1,72 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package schemas
+
+import (
+	"bytes"
+	"encoding/json"
+	"fmt"
+	"github.com/spf13/cobra"
+	"github.com/streamnative/pulsarctl/pkg/cmdutils"
+	"github.com/streamnative/pulsarctl/pkg/pulsar"
+	"io"
+)
+
+func Command(flagGrouping *cmdutils.FlagGrouping) *cobra.Command {
+	resourceCmd := cmdutils.NewResourceCmd(
+		"schemas",
+		"Operations related to Schemas associated with Pulsar topics",
+		"",
+		"schemas",
+	)
+
+	cmdutils.AddVerbCmd(flagGrouping, resourceCmd, getSchema)
+	cmdutils.AddVerbCmd(flagGrouping, resourceCmd, deleteSchema)
+	cmdutils.AddVerbCmd(flagGrouping, resourceCmd, uploadSchema)
+
+	return resourceCmd
+}
+
+func PrintSchema(w io.Writer, schema *pulsar.SchemaInfoWithVersion) {
+	name, err := json.MarshalIndent(schema.SchemaInfo.Name, "", "  ")
+	if err != nil {
+		fmt.Fprintf(w, "unexpected response type: %v\n", err)
+		return
+	}
+
+	schemaType, err := json.MarshalIndent(schema.SchemaInfo.Type, "", "  ")
+	if err != nil {
+		fmt.Fprintf(w, "unexpected response type: %v\n", err)
+		return
+	}
+
+	properties, err := json.MarshalIndent(schema.SchemaInfo.Properties, "", "  ")
+	if err != nil {
+		fmt.Fprintf(w, "unexpected response type: %v\n", err)
+		return
+	}
+	s, err := prettyPrint(schema.SchemaInfo.Schema)
+	fmt.Fprintf(w, "{\n  name: %s \n  schema: %s\n  type: %s \n  properties: %s\n}", string(name), string(s),
+		string(schemaType), string(properties))
+}
+
+func prettyPrint(b []byte) ([]byte, error) {
+	var out bytes.Buffer
+	err := json.Indent(&out, b, "", "    ")
+	return out.Bytes(), err
+}