handler

Documentation

Index

• Constants
• Variables
• type FanoutPool
  • func InitializeTestFanoutPool(ctx context.Context, podName metrics.PodName, ...) (*FanoutPool, error)
  • func NewFanoutPool(targets config.ReadonlyTargets, pubsubClient *pubsub.Client, ...) (*FanoutPool, error)
  • func (p *FanoutPool) SyncOnce(ctx context.Context) error
• type Handler
  • func NewHandler(sub *pubsub.Subscription, processor processors.Interface, ...) *Handler
  • func (h *Handler) IsAlive() bool
  • func (h *Handler) Start(ctx context.Context, done func(error))
  • func (h *Handler) Stop()
• type Option
  • func WithDeliveryTimeout(t time.Duration) Option
  • func WithHandlerConcurrency(c int) Option
  • func WithMaxConcurrentPerEvent(c int) Option
  • func WithPubsubReceiveSettings(s pubsub.ReceiveSettings) Option
  • func WithRetryPolicy(r RetryPolicy) Option
  • func WithTimeoutPerEvent(t time.Duration) Option
• type Options
  • func NewOptions(opts ...Option) (*Options, error)
• type RetryClient
  • func NewRetryClient(ctx context.Context, client *pubsub.Client, opts ...ceclient.Option) (RetryClient, error)
• type RetryPolicy
• type RetryPool
  • func InitializeTestRetryPool(targets config.ReadonlyTargets, podName metrics.PodName, ...) (*RetryPool, error)
  • func NewRetryPool(targets config.ReadonlyTargets, pubsubClient *pubsub.Client, ...) (*RetryPool, error)
  • func (p *RetryPool) SyncOnce(ctx context.Context) error
• type SyncPool
  • func StartSyncPool(ctx context.Context, syncPool SyncPool, syncSignal <-chan struct{}, ...) (SyncPool, error)

Constants

const (
	// DefaultHealthCheckPort is the default port for checking sync pool health.
	DefaultHealthCheckPort = 8080
)

Variables

var (
	DefaultCEClientOpts = []ceclient.Option{
		ceclient.WithUUIDs(),
		ceclient.WithTimeNow(),
		ceclient.WithTracePropagation(),
	}

	DefaultHTTPClient = &http.Client{
		Transport: &ochttp.Transport{
			Base: &http.Transport{
				MaxIdleConns:        1000,
				MaxIdleConnsPerHost: 500,
				MaxConnsPerHost:     500,
				IdleConnTimeout:     30 * time.Second,
			},
			Propagation: &tracecontext.HTTPFormat{},
		},
	}

	// ProviderSet provides the fanout and retry sync pools using the default client options. In
	// order to inject either pool, ProjectID, []Option, and config.ReadOnlyTargets must be
	// externally provided.
	ProviderSet = wire.NewSet(
		NewFanoutPool,
		NewRetryPool,
		clients.NewPubsubClient,
		NewRetryClient,
		wire.Value(DefaultHTTPClient),
		wire.Value(DefaultCEClientOpts),
	)
)

Types

type FanoutPool

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

FanoutPool is the sync pool for fanout handlers. For each broker in the config, it will attempt to create a handler. It will also stop/delete the handler if the corresponding broker is deleted in the config.

func InitializeTestFanoutPool

func InitializeTestFanoutPool(ctx context.Context, podName metrics.PodName, containerName metrics.ContainerName, targets config.ReadonlyTargets, pubsubClient *pubsub.Client, opts ...Option) (*FanoutPool, error)

func NewFanoutPool

func NewFanoutPool(
	targets config.ReadonlyTargets,
	pubsubClient *pubsub.Client,
	deliverClient *http.Client,
	retryClient RetryClient,
	statsReporter *metrics.DeliveryReporter,
	opts ...Option,
) (*FanoutPool, error)

NewFanoutPool creates a new fanout handler pool.

func (*FanoutPool) SyncOnce

func (p *FanoutPool) SyncOnce(ctx context.Context) error

SyncOnce syncs once the handler pool based on the targets config.

type Handler

type Handler struct {
	// PubsubEvents is the CloudEvents Pubsub protocol to pull
	// messages as events.
	Subscription *pubsub.Subscription

	// Processor is the processor to process events.
	Processor processors.Interface

	// Timeout is the timeout for processing each individual event.
	Timeout time.Duration
	// contains filtered or unexported fields
}

