mercure

package module
v1.1.2 Latest Latest
Warning

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

 Go to latest Published: Oct 11, 2026 License: AGPL-3.0 Imports: 49 Imported by: 7

README

Protocol and Reference Implementation

Mercure is a protocol for pushing data updates to web browsers and other HTTP clients in a convenient, fast, reliable, and battery-efficient way. It is especially useful to publish async and real-time updates of resources served through web APIs, to reactive web and mobile apps.

Awesome Artifact HUB PkgGoDev CI Coverage Status

Subscriptions Schema

The protocol is maintained in this repository and is also available as an Internet-Draft.

A reference, production-grade, implementation of a Mercure hub (the server) is also available in this repository. It's free software (AGPL) written in Go. It is provided along with a library that can be used in any Go application to implement the Mercure protocol directly (without a hub) and an official Docker image.

In addition, a managed and high-scalability version of the Mercure.rocks hub is available on Mercure.rocks.

Contributing

See CONTRIBUTING.md.

See license information.

Credits

Created by Kévin Dunglas. Graphic design by Laury Sorriaux. Sponsored by Les-Tilleuls.coop.

Documentation

Overview

Package mercure helps implement the Mercure protocol (https://mercure.rocks) in Go projects. It provides an implementation of a Mercure hub as an HTTP handler.

Example 
package main

import (
	"context"
	"log"
	"net/http"

	"github.com/dunglas/mercure"
)

func main() {
	ctx := context.Background()

	h, err := mercure.NewHub(
		ctx,
		mercure.WithIssuers([]mercure.Issuer{{
			Identifier: "https://example.com",
			Publisher:  mercure.Static{Key: []byte("!ChangeMe!"), Algorithm: "HS256"},
			Subscriber: mercure.Static{Key: []byte("!ChangeMe!"), Algorithm: "HS256"},
		}}),
		mercure.WithResourceIdentifier("https://example.com/.well-known/mercure"),
	)
	if err != nil {
		log.Fatal(err)
	}

	defer func() {
		if err := h.Stop(ctx); err != nil {
			panic(err)
		}
	}()

	http.Handle("/.well-known/mercure", h)
	log.Panic(http.ListenAndServe(":8080", nil))
}

Index

Examples

Constants

View Source 
const (
	DefaultWriteTimeout    = 600 * time.Second
	DefaultDispatchTimeout = 5 * time.Second
	DefaultHeartbeat       = 40 * time.Second

	// DefaultMaxRequestBodySize bounds the publish and QUERY subscribe request
	// bodies; larger requests are rejected with a 413 as the protocol requires.
	DefaultMaxRequestBodySize int64 = 1 << 20 // 1 MiB
)
View Source 
const BoltDefaultCleanupFrequency = 0.3
View Source 
const DefaultSubscriberListCacheSize = 100_000

DefaultSubscriberListCacheSize is the default size of the skipfilter cache.

Keys are fixed-size digests, so a full cache of 100,000 filters with no subscribers uses about 25MB whatever the size of the update topics.

View Source 
const DefaultTopicMatcherStoreCacheSize = 100_000

DefaultTopicMatcherStoreCacheSize bounds the (matcher_type, pattern, topics) -> bool match cache and each compiled-pattern cache. At ~100 B/entry, 100_000 keeps each cache under ~10 MB. Raise it via the `topic_matcher_cache <N>` Caddyfile directive for hubs handling a much larger topic / matcher universe.

View Source 
const EarliestLastEventID = "earliest"

EarliestLastEventID is the reserved value representing the earliest available event id.

View Source 
const PlaygroundURLPrefix = "/playground/"

PlaygroundURLPrefix is the root path the insecure playground echo endpoints are served under (only when the playground is enabled). It sits outside the reserved /.well-known/mercure namespace, so the resources it serves are valid, subscribable Mercure topics; nesting them under the hub path would make them reserved.

INSECURE + EXPERIMENTAL. Not covered by the backward compatibility promise.

View Source 
const ProtectedResourceMetadataPath = protectedResourceMetadataPath

ProtectedResourceMetadataPath is the RFC 9728 well-known path the hub serves its protected resource metadata at. It is exported so an embedding server (for example, the Caddy module) can route to it without re-deriving the path.

Variables

View Source 
var (
	// ErrInvalidAuthorizationHeader is returned when the Authorization header is invalid.
	ErrInvalidAuthorizationHeader = errors.New(`invalid "Authorization" HTTP header`)
	// ErrNoOrigin is returned when the cookie authorization mechanism is used and no Origin nor Referer headers are presents.
	ErrNoOrigin = errors.New(`an "Origin" or a "Referer" HTTP header must be present to use the cookie-based authorization mechanism`)
	// ErrOriginNotAllowed is returned when the Origin is not allowed to post updates.
	ErrOriginNotAllowed = errors.New("origin not allowed to post updates")
	// ErrInvalidJWT is returned when the access token is invalid.
	ErrInvalidJWT = errors.New("invalid JWT")
)
View Source 
var (
	ErrReservedTopic     = errors.New(`topic value resolves into the reserved "/.well-known/mercure" namespace`)
	ErrReservedWildcard  = errors.New(`topic value "*" is reserved for the wildcard matcher and cannot be published`)
	ErrInvalidEventID    = errors.New(`"id" field is too long, contains a forbidden control character or invalid UTF-8, starts with "#", or is the reserved value "earliest"`)
	ErrInvalidEventType  = errors.New(`"type" field contains a forbidden control character or invalid UTF-8`)
	ErrReservedEventType = errors.New(`"type" field uses the reserved value "mercure"`)
	ErrInvalidTopic      = errors.New("topic is too long, or contains a forbidden control character or invalid UTF-8")
	ErrTooManyTopics     = errors.New("too many topics in update")
	ErrMissingTopic      = errors.New("update carries no topic")
	ErrInvalidData       = errors.New(`"data" field is not valid UTF-8`)
)

Sentinel errors returned by Publish. Callers can branch on them via errors.Is.

View Source 
var ErrClosedTransport = errors.New("hub: read/write on closed Transport")

ErrClosedTransport is returned by the Transport's Dispatch and AddSubscriber methods after a call to Close.

View Source 
var ErrConflictingBaseURL = errors.New("topic matcher store already configured with a different base URL")

ErrConflictingBaseURL is returned by setBaseURL (via NewHub) when a store already configured with one base URL is reused by a hub with a different public URL. The base URL is immutable configuration; sharing a store across hubs that disagree on it would silently corrupt relative-pattern matching, so it is rejected at construction instead.

View Source 
var ErrDuplicateIssuer = errors.New("duplicate issuer identifier")

ErrDuplicateIssuer is returned when the same issuer identifier is configured more than once.

View Source 
var ErrInvalidBaseURL = errors.New("base URL must be an absolute URL")

ErrInvalidBaseURL is returned by setBaseURL (via NewHub) when the configured public URL is not an absolute URL. Relative URL patterns and topics are resolved against it, so an invalid value would make every relative-pattern match fail at request time with an opaque error; it is rejected at construction instead.

View Source 
var ErrInvalidPublicURL = errors.New("a public URL must be an absolute URL with a scheme and a host")

ErrInvalidPublicURL is returned when a public URL in the allowlist is not an absolute URL with a scheme and a host.

View Source 
var ErrInvalidResourceIdentifier = errors.New("the resource identifier must be an absolute URL without a fragment (RFC 9728)")

ErrInvalidResourceIdentifier is returned when the configured resource identifier is not an RFC 9728 protected resource identifier: an absolute URL without a fragment component.

View Source 
var ErrInvalidSubscriptionLimits = errors.New("subscription limits must be non-negative")

ErrInvalidSubscriptionLimits rejects negative subscription limits.

View Source 
var ErrIssuerMissingKey = errors.New("an issuer must configure a publisher or subscriber verifier")

ErrIssuerMissingKey is returned when an issuer has no verifier for either role, so no token could ever be verified for it.

View Source 
var ErrMissingAlgorithm = errors.New("a Static verifier requires a signing algorithm")

ErrMissingAlgorithm is returned when a Static verifier is configured without a signing algorithm.

View Source 
var ErrMissingIssuerIdentifier = errors.New("an issuer identifier is required in modern mode")

ErrMissingIssuerIdentifier is returned when an issuer is configured in modern mode with an empty identifier: the token iss claim is matched against it, so it must be set.

View Source 
var ErrMissingKey = errors.New("a Static verifier requires a key")

ErrMissingKey is returned when a Static verifier has no key material: an HMAC algorithm accepts a zero-length secret, so every token would verify.

View Source 
var ErrNegativeDrainTimeout = errors.New("the drain timeout must not be negative")

ErrNegativeDrainTimeout is returned when the drain timeout is negative.

View Source 
var ErrPEMKeyHMACAlgorithm = errors.New("a PEM-encoded key must not be used with an HMAC algorithm")

ErrPEMKeyHMACAlgorithm is returned when a Static verifier pairs a PEM key with an HMAC algorithm, which would make the public key the shared secret.

View Source 
var ErrSpanningPublishOrigin = errors.New(`a wildcard publish origin must stay within one registrable domain, such as "https://*.example.com"`)

ErrSpanningPublishOrigin is returned when a wildcard publish origin matches origins outside one registrable domain, which would let unrelated sites publish with the victim's cookie.

View Source 
var ErrUnexpectedSigningMethod = errors.New("unexpected signing method")

ErrUnexpectedSigningMethod is returned when the signing JWT method is not supported.

View Source 
var ErrUnsupportedMatcherType = errors.New("unsupported topic matcher type")

ErrUnsupportedMatcherType is returned when a matcher type is not one of the types defined by the protocol. The HTTP handlers map it to a 400 status code.

View Source 
var ErrUnsupportedProtocolVersion = errors.New("compatibility mode only supports protocol versions 7 and 8")

ErrUnsupportedProtocolVersion is returned when the version passed is unsupported.

View Source 
var SubscriberContextKey subscriberContextKeyType //nolint:gochecknoglobals
View Source 
var UpdateContextKey updateContextKeyType //nolint:gochecknoglobals

Functions

func NewRequestOriginContext added in v1.0.0 

func NewRequestOriginContext(ctx context.Context, scheme, host string) context.Context

NewRequestOriginContext returns a copy of ctx carrying the resolved request origin. An embedding server (for example the Caddy module) calls it once per request before invoking the hub, passing scheme and host taken from a trusted source; the hub validates them against the public-URL allowlist. The Caddy module reads them from Caddy's request placeholders, which report the request's own Host and TLS state, overridden by the X-Forwarded-Proto and X-Forwarded-Host headers of requests from trusted_proxies only.

func NewSlogHandler added in v0.21.0 

func NewSlogHandler(handler slog.Handler) slog.Handler

INTERNAL: NewSlogHandler returns a log/slog.Handler that automatically appends "update" and "subscriber" context, if applicable.

This function will be removed when https://github.com/caddyserver/caddy/pull/7346 will be available.

Types

type BoltTransport 

type BoltTransport struct {
	sync.RWMutex
	// contains filtered or unexported fields
}

BoltTransport implements the TransportInterface using the Bolt database.

func NewBoltTransport 

func NewBoltTransport(
	subscriberList *SubscriberList,
	logger *slog.Logger,
	path string,
	bucketName string,
	size uint64,
	cleanupFrequency float64,
) (*BoltTransport, error)

NewBoltTransport creates a new BoltTransport.

func (*BoltTransport) AddSubscriber 

func (t *BoltTransport) AddSubscriber(ctx context.Context, s *LocalSubscriber) error

AddSubscriber adds a new subscriber to the transport.

func (*BoltTransport) Close 

func (t *BoltTransport) Close(_ context.Context) (err error)

Close closes the Transport.

func (*BoltTransport) Dispatch 

func (t *BoltTransport) Dispatch(ctx context.Context, update *Update) error

Dispatch dispatches an update to all subscribers and persists it in Bolt DB.

func (*BoltTransport) GetSubscribers 

func (t *BoltTransport) GetSubscribers(_ context.Context) (string, []*Subscriber, error)

GetSubscribers get the list of active subscribers.

func (*BoltTransport) RemoveSubscriber added in v0.14.0 

func (t *BoltTransport) RemoveSubscriber(_ context.Context, s *LocalSubscriber) error

RemoveSubscriber removes a new subscriber from the transport.

type Event 

type Event struct {
	// The updates' data, encoded in the sever-sent event format: every line starts with the string "data: "
	// https://www.w3.org/TR/eventsource/#dispatchMessage
	Data string

	// The globally unique identifier corresponding to update
	ID string

	// The event type, will be attached to the "event" field
	Type string

	// The reconnection time
	Retry uint64
}

Event is the actual Server Sent Event that will be dispatched.

func (*Event) String 

func (e *Event) String() string

String serializes the event in a "text/event-stream" representation.

type Hub 

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

Hub stores channels with clients currently subscribed and allows to dispatch updates.

func NewHub 

func NewHub(ctx context.Context, options ...Option) (*Hub, error)

NewHub creates a new Hub instance.

func (*Hub) DebuggerConfigHandler added in v1.0.0 

func (h *Hub) DebuggerConfigHandler(w http.ResponseWriter, r *http.Request)

DebuggerConfigHandler serves the bundled UI's runtime configuration as JSON. It exposes only hub state the UI needs to adapt itself, never a token. Registered whenever the debugger is enabled (see WithDebugger).

func (*Hub) Drain added in v1.1.0 

func (h *Hub) Drain()

Drain closes subscribers at random times within the drain timeout; call it on termination only, never on reload.

func (*Hub) Playground added in v1.0.0 

func (h *Hub) Playground(w http.ResponseWriter, r *http.Request)

Playground exposes INSECURE endpoints to test discovery and authorization mechanisms. Add a query parameter named "body" to define the content to return in the response's body. Add a query parameter named "jwt" to set the authorization cookie (see WithCookieName) containing this token. Responses use JSON or JSON-LD for .json or .jsonld paths, and plain text otherwise.

func (*Hub) PlaygroundTokenHandler added in v1.0.0 

func (h *Hub) PlaygroundTokenHandler(w http.ResponseWriter, r *http.Request)

PlaygroundTokenHandler returns a freshly minted access token for the debugger UI to prefill, with the aud claim set to the resource identifier derived for this request (so it is valid on whatever public URL the playground answers on).

INSECURE: the token grants publish and subscribe on every topic; it exists only to make the playground self-serve. Registered only when the playground is enabled and a signing callback is configured (see WithPlayground, WithPlaygroundTokenFunc).

EXPERIMENTAL. Not covered by the backward compatibility promise.

func (*Hub) ProtectedResourceMetadataHandler added in v1.0.0 

func (h *Hub) ProtectedResourceMetadataHandler(w http.ResponseWriter, r *http.Request)

ProtectedResourceMetadataHandler serves the hub's RFC 9728 protected resource metadata document.

func (*Hub) Publish added in v0.21.0 

func (h *Hub) Publish(ctx context.Context, update *Update) error

Publish broadcasts the given update to all subscribers. Publish takes ownership of update: transports may deliver it asynchronously, so callers must not modify it, including its Topics slice, after the call. The transport may set its ID.

func (*Hub) PublishHandler 

func (h *Hub) PublishHandler(w http.ResponseWriter, r *http.Request)

PublishHandler allows publisher to broadcast updates to all subscribers.

func (*Hub) ServeHTTP 

func (h *Hub) ServeHTTP(w http.ResponseWriter, r *http.Request)

func (*Hub) Stop added in v0.11.1 

func (h *Hub) Stop(ctx context.Context) error

Stop stops the hub.

func (*Hub) SubscribeHandler 

func (h *Hub) SubscribeHandler(w http.ResponseWriter, r *http.Request)

SubscribeHandler creates a keep alive connection and sends the events to the subscribers.

func (*Hub) SubscriptionHandler 

func (h *Hub) SubscriptionHandler(w http.ResponseWriter, r *http.Request)

func (*Hub) SubscriptionsHandler 

func (h *Hub) SubscriptionsHandler(w http.ResponseWriter, r *http.Request)

type Issuer added in v1.0.0 

type Issuer struct {
	// Identifier is the exact value of the token iss claim.
	Identifier string
	// AuthorizationServer advertises Identifier in the hub's RFC 9728 protected
	// resource metadata. Leave false for self-issued tokens (a key shared out
	// of band, no authorization server).
	AuthorizationServer bool
	// Publisher and Subscriber verify tokens for each role. A nil Verifier
	// means the role is not accepted for this issuer.
	Publisher  Verifier
	Subscriber Verifier
}

Issuer binds a trusted issuer (RFC 9068 §4) to its per-role verification material. Configure issuers with WithIssuers.

type KeyFunc added in v1.0.0 

type KeyFunc struct {
	// Keyfunc supplies the verification key(s) for each token.
	Keyfunc jwt.Keyfunc
	// Algorithms pins the accepted JWS algorithms (RFC 8725): the algorithm is
	// never taken from the token header. Defaults to the asymmetric allowlist
	// when empty, so a public JWK can never be reinterpreted as an HMAC secret.
	Algorithms []string
}

KeyFunc verifies tokens with a caller-supplied keyfunc, typically backed by a JWK Set.

type LocalSubscriber added in v0.17.0 

type LocalSubscriber struct {
	Subscriber
	// contains filtered or unexported fields
}

LocalSubscriber represents a client subscribed to a list of topics on the current hub.

func NewLocalSubscriber added in v0.17.0 

func NewLocalSubscriber(lastEventID string, logger *slog.Logger, topicMatcherStore *TopicMatcherStore) *LocalSubscriber

NewLocalSubscriber creates a new subscriber.

func (*LocalSubscriber) Disconnect added in v0.17.0 

func (s *LocalSubscriber) Disconnect()

Disconnect disconnects the subscriber.

func (*LocalSubscriber) Dispatch added in v0.17.0 

func (s *LocalSubscriber) Dispatch(ctx context.Context, u *Update, fromHistory bool) bool

Dispatch an update to the subscriber. Security checks must (topics matching) be done before calling Dispatch, for instance by calling Match.

func (*LocalSubscriber) HistoryDispatched added in v0.17.0 

func (s *LocalSubscriber) HistoryDispatched(responseLastEventID string)

HistoryDispatched must be called when all messages coming from the history have been dispatched.

func (*LocalSubscriber) Ready added in v0.17.0 

func (s *LocalSubscriber) Ready(ctx context.Context) (n int)

Ready flips the ready flag to true and flushes queued live updates returning number of events flushed.

func (*LocalSubscriber) Receive added in v0.17.0 

func (s *LocalSubscriber) Receive() <-chan *Update

Receive returns a chan when incoming updates are dispatched.

type LocalTransport 

type LocalTransport struct {
	sync.RWMutex
	// contains filtered or unexported fields
}

LocalTransport implements the TransportInterface without database and simply broadcast the live Updates.

func NewLocalTransport 

func NewLocalTransport(sl *SubscriberList) *LocalTransport

NewLocalTransport creates a new LocalTransport.

func (*LocalTransport) AddSubscriber 

func (t *LocalTransport) AddSubscriber(ctx context.Context, s *LocalSubscriber) error

AddSubscriber adds a new subscriber to the transport.

func (*LocalTransport) Close 

func (t *LocalTransport) Close(_ context.Context) (err error)

Close closes the Transport.

func (*LocalTransport) Dispatch 

func (t *LocalTransport) Dispatch(ctx context.Context, update *Update) error

Dispatch dispatches an update to all subscribers.

func (*LocalTransport) GetSubscribers 

func (t *LocalTransport) GetSubscribers(_ context.Context) (string, []*Subscriber, error)

GetSubscribers gets the list of active subscribers.

func (*LocalTransport) RemoveSubscriber added in v0.14.0 

func (t *LocalTransport) RemoveSubscriber(_ context.Context, s *LocalSubscriber) error

RemoveSubscriber removes a subscriber from the transport.

type MatcherType added in v1.0.0 

type MatcherType string

MatcherType identifies how a topic matcher pattern is evaluated. The constant values are canonical: they double as the wire format in subscription API URLs, subscription event payloads and JWT claims. 

const (
	// MatcherTypeExact selects byte-for-byte topic comparison. The reserved
	// pattern "*" matches every topic.
	MatcherTypeExact MatcherType = "exact"

	// MatcherTypeURLPattern selects matching per the WHATWG URL Pattern
	// Living Standard, with the hub's public URL as the base URL.
	MatcherTypeURLPattern MatcherType = "urlpattern"
)

type Metrics 

type Metrics interface {
	// SubscriberConnected collects metrics about subscriber connections.
	SubscriberConnected(s *LocalSubscriber)
	// SubscriberDisconnected collects metrics about subscriber disconnections.
	SubscriberDisconnected(s *LocalSubscriber)
	// UpdatePublished collects metrics about update publications.
	UpdatePublished(u *Update)
}

type NopMetrics 

type NopMetrics struct{}

func (NopMetrics) SubscriberConnected 

func (NopMetrics) SubscriberConnected(_ *LocalSubscriber)

func (NopMetrics) SubscriberDisconnected 

func (NopMetrics) SubscriberDisconnected(_ *LocalSubscriber)

func (NopMetrics) UpdatePublished 

func (NopMetrics) UpdatePublished(_ *Update)

type Option 

type Option func(o *opt) error

Option instances allow to configure the library.

func WithAnonymous 

func WithAnonymous() Option

WithAnonymous allows subscribers with no valid JWT.

func WithCORSOrigins 

func WithCORSOrigins(origins []string) Option

WithCORSOrigins sets the allowed CORS origins.

func WithClientIPFunc added in v1.1.0 

func WithClientIPFunc(f func(*http.Request) string) Option

WithClientIPFunc sets how the hub identifies a client in logs and per-client subscription limits, such as a client IP resolved from trusted proxy headers. An empty result falls back to RemoteAddr.

func WithCookieName added in v0.14.0 

func WithCookieName(cookieName string) Option

WithCookieName sets the name of the authorization cookie (defaults to "__Secure-mercure_access_token"). The default "__Secure-" prefix makes user agents refuse the cookie over insecure transport; plain-HTTP deployments (local development) must configure a prefix-less name.

func WithDebug 

func WithDebug() Option

WithDebug enables the debug mode.

func WithDebugger added in v1.0.0 

func WithDebugger() Option

func WithDispatchTimeout 

func WithDispatchTimeout(timeout time.Duration) Option

WithDispatchTimeout sets maximum dispatch duration of an update.

func WithDrainTimeout added in v1.1.0 

func WithDrainTimeout(timeout time.Duration) Option

WithDrainTimeout sets the maximum duration of the drain started by Drain, defaults to 0 (disabled).

func WithHeartbeat 

func WithHeartbeat(interval time.Duration) Option

WithHeartbeat sets the frequency of the SSE keep-alive comments, defaults to 40s, set to 0 to disable.

func WithIssuers added in v1.0.0 

func WithIssuers(issuers []Issuer) Option

WithIssuers configures access-token verification, binding each trusted issuer (RFC 9068 §4) to its own verification material. It replaces the former per-role options: a token is verified only with the key(s) associated with its iss claim, so key material is never pooled across issuers.

Each Issuer provides a Publisher and/or Subscriber Verifier (a nil Verifier means that role is not accepted for the issuer). Setting AuthorizationServer advertises the issuer in the hub's RFC 9728 protected resource metadata; a self-issued issuer (a key shared out of band) leaves it false.

func WithLogger 

func WithLogger(logger *slog.Logger) Option

WithLogger sets the logger to use.

func WithMaxRequestBodySize added in v1.0.0 

func WithMaxRequestBodySize(size int64) Option

WithMaxRequestBodySize bounds the size, in bytes, of publish and QUERY subscribe request bodies; larger requests are rejected with a 413 status code. Defaults to DefaultMaxRequestBodySize, set to 0 to disable the in-hub limit (for example, when a reverse proxy already enforces one).

func WithMetrics 

func WithMetrics(m Metrics) Option

WithMetrics enables collection of Prometheus metrics.

func WithPlayground added in v1.0.0 

func WithPlayground() Option

WithPlayground enables the playground.

func WithPlaygroundTokenFunc added in v1.0.0 

func WithPlaygroundTokenFunc(f func(resourceIdentifier string) (string, error)) Option

WithPlaygroundTokenFunc sets the callback the playground UI uses to prefill an access token, exposed at .well-known/mercure/debug/playground-token. The callback receives the request's resource identifier (the token's aud claim, derived per request so the token is valid on whatever public URL the playground answers on). It has no effect unless the playground is also enabled (see WithPlayground).

