mdlsub

Documentation

Index

• Constants
• func CreateDefiner(callbacks map[string]stream.ConsumerCallbackFactory) httpserver.Definer
• func CreateMessageAttributes(modelId mdl.ModelId, typ string, version int) map[string]string
• func GetSubscriberConfigKey(name string) string
• func GetSubscriberFQN(name string, sourceModel SubscriberModel) string
• func GetSubscriberOutputConfigKey(name string) string
• func IsDelayOpError(err error) bool
• func NewGenericTransformer(transformer ModelTransformer) func(context.Context, cfg.Config, log.Logger) (ModelTransformer, error)
• func NewHandler(ctx context.Context, config cfg.Config, logger log.Logger, ...) (httpserver.HandlerWithInput, error)
• func NewSubscriberCallbackFactory(transformers ModelTransformers, outputs Outputs) stream.ConsumerCallbackFactory
• func NewSubscriberFactory(transformerFactoryMap TransformerMapTypeVersionFactories) kernel.ModuleMultiFactory
• func PublisherConfigPostProcessor(config cfg.GosoConf) (bool, error)
• func SubscriberConfigPostProcessor(config cfg.GosoConf) (bool, error)
• func SubscriberFactory(ctx context.Context, config cfg.Config, logger log.Logger, ...) (map[string]kernel.ModuleFactory, error)
• type DelayOpError
  • func NewDelayOpError(err error) DelayOpError
  • func (e DelayOpError) As(target interface{}) bool
  • func (e DelayOpError) Error() string
  • func (e DelayOpError) Unwrap() error
• type Handler
  • func (h *Handler) GetInput() interface{}
  • func (h *Handler) Handle(ctx context.Context, request *httpserver.Request) (*httpserver.Response, error)
• type Model
• type ModelDb
  • func (m ModelDb) GetId() interface{}
• type ModelSpecification
• type ModelTransformer
• type ModelTransformers
• type Output
• type OutputDb
  • func NewOutputDb(ctx context.Context, config cfg.Config, logger log.Logger) (*OutputDb, error)
  • func (p *OutputDb) Persist(_ context.Context, model Model, op string) error
• type OutputDdb
  • func NewOutputDdb(ctx context.Context, config cfg.Config, logger log.Logger, ...) *OutputDdb
  • func (p *OutputDdb) GetType() string
  • func (p *OutputDdb) Persist(ctx context.Context, model Model, op string) error
• type OutputFactory
• type OutputKvstore
  • func NewOutputKvstore(ctx context.Context, config cfg.Config, logger log.Logger, ...) (*OutputKvstore, error)
  • func (p *OutputKvstore) Persist(ctx context.Context, model Model, op string) error
• type Outputs
• type Publisher
  • func NewPublisher(ctx context.Context, config cfg.Config, logger log.Logger, name string) (Publisher, error)
  • func NewPublisherWithInterfaces(logger log.Logger, producer stream.Producer, settings *PublisherSettings) Publisher
  • func NewPublisherWithSettings(ctx context.Context, config cfg.Config, logger log.Logger, ...) (Publisher, error)
• type PublisherSettings
• type Settings
• type SubscriberApiSettings
• type SubscriberCallback
  • func (s *SubscriberCallback) Consume(ctx context.Context, input interface{}, attributes map[string]string) (bool, error)
  • func (s *SubscriberCallback) GetModel(attributes map[string]string) interface{}
• type SubscriberInputConfigPostProcessor
• type SubscriberModel
  • func UnmarshalSubscriberSourceModel(config cfg.Config, name string) SubscriberModel
• type SubscriberOutputConfigPostProcessor
• type SubscriberSettings
• type TransformerFactory
• type TransformerMapTypeVersionFactories
• type TransformerMapVersionFactories
• type VersionedModelTransformers

Constants

const (
	AttributeModelId          = "modelId"
	AttributeType             = "type"
	AttributeVersion          = "version"
	ConfigKeyMdlSubPublishers = "mdlsub.publishers"
	TypeCreate                = "create"
	TypeUpdate                = "update"
	TypeDelete                = "delete"
)

