redis

Documentation

Index

• func NewChannel[E typevent.Event](conf *Config, event string) typevent.Channel[E]
• type Codec
• type Config
  • func NewConfig(client *redis.Client, opts ...ConfigOption) *Config
• type ConfigOption
  • func WithCodec(codec Codec) ConfigOption
  • func WithKeyPrefix(prefix string) ConfigOption
• type JSONCodec
  • func (c *JSONCodec) Marshal(v any) ([]byte, error)
  • func (c *JSONCodec) Unmarshal(data []byte, v any) error
• type MsgpackCodec
  • func (c *MsgpackCodec) Marshal(v any) ([]byte, error)
  • func (c *MsgpackCodec) Unmarshal(data []byte, v any) error

Examples

• NewChannel

Functions

func NewChannel

func NewChannel[E typevent.Event](conf *Config, event string) typevent.Channel[E]

NewChannel returns a new typevent.Channel backed by redis Pub/Sub as the underlying event bus.

The channel will emit and subscribe to events of type E. The `name` param is a unique identifier for the event. All channels created with the same `name` must use the same type `E`. The package does intentionally not enforce this constraint, as it would require the use of reflection.

Redis Pub/Sub is used as the underlying event bus. The events emitted on the channel are passed to all channels subscribed on the same `name` on the same redis server, regardless of the DB they're connected to – see https://redis.io/docs/interact/pubsub/#database--scoping.

Example

server, err := miniredis.Run()
if err != nil {
	panic(err)
}
defer server.Close()

type event string

// Create a new channel using redis Pub/Sub as the underlying event bus.
client := redisclient.NewClient(&redisclient.Options{Addr: server.Addr()})
conf := redis.NewConfig(client)
channel := redis.NewChannel[event](conf, "CHANNEL_NAME")

// The WaitGroup is necessary to make the example test pass. In a real application,
// you would likely just let the handler run until its context is canceled.
wg := sync.WaitGroup{}
wg.Add(1)

// Register a subscriber for the channel.
sub, _ := channel.Subscribe(context.Background(), func(ctx context.Context, ev event) error {
	defer wg.Done()
	fmt.Printf("subscriber says: %s\n", ev)
	return nil
})
defer sub.Close()

// Emit an event on the channel.
channel.Emit("Hello World!")

wg.Wait() // again, just there to statisfy the example test

Output:

subscriber says: Hello World!

Types

type Codec

type Codec interface {
	Marshal(v any) ([]byte, error)
	Unmarshal(data []byte, v any) error
}

Codec is the Codec used to marshal and unmarshal events.

type Config

type Config struct {
	// contains filtered or unexported fields
}

Config contains the configuration for a redis channel.

func NewConfig

func NewConfig(client *redis.Client, opts ...ConfigOption) *Config

NewConfig returns a new Config for a redis channel.

type ConfigOption

type ConfigOption func(*Config)

ConfigOption is a configuration option for the redis channel.

func WithCodec

func WithCodec(codec Codec) ConfigOption

WithCodec sets the codec used to serialize and deserialize events.

func WithKeyPrefix

func WithKeyPrefix(prefix string) ConfigOption

WithKeyPrefix sets the prefix for all redis keys used by the channel.

type JSONCodec

type JSONCodec struct{}

JSONCodec is a codec that uses JSON as the encoding.

func (*JSONCodec) Marshal

func (c *JSONCodec) Marshal(v any) ([]byte, error)

Marshal turns the Event into a binary representation using JSON.

func (*JSONCodec) Unmarshal

func (c *JSONCodec) Unmarshal(data []byte, v any) error

Unmarshal unmarshals a binary representation of the Event using JSON.

type MsgpackCodec

type MsgpackCodec struct{}

MsgpackCodec is a codec that uses msgpack as the encoding.

func (*MsgpackCodec) Marshal

func (c *MsgpackCodec) Marshal(v any) ([]byte, error)

MarshalBinary turns the Event into a binary representation using msgpack.

func (*MsgpackCodec) Unmarshal

func (c *MsgpackCodec) Unmarshal(data []byte, v any) error

UnmarshalBinary unmarshals a binary representation of the Event using msgpack.