redis

Documentation

Overview

Package redis implements an events.PubSub implementation that uses Redis PubSub.

Example

package main

import (
	"context"

	"go.thethings.network/lorawan-stack/v3/pkg/config"
	"go.thethings.network/lorawan-stack/v3/pkg/events"
	"go.thethings.network/lorawan-stack/v3/pkg/events/redis"
	"go.thethings.network/lorawan-stack/v3/pkg/task"
)

type mockComponent struct {
	task.Starter
}

func (mockComponent) FromRequestContext(ctx context.Context) context.Context {
	return ctx
}

func main() {
	// The task starter is used for automatic re-subscription on failure.
	taskStarter := task.StartTaskFunc(task.DefaultStartTask)

	redisPubSub := redis.NewPubSub(context.TODO(), mockComponent{taskStarter}, config.RedisEvents{
		// Config here...
	})

	// Replace the default pubsub so that we will now publish to Redis.
	events.SetDefaultPubSub(redisPubSub)
}

Index

• func NewPubSub(ctx context.Context, component workerpool.Component, conf config.RedisEvents) events.PubSub
• type PubSub
  • func (ps *PubSub) Close(ctx context.Context) error
  • func (ps *PubSub) Publish(evs ...events.Event)
  • func (ps *PubSub) Subscribe(ctx context.Context, names []string, ids []*ttnpb.EntityIdentifiers, ...) error
• type PubSubStore
  • func (ps *PubSubStore) FetchHistory(ctx context.Context, names []string, ids []*ttnpb.EntityIdentifiers, ...) ([]events.Event, error)
  • func (ps *PubSubStore) FindRelated(ctx context.Context, correlationID string) ([]events.Event, error)
  • func (ps *PubSubStore) LoadEvent(ctx context.Context, uid string) (*ttnpb.Event, error)
  • func (ps *PubSubStore) Publish(evs ...events.Event)
  • func (ps *PubSubStore) SubscribeWithHistory(ctx context.Context, names []string, ids []*ttnpb.EntityIdentifiers, ...) error

Examples

• Package

Functions

func NewPubSub

func NewPubSub(ctx context.Context, component workerpool.Component, conf config.RedisEvents) events.PubSub

NewPubSub creates a new PubSub that publishes and subscribes to Redis.

Types

type PubSub

type PubSub struct {
	*basic.PubSub
	// contains filtered or unexported fields
}

PubSub with Redis backend.

func (*PubSub) Close

func (ps *PubSub) Close(ctx context.Context) error

Close the Redis publisher.

func (*PubSub) Publish

func (ps *PubSub) Publish(evs ...events.Event)

Publish an event to Redis.

func (*PubSub) Subscribe

func (ps *PubSub) Subscribe(
	ctx context.Context, names []string, ids []*ttnpb.EntityIdentifiers, hdl events.Handler,
) error

Subscribe implements the events.Subscriber interface.

type PubSubStore

type PubSubStore struct {
	*PubSub
	// contains filtered or unexported fields
}

PubSubStore is a PubSub with historical event storage.

func (*PubSubStore) FetchHistory

func (ps *PubSubStore) FetchHistory(
	ctx context.Context, names []string, ids []*ttnpb.EntityIdentifiers, after *time.Time, tail int,
) ([]events.Event, error)

FetchHistory fetches the tail (optional) of historical events matching the given names (optional) and identifiers (mandatory) after the given time (optional).

func (*PubSubStore) FindRelated

func (ps *PubSubStore) FindRelated(ctx context.Context, correlationID string) ([]events.Event, error)

FindRelated finds events with matching correlation IDs.

func (*PubSubStore) LoadEvent

func (ps *PubSubStore) LoadEvent(ctx context.Context, uid string) (*ttnpb.Event, error)

LoadEvent loads an event by its UID.

func (*PubSubStore) Publish

func (ps *PubSubStore) Publish(evs ...events.Event)

Publish an event to Redis.

func (*PubSubStore) SubscribeWithHistory

func (ps *PubSubStore) SubscribeWithHistory(
	ctx context.Context, names []string, ids []*ttnpb.EntityIdentifiers, after *time.Time, tail int, hdl events.Handler,
) error

SubscribeWithHistory is like FetchHistory, but after fetching historical events, this continues sending live events until the context is done.