const (
	ConfigKeyMdlSubSubscribers = "mdlsub.subscribers"
	MetricNameSuccess          = "ModelEventConsumeSuccess"
	MetricNameFailure          = "ModelEventConsumeFailure"
)

const (
	ConfigKeyMdlSub = "mdlsub"
)

const (
	OutputTypeDb = "db"
)

const (
	OutputTypeDdb = "ddb"
)

const (
	OutputTypeKvstore = "kvstore"
)

Functions

func CreateDefiner

func CreateDefiner(callbacks map[string]stream.ConsumerCallbackFactory) httpserver.Definer

func CreateMessageAttributes

func CreateMessageAttributes(modelId mdl.ModelId, typ string, version int) map[string]string

func GetSubscriberConfigKey

func GetSubscriberConfigKey(name string) string

func GetSubscriberFQN

func GetSubscriberFQN(name string, sourceModel SubscriberModel) string

func GetSubscriberOutputConfigKey

func GetSubscriberOutputConfigKey(name string) string

func IsDelayOpError

func IsDelayOpError(err error) bool

func NewGenericTransformer

func NewGenericTransformer(transformer ModelTransformer) func(context.Context, cfg.Config, log.Logger) (ModelTransformer, error)

func NewHandler

func NewHandler(ctx context.Context, config cfg.Config, logger log.Logger, callbackFactory stream.ConsumerCallbackFactory) (httpserver.HandlerWithInput, error)

func NewSubscriberCallbackFactory

func NewSubscriberCallbackFactory(transformers ModelTransformers, outputs Outputs) stream.ConsumerCallbackFactory

func NewSubscriberFactory

func NewSubscriberFactory(transformerFactoryMap TransformerMapTypeVersionFactories) kernel.ModuleMultiFactory

func PublisherConfigPostProcessor

func PublisherConfigPostProcessor(config cfg.GosoConf) (bool, error)

func SubscriberConfigPostProcessor

func SubscriberConfigPostProcessor(config cfg.GosoConf) (bool, error)

func SubscriberFactory

func SubscriberFactory(ctx context.Context, config cfg.Config, logger log.Logger, transformerFactories TransformerMapTypeVersionFactories) (map[string]kernel.ModuleFactory, error)

Types

type DelayOpError

type DelayOpError struct {
	Err error
}

func NewDelayOpError

func NewDelayOpError(err error) DelayOpError

func (DelayOpError) As

func (e DelayOpError) As(target interface{}) bool

func (DelayOpError) Error

func (e DelayOpError) Error() string

func (DelayOpError) Unwrap

func (e DelayOpError) Unwrap() error

type Handler

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

func (*Handler) GetInput

func (h *Handler) GetInput() interface{}

func (*Handler) Handle

func (h *Handler) Handle(ctx context.Context, request *httpserver.Request) (*httpserver.Response, error)

type Model

type Model interface {
	GetId() interface{}
}

type ModelDb

type ModelDb struct {
	Id *uint `gorm:"primary_key;"`
}

func (ModelDb) GetId

func (m ModelDb) GetId() interface{}

type ModelSpecification

type ModelSpecification struct {
	CrudType string
	Version  int
	ModelId  string
}

type ModelTransformer

type ModelTransformer interface {
	GetInput() interface{}
	Transform(ctx context.Context, inp interface{}) (out Model, err error)
}

type ModelTransformers

type ModelTransformers map[string]VersionedModelTransformers

type Output

type Output interface {
	Persist(ctx context.Context, model Model, op string) error
}

type OutputDb

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

func NewOutputDb

func NewOutputDb(ctx context.Context, config cfg.Config, logger log.Logger) (*OutputDb, error)

func (*OutputDb) Persist

func (p *OutputDb) Persist(_ context.Context, model Model, op string) error

type OutputDdb

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

func NewOutputDdb

