cloudevents

Documentation

Overview

Package cloudevents provides a CloudEvents library focused on target component requirements and on how responses should be composed.

Basics

The package provides a Replier object that should be instantiated as a singleton, and a set of pubic functions that are further divided into Knative managed and Custom managed responses.

Knative managed responses contain no payload, they just log and report ACK or NACK to the interacting channel.

Custom managed responses on the other hand provide information through payload. In the case of successful responses the payload is provided by the target, but when an error occurs, a custom managed response includes an EventError payload.

type EventError struct {
	Code        string		// short string to identify error nature
	Description string		// description from the error object
	Fields interface{}	// additional information
}

Replier

The Replier constructor can be customized by passing an array of ReplierOption objects, being that customization applied to all responses. Customization choices are:

responseType			how to compose the outgoing response event type.
responseContentType		which content-type header to choose.
responseSource			how to compose the outgoing event source.

responseErrorType		in case of error, same as responseType.
responseErrorContentType	in case of error, same as responseContentType.

payloadPolicy			if payload replies should be sent or not.

Knative Managed Functions

Ack() (*cloudevents.Event, cloudevents.Result)

Is the function to use when we want Knative to acknoledge that the message was delivered and no reponse payload is being returned.

ErrorKnativeManaged(event *cloudevents.Event, err error) (*cloudevents.Event, cloudevents.Result)

Sends a non properly delivered message back to the channel, which will decide if it should be retried, sent to the DLQ or forgotten. A summary of the incoming event that led to the error is added to the error message returned.

Custom Managed Functions

Ok(in *cloudevents.Event, payload interface{}, opts ...EventResponseOption) (*cloudevents.Event, cloudevents.Result)

Replies acknowledging the delivery of the message and providing a payload for further use. The incoming event is passed to help composing header fields for the response. EventResponseOptions will be sequentially processed and can be used to modify the response before sending out.

Ok responses will have a "category" header set to "success".

ErrorWithFields(in *cloudevents.Event, code string, reportedErr error, details interface{}, opts ...EventResponseOption) (*cloudevents.Event, cloudevents.Result)

Replies acknowledging the delivery of the message, which means Knative will consider the message as succeeded and wont retry nor send to the DLQ. The code parameter values are suggested at the package but can be provided by the target, being recommended that cardinality is kept low to make log analysis easy. If response options for type and content-type are not explicitly set the ones for the Ok response will be used.

Error responses will have a "category" header set to "error".

Index

• Constants
• type EventError
• type EventResponseOption
  • func ResponseWithDataContentType(dct string) EventResponseOption
  • func ResponseWithID(ID string) EventResponseOption
  • func ResponseWithProcessedHeaders() EventResponseOption
  • func ResponseWithStatefulHeaders(bridge string) EventResponseOption
  • func ResponseWithSubject(subject string) EventResponseOption
• type PayloadPolicy
• type Replier
  • func New(targetName string, logger *zap.SugaredLogger, opts ...ReplierOption) (*Replier, error)
  • func (r *Replier) Ack() (*cloudevents.Event, cloudevents.Result)
  • func (r *Replier) Error(in *cloudevents.Event, code string, reportedErr error, details interface{}, ...) (*cloudevents.Event, cloudevents.Result)
  • func (r *Replier) ErrorKnativeManaged(event *cloudevents.Event, err error) (*cloudevents.Event, cloudevents.Result)
  • func (r *Replier) Ok(in *cloudevents.Event, payload interface{}, opts ...EventResponseOption) (*cloudevents.Event, cloudevents.Result)
• type ReplierOption
  • func ReplierWithMappedErrorResponseType(resTypes map[string]string) ReplierOption
  • func ReplierWithMappedResponseType(resTypes map[string]string) ReplierOption
  • func ReplierWithPayloadPolicy(policy PayloadPolicy) ReplierOption
  • func ReplierWithProcessedHeaders() ReplierOption
  • func ReplierWithStatefulHeaders(bridge string) ReplierOption
  • func ReplierWithStaticDataContentType(contentType string) ReplierOption
  • func ReplierWithStaticErrorDataContentType(contentType string) ReplierOption
  • func ReplierWithStaticErrorResponseType(resType string) ReplierOption
  • func ReplierWithStaticResponseType(resType string) ReplierOption
• type ResponseHeaderValue
  • func MappedResponseType(eventTypes map[string]string) ResponseHeaderValue
  • func StaticResponse(value string) ResponseHeaderValue
  • func SuffixResponseType(suffix string) ResponseHeaderValue

Constants

const (
	ErrorCodeEventContext      = "event-context"
	ErrorCodeRequestParsing    = "request-parsing"
	ErrorCodeRequestValidation = "request-validation"
	ErrorCodeAdapterProcess    = "adapter-process"
	ErrorCodeParseResponse     = "response-parsing"

	ErrorCodeCloudEventProcessing = "cloudevents-processing"
)

Error code common categories.

const (
	ExtensionCategory             = "category"
	ExtensionCategoryValueError   = "error"
	ExtensionCategoryValueSuccess = "success"
)

CloudEvents extension names and values.

const (
	StatefulWorkflowHeader         = "statefulbridge"
	StatefulWorkflowInstanceHeader = "statefulid"

	ProcessedTypeHeader   = "processedtype"
	ProcessedSourceHeader = "processedsource"
	ProcessedIDHeader     = "processedid"
)

Optional headers