INSECURE: it hands every visitor an all-access token (publish and subscribe on every topic) signed with the hub's key, so only ever configure it on a throwaway playground hub. The Caddy module wires it automatically for a `playground` hub that has a symmetric signing key.

EXPERIMENTAL. Not covered by the backward compatibility promise.

func WithProtocolVersionCompatibility added in v0.14.0 

func WithProtocolVersionCompatibility(protocolVersionCompatibility int) Option

WithProtocolVersionCompatibility sets the version of the Mercure protocol to be backward compatible with (versions 7 and 8 are supported). The v8 behaviors (URI Template selectors in the `topic` parameter, bare-string JWT claims, alternate topics, v8 subscription routes) additionally require a hub binary built with the deprecated_topic tag.

func WithPublicURLs added in v1.0.0 

func WithPublicURLs(urls []string) Option

WithPublicURLs restricts the hub to the given public URLs, pinning their scheme as well as their host. A request whose derived origin (scheme + host) is not one of them is rejected with 421 Misdirected Request, and the hub never derives an identity from an unlisted origin. This is the safety gate for a catch-all site block, where the fronting server does not already constrain the Host. Each value is an absolute URL; only its scheme and host (with optional port) are significant.

func WithPublishOrigins 

func WithPublishOrigins(origins []string) Option

