pubsubmiddleware

Documentation

Index

• Constants
• func WithFanoutValidation(schema *openapi3.Schema, messageHandler ParsedFanoutMessageHandler, ...) pubsub.FanoutMessageHandler
• func WithValidation(schema *openapi3.Schema, messageHandler ParsedMessageHandler, ...) pubsub.MessageHandler
• type ErrorHandler
  • func DefaultErrorHandler(schema *openapi3.Schema, logger *logging.Logger) ErrorHandler
• type FanoutErrorHandler
  • func DefaultFanoutErrorHandler(schema *openapi3.Schema, logger *logging.Logger) FanoutErrorHandler
• type ParsedFanoutMessageHandler
  • func RouteBy(classifier func(pubsub.Message, interface{}) string, ...) ParsedFanoutMessageHandler
  • func WithParser(parser ValueParser, handler ParsedFanoutMessageHandler) ParsedFanoutMessageHandler
• type ParsedMessageHandler
• type ValueParser
  • func ParserFor(value interface{}) ValueParser

Constants

const DefaultValidationLoggingComponent = "pubsub/validation"

Functions

func WithFanoutValidation

func WithFanoutValidation(
	schema *openapi3.Schema,
	messageHandler ParsedFanoutMessageHandler,
	errorHandler FanoutErrorHandler,
) pubsub.FanoutMessageHandler

WithFanoutValidation wraps a fanout message handler with OAS validation. Messages are assumed to be JSON. The inner messageHandler is required, but errorHandler may be nil, in which case all messages that fail validation will be logged and the error re-classified as a non-retriable error. Callers are encouraged to provide a customized handler so that the component name associated with the logging is identifiable.

func WithValidation

func WithValidation(
	schema *openapi3.Schema,
	messageHandler ParsedMessageHandler,
	errorHandler ErrorHandler,
) pubsub.MessageHandler

WithValidation wraps a message handler with OAS validation. Messages are assumed to be JSON. The inner messageHandler is required, but errorHandler may be nil, in which case all messages that fail validation will be logged and Ack()'d. Callers are encouraged to provide a customized handler so that the component name associated with the logging is identifiable.

Types

type ErrorHandler

type ErrorHandler func(context.Context, pubsub.Message, error) (ackMessage bool)

func DefaultErrorHandler

func DefaultErrorHandler(schema *openapi3.Schema, logger *logging.Logger) ErrorHandler

type FanoutErrorHandler

type FanoutErrorHandler func(context.Context, pubsub.Message, error) pubsub.RetriableError

func DefaultFanoutErrorHandler

func DefaultFanoutErrorHandler(schema *openapi3.Schema, logger *logging.Logger) FanoutErrorHandler

type ParsedFanoutMessageHandler

type ParsedFanoutMessageHandler func(context.Context, pubsub.Message, interface{}) pubsub.RetriableError

func RouteBy

func RouteBy(
	classifier func(pubsub.Message, interface{}) string,
	handlers map[string]ParsedFanoutMessageHandler,
) ParsedFanoutMessageHandler

RouteBy uses a classifier function to chose exactly one handler to process a message. If no handler matches the classifier return, routing handler will return nil (no error). This is the same idea as pubsub.RouteBy, but for the parsed variant of handlers.

func WithParser

func WithParser(
	parser ValueParser,
	handler ParsedFanoutMessageHandler,
) ParsedFanoutMessageHandler

WithParser wraps a handler with a message parser, replacing the interface{} value passed to the inner handler with that produced by the parser. This might be used, for example to replace the generic map[string]interface{} from JSON parsing done for validation with a strongly typed struct specific to the message. Any error from parser is always interpreted as non-retriable, even if an error implementing RetriableError is returned!

type ParsedMessageHandler

type ParsedMessageHandler func(context.Context, pubsub.Message, interface{})

type ValueParser

type ValueParser func(context.Context, pubsub.Message, interface{}) (interface{}, error)

ValueParser is the type of the parser argument to WithParser. Functions of this type should return a replacement parsed value for message handlers given the input message and value, or an error if parsing fails.

func ParserFor

func ParserFor(
	value interface{},
) ValueParser

ParserFor generates a reflect-driven parser function for the type of the given sample value, which must be a pointer type.