Add topic operation commands

pkg/ctl/topic/args/args_check.go

@@ -23,7 +23,7 @@ import (
 
 func CheckTopicNameTwoArgs(args []string) error {
 	if len(args) != 2 {
-		return errors.New("need to specified the topic name and the partitions")
+		return errors.New("only two argument is allowed to be used as names")
 	}
 
 	return nil

pkg/ctl/topic/compact/compact.go

@@ -0,0 +1,64 @@
+package compact
+
+import (
+	"github.com/pkg/errors"
+	"github.com/streamnative/pulsarctl/pkg/cmdutils"
+	. "github.com/streamnative/pulsarctl/pkg/ctl/topic/errors"
+	. "github.com/streamnative/pulsarctl/pkg/pulsar"
+)
+
+func CompactCmd(vc *cmdutils.VerbCmd) {
+	var desc LongDescription
+	desc.CommandUsedFor = "This command is used for compacting a persistent topic."
+	desc.CommandPermission = "This command is requires tenant admin permissions."
+
+	var examples []Example
+	compact := Example{
+		Desc: "Compact a persistent topic <topic-name>",
+		Command: "pulsarctl topic compact <topic-name>",
+	}
+	desc.CommandExamples = append(examples, compact)
+
+	var out []Output
+	successOut := Output{
+		Desc:  "normal output",
+		Out: "Sending compact topic <topic-name> request successfully",
+	}
+	out = append(out, successOut, ArgError, TopicNotFoundError)
+	out = append(out, TopicNameErrors...)
+	out = append(out, NamespaceErrors...)
+	desc.CommandOutput  = out
+
+	vc.SetDescription(
+		"compact",
+		"Compact a topic",
+		desc.ToString())
+
+	vc.SetRunFuncWithNameArg(func() error {
+		return doCompact(vc)
+	})
+}
+
+func doCompact(vc *cmdutils.VerbCmd) error {
+	// for testing
+	if vc.NameError != nil {
+		return vc.NameError
+	}
+
+	topic, err := GetTopicName(vc.NameArg)
+	if err != nil {
+		return err
+	}
+
+	if !topic.IsPersistent() {
+		return errors.New("Need to provide a persistent topic.")
+	}
+
+	admin := cmdutils.NewPulsarClient()
+	err = admin.Topics().Compact(*topic)
+	if err == nil {
+		vc.Command.Printf("Sending compact topic %s request successfully/n", topic.String())
+	}
+
+	return err
+}

pkg/ctl/topic/compact/compact_status.go

@@ -0,0 +1,103 @@
+package compact
+
+import (
+	"time"
+
+	"github.com/pkg/errors"
+	"github.com/streamnative/pulsarctl/pkg/cmdutils"
+	. "github.com/streamnative/pulsarctl/pkg/ctl/topic/errors"
+	. "github.com/streamnative/pulsarctl/pkg/pulsar"
+)
+
+func CompactStatusCmd(vc *cmdutils.VerbCmd)  {
+	var desc LongDescription
+	desc.CommandUsedFor = "This command is used for getting status of compaction on a topic."
+	desc.CommandPermission  = "This command requires tenant admin permissions."
+
+	var examples []Example
+	compactStatus := Example{
+		Desc: "Get status of compaction of a persistent topic <topic-name>",
+		Command: "pulsarctl topic compact-status <topic-name>",
+	}
+	desc.CommandExamples =  append(examples, compactStatus)
+
+	var out []Output
+	successOut := Output{
+		Desc: "normal output",
+		Out:  "Compaction was a success",
+	}
+
+	notRun := Output{
+		Desc: "Compaction is not running",
+		Out: "Compaction has not been run for <topic-name> since broker startup",
+	}
+
+	running := Output{
+		Desc: "Compaction is running",
+		Out: "Compaction is currently running",
+	}
+
+	errorOut := Output{
+		Desc: "Compaction is error",
+		Out:  "Error in compaction",
+	}
+	out = append(out, successOut, notRun, running, errorOut, ArgError, TopicNotFoundError)
+	out = append(out, TopicNameErrors...)
+	out = append(out, NamespaceErrors...)
+	desc.CommandOutput = out
+
+	vc.SetDescription(
+		"compact-status",
+		"Get status of compaction on a topic",
+		desc.ToString())
+
+	var wait bool
+
+	vc.SetRunFuncWithNameArg(func() error {
+		return doCompactStatus(vc, wait)
+	})
+}
+
+func doCompactStatus(vc *cmdutils.VerbCmd, wait bool) error {
+	// for testing
+	if vc.NameError != nil {
+		return vc.NameError
+	}
+
+	topic, err := GetTopicName(vc.NameArg)
+	if err != nil {
+		return err
+	}
+
+	if topic.GetDomain().String() != "persistent" {
+		return errors.New("Need to provide a persistent topic.")
+	}
+
+	admin := cmdutils.NewPulsarClient()
+	status, err := admin.Topics().CompactStatus(*topic)
+	if err != nil {
+		return err
+	}
+
+	for wait && status.Status  == RUNNING {
+		time.Sleep( 1 * time.Second)
+		status,  err = admin.Topics().CompactStatus(*topic)
+		if err != nil {
+			return err
+		}
+	}
+
+	switch status.Status {
+	case NOT_RUN:
+		vc.Command.Printf("Compaction has not been run for %s since broker startup/n", topic.String())
+	case RUNNING:
+		vc.Command.Printf("Compaction is currently running/n")
+	case SUCCESS:
+		vc.Command.Printf("Compaction was a success/n")
+	case ERROR:
+		vc.Command.Printf("Error in Compaction/n")
+		err = errors.New(status.LastError)
+	}
+
+	return err
+}

pkg/ctl/topic/compact/compact_status_test.go

@@ -0,0 +1,29 @@
+package compact
+
+import (
+	"testing"
+
+	. "github.com/streamnative/pulsarctl/pkg/ctl/topic/test"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestCompactStatusArgsError(t *testing.T) {
+	args := []string{"compact-status"}
+	_, _, nameErr, _ := TestTopicCommands(CompactStatusCmd, args)
+	assert.NotNil(t, nameErr)
+	assert.Equal(t, "only one argument is allowed to be used as a name", nameErr.Error())
+}
+
+func TestCompactStatusNonExistingTopicError(t *testing.T) {
+	args := []string{"compact-status", "test-non-existing-compact-status"}
+	_, execErr, _, _ := TestTopicCommands(CompactStatusCmd, args)
+	assert.NotNil(t, execErr)
+	assert.Equal(t, "code: 404 reason: Topic not found", execErr.Error())
+}
+
+func TestCompactStatusNonPersistentTopicError(t *testing.T) {
+	args := []string{"compact-status", "non-persistent://public/default/test-non-persistent-topic-compact-status"}
+	_, execErr, _, _ := TestTopicCommands(CompactStatusCmd, args)
+	assert.NotNil(t, execErr)
+	assert.Equal(t, "Need to provide a persistent topic.", execErr.Error())
+}

pkg/ctl/topic/compact/compact_test.go

@@ -0,0 +1,52 @@
+package compact
+
+import (
+	"testing"
+
+	. "github.com/streamnative/pulsarctl/pkg/ctl/topic/crud"
+	. "github.com/streamnative/pulsarctl/pkg/ctl/topic/test"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestCompactCmd(t *testing.T)  {
+	args := []string{"create", "test-compact-topic", "0"}
+	_, execErr, _, _ := TestTopicCommands(CreateTopicCmd, args)
+	assert.Nil(t, execErr)
+
+	args = []string{"compact-status", "test-compact-topic"}
+	out, execErr, _, _ := TestTopicCommands(CompactStatusCmd, args)
+	assert.Nil(t, execErr)
+	assert.Equal(t, "Compaction has not been run for " +
+		"persistent://public/default/test-compact-topic since broker startup/n", out.String())
+
+	args = []string{"compact", "test-compact-topic"}
+	_, execErr, _, _ = TestTopicCommands(CompactCmd, args)
+	assert.Nil(t, execErr)
+
+	args = []string{"compact-status", "test-compact-topic"}
+	out, execErr, _, _ = TestTopicCommands(CompactStatusCmd, args)
+	assert.Nil(t, execErr)
+
+	assert.Equal(t, "Compaction is currently running/n", out.String())
+}
+
+func TestCompactArgError(t *testing.T) {
+	args := []string{"compact"}
+	_, _, nameErr, _ := TestTopicCommands(CompactCmd, args)
+	assert.NotNil(t, nameErr)
+	assert.Equal(t, "only one argument is allowed to be used as a name", nameErr.Error())
+}
+
+func TestCompactNonExistingTopic(t *testing.T) {
+	args := []string{"compact", "test-compact-non-existing-topic"}
+	_, execErr, _, _ := TestTopicCommands(CompactCmd, args)
+	assert.NotNil(t, execErr)
+	assert.Equal(t, "code: 404 reason: Topic not found", execErr.Error())
+}
+
+func TestCompactNonPersistentTopic(t *testing.T) {
+	args  := []string{"compact", "non-persistent://public/default/test-compact-non-persistent-topic"}
+	_, execErr, _, _ := TestTopicCommands(CompactCmd, args)
+	assert.NotNil(t, execErr)
+	assert.Equal(t, "Need to provide a persistent topic.", execErr.Error())
+}

pkg/ctl/topic/crud/create_test.go

@@ -53,7 +53,7 @@ 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())
+	assert.Equal(t, "only two argument is allowed to be used as names", nameErr.Error())
 }
 
 func TestCreateTopicWithInvalidPartitions(t *testing.T) {

pkg/ctl/topic/crud/update_test.go

@@ -59,7 +59,7 @@ func TestUpdateTopicArgsError(t *testing.T) {
 	args := []string{"update", "test-topic"}
 	_, _, nameErr, _ := TestTopicCommands(UpdateTopicCmd, args)
 	assert.NotNil(t, nameErr)
-	assert.Equal(t, "need to specified the topic name and the partitions", nameErr.Error())
+	assert.Equal(t, "only two argument is allowed to be used as names", nameErr.Error())
 }
 
 func TestUpdateTopicWithInvalidPartitions(t *testing.T) {

pkg/ctl/topic/errors/errors_topic.go

@@ -26,14 +26,19 @@ var ArgError = Output{
 
 var ArgsError = Output{
 	Desc: "the topic name and(or) the partitions is not specified",
-	Out:  "[✖]  need to specified the topic name and the partitions",
+	Out:  "[✖]  only two argument is allowed to be used as names",
 }
 
 var TopicAlreadyExistError = Output{
 	Desc: "the topic has been created",
 	Out:  "[✖]  code: 409 reason: Partitioned topic already exists",
 }
 
+var TopicNotFoundError = Output{
+	Desc: "the specified topic does not found",
+	Out:  "[✖]  code: 404 reason: Topic not found",
+}
+
 var TenantNotExistError = Output{
 	Desc: "the tenant of the namespace is not exist",
 	Out:  "[✖]  code: 404 reason: Tenant does not exist",