WithPublishOrigins sets the origins allowed to publish updates.

func WithResourceIdentifier added in v1.0.0 

func WithResourceIdentifier(resourceIdentifier string) Option

WithResourceIdentifier pins the hub's OAuth 2.0 resource identifier (RFC 9068 `aud` value, advertised through RFC 9728 protected resource metadata) to a single static value. When unset, the hub derives the identifier from each request (the public URL the client contacted), so a hub reachable through several public URLs needs no configuration; set this only to force one canonical audience shared across every URL. A value ending in "/.well-known/mercure" also becomes the base URL for matching relative URL patterns and topics.

func WithSubscriptionLimits added in v1.1.0 

func WithSubscriptionLimits(limits SubscriptionLimits) Option

WithSubscriptionLimits sets the concurrent HTTP subscription limits.

func WithSubscriptions 

func WithSubscriptions() Option

WithSubscriptions allows to dispatch updates when subscriptions are created or terminated.

func WithTopicMatcherStore added in v1.0.0 

func WithTopicMatcherStore(tms *TopicMatcherStore) Option

WithTopicMatcherStore sets the TopicMatcherStore instance to use.

func WithTransport 

func WithTransport(t Transport) Option

WithTransport sets the transport to use.

func WithWriteTimeout 

func WithWriteTimeout(timeout time.Duration) Option

