Help us shape the future of Go! Take the 2026 Go Developer Survey

azureeventhubs

package
v1.9.0 Latest Latest
Warning

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

 Go to latest Published: Oct 27, 2026 License: MIT Imports: 31 Imported by: 0

README

Azure Event Hubs

Parity grade: B · SDK azure-sdk-for-go/sdk/messaging/azeventhubs/v2@v2.0.2 · last audited 2026-10-20 (e332a0eae)

Coverage

Metric Value
PARITY entries audited 15 (8 ok, 7 partial)
Feature families 4 (3 ok, 1 partial)
Known gaps 9
Deferred items 0
Resource leaks clean
Known gaps
  • Kafka endpoint is not implemented.
  • Capture, geo-DR, schema registry, application groups, compaction and cluster features are not implemented.
  • SAS claims are not scoped to an entity; hub-scoped rules authorize any hub.
  • Hub names resolve across namespaces by first match.
  • Kafka endpoint for Event Hubs (port 9093) is not implemented.
  • Capture, geo-replication, compaction and application-group features are not implemented.
  • SAS claims are not scoped to the audience entity.
  • ARM namespace identity (SystemAssigned/UserAssigned), regenerateKeys and disasterRecoveryConfigs are not implemented.
  • No dashboard page.

More

Documentation

Overview

Package azureeventhubs emulates Azure Event Hubs: namespaces, event hubs, consumer groups and authorization rules shared with the ARM plane, an AMQP 1.0 data plane (pkgs/amqp1) for the official SDKs, and a small REST/Atom surface. The Kafka endpoint is not implemented. See PARITY.md.

Index

Constants

View Source 
const (
	DefaultNamespace      = "ehemulatorns"
	DefaultKeyName        = "RootManageSharedAccessKey"
	DefaultKeyValue       = "Kq0nF1bPjX5uV9c2yR7wZtHdLsGm3aEoQ8xT4iNvB6w="
	DefaultConsumerGroup  = "$Default"
	DefaultPartitionCount = 4
	MaxPartitionCount     = 1024
	DefaultRetention      = 24 * time.Hour
	MaxRetention          = 90 * 24 * time.Hour
	MaxEventBytes         = 1 << 20
)

Bounds and defaults.

View Source 
const (
	RightListen = "Listen"
	RightSend   = "Send"
	RightManage = "Manage"
)

Rights an authorization rule can grant.

View Source 
const (
	DefaultPort        = 10005
	DefaultAMQPPort    = 5673
	DefaultAMQPTLSPort = 5674
)

Default ports. The AMQP pair is offset by one from Service Bus's 5672/5671 so both emulators can run side by side.

View Source 
const SelectorFilterCode uint64 = 0x00000468C00000004

SelectorFilterCode is the numeric descriptor of the selector filter.

View Source 
const SelectorFilterName = "apache.org:selector-filter:string"

SelectorFilterName is the AMQP filter key Event Hubs receivers use.

Variables

View Source 
var (
	ErrNamespaceNotFound     = errors.New("azureeventhubs: namespace not found")
	ErrHubNotFound           = errors.New("azureeventhubs: event hub not found")
	ErrConsumerGroupNotFound = errors.New("azureeventhubs: consumer group not found")
	ErrRuleNotFound          = errors.New("azureeventhubs: authorization rule not found")
	ErrPartitionNotFound     = errors.New("azureeventhubs: partition not found")
	ErrInvalidPartitionCount = errors.New("azureeventhubs: partition count out of range")
	ErrPartitionCountShrink  = errors.New("azureeventhubs: partition count cannot be decreased")
	ErrInvalidRetention      = errors.New("azureeventhubs: invalid retention")
	ErrProtectedEntity       = errors.New("azureeventhubs: entity cannot be deleted")
	ErrInvalidName           = errors.New("azureeventhubs: invalid entity name")
	ErrEventTooLarge         = errors.New("azureeventhubs: event exceeds the maximum size")
	ErrReceiverStolen        = errors.New("azureeventhubs: receiver superseded by a higher epoch")
	ErrTooManyReceivers      = errors.New("azureeventhubs: too many concurrent receivers")
	ErrInvalidSelector       = errors.New("azureeventhubs: invalid selector filter")
	ErrNilAppContext         = errors.New("azureeventhubs: nil app context")
)

