scheduler

Documentation

Index

• Constants
• type AggregationType
• type Config
  • func ParseConfig(buffer []byte) (Config, error)
• type EgressTopicConfig
• type Ingress
  • func NewIngress(config IngressConfig) *Ingress
  • func (p *Ingress) Receive(ctx context.Context, data []byte, attributes map[string]string) error
• type IngressConfig
• type Router
  • func NewRouter(ctx context.Context, config Config) (Router, error)
• type SystemTime
  • func (SystemTime) Now() time.Time
• type Ticker
  • func NewTicker(config TickerConfig) *Ticker
  • func (t *Ticker) HandleMessage(ctx context.Context, msg storage.ScheduledMessage) error
  • func (h *Ticker) RunTicker(ctx context.Context)
• type TickerConfig
• type TimeSource

Constants

const (
	AggregationTypeNo     = iota
	AggregationTypeSkip   = iota
	AggregationTypeAppend = iota
)

Types

type AggregationType

type AggregationType int

type Config

type Config struct {
	ProjectID           string
	IngressSubscription string
	DefaultEgress       EgressTopicConfig
	Routing             map[string]EgressTopicConfig
}

func ParseConfig

func ParseConfig(buffer []byte) (Config, error)

type EgressTopicConfig

type EgressTopicConfig struct {
	Name        string
	Aggregation AggregationType
}

type Ingress

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

func NewIngress

func NewIngress(config IngressConfig) *Ingress

func (*Ingress) Receive

func (p *Ingress) Receive(ctx context.Context, data []byte, attributes map[string]string) error

type IngressConfig

type IngressConfig struct {
	Router     Router
	Storage    storage.SchedulerStorage
	TimeSource TimeSource
}

type Router

type Router interface {
	io.Closer
	Publish(ctx context.Context, msgType string, data []byte, attributes map[string]string) (msgqueue.PublishResult, error)
}

func NewRouter

func NewRouter(ctx context.Context, config Config) (Router, error)

type SystemTime

type SystemTime struct{}

func (SystemTime) Now

func (SystemTime) Now() time.Time

type Ticker

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

func NewTicker

func NewTicker(config TickerConfig) *Ticker

func (*Ticker) HandleMessage

func (t *Ticker) HandleMessage(ctx context.Context, msg storage.ScheduledMessage) error

func (*Ticker) RunTicker

func (h *Ticker) RunTicker(ctx context.Context)

Activates ticker loop

type TickerConfig

type TickerConfig struct {
	Router     Router
	Storage    storage.SchedulerStorage
	TimeSource TimeSource
}

type TimeSource

type TimeSource interface {
	Now() time.Time
}