cloudevents

package module
v1.1.1 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Sep 13, 2026 License: MIT Imports: 18 Imported by: 0

README

cloudevents

CI CodeQL Coverage Mutation Documentation Go Reference Release Go License

cloudevents is Golib's transport-independent CloudEvents interoperability envelope. It implements the stable 1.0 information model, deterministic JSON event and batch serialization, and selected HTTP and Kafka mappings. It does not replace richer domain, event-sourcing, outbox, queue, workflow, audit, or application envelopes.

This module is pre-release. The normative matrix is the authority for pinned revisions and support claims; an intended-support row is not complete until its listed evidence exists.

Quick start

data, err := cloudevents.NewJSONData([]byte(`{"order":"A-123"}`))
if err != nil {
    return err
}
event, err := cloudevents.NewEvent(cloudevents.Attributes{
    ID:              "evt-123",
    Source:          "/orders",
    Type:            "com.example.order.created.v1",
    DataContentType: "application/json",
}, data)
if err != nil {
    return err
}

wire, err := cloudevents.EncodeJSON(event)

EncodeJSON, EncodeJSONBatch, EncodeHTTP, and EncodeKafka are strict: they return ErrConversionLoss when a target representation would normalize a declared extension type, materialize metadata, or discard payload-byte distinctions. The corresponding Encode*WithReport functions return the encoded value plus a deterministic ConversionReport when the caller has an explicit policy for those changes. JSON payload bytes, including significant internal whitespace, otherwise remain exact across structured round trips.

Use DecodeJSON, DecodeJSONBatch, DecodeHTTP, or DecodeKafka with an explicit Limits value for untrusted input. DefaultLimits is conservative; applications remain responsible for choosing limits appropriate to their transport and threat model.

Ownership and I/O

  • Events, data, attributes, Kafka records, and returned bodies do not retain caller-owned mutable byte slices or maps.
  • HTTP decoders never close caller-owned bodies. Cancellation is checked before and after reading; prompt interruption requires a reader whose own Read operation observes cancellation.
  • Kafka helpers perform record mapping only. Topics, partitions, offsets, ordering, retries, acknowledgements, settlement, and broker I/O remain with the caller. DecodeKafka applies explicit key, total-header, header-name, and header-value limits before copying record metadata.
  • Schema validation is invoked only through a caller-supplied SchemaValidator. Receiving or constructing an event never resolves a schema or performs network I/O.

Formats and bindings

  • JSON event format: structured encode/decode with distinct absent, JSON null, text, empty, and binary data semantics.
  • JSON batch format: empty and non-empty batches with bounded event counts.
  • HTTP: binary, structured JSON, and JSON batch content modes.
  • Kafka: binary and structured JSON content modes. Kafka has no supported batch mode in the pinned official binding.

Queue and outbox representations are Golib transport mappings, not official CloudEvents bindings. Conversion rules and loss reporting are specified in conversion policy.

Adoption guidance

Use CloudEvents at an interoperability boundary where producers and consumers benefit from its portable context attributes and selected binding. Keep the application's canonical envelope when it owns richer invariants such as stream versions, transactional state, workflow execution, retry settlement, audit integrity, or authorization context.

Do not use this package as a broker, dispatcher, event taxonomy, compatibility policy, schema registry, audit log, or replacement for application validation.

Optional Golib integrations use target-oriented modules so consumers install only the dependency boundary they need:

Target Module
Audit metadata github.com/faustbrian/go-cloudevents/adapters/audit
Correlation identifiers github.com/faustbrian/go-cloudevents/adapters/correlation
Event sourcing github.com/faustbrian/go-cloudevents/adapters/event-sourcing
Direct JSON Schema github.com/faustbrian/go-cloudevents/adapters/jsonschema
Kafka records github.com/faustbrian/go-cloudevents/adapters/kafka
Transactional outbox github.com/faustbrian/go-cloudevents/adapters/outbox
Queue jobs github.com/faustbrian/go-cloudevents/adapters/queue
RabbitMQ Streams github.com/faustbrian/go-cloudevents/adapters/rabbitstream
Schema registry github.com/faustbrian/go-cloudevents/adapters/schema-registry
Golib telemetry github.com/faustbrian/go-cloudevents/adapters/telemetry
Tenancy github.com/faustbrian/go-cloudevents/adapters/tenancy
Workflow history github.com/faustbrian/go-cloudevents/adapters/workflow