Backend errors.

Functions

func SignSAS 

func SignSAS(encodedResource string, expiry int64, key string) string

SignSAS computes the SAS signature for an already URL-encoded resource.

Types

type ConfigProvider 

type ConfigProvider interface {
	GetAzureEventHubsSettings() Settings
}

ConfigProvider extracts the Event Hubs settings from the AppContext config.

type Handler 

type Handler struct {
	Backend *InMemoryBackend

	AMQPCert *tls.Certificate

	Port        int
	AMQPPort    int
	AMQPTLSPort int
	ValidateSAS bool
	// contains filtered or unexported fields
}

Handler owns the HTTP listener and the AMQP servers of the Event Hubs emulator.

func NewHandler 

func NewHandler(backend *InMemoryBackend) *Handler

NewHandler returns a Handler over backend with default ports.

func (*Handler) ExtractOperation 

func (h *Handler) ExtractOperation(c *echo.Context) string

ExtractOperation implements service.Service.

func (*Handler) ExtractResource 

func (h *Handler) ExtractResource(c *echo.Context) string

ExtractResource implements service.Service.

func (*Handler) GetSupportedOperations 

func (h *Handler) GetSupportedOperations() []string

GetSupportedOperations implements service.Service.

func (*Handler) Handler 

func (h *Handler) Handler() echo.HandlerFunc

Handler returns the echo handler of the HTTP listener.

func (*Handler) MatchPriority 

func (h *Handler) MatchPriority() int

MatchPriority implements service.Registerable.

func (*Handler) Name 

func (h *Handler) Name() string

Name implements service.Service.

func (*Handler) Reset 

func (h *Handler) Reset()

Reset implements service.Resettable.

func (*Handler) Restore 

func (h *Handler) Restore(ctx context.Context, data []byte) error

Restore implements persistence.Persistable by delegating to the backend.

func (*Handler) RouteMatcher 

func (h *Handler) RouteMatcher() service.Matcher

RouteMatcher never matches: Event Hubs listens on dedicated ports.

func (*Handler) ServeAMQP 

func (h *Handler) ServeAMQP(ln net.Listener, useTLS bool) error

ServeAMQP serves AMQP 1.0 on ln (TLS-wrapped when useTLS) until Shutdown.

func (*Handler) Shutdown 

func (h *Handler) Shutdown(ctx context.Context)

Shutdown implements service.Shutdowner.

func (*Handler) Snapshot 

func (h *Handler) Snapshot(ctx context.Context) []byte

Snapshot implements persistence.Persistable by delegating to the backend.

func (*Handler) StartWorker 

func (h *Handler) StartWorker(ctx context.Context) error

StartWorker binds the HTTP listener and the AMQP listeners.

type HubConfig 

type HubConfig struct {
	PartitionCount int           `json:"partitionCount"`
	Retention      time.Duration `json:"retention"`
}

HubConfig is the configurable part of an event hub.

type HubInfo 

type HubInfo struct {
	CreatedAt    time.Time
	Name         string
	Namespace    string
	PartitionIDs []string
	Config       HubConfig
}

HubInfo is the exported view of an event hub.

type HubRef 

type HubRef struct {
	Namespace string
	Hub       string
}

HubRef addresses an event hub within a namespace.

type InMemoryBackend 

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

InMemoryBackend is the Event Hubs state shared by the ARM, REST and AMQP planes.

func NewInMemoryBackend 

func NewInMemoryBackend() *InMemoryBackend

NewInMemoryBackend returns an empty backend using the wall clock.

func (*InMemoryBackend) Append 

func (b *InMemoryBackend) Append(ref HubRef, pid int, evs []NewEvent) (int, error)

Append stores events in one partition: pid, else the hash of the first event's key, else round-robin. It returns the partition used.

