buses

Documentation

Index

• Constants
• Variables
• func NewBusLoggerFromConfig(config *logging.Config) *zap.SugaredLogger
• func NewLoggingConfig() *logging.Config
• type ChannelReference
  • func NewChannelReference(channel *eventingv1alpha1.Channel) ChannelReference
  • func NewChannelReferenceFromNames(name, namespace string) ChannelReference
  • func NewChannelReferenceFromSubscription(subscription *eventingv1alpha1.Subscription) ChannelReference
  • func ParseChannel(host string) ChannelReference
  • func (r *ChannelReference) String() string
• type DispatchDefaults
• type Message
• type MessageDispatcher
  • func NewMessageDispatcher(logger *zap.SugaredLogger) *MessageDispatcher
  • func (d *MessageDispatcher) DispatchMessage(message *Message, destination, reply string, defaults DispatchDefaults) error
• type MessageReceiver
  • func NewMessageReceiver(receiverFunc func(ChannelReference, *Message) error, logger *zap.SugaredLogger) *MessageReceiver
  • func (r *MessageReceiver) HandleRequest(res http.ResponseWriter, req *http.Request)
  • func (r *MessageReceiver) Run(stopCh <-chan struct{})

Constants

const (
	MessageReceiverPort = 8080
)

MessageReceiver starts a server to receive new messages for the bus. The new message is emitted via the receiver function.

Variables

var ErrUnknownChannel = errors.New("unknown channel")

ErrUnknownChannel is returned when a message is received by a bus for a channel that does not exist.

Functions

func NewBusLoggerFromConfig

func NewBusLoggerFromConfig(config *logging.Config) *zap.SugaredLogger

NewBusLoggerFromConfig creates a new zap logger for the bus component based on the provided configuration

func NewLoggingConfig

func NewLoggingConfig() *logging.Config

NewLoggingConfig creates a static logging configuration appropriate for a bus. All logging levels are set to Info.

Types

type ChannelReference

type ChannelReference struct {
	Namespace string
	Name      string
}

ChannelReference references a Channel within the cluster by name and namespace.

func NewChannelReference

func NewChannelReference(channel *eventingv1alpha1.Channel) ChannelReference

NewChannelReference creates a ChannelReference from a Channel

func NewChannelReferenceFromNames

func NewChannelReferenceFromNames(name, namespace string) ChannelReference

NewChannelReferenceFromNames creates a ChannelReference for a name and namespace.

func NewChannelReferenceFromSubscription

func NewChannelReferenceFromSubscription(subscription *eventingv1alpha1.Subscription) ChannelReference

NewChannelReferenceFromSubscription creates a ChannelReference from a Subscription for a Channel.

func ParseChannel

func ParseChannel(host string) ChannelReference

ParseChannel converts the channel's hostname into a channel reference.

func (*ChannelReference) String

func (r *ChannelReference) String() string

type DispatchDefaults

type DispatchDefaults struct {
	Namespace string
}

DispatchDefaults provides default parameter values used when dispatching a message.

type Message

type Message struct {

	// Headers provide metadata about the message payload. All header keys
	// should be lowercase.
	Headers map[string]string

	// Payload is the raw binary content of the message. The payload format is
	// often described by the 'content-type' header.
	Payload []byte
}

Message represents an chunk of data within a bus. The message contains both a map of string headers and a binary payload.

A message may represent a CloudEvent.

type MessageDispatcher

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

MessageDispatcher dispatches messages to a destination over HTTP.

func NewMessageDispatcher

func NewMessageDispatcher(logger *zap.SugaredLogger) *MessageDispatcher

NewMessageDispatcher creates a new message dispatcher that can dispatch messages to HTTP destinations.

func (*MessageDispatcher) DispatchMessage

func (d *MessageDispatcher) DispatchMessage(message *Message, destination, reply string, defaults DispatchDefaults) error

DispatchMessage dispatches a message to a destination over HTTP.

The destination and reply are DNS names. For names with a single label, the default namespace is used to expand it into a fully qualified name within the cluster.

type MessageReceiver

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

func NewMessageReceiver

func NewMessageReceiver(receiverFunc func(ChannelReference, *Message) error, logger *zap.SugaredLogger) *MessageReceiver

NewMessageReceiver creates a message receiver passing new messages to the receiverFunc.

func (*MessageReceiver) HandleRequest

func (r *MessageReceiver) HandleRequest(res http.ResponseWriter, req *http.Request)

HandleRequest is an http Handler function. The request is converted to a Message and emitted to the receiver func.

The response status codes:

202 - the message was sent to subscibers
404 - the request was for an unknown channel
500 - an error occurred processing the request

func (*MessageReceiver) Run

func (r *MessageReceiver) Run(stopCh <-chan struct{})

Run starts receiving 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.