The released adapters/golib module is a deprecated compatibility facade. It remains available throughout v1, but is excluded from the recommended set because its broad bridge owns 26 module dependencies. New code and migrations should select the target module directly. The Kafka and schema validation recipe shows the target-oriented composition without depending on the facade.

See the canonical specification decision register, interoperability overview, security policy, security and cardinality review, fixture provenance, benchmark baseline, and changelog. Interoperability evidence covers the official Go SDK and the independent JavaScript SDK; importing the package never invokes either SDK or a runtime outside Go.

For shared package families, selection guidance, construction, ownership, and lifecycle vocabulary, see the versioned Golib ecosystem index and its protocols-and-descriptions package guidance.

Documentation

Start with the documentation index for specification, conversion, interoperability, security, and benchmark guidance.

Documentation

Overview

Package cloudevents implements the transport-independent CloudEvents 1.0 information model together with explicitly selected event formats, protocol bindings, and extensions.

The package is an interoperability envelope, not an event bus, event store, queue, outbox, workflow engine, schema registry, or canonical domain-event model. Constructors and decoders own retained mutable input. The core does no network I/O, schema resolution, global registration, telemetry, or background work.

Index

Constants

View Source 
const (
	JSONMediaType      = "application/cloudevents+json"
	JSONBatchMediaType = "application/cloudevents-batch+json"
)

Variables

View Source 
var (
	// ErrSchemaRequired reports that explicit validation was requested for an
	// event without a dataschema URI.
	ErrSchemaRequired = errors.New("cloudevents: schema URI required")
	// ErrDataRequired reports that explicit validation was requested for an
	// event without data.
	ErrDataRequired = errors.New("cloudevents: event data required")
	// ErrSchemaValidatorRequired reports a nil opt-in validator.
	ErrSchemaValidatorRequired = errors.New("cloudevents: schema validator required")
	// ErrContextRequired reports a nil context passed to an explicit operation.
	ErrContextRequired = errors.New("cloudevents: context required")
)
View Source 
var ErrConversionLoss = errors.New("cloudevents: conversion loss")

ErrConversionLoss identifies an encoding that would change declared event data or metadata without an explicit loss report.

View Source 
var ErrInvalidAdapterInput = errors.New("cloudevents golib adapter: invalid input")

ErrInvalidAdapterInput classifies an absent or malformed optional adapter input. Target adapters wrap this sentinel while preserving the safe cause.

View Source 
var ErrInvalidAttribute = errors.New("cloudevents: invalid attribute")

ErrInvalidAttribute identifies a context value outside the CloudEvents type system.

View Source 
var ErrInvalidData = errors.New("cloudevents: invalid data")

ErrInvalidData identifies event data that cannot be represented by its declared runtime kind.

View Source 
var ErrInvalidEvent = errors.New("cloudevents: invalid event")

ErrInvalidEvent identifies a CloudEvent that violates the supported specification contract.

View Source 
var ErrLimitExceeded = errors.New("cloudevents: limit exceeded")

ErrLimitExceeded identifies an input rejected before an unchecked allocation or conversion.

View Source 
var ErrMetadataCollision = errors.New("cloudevents golib adapter: metadata collision")

ErrMetadataCollision reports canonical metadata that conflicts with an existing CloudEvents value and therefore cannot be overwritten.

View Source 
var ErrSchemaMapping = errors.New("cloudevents golib adapter: schema mapping")

ErrSchemaMapping reports an absent, conflicting, or unsupported explicit schema selection in an optional adapter.

View Source 
var ErrSchemaViolation = errors.New("cloudevents golib adapter: schema violation")

ErrSchemaViolation reports a valid event payload that does not satisfy the explicitly selected schema.

View Source 
var ErrUnsupportedMode = errors.New("cloudevents: unsupported content mode")

ErrUnsupportedMode identifies a content mode or event format this module does not implement.

View Source 
var ErrUntrustedMetadata = errors.New("cloudevents golib adapter: metadata is untrusted")

ErrUntrustedMetadata reports an attempt to adopt inbound identity metadata without an explicit trust decision.

Functions

func EncodeHTTP 

func EncodeHTTP(events []Event, mode ContentMode) (http.Header, []byte, error)

EncodeHTTP maps events without implicit representation loss. Use EncodeHTTPWithReport to accept and inspect target-binding changes.

func EncodeJSON 

func EncodeJSON(event Event) ([]byte, error)

