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/component"
	"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"
)

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

	redisPubSub := redis.NewPubSub(context.TODO(), 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, taskStarter component.TaskStarter, ...) events.PubSub
• type PubSub
  • func (ps *PubSub) Close(ctx context.Context) error
  • func (ps *PubSub) Publish(evt 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(evt 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, taskStarter component.TaskStarter, 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(evt 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(evt 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.