Types

type EventError

type EventError struct {
	// Code is an identifiable moniker that classifies the error.
	Code        string
	Description string
	// Details that contain arbitrary data about the error.
	Details interface{}
}

EventError is returned as the payload when an error occurs.

type EventResponseOption

type EventResponseOption func(in, out *cloudevents.Event) error

EventResponseOption given the incoming and outgoing event, modifies the response event before being sent.

func ResponseWithDataContentType

func ResponseWithDataContentType(dct string) EventResponseOption

ResponseWithDataContentType is an option for modifying returned event content type.

func ResponseWithID

func ResponseWithID(ID string) EventResponseOption

ResponseWithID is an option for modifying returned event ID.

func ResponseWithProcessedHeaders

func ResponseWithProcessedHeaders() EventResponseOption

ResponseWithProcessedHeaders creates processed headers, propagating information about the incoming event headers into the outgoing event.

func ResponseWithStatefulHeaders

func ResponseWithStatefulHeaders(bridge string) EventResponseOption

ResponseWithStatefulHeaders creates stateful headers if not present.

func ResponseWithSubject

func ResponseWithSubject(subject string) EventResponseOption

ResponseWithSubject is an option for modifying returned event subject.

type PayloadPolicy

type PayloadPolicy string

PayloadPolicy defines when to return payloads with the response.

const (
	PayloadPolicyAlways PayloadPolicy = "always"
	PayloadPolicyErrors PayloadPolicy = "errors"
	PayloadPolicyNever  PayloadPolicy = "never"
)

Payload policy choices

type Replier

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

Replier helps normalizing CloudEvent responses.

func New

func New(targetName string, logger *zap.SugaredLogger, opts ...ReplierOption) (*Replier, error)

New returns a replier instance. Optional ReplierOption set can be returned to customize the replier behavior.

func (*Replier) Ack

func (r *Replier) Ack() (*cloudevents.Event, cloudevents.Result)

Ack when processing went ok and there is no payload as a response.

func (*Replier) Error

func (r *Replier) Error(in *cloudevents.Event, code string, reportedErr error, details interface{}, opts ...EventResponseOption) (*cloudevents.Event, cloudevents.Result)

Error replies with an error payload but dismisses the knative channel error management (retries and dead letter queues). This should be used when retrying would result in the same outcome, and when we want users to explicitly manage this error instead of relying on dead letter queue channel.

func (*Replier) ErrorKnativeManaged

func (r *Replier) ErrorKnativeManaged(event *cloudevents.Event, err error) (*cloudevents.Event, cloudevents.Result)

ErrorKnativeManaged lets the knative channel manage errors. This is intended when we rely on knative platform retries and dead letter queues.

func (*Replier) Ok

func (r *Replier) Ok(in *cloudevents.Event, payload interface{}, opts ...EventResponseOption) (*cloudevents.Event, cloudevents.Result)

Ok when returning a payload upon success.

type ReplierOption

type ReplierOption func(*Replier) error

ReplierOption modifies a newly created replier.

func ReplierWithMappedErrorResponseType

func ReplierWithMappedErrorResponseType(resTypes map[string]string) ReplierOption

ReplierWithMappedErrorResponseType option uses a map string to look up for error response type.

func ReplierWithMappedResponseType

func ReplierWithMappedResponseType(resTypes map[string]string) ReplierOption

ReplierWithMappedResponseType option uses a map string to look up response type.

func ReplierWithPayloadPolicy

func ReplierWithPayloadPolicy(policy PayloadPolicy) ReplierOption

ReplierWithPayloadPolicy option avoids returning events.

func ReplierWithProcessedHeaders

func ReplierWithProcessedHeaders() ReplierOption

ReplierWithProcessedHeaders adds response option to create processed headers.

func ReplierWithStatefulHeaders

func ReplierWithStatefulHeaders(bridge string) ReplierOption

ReplierWithStatefulHeaders adds response option to create stateful headers if not present.

func ReplierWithStaticDataContentType

func ReplierWithStaticDataContentType(contentType string) ReplierOption

ReplierWithStaticDataContentType sets the response content type for all replies.

func ReplierWithStaticErrorDataContentType

func ReplierWithStaticErrorDataContentType(contentType string) ReplierOption

ReplierWithStaticErrorDataContentType sets the response content type for error replies.

func ReplierWithStaticErrorResponseType

func ReplierWithStaticErrorResponseType(resType string) ReplierOption

ReplierWithStaticErrorResponseType option uses a static string for error response type.

func ReplierWithStaticResponseType

func ReplierWithStaticResponseType(resType string) ReplierOption

ReplierWithStaticResponseType option uses a static string for response type.

type ResponseHeaderValue

type ResponseHeaderValue func(event *cloudevents.Event) (string, error)

ResponseHeaderValue is a function that given an incoming CloudEvent returns a string to be used as a header value on the outgoing event.

func MappedResponseType

func MappedResponseType(eventTypes map[string]string) ResponseHeaderValue

MappedResponseType returns an static eventType based on the incoming eventType. When no type is mapped a default value is returned.

func StaticResponse

func StaticResponse(value string) ResponseHeaderValue

StaticResponse always returns the same fixed value. Should be used when returned types or sources do not vary depending on the incoming event data.

func SuffixResponseType

func SuffixResponseType(suffix string) ResponseHeaderValue

SuffixResponseType appends a string to the the incoming eventType.