EncodeJSON serializes event without implicit representation loss. Use EncodeJSONWithReport when the target JSON format cannot retain every declared in-memory distinction.

func EncodeJSONBatch 

func EncodeJSONBatch(events []Event) ([]byte, error)

EncodeJSONBatch serializes events without implicit representation loss.

func EncodeKafkaWithReport 

func EncodeKafkaWithReport(event Event, mode ContentMode, key []byte) (KafkaRecord, ConversionReport, error)

EncodeKafkaWithReport maps one Event to the stable Kafka protocol binding while reporting every representation change. Kafka does not define batch mode. The supplied key is copied unchanged.

func KafkaPartitionKey 

func KafkaPartitionKey(event Event) ([]byte, bool)

KafkaPartitionKey implements the official binding's opt-in partitionkey mapper. It does not modify the Event.

func ValidateSchema 

func ValidateSchema(ctx context.Context, event Event, validator SchemaValidator) error

ValidateSchema invokes a caller-supplied validator for an event that declares both dataschema and data. The validator receives an owned data copy.

Types

type AdapterLoss added in v1.1.0 

type AdapterLoss struct {
	Field  string
	Reason string
}

AdapterLoss describes one value that a selected adapter target cannot represent. It contains field names and reasons, never field values.

type AdapterReport added in v1.1.0 

type AdapterReport struct {
	Losses []AdapterLoss
}

AdapterReport makes every optional-adapter conversion loss explicit.

type Attribute 

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

Attribute is an immutable CloudEvents context attribute value.

func NewBinaryAttribute 

func NewBinaryAttribute(value []byte) Attribute

NewBinaryAttribute constructs a Binary context attribute and takes a copy of value.

func NewBooleanAttribute 

func NewBooleanAttribute(value bool) Attribute

NewBooleanAttribute constructs a Boolean context attribute.

func NewIntegerAttribute 

func NewIntegerAttribute(value int64) (Attribute, error)

NewIntegerAttribute constructs a 32-bit Integer context attribute.

func NewPartitionKeyAttribute 

func NewPartitionKeyAttribute(value string) (Attribute, error)

NewPartitionKeyAttribute constructs the selected partitioning extension.

func NewStringAttribute 

func NewStringAttribute(value string) (Attribute, error)

NewStringAttribute constructs a string-valued context attribute.

func NewTimestampAttribute 

func NewTimestampAttribute(value time.Time) Attribute

NewTimestampAttribute constructs an RFC 3339 Timestamp context attribute.

func NewTraceParentAttribute 

func NewTraceParentAttribute(value string) (Attribute, error)

NewTraceParentAttribute constructs the selected distributed-tracing extension's W3C traceparent value.

func NewTraceStateAttribute 

func NewTraceStateAttribute(value string) (Attribute, error)

NewTraceStateAttribute constructs the selected distributed-tracing extension's W3C tracestate value.

func NewURIAttribute 

func NewURIAttribute(value string) (Attribute, error)

NewURIAttribute constructs an absolute URI context attribute.

func NewURIReferenceAttribute 

func NewURIReferenceAttribute(value string) (Attribute, error)

NewURIReferenceAttribute constructs a URI-reference context attribute.

func (Attribute) Bytes 

func (a Attribute) Bytes() []byte

Bytes returns a copy of a Binary attribute. It returns nil for every other attribute type.

func (Attribute) Kind 

func (a Attribute) Kind() AttributeKind

Kind returns the CloudEvents abstract attribute type.

func (Attribute) String 

func (a Attribute) String() string

String returns the canonical string encoding of the attribute.

type AttributeKind 

type AttributeKind uint8

AttributeKind is a CloudEvents abstract context-attribute type. 

const (
	AttributeString AttributeKind
	AttributeBoolean
	AttributeInteger
	AttributeBinary
	AttributeURI
	AttributeURIReference
	AttributeTimestamp
)

type Attributes 

type Attributes struct {
	ID              string
	Source          string
	Type            string
	DataContentType string
	DataSchema      string
	Subject         string
	Time            *time.Time
	Extensions      map[string]Attribute
}

Attributes contains the standard and extension context attributes used to construct an Event. NewEvent takes ownership by copying mutable inputs.

type ContentMode 

type ContentMode uint8

ContentMode identifies a CloudEvents protocol-binding content mode. 

const (
	BinaryMode ContentMode
	StructuredMode
	BatchMode
)

type ConversionLoss 

type ConversionLoss struct {
	Field  string
	Reason string
}

