mpv2

Documentation

Index

• Constants
• Variables
• func MessageHandler(handler func(payload *mpv2.Payload[any], msg *message.Message) error) message.HandlerFunc
• func NoPublishMessageHandler(handler func(payload *mpv2.Payload[any], msg *message.Message) error) message.NoPublishHandlerFunc
• type Publisher
  • func NewPublisher(l *zap.Logger, host string, opts ...PublisherOption) *Publisher
  • func (p *Publisher) Close() error
  • func (p *Publisher) HTTPClient() *http.Client
  • func (p *Publisher) Publish(topic string, messages ...*message.Message) error
• type PublisherHandler
  • func PublisherMiddlewareEventParams(next PublisherHandler) PublisherHandler
  • func PublisherMiddlewareIgnoreError(next PublisherHandler) PublisherHandler
• type PublisherMiddleware
• type PublisherOption
  • func PublisherWithHTTPClient(v *http.Client) PublisherOption
  • func PublisherWithMiddlewares(v ...PublisherMiddleware) PublisherOption
  • func PublisherWithPath(v string) PublisherOption
• type Subscriber
  • func NewSubscriber(l *zap.Logger, opts ...SubscriberOption) *Subscriber
  • func (s *Subscriber) Close() error
  • func (s *Subscriber) ServeHTTP(w http.ResponseWriter, r *http.Request)
  • func (s *Subscriber) Subscribe(ctx context.Context, topic string) (<-chan *message.Message, error)
• type SubscriberOption
  • func SubscriberWithMessageFunc(v func(l *zap.Logger, r *http.Request, msg *message.Message) error) SubscriberOption
  • func SubscriberWithMiddlewares(v ...mpv2http.Middleware) SubscriberOption
  • func SubscriberWithUUIDFunc(v func() string) SubscriberOption

Constants

const (
	MetadataRequestQuery = "RequestQuery"
)

Variables

var (
	ErrMissingEventName = errors.New("missing event name")
	ErrErrorResponse    = errors.New("server responded with error status")
	ErrPublisherClosed  = errors.New("publisher is closed")
	ErrContextCanceled  = errors.New("request stopped without ACK received")
	ErrMessageNacked    = errors.New("message nacked")
	ErrClosed           = errors.New("subscriber already closed")
)

Functions

func MessageHandler

func MessageHandler(handler func(payload *mpv2.Payload[any], msg *message.Message) error) message.HandlerFunc

func NoPublishMessageHandler

func NoPublishMessageHandler(handler func(payload *mpv2.Payload[any], msg *message.Message) error) message.NoPublishHandlerFunc

Types

type Publisher

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

func NewPublisher

func NewPublisher(l *zap.Logger, host string, opts ...PublisherOption) *Publisher

func (*Publisher) Close

func (p *Publisher) Close() error

func (*Publisher) HTTPClient

func (p *Publisher) HTTPClient() *http.Client

func (*Publisher) Publish

func (p *Publisher) Publish(topic string, messages ...*message.Message) error

type PublisherHandler

type PublisherHandler func(l *zap.Logger, msg *message.Message) error

func PublisherMiddlewareEventParams

func PublisherMiddlewareEventParams(next PublisherHandler) PublisherHandler

PublisherMiddlewareEventParams moves the `debug_mode`, `session_id` & `engagement_time_msec` into the events params since this is required by the measurement protocol but make coding much more complex. That's why it's part of the payload in this library.

func PublisherMiddlewareIgnoreError

func PublisherMiddlewareIgnoreError(next PublisherHandler) PublisherHandler

PublisherMiddlewareIgnoreError ignores error responses from the gtm endpoint to prevent retries.

type PublisherMiddleware

type PublisherMiddleware func(next PublisherHandler) PublisherHandler

type PublisherOption

type PublisherOption func(*Publisher)

func PublisherWithHTTPClient

func PublisherWithHTTPClient(v *http.Client) PublisherOption

func PublisherWithMiddlewares

func PublisherWithMiddlewares(v ...PublisherMiddleware) PublisherOption

func PublisherWithPath

func PublisherWithPath(v string) PublisherOption

type Subscriber

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

func NewSubscriber

func NewSubscriber(l *zap.Logger, opts ...SubscriberOption) *Subscriber

func (*Subscriber) Close

func (s *Subscriber) Close() error

Close closes all subscriptions with their output channels and flush offsets etc. when needed.

func (*Subscriber) ServeHTTP

func (s *Subscriber) ServeHTTP(w http.ResponseWriter, r *http.Request)

func (*Subscriber) Subscribe

func (s *Subscriber) Subscribe(ctx context.Context, topic string) (<-chan *message.Message, error)

type SubscriberOption

type SubscriberOption func(*Subscriber)

func SubscriberWithMessageFunc

func SubscriberWithMessageFunc(v func(l *zap.Logger, r *http.Request, msg *message.Message) error) SubscriberOption

func SubscriberWithMiddlewares

func SubscriberWithMiddlewares(v ...mpv2http.Middleware) SubscriberOption

func SubscriberWithUUIDFunc

func SubscriberWithUUIDFunc(v func() string) SubscriberOption