func NewOutputDdb(ctx context.Context, config cfg.Config, logger log.Logger, settings *SubscriberSettings) *OutputDdb

func (*OutputDdb) GetType

func (p *OutputDdb) GetType() string

func (*OutputDdb) Persist

func (p *OutputDdb) Persist(ctx context.Context, model Model, op string) error

type OutputFactory

type OutputFactory func(ctx context.Context, config cfg.Config, logger log.Logger, settings *SubscriberSettings, transformers VersionedModelTransformers) (map[int]Output, error)

type OutputKvstore

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

func NewOutputKvstore

func NewOutputKvstore(ctx context.Context, config cfg.Config, logger log.Logger, settings *SubscriberSettings) (*OutputKvstore, error)

func (*OutputKvstore) Persist

func (p *OutputKvstore) Persist(ctx context.Context, model Model, op string) error

type Outputs

type Outputs map[string]map[int]Output

type Publisher

type Publisher interface {
	PublishBatch(ctx context.Context, typ string, version int, values []interface{}, customAttributes ...map[string]string) error
	Publish(ctx context.Context, typ string, version int, value interface{}, customAttributes ...map[string]string) error
}

func NewPublisher

func NewPublisher(ctx context.Context, config cfg.Config, logger log.Logger, name string) (Publisher, error)

func NewPublisherWithInterfaces

func NewPublisherWithInterfaces(logger log.Logger, producer stream.Producer, settings *PublisherSettings) Publisher

func NewPublisherWithSettings

func NewPublisherWithSettings(ctx context.Context, config cfg.Config, logger log.Logger, settings *PublisherSettings) (Publisher, error)

type PublisherSettings

type PublisherSettings struct {
	mdl.ModelId
	Producer   string `cfg:"producer" validate:"required_without=OutputType"`
	OutputType string `cfg:"output_type" validate:"required_without=Producer"`
	Shared     bool   `cfg:"shared"`
}

type Settings

type Settings struct {
	SubscriberApi SubscriberApiSettings          `cfg:"subscriber_api"`
	Subscribers   map[string]*SubscriberSettings `cfg:"subscribers"`
}

type SubscriberApiSettings

type SubscriberApiSettings struct {
	Enabled bool `cfg:"enabled" default:"true"`
}

type SubscriberCallback

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

func (*SubscriberCallback) Consume

func (s *SubscriberCallback) Consume(ctx context.Context, input interface{}, attributes map[string]string) (bool, error)

func (*SubscriberCallback) GetModel

func (s *SubscriberCallback) GetModel(attributes map[string]string) interface{}

type SubscriberInputConfigPostProcessor

type SubscriberInputConfigPostProcessor func(config cfg.GosoConf, name string, subscriberSettings *SubscriberSettings) cfg.Option

type SubscriberModel

type SubscriberModel struct {
	mdl.ModelId
	Shared bool `cfg:"shared"`
}

func UnmarshalSubscriberSourceModel

func UnmarshalSubscriberSourceModel(config cfg.Config, name string) SubscriberModel

type SubscriberOutputConfigPostProcessor

type SubscriberOutputConfigPostProcessor func(config cfg.GosoConf, name string, subscriberSettings *SubscriberSettings) cfg.Option

type SubscriberSettings

type SubscriberSettings struct {
	Input       string          `cfg:"input" default:"sns"`
	Output      string          `cfg:"output"`
	RunnerCount int             `cfg:"runner_count" default:"10" validate:"min=1"`
	SourceModel SubscriberModel `cfg:"source"`
	TargetModel SubscriberModel `cfg:"target"`
}

type TransformerFactory

type TransformerFactory func(ctx context.Context, config cfg.Config, logger log.Logger) (ModelTransformer, error)

type TransformerMapTypeVersionFactories

type TransformerMapTypeVersionFactories map[string]TransformerMapVersionFactories

type TransformerMapVersionFactories

type TransformerMapVersionFactories map[int]TransformerFactory

type VersionedModelTransformers

type VersionedModelTransformers map[int]ModelTransformer