mdlsub

Oct 2, 2024

Changes in this version

+ func AddOutput(name string, factory OutputFactory)

+ func FixtureSetFactory(transformerFactoryMap TransformerMapTypeVersionFactories) fixtures.FixtureSetsFactory

+ type FetchData struct

+ Data json.RawMessage

+ type FixtureSet struct

+ func NewFixtureSet(logger log.Logger, source SubscriberModel, core SubscriberCore, ...) *FixtureSet

+ func NewFixtureSetWithInterfaces(logger log.Logger, source SubscriberModel, core SubscriberCore, ...) *FixtureSet

+ func (f FixtureSet) Write(ctx context.Context) error

+ type FixtureSettings struct

+ DatasetName string

+ Host string

+ Path string

type ModelSpecification

+ func (m ModelSpecification) String() string

+ type SubscriberCore interface

+ GetLatestModelIdVersion func(modelId mdl.ModelId) (int, error)

+ GetModelIds func() []string

+ GetOutput func(spec *ModelSpecification) (Output, error)

+ GetTransformer func(spec *ModelSpecification) (ModelTransformer, error)

+ Persist func(ctx context.Context, spec *ModelSpecification, model Model) error

+ Transform func(ctx context.Context, spec *ModelSpecification, input any) (Model, error)

+ func NewSubscriberCore(ctx context.Context, config cfg.Config, logger log.Logger, ...) (SubscriberCore, error)

+ func NewSubscriberCoreWithInterfaces(transformers ModelTransformers, outputs Outputs) SubscriberCore

Oct 12, 2023

Changes in this version

+ const AttributeModelId

+ const AttributeType

+ const AttributeVersion

+ const ConfigKeyMdlSub

+ const ConfigKeyMdlSubPublishers

+ const ConfigKeyMdlSubSubscribers

+ const MetricNameFailure

+ const MetricNameSuccess

+ const OutputTypeDb

+ const OutputTypeDdb

+ const OutputTypeKvstore

+ const TypeCreate

+ const TypeDelete

+ const TypeUpdate

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

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

+ 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, ...) (apiserver.HandlerWithInput, error)

+ 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)

+ 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 struct

+ Err error

+ func NewDelayOpError(err error) DelayOpError

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

+ func (e DelayOpError) Error() string

+ func (e DelayOpError) Unwrap() error

+ type Handler struct

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

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

+ type Model interface

+ GetId func() interface{}

+ type ModelDb struct

+ Id *uint

+ func (m ModelDb) GetId() interface{}

+ type ModelSpecification struct

+ CrudType string

+ ModelId string

+ Version int

+ type ModelTransformer interface

+ GetInput func() interface{}

+ Transform func(ctx context.Context, inp interface{}) (out Model, err error)

+ type ModelTransformers map[string]VersionedModelTransformers

+ type Output interface

+ Persist func(ctx context.Context, model Model, op string) error

+ type OutputDb struct

+ func NewOutputDb(config cfg.Config, logger log.Logger) (*OutputDb, error)

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

+ type OutputDdb struct

+ 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 func(ctx context.Context, config cfg.Config, logger log.Logger, ...) (map[int]Output, error)

+ type OutputKvstore struct

+ 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 map[string]map[int]Output

+ type Publisher interface

+ Publish func(ctx context.Context, typ string, version int, value interface{}, ...) error

+ PublishBatch func(ctx context.Context, typ string, version int, values []interface{}, ...) error

+ type PublisherSettings struct

+ OutputType string

+ Producer string

+ Shared bool

+ type Settings struct

+ SubscriberApi SubscriberApiSettings

+ Subscribers map[string]*SubscriberSettings

+ type SubscriberApiSettings struct

+ Enabled bool

+ type SubscriberCallback struct

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

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

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

+ type SubscriberModel struct

+ Shared bool

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

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

+ type SubscriberSettings struct

+ Input string

+ Output string

+ RunnerCount int

+ SourceModel SubscriberModel

+ TargetModel SubscriberModel

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

+ type TransformerMapTypeVersionFactories map[string]TransformerMapVersionFactories

+ type TransformerMapVersionFactories map[int]TransformerFactory

+ type VersionedModelTransformers map[int]ModelTransformer