WithWriteTimeout sets maximum duration before closing the connection, defaults to 600s, set to 0 to disable.

type PrometheusMetrics 

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

PrometheusMetrics store Hub collected metrics.

func NewPrometheusMetrics 

func NewPrometheusMetrics(registry prometheus.Registerer) *PrometheusMetrics

NewPrometheusMetrics creates a Prometheus metrics collector. This method must be called only one time, or it will panic.

func (*PrometheusMetrics) SubscriberConnected 

func (m *PrometheusMetrics) SubscriberConnected(_ *LocalSubscriber)

func (*PrometheusMetrics) SubscriberDisconnected 

func (m *PrometheusMetrics) SubscriberDisconnected(_ *LocalSubscriber)

func (*PrometheusMetrics) SubscriptionRejected added in v1.1.0 

func (m *PrometheusMetrics) SubscriptionRejected(reason string)

SubscriptionRejected records a rejected stream using a bounded reason label (total, token, or client), without exposing tokens or client addresses. Custom Metrics implementations may optionally implement this method.

func (*PrometheusMetrics) UpdatePublished 

func (m *PrometheusMetrics) UpdatePublished(_ *Update)

type RequestOrigin added in v1.0.0 

type RequestOrigin struct {
	Scheme string
	Host   string
}

RequestOrigin is the scheme and host an embedding server resolved for the current request. The hub derives its per-request OAuth resource identifier and RFC 9728 protected resource metadata URL from it (see requestIdentity).

