pubsubx

Documentation

Index

• Constants
• Variables
• func AddConfigSchema(c interface{ ... }) error
• func NewLogrusLogger(log *logrus.Logger) watermill.LoggerAdapter
• func NewOTELSaramaTracer(option ...otelsarama.Option) kafka.SaramaTracer
• 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 OTELSaramaTracer
  • func (t OTELSaramaTracer) WrapConsumer(c sarama.Consumer) sarama.Consumer
  • func (t OTELSaramaTracer) WrapConsumerGroupHandler(h sarama.ConsumerGroupHandler) sarama.ConsumerGroupHandler
  • func (t OTELSaramaTracer) WrapPartitionConsumer(pc sarama.PartitionConsumer) sarama.PartitionConsumer
  • func (t OTELSaramaTracer) WrapSyncProducer(cfg *sarama.Config, p sarama.SyncProducer) sarama.SyncProducer
• type ProvidersConfig
• type PubSub
  • func New(l *logrusx.Logger, c *Config, opts ...PubSubOption) (PubSub, error)
• type PubSubOption
  • func WithPropagator(propagator propagation.TextMapPropagator) PubSubOption
  • func WithTracerProvider(provider trace.TracerProvider) PubSubOption
• type Publisher
  • func SetupInMemoryPublisher(l *logrusx.Logger, c *Config) (Publisher, error)
  • func SetupKafkaPublisher(l *logrusx.Logger, c *Config, opts *pubSubOptions) (Publisher, error)
• type Subscriber
  • func SetupInMemorySubscriber(l *logrusx.Logger, c *Config) (Subscriber, error)
  • func SetupKafkaSubscriber(l *logrusx.Logger, c *Config, opts *pubSubOptions, group string) (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 NewOTELSaramaTracer

func NewOTELSaramaTracer(option ...otelsarama.Option) kafka.SaramaTracer

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 OTELSaramaTracer

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

func (OTELSaramaTracer) WrapConsumer

func (t OTELSaramaTracer) WrapConsumer(c sarama.Consumer) sarama.Consumer

func (OTELSaramaTracer) WrapConsumerGroupHandler

func (t OTELSaramaTracer) WrapConsumerGroupHandler(h sarama.ConsumerGroupHandler) sarama.ConsumerGroupHandler

func (OTELSaramaTracer) WrapPartitionConsumer

func (t OTELSaramaTracer) WrapPartitionConsumer(pc sarama.PartitionConsumer) sarama.PartitionConsumer

func (OTELSaramaTracer) WrapSyncProducer

func (t OTELSaramaTracer) WrapSyncProducer(cfg *sarama.Config, p sarama.SyncProducer) sarama.SyncProducer

type ProvidersConfig

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

type PubSub

type PubSub interface {
	Publisher() Publisher
	Subscriber(group string) (Subscriber, error)
	// CLoses all publishers and subscribers.
	Close() error
}

func New

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

type PubSubOption

type PubSubOption func(*pubSubOptions)

func WithPropagator

func WithPropagator(propagator propagation.TextMapPropagator) PubSubOption

func WithTracerProvider

func WithTracerProvider(provider trace.TracerProvider) PubSubOption

WithTracerProvider specifies a tracer provider to use for creating a tracer. If none is specified, no tracer is configured

type Publisher

type Publisher interface {
	// Publish publishes a message to the topic.
	Publish(ctx context.Context, 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, opts *pubSubOptions) (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, opts *pubSubOptions, group string) (Subscriber, error)

TODO: add subscriber configs