handler

Documentation

Index

• type CompositeEventHandler
  • func (h *CompositeEventHandler) AddHandlers(handlers ...EventHandler) *CompositeEventHandler
  • func (h *CompositeEventHandler) Handle(event cloudevents.Event) cloudevents.Result
  • func (h *CompositeEventHandler) SupportEventTypes() []string
• type EventConsumer
  • func NewEventConsumer(cfg config.Kafka, topic string, hander EventHandler, faultWriter *kafka.Writer) (*EventConsumer, error)
  • func (ec *EventConsumer) Close()
  • func (ec *EventConsumer) Start(ctx context.Context, wg *sync.WaitGroup)
• type EventConsumerGroup
  • func NewEventConsumerGroup(cfg config.Kafka, hander EventHandler) (EventConsumerGroup, error)
  • func (ecs EventConsumerGroup) Close()
  • func (ecs EventConsumerGroup) Start(ctx context.Context, wg *sync.WaitGroup)
• type EventHandler
• type EventProducer
  • func NewEventProducer(cfg config.Kafka) (*EventProducer, error)
  • func (eb *EventProducer) HandleCloudEvent(ctx context.Context, event cloudevents.Event) cloudevents.Result

Types

type CompositeEventHandler

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

CompositeEventHandler is a public struct that composes multiple event handlers.

func (*CompositeEventHandler) AddHandlers

func (h *CompositeEventHandler) AddHandlers(handlers ...EventHandler) *CompositeEventHandler

AddHandlers adds child handlers to the composite event handler.

func (*CompositeEventHandler) Handle

func (h *CompositeEventHandler) Handle(event cloudevents.Event) cloudevents.Result

Handle handles the given event.

func (*CompositeEventHandler) SupportEventTypes

func (h *CompositeEventHandler) SupportEventTypes() []string

SupportEventTypes returns a list of supported event types.

type EventConsumer

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

func NewEventConsumer

func NewEventConsumer(cfg config.Kafka, topic string, hander EventHandler, faultWriter *kafka.Writer) (*EventConsumer, error)

func (*EventConsumer) Close

func (ec *EventConsumer) Close()

func (*EventConsumer) Start

func (ec *EventConsumer) Start(ctx context.Context, wg *sync.WaitGroup)

consumer workers

type EventConsumerGroup

type EventConsumerGroup map[string]*EventConsumer

func NewEventConsumerGroup

func NewEventConsumerGroup(cfg config.Kafka, hander EventHandler) (EventConsumerGroup, error)

func (EventConsumerGroup) Close

func (ecs EventConsumerGroup) Close()

func (EventConsumerGroup) Start

func (ecs EventConsumerGroup) Start(ctx context.Context, wg *sync.WaitGroup)

Start runs the EventConsumerGroup in parallel, starting each EventConsumer in a separate goroutine. It waits for all EventConsumers to finish before returning.

type EventHandler

type EventHandler interface {
	// Valid fn signatures are:
	// * func()
	// * func() protocol.Result
	// * func(context.Context)
	// * func(context.Context) protocol.Result
	// * func(event.Event)
	// * func(event.Event) transport.Result
	// * func(context.Context, event.Event)
	// * func(context.Context, event.Event) protocol.Result
	// * func(event.Event) *event.Event
	// * func(event.Event) (*event.Event, protocol.Result)
	// * func(context.Context, event.Event) *event.Event
	// * func(context.Context, event.Event) (*event.Event, protocol.Result)
	// Handle handles the given event.
	Handle(event cloudevents.Event) cloudevents.Result
	// SupportEventTypes returns a list of supported event types.
	SupportEventTypes() []string
}

EventHandler is an interface that defines the Handle method.

type EventProducer

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

EventProducer is the main structure for our event broker

func NewEventProducer

func NewEventProducer(cfg config.Kafka) (*EventProducer, error)

func (*EventProducer) HandleCloudEvent

func (eb *EventProducer) HandleCloudEvent(ctx context.Context, event cloudevents.Event) cloudevents.Result