Handler pulls Pubsub messages as events and processes them with chain of processors.

func NewHandler

func NewHandler(
	sub *pubsub.Subscription,
	processor processors.Interface,
	timeout time.Duration,
	retryPolicy RetryPolicy,
) *Handler

NewHandler creates a new Handler.

func (*Handler) IsAlive

func (h *Handler) IsAlive() bool

IsAlive indicates whether the handler is alive.

func (*Handler) Start

func (h *Handler) Start(ctx context.Context, done func(error))

Start starts the handler. done func will be called if the pubsub inbound is closed.

func (*Handler) Stop

func (h *Handler) Stop()

Stop stops the handlers.

type Option

type Option func(*Options)

Option is for providing individual option.

func WithDeliveryTimeout

func WithDeliveryTimeout(t time.Duration) Option

WithDeliveryTimeout sets the DeliveryTimeout.

func WithHandlerConcurrency

func WithHandlerConcurrency(c int) Option

WithHandlerConcurrency sets HandlerConcurrency.

func WithMaxConcurrentPerEvent

func WithMaxConcurrentPerEvent(c int) Option

WithMaxConcurrentPerEvent sets MaxConcurrencyPerEvent.

func WithPubsubReceiveSettings

func WithPubsubReceiveSettings(s pubsub.ReceiveSettings) Option

WithPubsubReceiveSettings sets PubsubReceiveSettings.

func WithRetryPolicy

func WithRetryPolicy(r RetryPolicy) Option

WithRetryPolicy sets the RetryPolicy.

func WithTimeoutPerEvent

func WithTimeoutPerEvent(t time.Duration) Option

WithTimeoutPerEvent sets TimeoutPerEvent.

type Options

type Options struct {
	// HandlerConcurrency is the number of goroutines
	// will be spawned in each handler.
	HandlerConcurrency int
	// MaxConcurrencyPerEvent is the max number of goroutines
	// will be spawned to handle an event.
	MaxConcurrencyPerEvent int
	// TimeoutPerEvent is the timeout for handling an event.
	TimeoutPerEvent time.Duration
	// DeliveryTimeout is the timeout for delivering an event to a consumer.
	DeliveryTimeout time.Duration
	// PubsubReceiveSettings is the pubsub receive settings.
	PubsubReceiveSettings pubsub.ReceiveSettings
	// RetryPolicy defines the retry policy for pubsub messages.
	RetryPolicy RetryPolicy
}

Options holds all the options for create handler pool.

func NewOptions

func NewOptions(opts ...Option) (*Options, error)

NewOptions creates a Options.

type RetryClient

type RetryClient ceclient.Client

func NewRetryClient

func NewRetryClient(ctx context.Context, client *pubsub.Client, opts ...ceclient.Option) (RetryClient, error)

NewRetryClient provides a retry CE client from a PubSub client and list of CE client options.

type RetryPolicy

type RetryPolicy struct {
	MinBackoff, MaxBackoff time.Duration
}

RetryPolicy defines the retry policy for pubsub messages. TODO: https://github.com/google/knative-gcp/issues/1100#issuecomment-638304147

type RetryPool

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

RetryPool is the sync pool for retry handlers. For each trigger in the config, it will attempt to create a handler. It will also stop/delete the handler if the corresponding trigger is deleted in the config.

func InitializeTestRetryPool

func InitializeTestRetryPool(targets config.ReadonlyTargets, podName metrics.PodName, containerName metrics.ContainerName, pubsubClient *pubsub.Client, opts ...Option) (*RetryPool, error)

func NewRetryPool

func NewRetryPool(
	targets config.ReadonlyTargets,
	pubsubClient *pubsub.Client,
	deliverClient *http.Client,
	statsReporter *metrics.DeliveryReporter,
	opts ...Option) (*RetryPool, error)

NewRetryPool creates a new retry handler pool.

func (*RetryPool) SyncOnce

func (p *RetryPool) SyncOnce(ctx context.Context) error

SyncOnce syncs once the handler pool based on the targets config.

type SyncPool

type SyncPool interface {
	SyncOnce(ctx context.Context) error
}

func StartSyncPool

func StartSyncPool(
	ctx context.Context,
	syncPool SyncPool,
	syncSignal <-chan struct{},
	maxStaleDuration time.Duration,
	healthCheckPort int,
) (SyncPool, error)

StartSyncPool starts the sync pool.