pubsubx

Documentation

Index

• Constants
• Variables
• func AddConfigSchema(c interface{ ... }) error
• func NewLogrusLogger(log *logrus.Logger) watermill.LoggerAdapter
• func SetupInMemoryPubSub(l *logrusx.Logger, c *Config) (*memoryPubSub, error)
• type Config
• type InMemoryConfig
• type KafkaConfig
• type LogrusLoggerAdapter
  • func (l *LogrusLoggerAdapter) Debug(msg string, fields watermill.LogFields)
  • func (l *LogrusLoggerAdapter) Error(msg string, err error, fields watermill.LogFields)
  • func (l *LogrusLoggerAdapter) Info(msg string, fields watermill.LogFields)
  • func (l *LogrusLoggerAdapter) Trace(msg string, fields watermill.LogFields)
  • func (l *LogrusLoggerAdapter) With(fields watermill.LogFields) watermill.LoggerAdapter
• type ProvidersConfig
• type PubSub
  • func New(l *logrusx.Logger, c *Config) (*PubSub, error)
• type Publisher
  • func SetupInMemoryPublisher(l *logrusx.Logger, c *Config) (Publisher, error)
  • func SetupKafkaPublisher(l *logrusx.Logger, c *Config) (Publisher, error)
• type Subscriber
  • func SetupInMemorySubscriber(l *logrusx.Logger, c *Config) (Subscriber, error)
  • func SetupKafkaSubscriber(l *logrusx.Logger, c *Config) (Subscriber, error)

Constants

const ConfigSchemaID = "clinia://pubsub-config"

Variables

var ConfigSchema string

Functions

func AddConfigSchema

func AddConfigSchema(c interface {
	AddResource(url string, r io.Reader) error
}) error

AddConfigSchema adds the tracing schema to the compiler. The interface is specified instead of `jsonschema.Compiler` to allow the use of any jsonschema library fork or version.

func NewLogrusLogger

func NewLogrusLogger(log *logrus.Logger) watermill.LoggerAdapter

NewLogrusLogger returns a LogrusLoggerAdapter that sends all logs to the passed logrus instance.

func SetupInMemoryPubSub

func SetupInMemoryPubSub(l *logrusx.Logger, c *Config) (*memoryPubSub, error)

Types

type Config

type Config struct {
	Provider  string          `json:"provider"`
	Providers ProvidersConfig `json:"providers"`
}

type InMemoryConfig

type InMemoryConfig struct{}

type KafkaConfig

type KafkaConfig struct {
	Brokers []string `json:"brokers"`
}

type LogrusLoggerAdapter

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

LogrusLoggerAdapter is a logrus adapter for watermill.

func (*LogrusLoggerAdapter) Debug

func (l *LogrusLoggerAdapter) Debug(msg string, fields watermill.LogFields)

Debug logs on level debug with optional fields.

func (*LogrusLoggerAdapter) Error

func (l *LogrusLoggerAdapter) Error(msg string, err error, fields watermill.LogFields)

Error logs on level error with err as field and optional fields.

func (*LogrusLoggerAdapter) Info

func (l *LogrusLoggerAdapter) Info(msg string, fields watermill.LogFields)

Info logs on level info with optional fields.

func (*LogrusLoggerAdapter) Trace

func (l *LogrusLoggerAdapter) Trace(msg string, fields watermill.LogFields)

Trace logs on level trace with optional fields.

func (*LogrusLoggerAdapter) With

func (l *LogrusLoggerAdapter) With(fields watermill.LogFields) watermill.LoggerAdapter

With returns a new LogrusLoggerAdapter that includes fields to be re-used between logging statements.

type ProvidersConfig

type ProvidersConfig struct {
	InMemory InMemoryConfig `json:"inmemory"`
	Kafka    KafkaConfig    `json:"kafka"`
}

type PubSub

type PubSub struct {
	Publisher  Publisher
	Subscriber Subscriber
}

func New

func New(l *logrusx.Logger, c *Config) (*PubSub, error)

type Publisher

type Publisher interface {
	// Publish publishes a message to the topic.
	Publish(topic string, messages ...*message.Message) error
	// Close closes the publisher.
	Close() error
}

Publisher is the interface that wraps the Publish method.

func SetupInMemoryPublisher

func SetupInMemoryPublisher(l *logrusx.Logger, c *Config) (Publisher, error)

func SetupKafkaPublisher

func SetupKafkaPublisher(l *logrusx.Logger, c *Config) (Publisher, error)

TODO: add publisher configs

type Subscriber

type Subscriber interface {
	// Subscribe subscribes to the topic.
	Subscribe(ctx context.Context, topic string) (<-chan *message.Message, error)
	// Close closes the subscriber.
	Close() error
}

func SetupInMemorySubscriber

func SetupInMemorySubscriber(l *logrusx.Logger, c *Config) (Subscriber, error)

func SetupKafkaSubscriber

func SetupKafkaSubscriber(l *logrusx.Logger, c *Config) (Subscriber, error)

TODO: add subscriber configs