broker

Documentation

Index

• Variables
• func Buckets125(low, high float64) []float64
• func GetTTL(ctx cloudevents.EventContext) (interface{}, string)
• func SendingContext(ctx context.Context, tctx cloudevents.HTTPTransportContext, targetURI *url.URL) context.Context
• func SetTTL(ctx cloudevents.EventContext, ttl interface{}) (cloudevents.EventContext, error)
• type Receiver
  • func New(logger *zap.Logger, client client.Client) (*Receiver, error)
  • func (r *Receiver) Start(stopCh <-chan struct{}) error

Variables

var (
	// MeasureTriggerEventsTotal is a counter which records the number of events received
	// by a Trigger.
	MeasureTriggerEventsTotal = stats.Int64(
		"knative.dev/eventing/trigger/measures/events_total",
		"Total number of events received by a Trigger",
		stats.UnitNone,
	)

	// MeasureTriggerDispatchTime records the time spent dispatching an event for
	// a Trigger, in milliseconds.
	MeasureTriggerDispatchTime = stats.Int64(
		"knative.dev/eventing/trigger/measures/dispatch_time",
		"Time spent dispatching an event to a Trigger",
		stats.UnitMilliseconds,
	)

	// MeasureTriggerFilterTime records the time spent filtering a message for a
	// Trigger, in milliseconds.
	MeasureTriggerFilterTime = stats.Int64(
		"knative.dev/eventing/trigger/measures/filter_time",
		"Time spent filtering a message for a Trigger",
		stats.UnitMilliseconds,
	)

	// TagResult is a tag key referring to the observed result of an operation.
	TagResult = mustNewTagKey("result")

	// TagFilterResult is a tag key referring to the observed result of a filter
	// operation.
	TagFilterResult = mustNewTagKey("filter_result")

	// TagBroker is a tag key referring to the Broker name serviced by this
	// filter process.
	TagBroker = mustNewTagKey("broker")

	// TagTrigger is a tag key referring to the Trigger name serviced by this
	// filter process.
	TagTrigger = mustNewTagKey("trigger")
)

Functions

func Buckets125

func Buckets125(low, high float64) []float64

Buckets125 generates an array of buckets with approximate powers-of-two buckets that also aligns with powers of 10 on every 3rd step. This can be used to create a view.Distribution.

func GetTTL

func GetTTL(ctx cloudevents.EventContext) (interface{}, string)

GetTTL finds the TTL in the EventContext using a case insensitive comparison for the key. The second return param, is the case preserved key that matched. Depending on the encoding/transport, the extension case could be changed.

func SendingContext

func SendingContext(ctx context.Context, tctx cloudevents.HTTPTransportContext, targetURI *url.URL) context.Context

SendingContext creates the context to use when sending a Cloud Event with ceclient.Client. It sets the target and attaches a filtered set of headers from the initial request.

func SetTTL

func SetTTL(ctx cloudevents.EventContext, ttl interface{}) (cloudevents.EventContext, error)

SetTTL sets the TTL into the EventContext. ttl should be a positive integer.

Types

type Receiver

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

Receiver parses Cloud Events, determines if they pass a filter, and sends them to a subscriber.

func New

func New(logger *zap.Logger, client client.Client) (*Receiver, error)

New creates a new Receiver and its associated MessageReceiver. The caller is responsible for Start()ing the returned MessageReceiver.

func (*Receiver) Start

func (r *Receiver) Start(stopCh <-chan struct{}) error

Start begins to receive messages for the receiver.

Only HTTP POST requests to the root path (/) are accepted. If other paths or methods are needed, use the HandleRequest method directly with another HTTP server.

This method will block until a message is received on the stop channel.