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 BatchConsumerOptions
• type Config
• type ConsumerModel
• 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
• type Subscriber
• type SubscriberOption
  • func WithBatchConsumerModel(batchOptions *BatchConsumerOptions) SubscriberOption
  • func WithDefaultConsumerModel() SubscriberOption

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 BatchConsumerOptions

type BatchConsumerOptions struct {
	// MaxBatchSize max amount of elements the batch will contain.
	// Default value is 100 if nothing is specified.
	MaxBatchSize int16
	// MaxWaitTime max time that it will be waited until MaxBatchSize elements are received.
	// Default value is 100ms if nothing is specified.
	MaxWaitTime time.Duration
}

type Config

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

type ConsumerModel

type ConsumerModel string

var (

	// Default is a model when only one message is sent to the customer and customer needs to ACK the message
	// to receive the next.
	ConsumerModelDefault ConsumerModel = ""
	// Batch works by sending multiple messages in a batch
	// You can ack all of them at once, or one by one.
	ConsumerModelBatch ConsumerModel = "batch"
)

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, opts ...SubscriberOption) (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.

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
}

type SubscriberOption

type SubscriberOption func(*subscriberOptions)

func WithBatchConsumerModel

func WithBatchConsumerModel(batchOptions *BatchConsumerOptions) SubscriberOption

func WithDefaultConsumerModel

func WithDefaultConsumerModel() SubscriberOption