Add partitioned topic command CURD

main.go

@@ -7,6 +7,7 @@ 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/tenant"
 	"os"
@@ -65,6 +66,7 @@ func addCommands(flagGrouping *cmdutils.FlagGrouping) {
 	rootCmd.AddCommand(tenant.Command(flagGrouping))
 	rootCmd.AddCommand(completion.Command(rootCmd))
 	rootCmd.AddCommand(functions.Command(flagGrouping))
+	rootCmd.AddCommand(topic.Command(flagGrouping))
 }
 
 func main() {

pkg/cmdutils/config.go

@@ -1,9 +1,9 @@
 package cmdutils
 
 import (
+	`log`
 	"github.com/spf13/pflag"
 	"github.com/streamnative/pulsarctl/pkg/pulsar"
-	`log`
 )
 
 var PulsarCtlConfig = ClusterConfig{}

pkg/cmdutils/verb.go

@@ -28,6 +28,12 @@ func AddVerbCmd(flagGrouping *FlagGrouping, parentResourceCmd *cobra.Command, ne
 	parentResourceCmd.AddCommand(verb.Command)
 }
 
+func AddVerbCmds(flagGrouping *FlagGrouping, parentResourceCmd *cobra.Command, newVerbCmd ...func(cmd *VerbCmd))  {
+	for _, cmd := range newVerbCmd {
+		AddVerbCmd(flagGrouping, parentResourceCmd, cmd)
+	}
+}
+
 // SetDescription sets usage along with short and long descriptions as well as aliases
 func (vc *VerbCmd) SetDescription(use, short, long string, aliases ...string) {
 	vc.Command.Use = use
@@ -51,7 +57,7 @@ func (vc *VerbCmd) SetRunFuncWithNameArg(cmd func() error) {
 	}
 }
 
-func (vc *VerbCmd) SetRunFuncWithNameArgs(cmd func() error, checkArgs func(args []string) error) {
+func (vc *VerbCmd) SetRunFuncWithMultiNameArgs(cmd func() error, checkArgs func(args []string) error) {
 	vc.Command.Run = func(_ *cobra.Command, args []string) {
 		vc.NameArgs, vc.NameError = GetNameArgs(args, checkArgs)
 		run(cmd)

pkg/ctl/cluster/create_failure_domain.go

@@ -50,7 +50,7 @@ func createFailureDomainCmd(vc *cmdutils.VerbCmd) {
 
 	var failureDomainData pulsar.FailureDomainData
 
-	vc.SetRunFuncWithNameArgs(func() error {
+	vc.SetRunFuncWithMultiNameArgs(func() error {
 		return doCreateFailureDomain(vc, &failureDomainData)
 	}, checkFailureDomainArgs)
 

pkg/ctl/cluster/delete_failure_domain.go

@@ -46,7 +46,7 @@ func deleteFailureDomainCmd(vc *cmdutils.VerbCmd) {
 		desc.ToString(),
 		"dfd")
 
-	vc.SetRunFuncWithNameArgs(func() error {
+	vc.SetRunFuncWithMultiNameArgs(func() error {
 		return doDeleteFailureDomain(vc)
 	}, checkFailureDomainArgs)
 }

pkg/ctl/cluster/get_failure_domain.go

@@ -38,7 +38,7 @@ func getFailureDomainCmd(vc *cmdutils.VerbCmd) {
 		desc.ToString(),
 		"gfd")
 
-	vc.SetRunFuncWithNameArgs(func() error {
+	vc.SetRunFuncWithMultiNameArgs(func() error {
 		return doGetFailureDomain(vc)
 	}, checkFailureDomainArgs)
 }

pkg/ctl/cluster/list_failure_domain_test.go

@@ -44,7 +44,7 @@ func TestListFailureDomainsCmd(t *testing.T) {
 	assert.Equal(t, "list-failure-broker-B", brokerMap["list-failure-B"].BrokerList[0])
 }
 
-func TestListFailureArgsError(t *testing.T)  {
+func TestListFailureArgsError(t *testing.T) {
 	args := []string{"list-failure-domains"}
 	_, _, nameErr, _ := TestClusterCommands(listFailureDomainCmd, args)
 	assert.NotNil(t, nameErr)

pkg/ctl/cluster/update_failure_domain.go

@@ -50,7 +50,7 @@ func updateFailureDomainCmd(vc *cmdutils.VerbCmd) {
 
 	var failureDomainData pulsar.FailureDomainData
 
-	vc.SetRunFuncWithNameArgs(func() error {
+	vc.SetRunFuncWithMultiNameArgs(func() error {
 		return doUpdateFailureDomain(vc, &failureDomainData)
 	}, checkFailureDomainArgs)
 

pkg/ctl/functions/putstate.go

@@ -101,7 +101,7 @@ func putstateFunctionsCmd(vc *cmdutils.VerbCmd) {
 	functionData := &pulsar.FunctionData{}
 
 	// set the run function
-	vc.SetRunFuncWithNameArgs(func() error {
+	vc.SetRunFuncWithMultiNameArgs(func() error {
 		return doPutStateFunction(vc, functionData)
 	}, checkPutStateArgs)
 

pkg/ctl/topic/args/args_check.go

@@ -0,0 +1,13 @@
+package args
+
+import (
+	"github.com/pkg/errors"
+)
+
+func CheckTopicNameTwoArgs(args []string) error {
+	if len(args) != 2 {
+		return errors.New("need to specified the topic name and the partitions")
+	}
+
+	return nil
+}

pkg/ctl/topic/crud/create.go

@@ -0,0 +1,75 @@
+package crud
+
+import (
+	"github.com/pkg/errors"
+	"github.com/streamnative/pulsarctl/pkg/cmdutils"
+	. "github.com/streamnative/pulsarctl/pkg/ctl/topic/args"
+	. "github.com/streamnative/pulsarctl/pkg/ctl/topic/errors"
+	"github.com/streamnative/pulsarctl/pkg/pulsar"
+	"strconv"
+)
+
+func CreateTopicCmd(vc *cmdutils.VerbCmd) {
+	var desc pulsar.LongDescription
+	desc.CommandUsedFor = "This command is used for creating topic."
+	desc.CommandPermission = "This command requires namespace admin permissions."
+
+	var examples []pulsar.Example
+	createNonPartitions := pulsar.Example{
+		Desc:    "Create a non-partitioned topic <topic-name>",
+		Command: "pulsarctl topics create <topic-name> 0",
+	}
+	examples = append(examples, createNonPartitions)
+
+	create := pulsar.Example{
+		Desc:    "Create a partitioned topic <topic-name> with <partitions-num> partitions",
+		Command: "pulsarctl topics create <topic-name> <partition-num>",
+	}
+	examples = append(examples, create)
+	desc.CommandExamples = examples
+
+	var out []pulsar.Output
+	successOut := pulsar.Output{
+		Desc: "normal output",
+		Out:  "Create topic <topic-name> with <partition-num> partitions successfully",
+	}
+	out = append(out, successOut, ArgsError, TopicAlreadyExistError)
+	out = append(out, TopicNameErrors...)
+	out = append(out, NamespaceErrors...)
+	desc.CommandOutput = out
+
+	vc.SetDescription(
+		"create",
+		"Create a topic with n partitions",
+		desc.ToString(),
+		"c")
+
+	vc.SetRunFuncWithMultiNameArgs(func() error {
+		return doCreateTopic(vc)
+	}, CheckTopicNameTwoArgs)
+}
+
+func doCreateTopic(vc *cmdutils.VerbCmd) error {
+	// for testing
+	if vc.NameError != nil {
+		return vc.NameError
+	}
+
+	topic, err := pulsar.GetTopicName(vc.NameArgs[0])
+	if err != nil {
+		return err
+	}
+
+	partitions, err := strconv.Atoi(vc.NameArgs[1])
+	if err != nil || partitions < 0 {
+		return errors.Errorf("invalid partition number '%s'", vc.NameArgs[1])
+	}
+
+	admin := cmdutils.NewPulsarClient()
+	err = admin.Topics().Create(*topic, partitions)
+	if err == nil {
+		vc.Command.Printf("Create topic %s with %d partitions successfully\n", topic.String(), partitions)
+	}
+
+	return err
+}

pkg/ctl/topic/crud/create_test.go

@@ -0,0 +1,68 @@
+package crud
+
+import (
+	. "github.com/streamnative/pulsarctl/pkg/ctl/topic/test"
+	"github.com/stretchr/testify/assert"
+	"testing"
+)
+
+func TestCreateTopicCmd(t *testing.T) {
+	args := []string{"create", "test-create-topic", "2"}
+	_, execErr, argsErr, err := TestTopicCommands(CreateTopicCmd, args)
+	assert.Nil(t, execErr)
+	assert.Nil(t, argsErr)
+	assert.Nil(t, err)
+}
+
+func TestCreateNonPersistentTopic(t *testing.T) {
+	args := []string{"create", "non-persistent://public/default/test-create-topic", "2"}
+	_, execErr, argsErr, err := TestTopicCommands(CreateTopicCmd, args)
+	assert.Nil(t, execErr)
+	assert.Nil(t, argsErr)
+	assert.Nil(t, err)
+}
+
+func TestCreateTopicAlreadExists(t *testing.T) {
+	args := []string{"create", "test-duplicate-topic", "2"}
+	_, _, _, err := TestTopicCommands(CreateTopicCmd, args)
+	assert.Nil(t, err)
+
+	_, execErr, _, _ := TestTopicCommands(CreateTopicCmd, args)
+	assert.NotNil(t, execErr)
+	assert.Equal(t, "code: 409 reason: Partitioned topic already exists", execErr.Error())
+}
+
+func TestCreateTopicArgsError(t *testing.T) {
+	args := []string{"create", "topic"}
+	_, _, nameErr, _ := TestTopicCommands(CreateTopicCmd, args)
+	assert.NotNil(t, nameErr)
+	assert.Equal(t, "need to specified the topic name and the partitions", nameErr.Error())
+}
+
+func TestCreateTopicWithInvalidPartitions(t *testing.T) {
+	args := []string{"create", "topic", "a"}
+	_, execErr, _, _ := TestTopicCommands(CreateTopicCmd, args)
+	assert.NotNil(t, execErr)
+	assert.Equal(t, "invalid partition number 'a'", execErr.Error())
+
+	args = []string{"create", "topic", "--", "-1"}
+	_, execErr, _, _ = TestTopicCommands(CreateTopicCmd, args)
+	assert.NotNil(t, execErr)
+	assert.Equal(t, "invalid partition number '-1'", execErr.Error())
+}
+
+func TestCreateNonPartitionedTopic(t *testing.T) {
+	args := []string{"create", "test-create-non-partitioned-topic", "0"}
+	_, execErr, argsErr, err := TestTopicCommands(CreateTopicCmd, args)
+	assert.Nil(t, execErr)
+	assert.Nil(t, argsErr)
+	assert.Nil(t, err)
+}
+
+func TestCreateNonPersistentNonPartitionedTopic(t *testing.T) {
+	args := []string{"create", "non-persistent://public/default/test-create-non-partitioned-topic", "0"}
+	_, execErr, argsErr, err := TestTopicCommands(CreateTopicCmd, args)
+	assert.Nil(t, execErr)
+	assert.Nil(t, argsErr)
+	assert.Nil(t, err)
+}

pkg/ctl/topic/crud/delete.go

@@ -0,0 +1,91 @@
+package crud
+
+import (
+	"github.com/spf13/pflag"
+	"github.com/streamnative/pulsarctl/pkg/cmdutils"
+	. "github.com/streamnative/pulsarctl/pkg/ctl/topic/errors"
+	"github.com/streamnative/pulsarctl/pkg/pulsar"
+)
+
+func DeleteTopicCmd(vc *cmdutils.VerbCmd) {
+	var desc pulsar.LongDescription
+	desc.CommandUsedFor = "This command is used for deleting an existing topic."
+	desc.CommandPermission = "This command requires namespace admin permissions."
+
+	var examples []pulsar.Example
+	deleteTopic := pulsar.Example{
+		Desc:    "Delete a partitioned topic <topic-name>",
+		Command: "pulsarctl topics delete <topic-name>",
+	}
+
+	deleteNonPartitionedTopic := pulsar.Example{
+		Desc:    "Delete a non-partitioned topic <topic-name>",
+		Command: "pulsarctl topics delete --non-partitioned <topic-name>",
+	}
+
+	desc.CommandExamples = append(examples, deleteTopic, deleteNonPartitionedTopic)
+	var out []pulsar.Output
+	successOut := pulsar.Output{
+		Desc: "normal output",
+		Out:  "Delete topic <topic-name> successfully",
+	}
+
+	partitionedTopicNotExistError := pulsar.Output{
+		Desc: "the partitioned topic does not exist",
+		Out:  "[✖]  code: 404 reason: Partitioned topic does not exist",
+	}
+
+	nonPartitionedTopicNotExistError := pulsar.Output{
+		Desc: "the non-partitioned topic does not exist",
+		Out:  "[✖]  code: 404 reason: Topic not found",
+	}
+	out = append(out, successOut, ArgError,
+		partitionedTopicNotExistError, nonPartitionedTopicNotExistError)
+	out = append(out, TopicNameErrors...)
+	out = append(out, NamespaceErrors...)
+	desc.CommandOutput = out
+
+	vc.SetDescription(
+		"delete",
+		"Delete a topic",
+		desc.ToString(),
+		"d")
+
+	var force bool
+	var deleteSchema bool
+	var nonPartitioned bool
+
+	vc.FlagSetGroup.InFlagSet("Delete Topic", func(set *pflag.FlagSet) {
+		set.BoolVarP(&nonPartitioned, "non-partitioned", "n", false,
+			"Delete a non-partitioned topic")
+		set.BoolVarP(&force, "force", "f", false,
+			"Close all producer/consumer/replicator and delete topic forcefully")
+		set.BoolVarP(&deleteSchema, "delete-schema", "d", false,
+			"Delete schema while deleting topic")
+	})
+
+	vc.SetRunFuncWithNameArg(func() error {
+		return doDeleteTopic(vc, force, deleteSchema, nonPartitioned)
+	})
+}
+
+// TODO add delete schema
+func doDeleteTopic(vc *cmdutils.VerbCmd, force, deleteSchema, nonPartitioned bool) error {
+	// for testing
+	if vc.NameError != nil {
+		return vc.NameError
+	}
+
+	topic, err := pulsar.GetTopicName(vc.NameArg)
+	if err != nil {
+		return err
+	}
+
+	admin := cmdutils.NewPulsarClient()
+	err = admin.Topics().Delete(*topic, force, nonPartitioned)
+	if err == nil {
+		vc.Command.Printf("Delete topic %s successfully\n", topic.String())
+	}
+
+	return err
+}