release v3.5.1

Taskfile.yml

@@ -4,7 +4,7 @@ version: '2'
 
 vars:
   BINARY_NAME: kubemqctl
-  VERSION: v3.5.0
+  VERSION: v3.5.1
 
 tasks:
   check_update:

cmd/commands/send.go

@@ -10,7 +10,6 @@ import (
 	"github.com/kubemq-io/kubemqctl/pkg/targets"
 	"github.com/kubemq-io/kubemqctl/pkg/utils"
 	"github.com/spf13/cobra"
-	"io/ioutil"
 	"time"
 )
 
@@ -21,7 +20,7 @@ type CommandsSendOptions struct {
 	body      string
 	metadata  string
 	timeout   int
-	fileName  string
+	fileName  bool
 	build     bool
 }
 
@@ -60,7 +59,7 @@ func NewCmdCommandsSend(ctx context.Context, cfg *config.Config) *cobra.Command
 	}
 	cmd.PersistentFlags().StringVarP(&o.metadata, "metadata", "m", "", "Set metadata body")
 	cmd.PersistentFlags().IntVarP(&o.timeout, "timeout", "o", 30, "Set command timeout")
-	cmd.PersistentFlags().StringVarP(&o.fileName, "file", "f", "", "set load body from file")
+	cmd.PersistentFlags().BoolVarP(&o.fileName, "file", "f", false, "set load body from file")
 	cmd.PersistentFlags().BoolVarP(&o.build, "build", "b", false, "build kubemq targets request")
 	return cmd
 }
@@ -81,8 +80,8 @@ func (o *CommandsSendOptions) Complete(args []string, transport string) error {
 		o.body = string(data)
 		return nil
 	}