type Static added in v1.0.0 

type Static struct {
	// Key is the verification key: an HMAC secret or a PEM-encoded public key.
	Key []byte
	// Algorithm is the single accepted JWS algorithm (e.g. "HS256", "RS256").
	Algorithm string
}

Static verifies tokens with a single embedded key.

type Subscriber 

type Subscriber struct {
	ID                 string
	EscapedID          string
	Claims             *claims
	RequestLastEventID string
	// RequestLastEventIDSet reports whether the request carried a
	// Last-Event-ID header or a last_event_id query parameter at all, even
	// with an empty value: the protocol requires answering with a
	// Mercure-Last-Event-ID response field whenever one was present.
	RequestLastEventIDSet bool

	// SubscribedMatchers are the topic matchers from the topic and
	// match_urlpattern query parameters (or from the v8 `topic` parameter,
	// which resolves to a deprecated matcher under compatibility mode).
	SubscribedMatchers []TopicMatcher
	// AllowedPrivateMatchers are the topic matchers from the JWT claims.
	AllowedPrivateMatchers []TopicMatcher
	// EscapedMatchers are precomputed "escapedType/escapedPattern" slugs for
	// subscription URLs.
	EscapedMatchers []string
	// SubscriptionPayloads holds the `payload` resolved per
	// subscribed matcher at registration time, indexed parallel to
	// SubscribedMatchers. Precomputing lets the subscription API render
	// payloads for subscribers a transport reconstructed from its
	// persistence layer without doing live matcher dispatch on the
	// deserialized object.
	SubscriptionPayloads []any
	// contains filtered or unexported fields
}

