gtag

Documentation

Index

• Constants
• Variables
• func MPv2MessageHandler(msg *message.Message) ([]*message.Message, error)
• func MessageHandler(handler func(payload *gtag.Payload, msg *message.Message) error) message.HandlerFunc
• func NoPublishMessageHandler(handler func(payload *gtag.Payload, msg *message.Message) error) func(msg *message.Message) error
• type Publisher
  • func NewPublisher(l *zap.Logger, host string, opts ...PublisherOption) *Publisher
  • func (p *Publisher) Client() *http.Client
  • func (p *Publisher) Close() error
  • func (p *Publisher) Publish(topic string, messages ...*message.Message) error
• type PublisherHandler
  • func PublisherMiddlewareIgnoreError(next PublisherHandler) PublisherHandler
• type PublisherMarshalMessageFunc
• type PublisherMiddleware
• type PublisherOption
  • func PublisherWithClient(v *http.Client) PublisherOption
  • func PublisherWithMaxResponseCode(v int) 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 ...gtaghttp.Middleware) SubscriberOption
  • func SubscriberWithUUIDFunc(v func() string) SubscriberOption

Constants

const (
	MetadataEventName = "X-Event-Name"
)

Variables

var (
	ErrMissingEventName = errors.New("missing event name")
	ErrContextCanceled  = errors.New("request stopped without ACK received")
	ErrMessageNacked    = errors.New("message nacked")
	ErrClosed           = errors.New("subscriber already closed")
)

var (
	ErrErrorResponse   = errors.New("server responded with error status")
	ErrPublisherClosed = errors.New("publisher is closed")
)

Functions

func MPv2MessageHandler

func MPv2MessageHandler(msg *message.Message) ([]*message.Message, error)

func MessageHandler

func MessageHandler(handler func(payload *gtag.Payload, msg *message.Message) error) message.HandlerFunc

func NoPublishMessageHandler

func NoPublishMessageHandler(handler func(payload *gtag.Payload, msg *message.Message) error) func(msg *message.Message) error

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) Client

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

func (*Publisher) Close

func (p *Publisher) Close() error

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 PublisherMiddlewareIgnoreError

func PublisherMiddlewareIgnoreError(next PublisherHandler) PublisherHandler

type PublisherMarshalMessageFunc

type PublisherMarshalMessageFunc func(url string, msg *message.Message) (*http.Request, error)

PublisherMarshalMessageFunc transforms the message into a HTTP request to be sent to the specified url.

type PublisherMiddleware

type PublisherMiddleware func(next PublisherHandler) PublisherHandler

type PublisherOption

type PublisherOption func(*Publisher)

func PublisherWithClient

func PublisherWithClient(v *http.Client) PublisherOption

func PublisherWithMaxResponseCode

func PublisherWithMaxResponseCode(v int) 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 ...gtaghttp.Middleware) SubscriberOption

func SubscriberWithUUIDFunc

func SubscriberWithUUIDFunc(v func() string) SubscriberOption