ConversionLoss identifies one declared field whose representation changes in a target event format or protocol binding. It never contains field data.

type ConversionReport 

type ConversionReport struct {
	Losses []ConversionLoss
}

ConversionReport makes every representation change explicit. Losses are sorted by field and reason so callers can compare and persist reports.

func EncodeHTTPWithReport 

func EncodeHTTPWithReport(events []Event, mode ContentMode) (http.Header, []byte, ConversionReport, error)

EncodeHTTPWithReport maps events to HTTP headers and an owned body while reporting every representation change. Binary and structured modes require exactly one event; batch mode accepts zero or more events.

func EncodeJSONBatchWithReport 

func EncodeJSONBatchWithReport(events []Event) ([]byte, ConversionReport, error)

EncodeJSONBatchWithReport serializes events using the normative JSON batch format and reports every per-event representation change.

func EncodeJSONWithReport 

func EncodeJSONWithReport(event Event) ([]byte, ConversionReport, error)

EncodeJSONWithReport serializes event using the CloudEvents JSON event format and reports every representation change. Member names are sorted lexicographically as a package determinism policy; CloudEvents does not require that ordering.

type Data 

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

Data is an immutable CloudEvents data value.

func NewBinaryData 

func NewBinaryData(value []byte) Data

NewBinaryData constructs binary event data and takes a copy of value.

func NewJSONData 

func NewJSONData(value []byte) (Data, error)

NewJSONData constructs JSON-valued event data and takes a copy of value.

func NewTextData 

func NewTextData(value string) (Data, error)

NewTextData constructs textual event data.

func (Data) Bytes 

func (d Data) Bytes() []byte

Bytes returns a copy of the event data's wire representation.

func (Data) Kind 

func (d Data) Kind() DataKind

Kind returns the runtime data kind.

func (Data) Present 

func (d Data) Present() bool

Present reports whether the event has a data value. Present empty or null values return true.

type DataKind 

type DataKind uint8

DataKind distinguishes absent data from JSON, textual, and binary runtime values. The distinction controls normative event-format serialization. 

const (
	DataAbsent DataKind = iota
	DataJSON
	DataText
	DataBinary
)

type Event 

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

Event is an immutable CloudEvent using the stable 1.0 specification.

func DecodeJSON 

func DecodeJSON(value []byte, limits Limits) (Event, error)

DecodeJSON parses one CloudEvent in the JSON event format without performing I/O. It takes ownership of all retained input.

func DecodeJSONBatch 

func DecodeJSONBatch(value []byte, limits Limits) ([]Event, error)

DecodeJSONBatch parses the normative JSON batch format. Empty batches are valid. Every returned Event owns its storage.

func NewEvent 

func NewEvent(attributes Attributes, data Data) (Event, error)

NewEvent validates and constructs an immutable CloudEvent.

func (Event) Data 

func (e Event) Data() Data

Data returns an immutable copy of the event data.

func (Event) DataContentType 

func (e Event) DataContentType() (string, bool)

DataContentType returns the data media type and whether it is present.

func (Event) DataSchema 

func (e Event) DataSchema() (string, bool)

DataSchema returns the absolute schema URI and whether it is present.

func (Event) Extension 

func (e Event) Extension(name string) (Attribute, bool)

Extension returns an extension context attribute by name.

func (Event) Extensions 

func (e Event) Extensions() map[string]Attribute

Extensions returns an independently owned copy of all extension context attributes, including attributes unknown to this package.

func (Event) ID 

func (e Event) ID() string

ID returns the producer-scoped event identifier.

func (Event) Source 

func (e Event) Source() string

Source returns the event source URI-reference.

func (Event) SpecVersion 

func (e Event) SpecVersion() string

SpecVersion returns the stable CloudEvents specification version.

func (Event) Subject 

func (e Event) Subject() (string, bool)

Subject returns the event subject and whether it is present.

func (Event) Time 

func (e Event) Time() (time.Time, bool)

Time returns the occurrence timestamp and whether it is present.

func (Event) Type 

func (e Event) Type() string

Type returns the producer-defined event type.

func (Event) Validate 

func (e Event) Validate() error

Validate reports whether Event is assigned and satisfies the supported stable CloudEvents contract.

type HTTPMessage 

type HTTPMessage struct {
	Mode   ContentMode
	Events []Event
}

HTTPMessage is a decoded CloudEvents HTTP message. Binary and structured messages contain one Event; batch messages may contain zero or more.