Subscriber represents a client subscribed to a list of topics on a remote or on the current hub.

func NewSubscriber 

func NewSubscriber(logger *slog.Logger, topicMatcherStore *TopicMatcherStore) *Subscriber

func (*Subscriber) LogValue added in v0.21.0 

func (s *Subscriber) LogValue() slog.Value

func (*Subscriber) Match added in v0.14.0 

func (s *Subscriber) Match(u *Update) bool

Match checks if the current subscriber can receive the given update.

func (*Subscriber) MatchTopics added in v0.14.0 

func (s *Subscriber) MatchTopics(topics []string, private bool) bool

MatchTopics checks if the current subscriber can access to at least one of the given topics.

func (*Subscriber) SetMatchers added in v1.0.0 

func (s *Subscriber) SetMatchers(subscribed, allowedPrivate []TopicMatcher)

SetMatchers sets the subscribed and allowed-private topic matchers and recomputes the derived subscription slugs and payloads, keeping the parallel SubscribedMatchers / EscapedMatchers / SubscriptionPayloads slices consistent. Transport implementations that reconstruct a Subscriber on another node use it to populate matchers programmatically, instead of relying on the exported fields (whose parallel-slice invariant is not otherwise enforced).

type SubscriberList added in v0.14.0 

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

func NewSubscriberList added in v0.14.0 

