stream

Documentation

Index

• func Init(ctx context.Context, pgURL string) error
• func Run(ctx context.Context, logger loglib.Logger, config *Config, ...) error
• func TearDown(ctx context.Context, pgURL string) error
• type Config
  • func (c *Config) IsValid() error
• type KafkaListenerConfig
• type KafkaProcessorConfig
• type ListenerConfig
• type PostgresListenerConfig
• type ProcessorConfig
• type SearchProcessorConfig
• type WebhookProcessorConfig
• type WebhookSubscriptionStoreConfig

Functions

func Init

func Init(ctx context.Context, pgURL string) error

Init initialises the pgstream state in the postgres database provided, along with creating the relevant replication slot.

func Run

func Run(ctx context.Context, logger loglib.Logger, config *Config, instrumentation *otel.Instrumentation) error

Run will run the configured pgstream processes. This call is blocking.

func TearDown

func TearDown(ctx context.Context, pgURL string) error

TearDown removes the pgstream state from the postgres database provided, as well as removing the replication slot.

Types

type Config

type Config struct {
	Listener  ListenerConfig
	Processor ProcessorConfig
}

func (*Config) IsValid

func (c *Config) IsValid() error

type KafkaListenerConfig

type KafkaListenerConfig struct {
	Reader       kafka.ReaderConfig
	Checkpointer kafkacheckpoint.Config
}

type KafkaProcessorConfig

type KafkaProcessorConfig struct {
	Writer *kafkaprocessor.Config
}

type ListenerConfig

type ListenerConfig struct {
	Postgres *PostgresListenerConfig
	Kafka    *KafkaListenerConfig
}

type PostgresListenerConfig

type PostgresListenerConfig struct {
	Replication pgreplication.Config
}

type ProcessorConfig

type ProcessorConfig struct {
	Kafka      *KafkaProcessorConfig
	Search     *SearchProcessorConfig
	Webhook    *WebhookProcessorConfig
	Translator *translator.Config
}

type SearchProcessorConfig

type SearchProcessorConfig struct {
	Indexer search.IndexerConfig
	Store   store.Config
	Retrier search.StoreRetryConfig
}

type WebhookProcessorConfig

type WebhookProcessorConfig struct {
	Notifier           notifier.Config
	SubscriptionServer server.Config
	SubscriptionStore  WebhookSubscriptionStoreConfig
}

type WebhookSubscriptionStoreConfig

type WebhookSubscriptionStoreConfig struct {
	URL                  string
	CacheEnabled         bool
	CacheRefreshInterval time.Duration
}