FranklinWH-Toolbox/service/cache/redisx/streamx.go

package redisx

import (
	"context"
	"time"

	"github.com/redis/go-redis/v9"
)

// Stream（流）封装

// 添加消息
func XAdd(ctx context.Context, stream string, values map[string]interface{}) (string, error) {
	return getClient().XAdd(ctx, &redis.XAddArgs{
		Stream: stream,
		Values: values,
	}).Result()
}

// 读取消息
func XRead(ctx context.Context, streams map[string]string, count int64) ([]redis.XStream, error) {
	args := &redis.XReadArgs{
		Streams: flattenStreams(streams),
		Count:   count,
		Block:   0,
	}
	return getClient().XRead(ctx, args).Result()
}

// 创建消费者组
func XGroupCreate(ctx context.Context, stream, group, start string) error {
	return getClient().XGroupCreateMkStream(ctx, stream, group, start).Err()
}

// 从消费者组读取
func XReadGroup(ctx context.Context, group, consumer string, streams map[string]string, count int64) ([]redis.XStream, error) {
	args := &redis.XReadGroupArgs{
		Group:    group,
		Consumer: consumer,
		Streams:  flattenStreams(streams),
		Count:    count,
		Block:    5 * time.Second,
	}
	return getClient().XReadGroup(ctx, args).Result()
}

// 确认消息
func XAck(ctx context.Context, stream, group string, ids ...string) error {
	return getClient().XAck(ctx, stream, group, ids...).Err()
}

func flattenStreams(m map[string]string) []string {
	arr := make([]string, 0, len(m)*2)
	for k, v := range m {
		arr = append(arr, k, v)
	}
	return arr
}