pubsub

Documentation

Overview

Package pubsub wraps nats calls

Index

• Constants
• Variables
• func NewAssignmentURN(id string) string
• func NewLoadBalancerURN(id string) string
• func NewMessage(tenantURN string, opts ...MsgOption) (*pubsubx.Message, error)
• func NewMetadataURN(id string) string
• func NewOriginURN(id string) string
• func NewPoolURN(id string) string
• func NewPortURN(id string) string
• func NewTenantURN(id string) string
• type Client
  • func NewClient(opts ...Option) *Client
  • func (c *Client) AddStream() (*nats.StreamInfo, error)
  • func (c *Client) ChanSubscribe(_ context.Context, sub string, ch chan *nats.Msg, stream string) (*nats.Subscription, error)
  • func (c *Client) PublishCreate(ctx context.Context, actor, location string, data *pubsubx.Message) error
  • func (c *Client) PublishDelete(ctx context.Context, actor, location string, data *pubsubx.Message) error
  • func (c *Client) PublishUpdate(ctx context.Context, actor, location string, data *pubsubx.Message) error
• type MsgOption
  • func WithActorURN(u string) MsgOption
  • func WithAdditionalData(d map[string]interface{}) MsgOption
  • func WithAdditionalSubjectURNs(a ...string) MsgOption
  • func WithEventType(e string) MsgOption
  • func WithSource(s string) MsgOption
  • func WithSubjectFields(f map[string]string) MsgOption
  • func WithSubjectURN(s string) MsgOption
• type Option
  • func WithJetreamContext(js nats.JetStreamContext) Option
  • func WithLogger(l *zap.SugaredLogger) Option
  • func WithStreamName(s string) Option
  • func WithSubjectPrefix(p string) Option

Constants

const (
	// CreateEventType is the create event type string
	CreateEventType = "create"
	// DeleteEventType is the delete event type string
	DeleteEventType = "delete"
	// UpdateEventType is the update event type string
	UpdateEventType = "update"
)

const (
	// DefaultMessageSource is the default source for messages
	DefaultMessageSource = "load-balancer-api"
)

Variables

var (
	// ErrMissingEventSubjectURN is returned when the event subject urn is missing
	ErrMissingEventSubjectURN = errors.New("missing event subject urn")

	// ErrMissingEventActorURN is returned when the event actor urn is missing
	ErrMissingEventActorURN = errors.New("missing event actor urn")

	// ErrMissingEventSource is returned when the event source is missing
	ErrMissingEventSource = errors.New("missing event source")
)

Functions

func NewAssignmentURN

func NewAssignmentURN(id string) string

NewAssignmentURN creates a new assignment URN

func NewLoadBalancerURN

func NewLoadBalancerURN(id string) string

NewLoadBalancerURN creates a new load balancer URN

func NewMessage

func NewMessage(tenantURN string, opts ...MsgOption) (*pubsubx.Message, error)

NewMessage functionally generates a new pubsub message and appends the tenantURN to the list of additional subject urns

func NewMetadataURN

func NewMetadataURN(id string) string

NewMetadataURN creates a new metadata URN

func NewOriginURN

func NewOriginURN(id string) string

NewOriginURN creates a new origin URN

func NewPoolURN

func NewPoolURN(id string) string

NewPoolURN creates a new pool URN

func NewPortURN

func NewPortURN(id string) string

NewPortURN creates a new port URN

func NewTenantURN

func NewTenantURN(id string) string

NewTenantURN creates a new tenant URN

Types

type Client

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

Client is an event bus client with some configuration

func NewClient

func NewClient(opts ...Option) *Client

NewClient configures and establishes a new event bus client connection

func (*Client) AddStream

func (c *Client) AddStream() (*nats.StreamInfo, error)

AddStream checks if a stream exists and attempts to create it if it doesn't. Currently we don't currently check that the stream is configured identically to the desired configuration.

func (*Client) ChanSubscribe

func (c *Client) ChanSubscribe(_ context.Context, sub string, ch chan *nats.Msg, stream string) (*nats.Subscription, error)

ChanSubscribe creates a subcription and returns messages on a channel

func (*Client) PublishCreate

func (c *Client) PublishCreate(ctx context.Context, actor, location string, data *pubsubx.Message) error

PublishCreate publishes a create event

func (*Client) PublishDelete

func (c *Client) PublishDelete(ctx context.Context, actor, location string, data *pubsubx.Message) error

PublishDelete publishes a delete event

func (*Client) PublishUpdate

func (c *Client) PublishUpdate(ctx context.Context, actor, location string, data *pubsubx.Message) error

PublishUpdate publishes an update event

type MsgOption

type MsgOption func(m *pubsubx.Message)

MsgOption is a functional argument for NewMessage

func WithActorURN

func WithActorURN(u string) MsgOption

WithActorURN sets the actor urn of the message

func WithAdditionalData

func WithAdditionalData(d map[string]interface{}) MsgOption

WithAdditionalData sets the additional data of the message

func WithAdditionalSubjectURNs

func WithAdditionalSubjectURNs(a ...string) MsgOption

WithAdditionalSubjectURNs sets the additional subject urns of the message

func WithEventType

func WithEventType(e string) MsgOption

WithEventType sets the event type of the message

func WithSource

func WithSource(s string) MsgOption

WithSource sets the source of the message

func WithSubjectFields

func WithSubjectFields(f map[string]string) MsgOption

WithSubjectFields sets the subject fields of the message

func WithSubjectURN

func WithSubjectURN(s string) MsgOption

WithSubjectURN sets the subject urn of the message

type Option

type Option func(c *Client)

Option is a functional configuration option for governor eventing

func WithJetreamContext

func WithJetreamContext(js nats.JetStreamContext) Option

WithJetreamContext sets the nats jetstream context

func WithLogger

func WithLogger(l *zap.SugaredLogger) Option

WithLogger sets the client logger

func WithStreamName

func WithStreamName(s string) Option

WithStreamName sets the nats stream name

func WithSubjectPrefix

func WithSubjectPrefix(p string) Option

WithSubjectPrefix sets the nats subject prefix