func DecodeHTTP 

func DecodeHTTP(ctx context.Context, header http.Header, body io.Reader, limits Limits) (HTTPMessage, error)

DecodeHTTP maps an HTTP header and body to CloudEvents. The caller retains ownership of body; DecodeHTTP never closes it. Cancellation can interrupt cancellation-aware readers and is checked before and after the bounded read.

type Issue 

type Issue struct {
	Field string
	Code  IssueCode
}

Issue identifies one invalid field without retaining or disclosing its value.

type IssueCode 

type IssueCode string

IssueCode is a stable, value-free validation diagnostic. 

const (
	IssueRequired            IssueCode = "required"
	IssueInvalidString       IssueCode = "invalid_string"
	IssueInvalidURIReference IssueCode = "invalid_uri_reference"
	IssueAbsoluteURIRequired IssueCode = "absolute_uri_required"
	IssueInvalidMediaType    IssueCode = "invalid_media_type"
	IssueInvalidName         IssueCode = "invalid_name"
	IssueReservedName        IssueCode = "reserved_name"
	IssueInvalidAttribute    IssueCode = "invalid_attribute"
)

type KafkaHeader 

type KafkaHeader struct {
	Key   string
	Value []byte
}

KafkaHeader is a Kafka record header. Value is always caller-owned on public input and output boundaries.

type KafkaMessage 

type KafkaMessage struct {
	Mode             ContentMode
	Event            Event
	Key              []byte
	TransportHeaders []KafkaHeader
}

KafkaMessage is a decoded binding result. Transport headers are headers not owned by the CloudEvents binding and are returned without interpretation.

func DecodeKafka 

func DecodeKafka(record KafkaRecord, limits Limits) (KafkaMessage, error)

DecodeKafka decodes the stable Kafka protocol binding without broker I/O.

type KafkaRecord 

type KafkaRecord struct {
	Key     []byte
	Value   []byte
	Headers []KafkaHeader
}

KafkaRecord is the transport-neutral portion of a Kafka record owned by the CloudEvents binding. Topic, partition, offset, timestamp, retries, and broker settlement remain owned by the Kafka caller.

func EncodeKafka 

func EncodeKafka(event Event, mode ContentMode, key []byte) (KafkaRecord, error)

EncodeKafka maps one Event without implicit representation loss. Use EncodeKafkaWithReport to accept and inspect target-binding changes.

type Limits 

type Limits struct {
	MaxEventBytes            int64
	MaxDataBytes             int64
	MaxAttributes            int
	MaxAttributeNameBytes    int
	MaxAttributeValueBytes   int
	MaxDepth                 int
	MaxBatchEvents           int
	MaxKafkaKeyBytes         int
	MaxKafkaHeaders          int
	MaxKafkaHeaderNameBytes  int
	MaxKafkaHeaderValueBytes int
}

Limits bounds untrusted event formats and protocol bindings before semantic parsing. Values are byte counts unless stated otherwise. Kafka-specific limits cover the record metadata copied by DecodeKafka, including metadata not owned by the CloudEvents binding.

func DefaultLimits 

func DefaultLimits() Limits

DefaultLimits accepts the CloudEvents interoperability floor while bounding allocations for ordinary library use.

type SchemaValidator 

type SchemaValidator interface {
	Validate(ctx context.Context, uri string, contentType string, data []byte) error
}

SchemaValidator is implemented by an explicit JSON Schema or schema-registry adapter. The core package never resolves schemas or performs I/O itself.

type ValidationError 

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

ValidationError contains canonical, field-sorted validation diagnostics.

func (*ValidationError) Error 

func (e *ValidationError) Error() string

Error returns a deterministic diagnostic that never contains rejected values.

func (*ValidationError) Issues 

func (e *ValidationError) Issues() []Issue

Issues returns an owned copy of the canonical diagnostics.

func (*ValidationError) Unwrap 

func (e *ValidationError) Unwrap() error

Unwrap makes every ValidationError match ErrInvalidEvent.

Source Files

View all Source files

Directories

Path Synopsis
adapters
audit module
correlation module
event-sourcing module
golib module
jsonschema module
kafka module
outbox module
queue module
rabbitstream module
schema-registry module
telemetry module
tenancy module
workflow module
internal
adapter
Package adapter contains shared mechanics for independently released CloudEvents target adapters.
 Package adapter contains shared mechanics for independently released CloudEvents target adapters.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.