func NewSubscriberList(cacheSize int) *SubscriberList

func (*SubscriberList) Add added in v0.14.0 

func (sl *SubscriberList) Add(s *LocalSubscriber)

func (*SubscriberList) Len added in v0.14.0 

func (sl *SubscriberList) Len() int

func (*SubscriberList) MatchAny added in v0.14.0 

func (sl *SubscriberList) MatchAny(u *Update) []*LocalSubscriber

func (*SubscriberList) Remove added in v0.14.0 

func (sl *SubscriberList) Remove(s *LocalSubscriber)

func (*SubscriberList) Walk added in v0.14.0 

func (sl *SubscriberList) Walk(start uint64, callback func(s *LocalSubscriber) bool) uint64

type SubscriptionLimits added in v1.1.0 

type SubscriptionLimits struct {
	Total     int `json:"total"`
	PerToken  int `json:"per_token"`
	PerClient int `json:"per_client"`
}

SubscriptionLimits bounds concurrent HTTP event streams on one hub instance. A zero field disables that limit, and all are zero by default. Client limits use the address from WithClientIPFunc, or RemoteAddr, never untrusted forwarding headers; token limits count verified tokens.

type TopicMatcher added in v1.0.0 

type TopicMatcher struct {
	Type    MatcherType
	Pattern string
}

TopicMatcher pairs a matcher type with a pattern string. It is the exported value type transports use to (re)construct a Subscriber's matchers via Subscriber.SetMatchers.

type TopicMatcherStore added in v1.0.0 

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

TopicMatcherStore caches match results and compiled patterns. The match cache is a single unsharded otter instance; otter v2 is designed for high concurrency.

func NewTopicMatcherStore added in v1.0.0 

func NewTopicMatcherStore(cacheSize int) (*TopicMatcherStore, error)