func (*InMemoryBackend) ConsumerGroup 

func (b *InMemoryBackend) ConsumerGroup(ref HubRef, name string) (time.Time, error)

ConsumerGroup returns a consumer group's creation time.

func (*InMemoryBackend) CreateConsumerGroup 

func (b *InMemoryBackend) CreateConsumerGroup(ref HubRef, name string) (time.Time, error)

CreateConsumerGroup creates a consumer group (idempotent).

func (*InMemoryBackend) CreateHub 

func (b *InMemoryBackend) CreateHub(ref HubRef, cfg HubConfig) (HubInfo, error)

CreateHub creates or updates an event hub; the partition count may only grow.

func (*InMemoryBackend) DeleteConsumerGroup 

func (b *InMemoryBackend) DeleteConsumerGroup(ref HubRef, name string) error

DeleteConsumerGroup removes a consumer group; $Default is protected.

func (*InMemoryBackend) DeleteHub 

func (b *InMemoryBackend) DeleteHub(ref HubRef) error

DeleteHub removes an event hub and disconnects its receivers.

func (*InMemoryBackend) DeleteNamespace 

func (b *InMemoryBackend) DeleteNamespace(name string) error

DeleteNamespace removes a namespace and everything in it.

func (*InMemoryBackend) DeleteRule 

func (b *InMemoryBackend) DeleteRule(nsName, hubName, name string) error

DeleteRule removes an authorization rule; the namespace root rule is protected.

func (*InMemoryBackend) EnsureNamespace 

func (b *InMemoryBackend) EnsureNamespace(name string) error

EnsureNamespace creates the namespace (with its root rule) if absent.

func (*InMemoryBackend) GetRule 

func (b *InMemoryBackend) GetRule(nsName, hubName, name string) (Rule, error)

GetRule returns an authorization rule including its keys.

func (*InMemoryBackend) Hub 

func (b *InMemoryBackend) Hub(ref HubRef) (HubInfo, error)

Hub returns an event hub's description.

func (*InMemoryBackend) HubCount 

func (b *InMemoryBackend) HubCount() int

HubCount returns the number of hubs across namespaces.

func (*InMemoryBackend) ListConsumerGroups 

func (b *InMemoryBackend) ListConsumerGroups(ref HubRef) ([]string, error)

ListConsumerGroups returns the hub's consumer group names, sorted.

func (*InMemoryBackend) ListHubs 

func (b *InMemoryBackend) ListHubs(nsName string) ([]HubInfo, error)

ListHubs returns the namespace's event hubs sorted by name.

func (*InMemoryBackend) NamespaceExists 

func (b *InMemoryBackend) NamespaceExists(name string) bool

NamespaceExists reports whether the namespace exists.

func (*InMemoryBackend) NamespaceNames 

func (b *InMemoryBackend) NamespaceNames() []string

NamespaceNames returns the namespace names sorted.

func (*InMemoryBackend) Next 

func (b *InMemoryBackend) Next(
	ctx context.Context, ref HubRef, pid int, cursor int64, stop <-chan struct{},
) (*event, error)

Next blocks until the partition holds an event at or after cursor, ctx ends, or stop closes (ErrReceiverStolen is reported through the caller).

func (*InMemoryBackend) PartitionInfo 

func (b *InMemoryBackend) PartitionInfo(ref HubRef, pid int) (PartitionInfo, error)

PartitionInfo returns a partition's runtime state after retention.

func (*InMemoryBackend) PlainAuthorized 

func (b *InMemoryBackend) PlainAuthorized(user, password string) bool

PlainAuthorized reports whether a SASL PLAIN user/password pair is a rule key.

func (*InMemoryBackend) PutRule 

func (b *InMemoryBackend) PutRule(nsName, hubName, name string, rights []string) (Rule, error)

PutRule creates or updates an authorization rule scoped to the namespace or, when hubName is set, to one event hub; keys survive updates.

func (*InMemoryBackend) RegisterReceiver 

func (b *InMemoryBackend) RegisterReceiver(
	ref HubRef,
	cg string,
	pid int,
	name string,
	epoch *int64,
) (*ReceiverReg, error)

