redis

package

v0.9.19 Latest Latest Go to latest Published: Mar 26, 2026 License: Apache-2.0 Imports: 13 Imported by: 0

Details

Valid go.mod file
Redistributable license
Tagged version
Stable version
Learn more about best practices

Repository

github.com/metraction/pharos

Links

Open Source Insights

Documentation ¶

Index ¶

func NewQueueLimit(ctx context.Context, rdb *redis.Client, queueName string, limit int64, ...) func(in any) bool
func NewRedis(ctx context.Context, cfg *model.Config) *redis.Client
func NewRedisConsumerGroupSource[T any](ctx context.Context, rdb *redis.Client, streamName string, groupName string, ...) streams.Source
func NewRedisStreamSink[T any](ctx context.Context, rdb *redis.Client, streamName string) streams.Sink
func SetupTestRedis(t *testing.T) (*miniredis.Miniredis, *redis.Client, *model.Config)
type RedisGtrsClient
- func NewRedisGtrsClient[T any, R any](ctx context.Context, redisCfg *model.Config, requestQueue string, ...) (*RedisGtrsClient[T, R], error)
- func NewRedisGtrsClientStefan[T any, R any](ctx context.Context, redisEndpoint, streamName string) (*RedisGtrsClient[T, R], error)
type RedisGtrsServer
- func NewRedisGtrsServer[T any, R any](ctx context.Context, redisCfg model.Redis, requestQueue string, ...) (*RedisGtrsServer[T, R], error)
- func (c *RedisGtrsServer[T, R]) ProcessRequest(ctx context.Context, handler func(T) R)

Constants ¶

This section is empty.

Variables ¶

This section is empty.

Functions ¶

func NewQueueLimit ¶

func NewQueueLimit(ctx context.Context, rdb *redis.Client, queueName string, limit int64, cb ...func(in any)) func(in any) bool

func NewRedis ¶

func NewRedis(ctx context.Context, cfg *model.Config) *redis.Client

func NewRedisConsumerGroupSource ¶

func NewRedisConsumerGroupSource[T any](ctx context.Context, rdb *redis.Client, streamName string, groupName string, consumerName string, groupStartID string, blockTimeout time.Duration, messageCount int64) streams.Source

func NewRedisStreamSink ¶

func NewRedisStreamSink[T any](ctx context.Context, rdb *redis.Client, streamName string) streams.Sink

func SetupTestRedis ¶

func SetupTestRedis(t *testing.T) (*miniredis.Miniredis, *redis.Client, *model.Config)

Types ¶

type RedisGtrsClient ¶

type RedisGtrsClient[T any, R any] struct {
	// contains filtered or unexported fields
}

func NewRedisGtrsClient ¶

func NewRedisGtrsClient[T any, R any](ctx context.Context, redisCfg *model.Config, requestQueue string, replyQueue string) (*RedisGtrsClient[T, R], error)

func NewRedisGtrsClientStefan ¶

func NewRedisGtrsClientStefan[T any, R any](ctx context.Context, redisEndpoint, streamName string) (*RedisGtrsClient[T, R], error)

NEW: Stefan

func (*RedisGtrsClient[T, R]) Close ¶

func (rx *RedisGtrsClient[T, R]) Close()

NEW: Stefan

func (*RedisGtrsClient[T, R]) Connect ¶

func (rx *RedisGtrsClient[T, R]) Connect(ctx context.Context) error

NEW: Stefan

func (*RedisGtrsClient[T, R]) ReceiveResponse ¶

func (c *RedisGtrsClient[T, R]) ReceiveResponse(ctx context.Context, corrID string, timeout time.Duration) (R, error)

func (*RedisGtrsClient[T, R]) ReceiveStefan ¶

func (rx *RedisGtrsClient[T, R]) ReceiveStefan(ctx context.Context, groupName, consumerName, mode string, handlerFunc func(gtrs.Message[T]) error) error

NEW: Stefan

func (*RedisGtrsClient[T, R]) RequestReply ¶

func (c *RedisGtrsClient[T, R]) RequestReply(ctx context.Context, payload T) (R, error)

TODO: As simple consumer doesn't do ACK reply queue might fillup and cause long read on client side. It should start waiting for response and read only new messages

func (*RedisGtrsClient[T, R]) SendRequest ¶

func (c *RedisGtrsClient[T, R]) SendRequest(ctx context.Context, payload T) (error, string)

type RedisGtrsServer ¶

type RedisGtrsServer[T any, R any] struct {
	// contains filtered or unexported fields
}

func NewRedisGtrsServer ¶

func NewRedisGtrsServer[T any, R any](ctx context.Context, redisCfg model.Redis, requestQueue string, replyQueue string) (*RedisGtrsServer[T, R], error)

func (*RedisGtrsServer[T, R]) ProcessRequest ¶

func (c *RedisGtrsServer[T, R]) ProcessRequest(ctx context.Context, handler func(T) R)

Source Files ¶

View all Source files

?	: This menu
/	: Search site
f or F	: Jump to
y or Y	: Canonical URL