-	if o.fileName != "" {
-		data, err := ioutil.ReadFile(o.fileName)
+	if o.fileName {
+		data, err := targets.BuildFile()
 		if err != nil {
 			return err
 		}

cmd/events/send.go

@@ -11,7 +11,6 @@ import (
 	"github.com/kubemq-io/kubemqctl/pkg/targets"
 	"github.com/kubemq-io/kubemqctl/pkg/utils"
 	"github.com/spf13/cobra"
-	"io/ioutil"
 	"time"
 )
 
@@ -23,7 +22,7 @@ type EventsSendOptions struct {
 	metadata  string
 	messages  int
 	isStream  bool
-	fileName  string
+	fileName  bool
 	build     bool
 }
 
@@ -66,7 +65,7 @@ func NewCmdEventsSend(ctx context.Context, cfg *config.Config) *cobra.Command {
 	cmd.PersistentFlags().StringVarP(&o.metadata, "metadata", "", "", "set body metadata field")
 	cmd.PersistentFlags().IntVarP(&o.messages, "messages", "m", 1, "set how many 'events' messages to send")
 	cmd.PersistentFlags().BoolVarP(&o.isStream, "stream", "s", false, "set stream of all messages at once")
-	cmd.PersistentFlags().StringVarP(&o.fileName, "file", "f", "", "set body body from file")
+	cmd.PersistentFlags().BoolVarP(&o.fileName, "file", "f", false, "set body body from file")
 	cmd.PersistentFlags().BoolVarP(&o.build, "build", "b", false, "build kubemq targets request")
 	return cmd
 }
@@ -87,8 +86,8 @@ func (o *EventsSendOptions) Complete(args []string, transport string) error {
 		o.body = string(data)
 		return nil
 	}
-	if o.fileName != "" {
-		data, err := ioutil.ReadFile(o.fileName)
+	if o.fileName {
+		data, err := targets.BuildFile()
 		if err != nil {
 			return err
 		}

cmd/events_store/list.go

@@ -4,7 +4,7 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
-	"github.com/go-resty/resty"
+	"github.com/go-resty/resty/v2"
 	"github.com/kubemq-io/kubemqctl/pkg/config"
 	"github.com/kubemq-io/kubemqctl/pkg/k8s"
 	"github.com/kubemq-io/kubemqctl/pkg/utils"
@@ -68,7 +68,7 @@ func (o *EventsStoreListOptions) Run(ctx context.Context) error {
 	resp := &Response{}
 	q := &Queues{}
 
-	r, err := resty.R().SetResult(resp).SetError(resp).Get(fmt.Sprintf("%s/v1/stats/events_stores", o.cfg.GetApiHttpURI()))
+	r, err := resty.New().R().SetResult(resp).SetError(resp).Get(fmt.Sprintf("%s/v1/stats/events_stores", o.cfg.GetApiHttpURI()))
 	if err != nil {
 		return err
 	}

cmd/events_store/send.go

@@ -11,7 +11,6 @@ import (
 	"github.com/kubemq-io/kubemqctl/pkg/targets"
 	"github.com/kubemq-io/kubemqctl/pkg/utils"
 	"github.com/spf13/cobra"
-	"io/ioutil"
 	"time"
 )
 
@@ -23,7 +22,7 @@ type EventsStoreSendOptions struct {
 	metadata  string
 	messages  int
 	isStream  bool
-	fileName  string
+	fileName  bool
 	build     bool
 }
 
@@ -66,7 +65,7 @@ func NewCmdEventsStoreSend(ctx context.Context, cfg *config.Config) *cobra.Comma
 	cmd.PersistentFlags().StringVarP(&o.metadata, "metadata", "", "", "set body metadata field")
 	cmd.PersistentFlags().IntVarP(&o.messages, "messages", "m", 1, "set how many 'events store' messages to send")
 	cmd.PersistentFlags().BoolVarP(&o.isStream, "stream", "s", false, "set stream of all messages at once")
-	cmd.PersistentFlags().StringVarP(&o.fileName, "file", "f", "", "set load body from file")
+	cmd.PersistentFlags().BoolVarP(&o.fileName, "file", "f", false, "set load body from file")
 	cmd.PersistentFlags().BoolVarP(&o.build, "build", "b", false, "build kubemq targets request")
 	return cmd
 }
@@ -87,8 +86,8 @@ func (o *EventsStoreSendOptions) Complete(args []string, transport string) error
 		o.body = string(data)
 		return nil
 	}
-	if o.fileName != "" {
-		data, err := ioutil.ReadFile(o.fileName)
+	if o.fileName {
+		data, err := targets.BuildFile()
 		if err != nil {
 			return err
 		}

cmd/queries/send.go

@@ -10,7 +10,6 @@ import (
 	"github.com/kubemq-io/kubemqctl/pkg/targets"
 	"github.com/kubemq-io/kubemqctl/pkg/utils"
 	"github.com/spf13/cobra"
-	"io/ioutil"
 	"time"
 )
 
@@ -23,7 +22,7 @@ type QueriesSendOptions struct {
 	timeout   int
 	cacheKey  string
 	cacheTTL  time.Duration
-	fileName  string
+	fileName  bool
 	build     bool
 }
 
@@ -67,7 +66,7 @@ func NewCmdQueriesSend(ctx context.Context, cfg *config.Config) *cobra.Command {
 	cmd.PersistentFlags().StringVarP(&o.cacheKey, "cache-key", "c", "", "set query cache key")
 	cmd.PersistentFlags().IntVarP(&o.timeout, "timeout", "o", 30, "set query timeout")
 	cmd.PersistentFlags().DurationVarP(&o.cacheTTL, "cache-duration", "d", 10*time.Minute, "set cache duration timeout")
-	cmd.PersistentFlags().StringVarP(&o.fileName, "file", "f", "", "set load body from file")
+	cmd.PersistentFlags().BoolVarP(&o.fileName, "file", "f", false, "set load body from file")
 	cmd.PersistentFlags().BoolVarP(&o.build, "build", "b", false, "build kubemq targets request")
 	return cmd
 }
@@ -88,8 +87,8 @@ func (o *QueriesSendOptions) Complete(args []string, transport string) error {
 		o.body = string(data)
 		return nil
 	}
-	if o.fileName != "" {
-		data, err := ioutil.ReadFile(o.fileName)
+	if o.fileName {
+		data, err := targets.BuildFile()
 		if err != nil {
 			return err
 		}

cmd/queue/attach.go

@@ -4,7 +4,7 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
-	"github.com/go-resty/resty"
+	"github.com/go-resty/resty/v2"
 	"github.com/kubemq-io/kubemqctl/pkg/attach"
 	"github.com/kubemq-io/kubemqctl/pkg/config"
 	"github.com/kubemq-io/kubemqctl/pkg/k8s"
@@ -75,7 +75,7 @@ func (o *QueueAttachOptions) Complete(args []string, transport string) error {
 		resp := &Response{}
 		queues := &Queues{}
 
-		r, err := resty.R().SetResult(resp).SetError(resp).Get(fmt.Sprintf("%s/v1/stats/queues", o.cfg.GetApiHttpURI()))
+		r, err := resty.New().R().SetResult(resp).SetError(resp).Get(fmt.Sprintf("%s/v1/stats/queues", o.cfg.GetApiHttpURI()))
 		if err != nil {
 			return err
 		}

cmd/queue/helper.go

@@ -87,3 +87,29 @@ func printItems(items []*kubemq.QueueMessage) {
 func printQueueMessage(msg *kubemq.QueueMessage) {
 	fmt.Println(newQueueMessageObject(msg))
 }
+
+type resultObj struct {
+	MessageID    string `json:"message_id"`
+	SentAt       string `json:"sent_at,omitempty"`
+	ExpirationAt string `json:"expiration_at,omitempty"`
+	DelayedTo    string `json:"delayed_to,omitempty"`
+	IsError      bool   `json:"is_error,omitempty"`
+	Error        string `json:"error,omitempty"`
+}
+
+func printQueueMessageResult(res *kubemq.SendQueueMessageResult) {
+	obj := &resultObj{
+		MessageID: res.MessageID,
+		SentAt:    time.Unix(0, res.SentAt).Format("2006-01-02 15:04:05.999"),
+		IsError:   res.IsError,
+		Error:     res.Error,
+	}
+	if res.ExpirationAt > 0 {
+		obj.ExpirationAt = time.Unix(0, res.ExpirationAt).Format("2006-01-02 15:04:05.999")
+	}
+	if res.DelayedTo > 0 {
+		obj.DelayedTo = time.Unix(0, res.DelayedTo).Format("2006-01-02 15:04:05.999")
+	}
+	data, _ := json.MarshalIndent(obj, "", "    ")
+	fmt.Println(string(data))
+}

cmd/queue/list.go

@@ -4,7 +4,7 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
-	"github.com/go-resty/resty"
+	"github.com/go-resty/resty/v2"
 	"github.com/kubemq-io/kubemqctl/pkg/config"
 	"github.com/kubemq-io/kubemqctl/pkg/k8s"
 	"github.com/kubemq-io/kubemqctl/pkg/utils"
@@ -68,7 +68,7 @@ func (o *QueueListOptions) Run(ctx context.Context) error {
 	resp := &Response{}
 	q := &Queues{}
 
-	r, err := resty.R().SetResult(resp).SetError(resp).Get(fmt.Sprintf("%s/v1/stats/queues", o.cfg.GetApiHttpURI()))
+	r, err := resty.New().R().SetResult(resp).SetError(resp).Get(fmt.Sprintf("%s/v1/stats/queues", o.cfg.GetApiHttpURI()))
 	if err != nil {
 		return err
 	}

cmd/queue/send.go

@@ -9,7 +9,6 @@ import (
 	"github.com/kubemq-io/kubemqctl/pkg/targets"
 	"github.com/kubemq-io/kubemqctl/pkg/utils"
 	"github.com/spf13/cobra"
-	"io/ioutil"
 )
 
 type QueueSendOptions struct {
@@ -23,7 +22,7 @@ type QueueSendOptions struct {
 	metadata   string
 	deadLetter string
 	messages   int
-	fileName   string
+	fileName   bool
 	build      bool
 }
 
@@ -75,7 +74,7 @@ func NewCmdQueueSend(ctx context.Context, cfg *config.Config) *cobra.Command {
 	cmd.PersistentFlags().IntVarP(&o.messages, "messages", "m", 1, "set dead-letter max receive count")
 	cmd.PersistentFlags().StringVarP(&o.deadLetter, "dead-letter-queue", "q", "", "set dead-letter queue name")
 	cmd.PersistentFlags().StringVarP(&o.metadata, "metadata", "", "", "set queue message metadata field")
-	cmd.PersistentFlags().StringVarP(&o.fileName, "file", "f", "", "set load message body from file")
+	cmd.PersistentFlags().BoolVarP(&o.fileName, "file", "f", false, "set load message body from file")
 	cmd.PersistentFlags().BoolVarP(&o.build, "build", "b", false, "build kubemq targets request")
 
 	return cmd
@@ -97,8 +96,8 @@ func (o *QueueSendOptions) Complete(args []string, transport string) error {
 		o.body = string(data)
 		return nil
 	}
-	if o.fileName != "" {
-		data, err := ioutil.ReadFile(o.fileName)
+	if o.fileName {
+		data, err := targets.BuildFile()
 		if err != nil {
 			return err
 		}
@@ -137,18 +136,15 @@ func (o *QueueSendOptions) Run(ctx context.Context) error {
 			SetPolicyDelaySeconds(o.delay).
 			SetPolicyMaxReceiveCount(o.maxReceive).
 			SetPolicyMaxReceiveQueue(o.deadLetter)
-		fmt.Println("Sending Queue Message:")
+		utils.Println("Sending Queue Message:")
+		printQueueMessage(msg)
 		res, err := msg.Send(ctx)
 		if err != nil {
-			return fmt.Errorf("sending queue message, %s", err.Error())
+			return fmt.Errorf("error sending queue message, %s", err.Error())
 		}
-
 		if res != nil {
-			if res.IsError {
-				return fmt.Errorf("sending queue message response, %s", res.Error)
-			}
-
-			printQueueMessage(msg)
+			utils.Println("Response:")
+			printQueueMessageResult(res)
 		}
 	}