subscribe

Documentation

Index

• Constants
• Variables
• func Enqueue(queue *lane.Deque, id interface{}, ...) error
• func Remarshal(in, out interface{}) error
• func Trigger(ctx context.Context, opts *TriggerOptions) error
• type Config
• type Factory
  • func (Factory) New(m *plugins.Manager, config interface{}) plugins.Plugin
  • func (Factory) Validate(_ *plugins.Manager, config []byte) (interface{}, error)
• type NewSubscriberFunc
• type NewSubscriberOptions
• type Plugin
  • func (p *Plugin) Reconfigure(ctx context.Context, config interface{})
  • func (p *Plugin) Start(ctx context.Context) error
  • func (p *Plugin) Stop(ctx context.Context)
• type Subscriber
• type SubscriberConfig
  • func (s *SubscriberConfig) GetName() string
• type SubscriberProvider
• type TriggerOptions

Constants

const (
	PluginName = "subscribe"
)

Variables

var (
	SubscribePluginProviders = map[string]NewSubscriberFunc{}
)

Functions

func Enqueue

func Enqueue(queue *lane.Deque, id interface{}, workFunc func(id interface{}, args ...interface{}) error, args ...interface{}) error

Enqueue ensures at most n-number of calls to a function are queued up This prevents concurrent calls from queueing up past a certain number typeically 1, and allows for at least 1 call to be queued up if it is made during another call's execution

func Remarshal

func Remarshal(in, out interface{}) error

hacky way to map one interface to another

func Trigger

func Trigger(ctx context.Context, opts *TriggerOptions) error

triggers the named plugin

Types

type Config

type Config struct {
	Subscribers map[string]*SubscriberConfig `yaml:"subscribers" json:"subscribers"`
	// contains filtered or unexported fields
}

type Factory

type Factory struct{}

func (Factory) New

func (Factory) New(m *plugins.Manager, config interface{}) plugins.Plugin

New

func (Factory) Validate

func (Factory) Validate(_ *plugins.Manager, config []byte) (interface{}, error)

Validate

type NewSubscriberFunc

type NewSubscriberFunc func(opts *NewSubscriberOptions) (Subscriber, error)

type NewSubscriberOptions

type NewSubscriberOptions struct {
	Config  *SubscriberConfig
	Manager *plugins.Manager
	Logger  logging.Logger
}

type Plugin

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

func (*Plugin) Reconfigure

func (p *Plugin) Reconfigure(ctx context.Context, config interface{})

Reconfigure

func (*Plugin) Start

func (p *Plugin) Start(ctx context.Context) error

Start

func (*Plugin) Stop

func (p *Plugin) Stop(ctx context.Context)

Stop

type Subscriber

type Subscriber interface {
	Connect(ctx context.Context) error
	Subscribe(ctx context.Context) error
	Unsubscribe(ctx context.Context) error
	Disconnect(ctx context.Context) error
}

type SubscriberConfig

type SubscriberConfig struct {
	Provider string      `yaml:"provider" json:"provider"`
	Topic    string      `yaml:"topic" json:"topic"`
	Plugin   string      `yaml:"plugin" json:"plugin"`
	Config   interface{} `yaml:"config" json:"config"`
	// contains filtered or unexported fields
}

func (*SubscriberConfig) GetName

func (s *SubscriberConfig) GetName() string

type SubscriberProvider

type SubscriberProvider string

type TriggerOptions

type TriggerOptions struct {
	Manager *plugins.Manager
	Name    string
	DQ      *lane.Deque
}