kmsvc-cli/internal/cli/messages.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
}