NewTopicMatcherStore creates a TopicMatcherStore. If cacheSize > 0, match results, compiled templates and compiled URL patterns are cached; otherwise nothing is memoised.

type Transport 

type Transport interface {
	// Dispatch dispatches an update to all subscribers.
	//
	// It trusts u to be well-formed. A caller that builds u from untrusted
	// input (e.g. a publisher request) and dispatches it directly instead of
	// through Hub.Publish MUST call u.Validate with the matching base URL and
	// reject the update on error, otherwise a CR, LF, or NUL in ID or Type can inject arbitrary SSE
	// fields into subscribers' streams (CWE-93). Hub-internal updates such as
	// subscription events are trusted and skip Validate (they use reserved
	// topics that Validate rejects by design).
	Dispatch(ctx context.Context, u *Update) error

	// AddSubscriber adds a new subscriber to the transport.
	// It must call s.HistoryDispatched exactly once when s.RequestLastEventIDSet is true, even for an empty RequestLastEventID: the subscribe handler blocks on it before sending headers.
	AddSubscriber(ctx context.Context, s *LocalSubscriber) error

	// RemoveSubscriber removes a subscriber from the transport.
	RemoveSubscriber(ctx context.Context, s *LocalSubscriber) error

	// Close closes the Transport.
	Close(ctx context.Context) error
}

Transport provides methods to dispatch and persist updates.

type TransportError added in v0.12.2 

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

TransportError is returned when the Transport's DSN is invalid.

func (*TransportError) Error added in v0.12.2 

func (e *TransportError) Error() string

func (*TransportError) Unwrap added in v0.12.2 

func (e *TransportError) Unwrap() error

type TransportHealthChecker added in v0.23.0 

type TransportHealthChecker interface {
	// Ready reports whether the transport can currently serve traffic.
	// Returns nil if healthy, or an error describing the problem.
	// This is typically used for readiness probes (e.g. Kubernetes).
	Ready(ctx context.Context) error

	// Live reports whether the transport is fundamentally operational.
	// Returns nil if alive, or an error if the transport has been unhealthy
	// for an extended period and should be restarted.
	// This is typically used for liveness probes (e.g. Kubernetes).
	Live(ctx context.Context) error
}

TransportHealthChecker may be implemented by transports that support health checking. Transports that do not implement this interface are assumed to always be healthy.

type TransportSubscribers 

type TransportSubscribers interface {
	// GetSubscribers gets the last event ID and the list of active subscribers at this time.
	GetSubscribers(ctx context.Context) (string, []*Subscriber, error)
}

TransportSubscribers provides a method to retrieve the list of active subscribers.

type TransportTopicMatcherStore added in v1.0.0 

type TransportTopicMatcherStore interface {
	SetTopicMatcherStore(store *TopicMatcherStore)
}

TransportTopicMatcherStore provides a method to pass the TopicMatcherStore to the transport.

type Update 

type Update struct {
	// The Server-Sent Event to send.
	Event

	// The topics' Internationalized Resource Identifier (RFC3987) (will most
	// likely be URLs). The first one is the canonical topic; any others are
	// alternate topics. The update is dispatched to subscribers matching
	// either the canonical topic or any alternate; a private update's
	// audience is the union of the audiences of all its topics.
	Topics []string

	// Private updates can only be dispatched to subscribers authorized to receive them.
	Private bool

	// To print debug information
	Debug bool
}

Update represents an update to send to subscribers.

func (*Update) AssignUUID added in v0.21.0 

func (u *Update) AssignUUID()

AssignUUID generates a new UUID an assign it to the given update if no ID is already set.

func (*Update) LogValue added in v0.21.0 

func (u *Update) LogValue() slog.Value

func (*Update) SpanAttributes added in v0.24.0 

func (u *Update) SpanAttributes() []attribute.KeyValue

SpanAttributes returns the OpenTelemetry attributes describing this update.

func (*Update) Validate added in v0.24.2 

func (u *Update) Validate(baseURL string) error

Validate enforces the publish-side input rules that protect subscribers from update forgery and SSE field injection. Hub.Publish calls it, so the bundled hub and PublishHandler are already covered.

A caller that builds an Update from untrusted input (e.g. a publisher request) and dispatches it through a Transport directly, bypassing Hub.Publish, MUST call Validate with the same base URL used for topic matching and reject the update on error. The base must be an absolute URL with a host. Skipping it lets a CR, LF, or NUL in ID or Type inject arbitrary SSE fields into subscribers' streams (CWE-93). Validate also rejects the reserved "/.well-known/mercure" topic namespace, so it is meant for publisher input, not hub-internal updates such as subscription events.

type Verifier added in v1.0.0 

type Verifier interface {
	// contains filtered or unexported methods
}

Verifier supplies the material to verify an access token for one role of one issuer. It is a sealed interface with two implementations, Static and KeyFunc.

Source Files

View all Source files

Directories

Path Synopsis
caddy module

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.