Documentation
¶
Index ¶
- Constants
- Variables
- func CreateDispatcherService(ctx context.Context, client runtimeClient.Client, ...) (*corev1.Service, error)
- func CreateK8sService(ctx context.Context, client runtimeClient.Client, c *eventingv1alpha1.Channel) (*corev1.Service, error)
- func CreateVirtualService(ctx context.Context, client runtimeClient.Client, ...) (*istiov1alpha3.VirtualService, error)
- func DispatcherLabels(ccpName string) map[string]string
- func NewLoggingConfig() *logging.Config
- func NewProvisionerLoggerFromConfig(config *logging.Config) *zap.SugaredLogger
- func RemoveFinalizer(o metav1.Object, finalizerName string)
- func UpdateChannel(ctx context.Context, client runtimeClient.Client, u *eventingv1alpha1.Channel) error
- func UpdateClusterChannelProvisionerStatus(ctx context.Context, client runtimeClient.Client, ...) error
- type AddFinalizerResult
- type ChannelReference
- type DispatchDefaults
- type Dispatcher
- type Message
- type MessageDispatcher
- type MessageReceiver
Constants ¶
const ( PortName = "http" PortNumber = 80 // EventingChannelLabel carries the name of knative's label for the channel EventingChannelLabel = "eventing.knative.dev/channel" // EventingProvisionerLabel carries the name of knative's label for the provisioner EventingProvisionerLabel = "eventing.knative.dev/provisioner" // OldEventingChannelLabel carries the name of knative's old label for the channel OldEventingChannelLabel = "channel" // OldEventingProvisionerLabel carries the name of knative's old label for the provisioner OldEventingProvisionerLabel = "provisioner" )
const ( // MessageHistoryHeader is the header containing all channel hosts traversed by the message // This is an experimental header: https://github.com/knative/eventing/issues/638 MessageHistoryHeader = "ce-knativehistory" MessageHistorySeparator = "; " )
const (
MessageReceiverPort = 8080
)
MessageReceiver starts a server to receive new messages for the channel dispatcher. 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 channel dispatcher for a channel that does not exist.
Functions ¶
func CreateDispatcherService ¶
func CreateDispatcherService(ctx context.Context, client runtimeClient.Client, ccp *eventingv1alpha1.ClusterChannelProvisioner) (*corev1.Service, error)
func CreateK8sService ¶
func CreateK8sService(ctx context.Context, client runtimeClient.Client, c *eventingv1alpha1.Channel) (*corev1.Service, error)
func CreateVirtualService ¶
func CreateVirtualService(ctx context.Context, client runtimeClient.Client, channel *eventingv1alpha1.Channel, svc *corev1.Service) (*istiov1alpha3.VirtualService, error)
func DispatcherLabels ¶
func NewLoggingConfig ¶
NewLoggingConfig creates a static logging configuration appropriate for a provisioner. All logging levels are set to Info.
func NewProvisionerLoggerFromConfig ¶
func NewProvisionerLoggerFromConfig(config *logging.Config) *zap.SugaredLogger
NewProvisionerLoggerFromConfig creates a new zap logger for the provisioner component based on the provided configuration
func RemoveFinalizer ¶
func UpdateChannel ¶
func UpdateChannel(ctx context.Context, client runtimeClient.Client, u *eventingv1alpha1.Channel) error
func UpdateClusterChannelProvisionerStatus ¶
func UpdateClusterChannelProvisionerStatus(ctx context.Context, client runtimeClient.Client, u *eventingv1alpha1.ClusterChannelProvisioner) error
Types ¶
type AddFinalizerResult ¶
type AddFinalizerResult bool
AddFinalizerResult is used indicate whether a finalizer was added or already present.
const ( FinalizerAlreadyPresent AddFinalizerResult = false FinalizerAdded AddFinalizerResult = true )
func AddFinalizer ¶
func AddFinalizer(o metav1.Object, finalizerName string) AddFinalizerResult
AddFinalizer adds finalizerName to the Object.
type ChannelReference ¶
ChannelReference references a Channel within the cluster by name and namespace.
func ParseChannel ¶
func ParseChannel(host string) (ChannelReference, error)
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 Dispatcher ¶
type Dispatcher interface {
// 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.
DispatchMessage(message *Message, destination, reply string, defaults DispatchDefaults) error
}
type Message ¶
type Message struct {
// Headers provide metadata about the message payload. All header keys
// should be lowercase.
Headers map[string]string `json:"headers,omitempty"`
// Payload is the raw binary content of the message. The payload format is
// often described by the 'content-type' header.
Payload []byte `json:"payload,omitempty"`
}
Message represents an chunk of data within a channel dispatcher. The message contains both a map of string headers and a binary payload. This struct gets mashaled/unmarshaled in order to preserve and pass Header information to the event subscriber.
A message may represent a CloudEvent.
func (*Message) AppendToHistory ¶
AppendToHistory append a new host at the end of the list of hosts of the message history
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
}
Message receiver receives messages.
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 subscribers 404 - the request was for an unknown channel 500 - an error occurred processing the request
func (*MessageReceiver) Start ¶
func (r *MessageReceiver) Start(stopCh <-chan struct{}) error
Start begings 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.
Source Files
Directories