RegisterReceiver admits a receiver to a partition of a consumer group, applying epoch (owner level) arbitration.

func (*InMemoryBackend) Reset 

func (b *InMemoryBackend) Reset()

Reset drops all state and disconnects every receiver.

func (*InMemoryBackend) Resolve 

func (b *InMemoryBackend) Resolve(ref HubRef, pid int, sel Selector) (int64, error)

Resolve turns a selector into the first sequence number a receiver should get.

func (*InMemoryBackend) ResolveHub 

func (b *InMemoryBackend) ResolveHub(name string) (HubRef, bool)

ResolveHub finds the namespace owning a hub of the given name; AMQP addresses carry no namespace, so the first namespace by name wins.

func (*InMemoryBackend) Restore 

func (b *InMemoryBackend) Restore(ctx context.Context, data []byte) error

Restore implements persistence.Persistable.

func (*InMemoryBackend) Snapshot 

func (b *InMemoryBackend) Snapshot(ctx context.Context) []byte

Snapshot implements persistence.Persistable.

func (*InMemoryBackend) Sweep 

func (b *InMemoryBackend) Sweep(now time.Time) int

Sweep applies retention to every hub; it returns the number of hubs visited.

func (*InMemoryBackend) UnregisterReceiver 

func (b *InMemoryBackend) UnregisterReceiver(reg *ReceiverReg)

UnregisterReceiver removes a receiver registration.

type NewEvent 

type NewEvent struct {
	Key string
	Raw []byte
}

NewEvent is one event to append: the encoded AMQP message minus the broker-owned annotations, and its partition key.

type PartitionInfo 

type PartitionInfo struct {
	LastEnqueuedTime time.Time
	Hub              string
	Partition        string
	BeginSequence    int64
	LastSequence     int64
	LastOffset       int64
	Empty            bool
}

PartitionInfo is the runtime state of one partition.

type Provider 

type Provider struct{}

Provider implements service.Provider for Azure Event Hubs.

func (*Provider) Init 

func (p *Provider) Init(ctx *service.AppContext) (service.Registerable, error)

Init builds the backend and handler; ports are bound by Handler.StartWorker.

func (*Provider) Name 

func (p *Provider) Name() string

Name returns the service provider name.

type ReceiverReg 

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

ReceiverReg is a partition receiver attached to a consumer group; kicked closes when a newer epoch takes over.

func (*ReceiverReg) Kicked 

func (r *ReceiverReg) Kicked() <-chan struct{}

Kicked is closed when a higher-epoch receiver supersedes this one.

func (*ReceiverReg) Reason 

func (r *ReceiverReg) Reason() string

Reason describes why the receiver was disconnected; valid once Kicked is closed.

type Rule 

type Rule struct {
	Name         string
	PrimaryKey   string
	SecondaryKey string
	Rights       []string
}

Rule is the exported view of an authorization rule.

type Selector 

type Selector struct {
	Value     int64
	Kind      selectorKind
	Latest    bool
	Inclusive bool
}

Selector is a parsed start position (offset, sequence number or enqueued time).

func ParseSelector 

func ParseSelector(expr string) (Selector, error)

ParseSelector parses an `amqp.annotation.x-opt-* > 'v'` filter expression.

type Settings 

type Settings struct {
	Port int `` //nolint:lll // config struct tags are intentionally verbose
	/* 194-byte string literal not displayed */
	AMQPPort int `` //nolint:lll // config struct tags are intentionally verbose
	/* 166-byte string literal not displayed */
	AMQPTLSPort int `` //nolint:lll // config struct tags are intentionally verbose
	/* 173-byte string literal not displayed */
	ValidateSAS bool `` //nolint:lll // config struct tags are intentionally verbose
	/* 189-byte string literal not displayed */
}

Settings is the Kong-embedded configuration of the Event Hubs emulator.

func DefaultSettings 

func DefaultSettings() Settings

DefaultSettings returns the settings used when no ConfigProvider exists.

Source Files

View all Source files

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.