pubsub

Documentation

Overview

Package pubsub provides cluster-wide publishing and subscribing of events

Index

• Variables
• func WriteSSEEvent(w io.Writer, data []byte, event EventType, base64encode bool)
• type Broker
  • func NewBroker(logger logr.Logger, db pool) *Broker
  • func (b *Broker) Publish(event Event)
  • func (r Broker) Register(table string, getter Getter)
  • func (r Broker) RegisterFunc(table string, getter GetterFunc)
  • func (b *Broker) Start(ctx context.Context) error
  • func (b *Broker) Started() <-chan struct{}
  • func (b *Broker) Subscribe(ctx context.Context, prefix string) (<-chan Event, error)
• type DBAction
• type Event
  • func NewCreatedEvent(payload any) Event
  • func NewDeletedEvent(payload any) Event
  • func NewUpdatedEvent(payload any) Event
• type EventType
• type FakePublisher
  • func (f *FakePublisher) Publish(Event)
• type Getter
• type GetterFunc
  • func (f GetterFunc) GetByID(ctx context.Context, workspaceID string, action DBAction) (any, error)
• type PubSubService
• type Publisher
• type Subscriber
• type Table

Variables

var ErrSubscriptionTerminated = errors.New("broker terminated the subscription")

ErrSubscriptionTerminated is for use by subscribers to indicate that their subscription has been terminated by the broker.

Functions

func WriteSSEEvent

func WriteSSEEvent(w io.Writer, data []byte, event EventType, base64encode bool)

WriteSSEEvent writes an server-side-event to w. The data is optionally base64 encoded before being written.

Types

type Broker

type Broker struct {
	logr.Logger
	// contains filtered or unexported fields
}

Broker is a pubsub Broker implemented using postgres' listen/notify

func NewBroker

func NewBroker(logger logr.Logger, db pool) *Broker

func (*Broker) Publish

func (b *Broker) Publish(event Event)

Publish sends an event to subscribers.

func (Broker) Register

func (r Broker) Register(table string, getter Getter)

Register a table and getter with the pubsub broker, to enable the broker to convert a database event into an OTF event.

func (Broker) RegisterFunc

func (r Broker) RegisterFunc(table string, getter GetterFunc)

Register a table and getter function with the pubsub broker, to enable the broker to convert a database event into an OTF event.

func (*Broker) Start

func (b *Broker) Start(ctx context.Context) error

Start the pubsub daemon; listen to notifications from postgres and forward to local pubsub broker. The listening channel is closed once the broker has started listening; from this point onwards published messages will be forwarded.

func (*Broker) Started

func (b *Broker) Started() <-chan struct{}

func (*Broker) Subscribe

func (b *Broker) Subscribe(ctx context.Context, prefix string) (<-chan Event, error)

Subscribe subscribes the caller to a stream of events. Prefix is an identifier prefixed to a random string to helpfully identify the subscriber in metrics.

type DBAction

type DBAction string

DBAction is the action carried out on a database row

type Event

type Event struct {
	Type    EventType
	Payload any
	Local   bool // for local node only and not to be published to rest of cluster
}

Event represents an event in the lifecycle of an otf resource

func NewCreatedEvent

func NewCreatedEvent(payload any) Event

func NewDeletedEvent

func NewDeletedEvent(payload any) Event

func NewUpdatedEvent

func NewUpdatedEvent(payload any) Event

type EventType

type EventType string

EventType identifies the type of event

const (
	EventError       EventType = "error"
	EventInfo        EventType = "info"
	EventLogChunk    EventType = "log_update"
	EventLogFinished EventType = "log_finished"
	EventVCS         EventType = "vcs_event"

	CreatedEvent EventType = "created"
	UpdatedEvent EventType = "updated"
	DeletedEvent EventType = "deleted"

	InsertDBAction = "INSERT"
	UpdateDBAction = "UPDATE"
	DeleteDBAction = "DELETE"
)

type FakePublisher

type FakePublisher struct{}

func (*FakePublisher) Publish

func (f *FakePublisher) Publish(Event)

type Getter

type Getter interface {
	GetByID(context.Context, string, DBAction) (any, error)
}

Getter retrieves an event payload using its ID.

type GetterFunc

type GetterFunc func(context.Context, string, DBAction) (any, error)

GetterFunc is a function wrapper for Getter.

func (GetterFunc) GetByID

func (f GetterFunc) GetByID(ctx context.Context, workspaceID string, action DBAction) (any, error)

type PubSubService

type PubSubService interface {
	Publisher
	Subscriber
}

PubSubService provides low-level access to pub-sub behaviours. Access is unauthenticated.

type Publisher

type Publisher interface {
	// Publish an event
	Publish(Event)
}

type Subscriber

type Subscriber interface {
	// Subscribe subscribes the caller to OTF events. Name uniquely identifies the
	// caller.
	Subscribe(ctx context.Context, name string) (<-chan Event, error)
}

Subscriber is capable of creating a subscription to events.

type Table

type Table string