cadmin/server/cmd/queue.go

package cmd

import (
	"context"
	"gadmin/internal/admin/consts"
	"gadmin/internal/admin/library/nats"
	"gadmin/internal/queues"
	"gadmin/internal/queues/subscribes"
	"os"
	"os/signal"

	"github.com/sirupsen/logrus"
	"github.com/spf13/cobra"
)

var queueCmd = &cobra.Command{
	Use:   "queue",
	Short: "消息队列消费者监听",
	Long:  ``,
	Run: func(cmd *cobra.Command, args []string) {
		logrus.Info("queue called")

		ctx, cancel := context.WithCancel(context.Background())
		defer cancel()
		go func(ctx context.Context) {
			client := nats.New(os.Getenv("NATS_URL"), os.Getenv("NATS_STREAM"))
			if client == nil {
				logrus.Panic("nats initialization failed")
			}

			subMap := map[string]queues.SubStrategy{
				client.GenTopicLabel(consts.NatsTopicAdminLog): subscribes.AdminLog, // 后台日志
			}
			client.AddSubscriber(subMap).Subscribe(ctx)
		}(ctx)

		c := make(chan os.Signal, 1)
		signal.Notify(c, os.Interrupt, os.Kill)
		sig := <-c
		logrus.Printf("queue closing down (signal: %v)", sig)
	},
}

func init() {
	rootCmd.AddCommand(queueCmd)
}