source

Documentation

Index

• Variables
• func Close() error
• func Listen(ctx context.Context) error
• func Register(ctx context.Context, name string, options ...Option) error
• func RegisterConnector(driver string, c connector)
• func Source(name string) eventstream.Sourcer
• func Subscribe(ctx context.Context, name string, streams ...eventstream.Streamer) error
• type Config
  • func (c *Config) Decode(v any) error
  • func (c *Config) UnmarshalJSON(data []byte) (err error)
• type Option
  • func WithConfig(conf *Config) Option
  • func WithConnect(driver, connect string) Option
  • func WithConnectURL(connect string) Option
  • func WithDebug(debug bool) Option
  • func WithFormat(format string) Option
  • func WithObjectConfig(obj any) Option
  • func WithRawConfig(raw json.RawMessage) Option

Variables

var ErrUndefinedSourceDriver = errors.New(`undefined source driver`)

ErrUndefinedSourceDriver in case if source not exists

Functions

func Close

func Close() error

Close all listeners and source connections, in the global registry

func Listen

func Listen(ctx context.Context) error

Listen method launch into the background all sources where the supervised daemon mode is required, in the global registry

func Register

func Register(ctx context.Context, name string, options ...Option) error

Register data source connection with `name` and options The source defines by connection options, in the global registry

func RegisterConnector

func RegisterConnector(driver string, c connector)

RegisterConnector function which creates new stream coneection by config and bind the connector to the `driver` name, in the global registry

func Source

func Source(name string) eventstream.Sourcer

Source returns the source object registered with the `name`, in the global registry

func Subscribe

func Subscribe(ctx context.Context, name string, streams ...eventstream.Streamer) error

Subscribe some handler interface to processing the stream with `name`, in the global registry

Types

type Config

type Config struct {
	Debug   bool
	Connect string
	Driver  string
	Format  string
	Raw     json.RawMessage
}

Config of the source connection

func (*Config) Decode

func (c *Config) Decode(v any) error

Decode raw data to the target object

func (*Config) UnmarshalJSON

func (c *Config) UnmarshalJSON(data []byte) (err error)

UnmarshalJSON data

type Option

type Option func(cnf *Config)

Option of the connection

func WithConfig

func WithConfig(conf *Config) Option

WithConfig custom

func WithConnect

func WithConnect(driver, connect string) Option

WithConnect to the source

func WithConnectURL

func WithConnectURL(connect string) Option

WithConnectURL to the source

func WithDebug

func WithDebug(debug bool) Option

WithDebug mode

func WithFormat

func WithFormat(format string) Option

WithFormat of the messages

func WithObjectConfig

func WithObjectConfig(obj any) Option

WithObjectConfig converts Object to JSON storage config

func WithRawConfig

func WithRawConfig(raw json.RawMessage) Option

WithRawConfig storage config