147 lines
4.3 KiB
Go
147 lines
4.3 KiB
Go
package cli
|
|
|
|
import (
|
|
"fmt"
|
|
|
|
kmsvc "forgejo.riotpiao.com/rock/kmsvc-sdk"
|
|
"github.com/spf13/cobra"
|
|
)
|
|
|
|
func newMessageSendCmd(flags *globalFlags) *cobra.Command {
|
|
var queue, body, groupID, dedupID string
|
|
var delaySeconds int32
|
|
|
|
cmd := &cobra.Command{
|
|
Use: "send-message",
|
|
Short: "Send a message to a queue",
|
|
RunE: func(cmd *cobra.Command, args []string) error {
|
|
client, err := buildClient(cmd.Context(), flags)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer client.Close()
|
|
|
|
out, err := client.SendMessage(cmd.Context(), kmsvc.SendMessageInput{
|
|
QueueName: queue,
|
|
Body: []byte(body),
|
|
MessageGroupID: groupID,
|
|
MessageDeduplicationID: dedupID,
|
|
DelaySeconds: delaySeconds,
|
|
})
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
fmt.Fprintf(cmd.OutOrStdout(), "message_id=%s sequence_number=%s\n", out.MessageID, out.SequenceNumber)
|
|
return nil
|
|
},
|
|
}
|
|
|
|
cmd.Flags().StringVar(&queue, "queue", "", "queue name (required)")
|
|
cmd.Flags().StringVar(&body, "body", "", "message body (required)")
|
|
cmd.Flags().StringVar(&groupID, "group-id", "", "FIFO message group ID")
|
|
cmd.Flags().StringVar(&dedupID, "dedup-id", "", "FIFO message deduplication ID")
|
|
cmd.Flags().Int32Var(&delaySeconds, "delay", 0, "delay before the message becomes visible, in seconds")
|
|
cmd.MarkFlagRequired("queue")
|
|
cmd.MarkFlagRequired("body")
|
|
|
|
return cmd
|
|
}
|
|
|
|
func newMessageReceiveCmd(flags *globalFlags) *cobra.Command {
|
|
var queue string
|
|
var maxMessages, waitSeconds, visibilityTimeout int32
|
|
|
|
cmd := &cobra.Command{
|
|
Use: "receive-message",
|
|
Short: "Receive messages from a queue (long-poll)",
|
|
RunE: func(cmd *cobra.Command, args []string) error {
|
|
client, err := buildClient(cmd.Context(), flags)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer client.Close()
|
|
|
|
msgs, err := client.ReceiveMessage(cmd.Context(), queue, kmsvc.ReceiveOptions{
|
|
MaxNumberOfMessages: maxMessages,
|
|
WaitTimeSeconds: waitSeconds,
|
|
VisibilityTimeoutSeconds: visibilityTimeout,
|
|
})
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
return renderMessages(cmd.OutOrStdout(), flags.output, msgs)
|
|
},
|
|
}
|
|
|
|
cmd.Flags().StringVar(&queue, "queue", "", "queue name (required)")
|
|
cmd.Flags().Int32Var(&maxMessages, "max-messages", 1, "maximum number of messages to return (1-10)")
|
|
cmd.Flags().Int32Var(&waitSeconds, "wait", 0, "long-poll wait time in seconds (0-20)")
|
|
cmd.Flags().Int32Var(&visibilityTimeout, "visibility-timeout", 0, "override the queue's default visibility timeout, in seconds")
|
|
cmd.MarkFlagRequired("queue")
|
|
|
|
return cmd
|
|
}
|
|
|
|
func newMessageDeleteCmd(flags *globalFlags) *cobra.Command {
|
|
var queue, receiptHandle string
|
|
|
|
cmd := &cobra.Command{
|
|
Use: "delete-message",
|
|
Short: "Delete (acknowledge) a message",
|
|
RunE: func(cmd *cobra.Command, args []string) error {
|
|
client, err := buildClient(cmd.Context(), flags)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer client.Close()
|
|
|
|
if err := client.DeleteMessage(cmd.Context(), queue, receiptHandle); err != nil {
|
|
return err
|
|
}
|
|
fmt.Fprintln(cmd.OutOrStdout(), "deleted")
|
|
return nil
|
|
},
|
|
}
|
|
|
|
cmd.Flags().StringVar(&queue, "queue", "", "queue name (required)")
|
|
cmd.Flags().StringVar(&receiptHandle, "receipt-handle", "", "receipt handle from receive (required)")
|
|
cmd.MarkFlagRequired("queue")
|
|
cmd.MarkFlagRequired("receipt-handle")
|
|
|
|
return cmd
|
|
}
|
|
|
|
func newMessageChangeVisibilityCmd(flags *globalFlags) *cobra.Command {
|
|
var queue, receiptHandle string
|
|
var timeout int32
|
|
|
|
cmd := &cobra.Command{
|
|
Use: "change-message-visibility",
|
|
Short: "Change the visibility timeout of an in-flight message",
|
|
RunE: func(cmd *cobra.Command, args []string) error {
|
|
client, err := buildClient(cmd.Context(), flags)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer client.Close()
|
|
|
|
if err := client.ChangeMessageVisibility(cmd.Context(), queue, receiptHandle, timeout); err != nil {
|
|
return err
|
|
}
|
|
fmt.Fprintln(cmd.OutOrStdout(), "updated")
|
|
return nil
|
|
},
|
|
}
|
|
|
|
cmd.Flags().StringVar(&queue, "queue", "", "queue name (required)")
|
|
cmd.Flags().StringVar(&receiptHandle, "receipt-handle", "", "receipt handle from receive (required)")
|
|
cmd.Flags().Int32Var(&timeout, "timeout", 0, "new visibility timeout, in seconds (required)")
|
|
cmd.MarkFlagRequired("queue")
|
|
cmd.MarkFlagRequired("receipt-handle")
|
|
cmd.MarkFlagRequired("timeout")
|
|
|
|
return cmd
|
|
}
|