brokerchaos

Documentation

Index

• type ConfigProvider
• type Evaluator
  • func NewEvaluator(provider ConfigProvider) Evaluator
• type KafkaConsumerMiddleware
  • func NewKafkaConsumerMiddleware(eval Evaluator) *KafkaConsumerMiddleware
  • func (m *KafkaConsumerMiddleware) Intercept(ctx context.Context, msg *sarama.ConsumerMessage) (drop bool, err error)
• type MessageContext
• type ProtocolType
• type RabbitMQMiddleware
  • func NewRabbitMQMiddleware(eval Evaluator) *RabbitMQMiddleware
  • func (m *RabbitMQMiddleware) Intercept(ctx context.Context, delivery *amqp.Delivery) (drop bool, err error)

Types

type ConfigProvider

type ConfigProvider interface {
	GetActivePolicies() []config.Policy
	IsCommandIgnored(protocol string, cmd string) bool
}

type Evaluator

type Evaluator interface {
	Evaluate(ctx context.Context, msgCtx *MessageContext) (bool, time.Duration, error)
}

func NewEvaluator

func NewEvaluator(provider ConfigProvider) Evaluator

type KafkaConsumerMiddleware

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

func NewKafkaConsumerMiddleware

func NewKafkaConsumerMiddleware(eval Evaluator) *KafkaConsumerMiddleware

func (*KafkaConsumerMiddleware) Intercept

func (m *KafkaConsumerMiddleware) Intercept(ctx context.Context, msg *sarama.ConsumerMessage) (drop bool, err error)

type MessageContext

type MessageContext struct {
	Topic     string
	Protocol  ProtocolType
	Partition int32
	GetHeader func(key string) (string, bool)
}

type ProtocolType

type ProtocolType string

const (
	ProtocolKafka    ProtocolType = "kafka"
	ProtocolRabbitMQ ProtocolType = "rabbitmq"
)

type RabbitMQMiddleware

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

func NewRabbitMQMiddleware

func NewRabbitMQMiddleware(eval Evaluator) *RabbitMQMiddleware

func (*RabbitMQMiddleware) Intercept

func (m *RabbitMQMiddleware) Intercept(ctx context.Context, delivery *amqp.Delivery) (drop bool, err error)