route

Documentation

Index

• type Config
  • func (pc *Config) Hash(ctx context.Context) string
  • func (rc *Config) Validate(ctx context.Context) error
• type InvalidRouteError
  • func (e *InvalidRouteError) Error() string
  • func (e *InvalidRouteError) Unwrap() error
• type PluginConfig
  • func (pc *PluginConfig) Hash(ctx context.Context) string
  • func (pc *PluginConfig) Validate(ctx context.Context) error
• type Route
  • func (rte *Route) Run(r receiver.Receiver, f filter.Filterer, s sender.Sender) error
  • func (rte *Route) Stop(ctx context.Context) error
• type RouteNotFoundError
  • func (e *RouteNotFoundError) Error() string
• type RouteStorer
• type Router

Types

type Config

type Config struct {
	Id           string         `json:"id,omitempty"`           // route ID
	TenantId     tenant.Id      `json:"tenant,omitempty"`       // TenantId. Derived from URL path. Should not be marshaled
	UserId       string         `json:"userId,omitempty"`       // user ID / author of route
	Name         string         `json:"name,omitempty"`         // optional unique name for route
	Origin       string         `json:"origin,omitempty"`       // optional reference to route owner, e.g. Flow ID in case of Gears
	Receiver     PluginConfig   `json:"receiver,omitempty"`     // source plugin configuration
	Sender       PluginConfig   `json:"sender,omitempty"`       // destination plugin configuration
	FilterChain  []PluginConfig `json:"filterChain,omitempty"`  // filter chain configuration
	DeliveryMode string         `json:"deliveryMode,omitempty"` // possible values: fire_and_forget, at_least_once, exactly_once
	Debug        bool           `json:"debug,omitempty"`        // if true generate debug logs and metrics for events taking this route
	Created      int64          `json:"created,omitempty"`      // time on when route was created, in unix timestamp seconds
	Modified     int64          `json:"modified,omitempty"`     // last time when route was modified, in unix timestamp seconds
}

func (*Config) Hash

func (pc *Config) Hash(ctx context.Context) string

Hash returns the md5 hash of a route config

func (*Config) Validate

func (rc *Config) Validate(ctx context.Context) error

Validate returns an error if the route config is invalid and nil otherwise

type InvalidRouteError

type InvalidRouteError struct {
	Err error
}

func (*InvalidRouteError) Error

func (e *InvalidRouteError) Error() string

func (*InvalidRouteError) Unwrap

func (e *InvalidRouteError) Unwrap() error

type PluginConfig

type PluginConfig struct {
	Plugin string      `json:"plugin,omitempty"` // plugin or filter type, e.g. kafka, kds, sqs, webhook, filter
	Name   string      `json:"name,omitempty"`   // plugin label to allow multiple instances of otherwise identical plugin configurations
	Config interface{} `json:"config,omitempty"` // plugin specific configuration parameters
}

func (*PluginConfig) Hash

func (pc *PluginConfig) Hash(ctx context.Context) string

Hash returns the md5 hash of the plugin config

func (*PluginConfig) Validate

func (pc *PluginConfig) Validate(ctx context.Context) error

Validate returns an error if the plugin config is invalid and nil otherwise

type Route

type Route struct {
	sync.Mutex
	// contains filtered or unexported fields
}

func (*Route) Run

func (rte *Route) Run(r receiver.Receiver, f filter.Filterer, s sender.Sender) error

func (*Route) Stop

func (rte *Route) Stop(ctx context.Context) error

type RouteNotFoundError

type RouteNotFoundError struct {
	TenantId tenant.Id
	RouteId  string
}

func (*RouteNotFoundError) Error

func (e *RouteNotFoundError) Error() string

type RouteStorer

type RouteStorer interface {
	//Get all routes of all tenants
	GetAllRoutes(context.Context) ([]Config, error)

	//If route is not found, the function should return a item not found error
	GetRoute(context.Context, tenant.Id, string) (Config, error)

	GetAllTenantRoutes(ctx context.Context, id tenant.Id) ([]Config, error)

	//SetRoute will add the route if it new or update the route if
	//it is an existing one. It will also update the create time and
	//modified time of the route where appropriate.
	SetRoute(context.Context, Config) error

	SetRoutes(context.Context, []Config) error

	DeleteRoute(context.Context, tenant.Id, string) error

	DeleteRoutes(context.Context, tenant.Id, []string) error
}

All route operations are synchronous. The storer should respect the cancellation from the context and cancel its operation gracefully when desired.

type Router

type Router interface {
	Run(r receiver.Receiver, f filter.Filterer, s sender.Sender) error
	Stop(ctx context.Context) error
}