joaju

package module
v0.2.0 Latest Latest
Warning

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

 Go to latest Published: Aug 15, 2026 License: MIT Imports: 17 Imported by: 0

README

Arandu

arandu-io/joaju

The WebSocket server for Arandu.

Build Status Go Reference Latest Version License

About the WebSocket server

The wire format is the Pusher protocol, so a browser client that already talks to Pusher or to Reverb talks to this. Nine HTTP routes and six kinds of channel, carrying the names Reverb uses, because a reader who knows Reverb should recognize this repository without reading it first.

server, err := joaju.NewServer(cfg)

Server is an http.Handler and it authenticates nobody: it reads the subject the framework's middleware put on the request and asks a Policy about it. Subscribing to a channel is a read, and reads have no exception — a connection cannot be built without a Grant, a channel name cannot either, and the tenant is taken from the Grant rather than from the wire.

The transport is the ws subpackage: this project's own RFC 6455, not a fork and with no borrowed code. It is why this repository takes no third-party dependency at all.

It is written, and go build, go vet and go test -race are green. The Autobahn conformance suite has not been run against it yet.

Learning Arandu

The API reference is generated from the doc comments and lives on pkg.go.dev. Every exported symbol carries one, and that is deliberate: it is the documentation that cannot drift from the code, because it sits in the same file.

The CLI documents itself. aru help lists every command, and each one explains what it writes and what to do with it. aru doctor explains what it found and what breaks, not which rule was violated.

A guide and a website do not exist yet, and that is a decision rather than a gap: a guide written against an API that still moves is work done twice, and the second time is worse — there is wrong documentation published. The site is the next phase, and it will be an Arandu application.

Contributing

See CONTRIBUTING.md. Before opening a pull request, the three commands at the top of that file have to pass, and CI runs exactly them.

Security Vulnerabilities

Please review our security policy on how to report a vulnerability. Never open a public issue for one.

License

Open-sourced software licensed under the MIT license.

Documentation

Overview

Package joaju is the WebSocket server: it holds open sockets, keeps them grouped into channels, and delivers a message to every socket in a group.

This file declares the types the rest of the repository is written against. Nothing here opens a socket, parses a frame or serves an HTTP route; the packages that do those things take and return the values declared here.

Where the shapes come from

The structure is laravel/reverb's -- the route table, the channel kinds, the method names on a channel -- because a reader who knows Reverb should recognize this repository without reading it first (RULE 10). The implementation is not Reverb's and cannot be: Reverb is PHP on a ReactPHP event loop, whose default stream_select stops at 1024 connections. In Go a connection is a goroutine and the ceiling is the file descriptor limit, so there is no event loop to pick (ADR 0052).

The wire format is the Pusher protocol, which is why several names here are Pusher's rather than Go's: Member has user_id and user_info, the event names carry the "pusher:" and "pusher_internal:" prefixes, and a channel's kind is read off its name prefix. Those are bytes a browser client already knows how to read, and renaming them would mean shipping a client nobody else can talk to.

The authorization shape, which is the point

Two decisions happen, and each one is a Policy that issues an auth.Grant: 

[Connect]                    opening the socket    -> [ConnectPolicy]
broadcasting.ChannelJoin     listening on one      -> [SubscriptionPolicy]

They are two decisions and not one because they answer different questions. The first is whether this subject may hold a socket at all, and it is where an origin allowlist belongs -- see Handshake. The second is whether this subject may hear what goes out on one particular channel, and it is asked again for every channel, because subscribing is a read and RULE 17 opens no exception for reads.

A Connection cannot be built without a Grant, and a ChannelName cannot be built without one either. That is RULE 14 as a type rather than a review comment: the tenant is read off the Grant by NewChannelName, and the name the client sent is refused outright if it so much as contains the separator the tenant goes before. A client that could put "acme:" in front of a channel is a client choosing whose events it hears.

There is deliberately no constructor that builds a ChannelName out of a name that already carries a tenant. The server that receives a relayed message over Redis pub/sub has such a name in hand, and the temptation is to parse the tenant back out of it -- which is the tenant coming off the wire. It compares the string against the ChannelName.String of the channels it already holds instead, and those were each built from a Grant.

What hesape already answers, and is not answered again here

github.com/arandu-io/hesape/broadcasting owns the channel vocabulary shared with the SSE fanout, and this package imports it rather than restating it (RULE 9): broadcasting.PrivateChannelPrefix and broadcasting.PresenceChannelPrefix, broadcasting.TenantSeparator, broadcasting.RequestedChannel, broadcasting.TenantChannel and the broadcasting.ChannelJoin action are all used here as they stand. Only the three cache-channel prefixes are new, because SSE has no cache channel.

What Reverb has here and this package does not

Reverb's Channel::findById is Channel.Find. Reverb needs both because its find() takes a Connection object and compares identity; here a connection is identified by its SocketID and one lookup covers both callers.

Reverb's Channel::broadcastInternally is not on Channel. It exists there so that a CacheChannel replaying a payload it received from another server does not cache it a second time, and the distinction it draws is between an event this server was handed and an event it was relayed. That belongs to the relay and not to the channel interface, and putting it here would give every implementation a second broadcast method to explain (RULE 9).

Reverb's ChannelManager::for($app) is not here either: multi-application is out of the first version, because a Go binary is one process and running one per application is both simpler to operate and free of the cross-application state RULE 14 exists to prevent (ADR 0052).

Pusher's private-encrypted- channels have no ChannelType. Reverb has no class for them, and end-to-end encrypted channels are a key-distribution feature, not a channel kind.

The routes these types serve

The nine routes of Reverb's Factory::pusherRoutes, which the HTTP layer of this repository answers with the values declared here: 

GET  /app/{appKey}                                       the socket itself
POST /apps/{appId}/events                                publish one
POST /apps/{appId}/batch_events                          publish many
GET  /apps/{appId}/connections                           metrics
GET  /apps/{appId}/channels                              list
GET  /apps/{appId}/channels/{channel}                    one
GET  /apps/{appId}/channels/{channel}/users              presence members
POST /apps/{appId}/users/{userId}/terminate_connections  disconnect
GET  /up                                                 health

Every one of the seven that reads or writes channel state needs a Grant, and Broker is why: there is no method on it that reaches a channel without one.

The transport

Subpackage ws is this project's own implementation of RFC 6455, and it is why the dependency graph has one entry in it. It is not a fork and it borrows no code: the surface this server actually uses is ten symbols, so writing them costs less than carrying a library's client stack, proxy support and compression -- and less than tracking somebody else's security advisories forever (ADR 0052).

Index

Constants

View Source 
const (
	// ReasonClient is the client hanging up, which is the ordinary case.
	ReasonClient = "client"
	// ReasonTimeout is nothing heard within PongTimeout.
	ReasonTimeout = "timeout"
	// ReasonTerminated is POST /apps/{id}/users/{userId}/terminate_connections.
	ReasonTerminated = "terminated"
	// ReasonShutdown is the server closing.
	ReasonShutdown = "shutdown"
	// ReasonLimit is the connection limit, refused after the upgrade.
	ReasonLimit = "limit"
)

Reasons a connection ended, as Observer.ConnectionClosed receives them.

View Source 
const (
	// CacheChannelPrefix marks a public cache channel.
	CacheChannelPrefix = "cache-"
	// PrivateCacheChannelPrefix marks a cache channel that must be authorized.
	PrivateCacheChannelPrefix = broadcasting.PrivateChannelPrefix + CacheChannelPrefix
	// PresenceCacheChannelPrefix marks a cache channel that publishes its
	// members.
	PresenceCacheChannelPrefix = broadcasting.PresenceChannelPrefix + CacheChannelPrefix
)

The prefixes a Pusher cache channel carries.

broadcasting.PrivateChannelPrefix and broadcasting.PresenceChannelPrefix already name the other two, and this package uses those rather than spelling them a second time (RULE 9). Only these three are new, because the SSE fanout hesape/broadcasting was written for has no cache channel.

A cache channel replays the last event it carried to whoever subscribes next, so a client that connects after the event still sees the current state instead of waiting for the next one.

View Source 
const (
	// ProtocolPrefix is the namespace of the protocol's own events.
	ProtocolPrefix = "pusher:"
	// InternalPrefix is the namespace of events only the server may send.
	InternalPrefix = "pusher_internal:"
)

The two reserved namespaces of the Pusher protocol.

An event a client publishes may use neither: the first is the protocol's own traffic, and the second is the server talking about channel state. A client that could send pusher_internal:member_added could invent members.

View Source 
const (
	// EventConnectionEstablished carries the [SocketID] to a client that has
	// just connected. It is the first frame the server sends, and the client
	// needs it: the socket id is what it puts in a publish so its own broadcast
	// does not come back to it.
	EventConnectionEstablished = ProtocolPrefix + "connection_established"
	// EventError reports a refusal. A subscription denied by a Policy is one.
	EventError = ProtocolPrefix + "error"
	// EventSubscribe is a client asking to listen on a channel.
	EventSubscribe = ProtocolPrefix + "subscribe"
	// EventUnsubscribe is a client asking to stop.
	EventUnsubscribe = ProtocolPrefix + "unsubscribe"
	// EventPing is a client checking the socket is alive.
	EventPing = ProtocolPrefix + "ping"
	// EventPong answers [EventPing].
	EventPong = ProtocolPrefix + "pong"

	// EventSubscriptionSucceeded confirms a subscription and carries
	// [Channel.Data], which for a presence channel is the member list.
	EventSubscriptionSucceeded = InternalPrefix + "subscription_succeeded"
	// EventMemberAdded tells a presence channel a member joined.
	EventMemberAdded = InternalPrefix + "member_added"
	// EventMemberRemoved tells a presence channel a member left.
	EventMemberRemoved = InternalPrefix + "member_removed"
)

The protocol event names.

The four the Reverb clone spells are EventConnectionEstablished, EventError, EventMemberAdded and EventMemberRemoved. The rest are the Pusher protocol's, and are here so that the two ends of this repository -- the code that reads a frame and the code that writes one -- name them once.

View Source 
const (
	// DefaultMaxMessageSize is the largest frame a client may send, and is the
	// Pusher protocol's own limit of 10 KiB. It is a limit on what a client
	// sends, not on what the server sends: a presence channel's member list is
	// larger than this and goes out fine.
	DefaultMaxMessageSize int64 = 10 << 10
	// DefaultMaxBodySize is the largest body an API route reads. A batch
	// publish is the big one.
	DefaultMaxBodySize int64 = 1 << 20
	// DefaultOutboundQueue is how many frames may be waiting on one socket
	// before the client is judged to have fallen behind.
	DefaultOutboundQueue = 64
	// DefaultMaxConnections is how many sockets one tenant may hold open.
	//
	// Zero in [ServerConfig] means this, and this is not "no limit": a server
	// that accepts sockets until the file descriptors run out stops answering
	// for EVERY tenant, and the one that exhausted them is not necessarily the
	// one that notices. Reverb calls the refusal ConnectionLimitExceeded and
	// leaves the number to configuration; the number here is a default that a
	// deployment raises knowingly.
	DefaultMaxConnections = 10_000
	// DefaultWriteTimeout is how long one frame may take to reach the client.
	DefaultWriteTimeout = 10 * time.Second
	// DefaultPingInterval is how often the server sends a WebSocket ping.
	DefaultPingInterval = 30 * time.Second
	// DefaultPongTimeout is how long the server waits to hear anything from a
	// client before it hangs up. It has to be longer than
	// [DefaultPingInterval], or the read deadline expires before the pong the
	// ping asked for can arrive.
	DefaultPongTimeout = 70 * time.Second
)

The defaults NewServer fills in for a zero ServerConfig field.

View Source 
const ClientEventPrefix = "client-"

ClientEventPrefix is the namespace of an event a client published rather than received: "client-".

It is the third reserved prefix and the only one that travels inwards. ProtocolPrefix and InternalPrefix name what the server says; this one names what a browser says to every other browser on the channel, which is why ClientEvents exists and why it is off in its zero value.

View Source 
const Connect auth.Action = "joaju.connect"

Connect is the auth.Action the Grant behind a Connection is issued for.

It is the first of the two decisions this package makes, and it is separate from broadcasting.ChannelJoin on purpose. This one answers whether a subject may hold a socket at all; the other answers whether they may hear one channel, and it is asked again for every channel they name. Issuing one Grant for both would mean the per-channel policy never runs, which is RULE 17 undone by a constant.

View Source 
const DefaultMetricsTimeout = 500 * time.Millisecond

DefaultMetricsTimeout is how long a metrics route waits for the rest of the fleet before it answers with what it has.

Unlike DefaultRetryInterval this one is a default and not the whole answer, because it is the number here that depends on the deployment: instances sharing a datacentre answer in single-digit milliseconds and instances spread across a region do not. What it may not be is absent. An instance that has stopped answering must not hold a route open, and most of the fleet's numbers now is worth more than all of them never.

View Source 
const DefaultRetryInterval = time.Second

DefaultRetryInterval is how long a Relay waits before opening a subscription again after the one it had was lost.

It is a constant and not a knob because a knob would be a second way to answer "how fast does this reconnect" (RULE 9). A second is short enough that a Redis restart costs one message and long enough that a Redis that is down is not asked a thousand times a second by every instance at once.

View Source 
const EventCacheMiss = ProtocolPrefix + "cache_miss"

EventCacheMiss tells a subscriber of a cache channel that there is nothing to replay yet, so it stops waiting for the event it would have been sent.

It is the one protocol event name joaju does not already declare, because it belongs to the cache channels and those exist only here, in the wire format.

View Source 
const MetricsTopic = TopicPrefix + "metrics"

MetricsTopic is the pub/sub channel one instance asks the others what they are holding on.

It is one topic for the whole fleet and not one per tenant, because the question is asked by an instance and not by a customer: every instance subscribes once, when its server is built, and a topic per tenant would mean a subscription that changes every time a customer's first socket opens. The tenant travels inside the question, where it is a filter and never an authority -- see [Relay.answer].

It cannot be mistaken for a channel's Topic. A channel's topic is TopicPrefix, the tenant, broadcasting.TenantSeparator and the name, so it always carries a second colon; this carries none.

View Source 
const TopicPrefix = "joaju:"

TopicPrefix is what every Redis pub/sub channel this package publishes on begins with.

It is there so that relay traffic is distinguishable from everything else on the same Redis -- in particular from what hesape's RedisBroadcaster publishes, which is an application talking to a relay and not two relays talking to each other. The two carry different payloads, and a subscriber that heard both would decode one of them wrong.

Variables

View Source 
var (
	// ErrOverQuota is 4004, and is what a server past its connection limit
	// answers a new socket.
	ErrOverQuota = ProtocolError{Code: CodeOverQuota, Message: "Application is over connection quota"}
	// ErrUnauthorized is 4009, and is what every refusal from a
	// [ConnectPolicy] or a [SubscriptionPolicy] becomes on the wire.
	ErrUnauthorized = ProtocolError{Code: CodeUnauthorized, Message: "Connection is unauthorized"}
	// ErrOriginNotAllowed is 4009: the Origin the browser sent is one a
	// [ConnectPolicy] refused. See [Handshake].
	ErrOriginNotAllowed = ProtocolError{Code: CodeUnauthorized, Message: "Origin not allowed"}
	// ErrNotSubscribed is 4009: a client tried to publish on a channel it is
	// not on.
	ErrNotSubscribed = ProtocolError{Code: CodeUnauthorized, Message: "The client is not a member of the specified channel."}
	// ErrClientEventChannel is 4009: client events exist only on the channels a
	// [SubscriptionPolicy] guarded. See [ClientEvents].
	ErrClientEventChannel = ProtocolError{Code: CodeUnauthorized, Message: "Client event rejected - only supported on private and presence channels"}
	// ErrInvalidMessage is 4200, and is what anything unreadable becomes. It is
	// also the answer to an unknown event name, which is Reverb's.
	ErrInvalidMessage = ProtocolError{Code: CodeInvalidMessage, Message: "Invalid message format"}
	// ErrRateLimited is 4301: the client is sending faster than it may.
	ErrRateLimited = ProtocolError{Code: CodeRateLimited, Message: "Rate limit exceeded"}
	// ErrClientEventsDisabled is 4301: this server does not relay client
	// events, which is the default. See [ClientEvents].
	ErrClientEventsDisabled = ProtocolError{Code: CodeRateLimited, Message: "The app does not have client messaging enabled."}
)

The refusals this package sends, with the wording laravel/reverb uses so that a client's own message table keeps matching.

View Source 
var ErrConnectionLimit = errors.New("joaju: this tenant is holding as many connections as it may")

ErrWrongTenant is what a Broker answers when the Grant it was handed and the ChannelName it was handed disagree about whose channel it is.

It cannot happen through NewChannelName, which reads the tenant off the Grant. It can happen when a name built under one Grant is used with another, which is the mistake worth a distinct error: both values are valid, and nothing but this comparison notices. ErrConnectionLimit is the tenant already holding as many sockets as ServerConfig.MaxConnections allows.

It is refused AFTER the upgrade, not before, and that is on purpose: the client is a websocket by then, so it can be told why in a protocol frame it understands instead of getting an HTTP status it has no handler for. Reverb answers the same way, with ConnectionLimitExceeded.

It is not an auth.ErrForbidden. Nothing was refused on authority -- the same client with the same Grant is admitted the moment somebody disconnects.

View Source 
var ErrNoChannel = errors.New("joaju: no such channel")

ErrNoChannel is what Broker.Find answers when no channel is held under the name.

View Source 
var ErrNoGrant = fmt.Errorf("%w: joaju: no grant", auth.ErrForbidden)

ErrNoGrant is what a constructor answers when the Grant it was handed authorized nothing.

It wraps auth.ErrForbidden, so a caller that already distinguishes a refusal from a failure keeps doing so.

View Source 
var ErrRelayClosed = errors.New("joaju: the relay is closed")

ErrRelayClosed is what a Relay answers once Relay.Close has been called.

View Source 
var ErrSocketClosed = errors.New("joaju: the socket is closed")

ErrSocketClosed is what Sink.Send answers once the socket is gone.

A client that cannot keep up gets it too, wrapped: the socket is closed first, because the alternative is a broadcast that blocks on the slowest subscriber and takes the channel down with it.

View Source 
var ErrWrongTenant = fmt.Errorf("%w: joaju: the grant and the channel belong to different tenants", auth.ErrForbidden)

Functions

func Encode 

func Encode(f Frame) ([]byte, error)

Encode is the bytes of one frame, ready to be written to a socket.

func MetricsReplyTopic 

func MetricsReplyTopic(id InstanceID) string

MetricsReplyTopic is where the answers to one instance's question go.

An answer is addressed to the instance that asked rather than published back to the fleet, and what an answer carries is the reason: the members of a presence channel are people's ids, and a shared reply topic would put one customer's member list in front of every instance in the fleet, including the ones holding none of that customer's sockets.

The separator is "/" and not broadcasting.TenantSeparator so that no reply topic can be read as a channel's Topic: a tenant is letters, digits, "-" and "_", so a topic with a "/" in that position was never a channel's.

func Topic 

func Topic(name ChannelName) string

Topic is the Redis pub/sub channel a joaju channel is relayed on.

It is TopicPrefix followed by ChannelName.String, so the tenant is in the middle of it -- "joaju:acme:orders" and "joaju:globex:orders" are two topics, and two customers who both called a channel "orders" never hear each other. That is RULE 14 reaching the one place where a channel name leaves this process, and it holds by construction: a ChannelName cannot be built without a Grant to read the tenant off.

The zero ChannelName has no topic and answers the empty string, which Relay.Join and Relay.Publish refuse before they get here.

Types

type Broker 

type Broker interface {
	// Find is the channel held under the name, or [ErrNoChannel].
	//
	// It returns [ErrWrongTenant] when auth.Tenant(g) is not
	// [ChannelName.Tenant] -- which [NewChannelName] cannot produce, and a
	// name carried over from another Grant can.
	Find(ctx context.Context, g auth.Grant, name ChannelName) (Channel, error)

	// FindOrCreate is Find, creating the channel of the kind
	// [ChannelName.Type] names when it is not there yet.
	//
	// It is what a subscription calls: the first subscriber to a channel is
	// what brings it into existence. Creating one discloses nothing and grants
	// nothing -- [Channel.Subscribe] still asks a [SubscriptionPolicy].
	FindOrCreate(ctx context.Context, g auth.Grant, name ChannelName) (Channel, error)

	// Remove drops the channel. An implementation calls it when the last
	// subscriber leaves, which is what Reverb's Channel::unsubscribe does.
	Remove(ctx context.Context, g auth.Grant, name ChannelName) error

	// All is every channel in the Grant's tenant, and never another tenant's.
	//
	// It answers GET /apps/{appId}/channels. A broker holding several tenants
	// filters here; the filter is the whole reason this takes a Grant.
	All(ctx context.Context, g auth.Grant) ([]Channel, error)
}

Broker holds the channels and hands them out. It is the only way to one.

It is Reverb's ChannelManager and its ChannelBroker at once. Those are two classes there because one picks the subclass by prefix and the other keeps the instances; here picking is ChannelName.Type and keeping is this, and a caller that could get a channel from either of two places would have two answers to compare (RULE 9).

Every method takes an auth.Grant, and that is RULE 17 rather than ceremony. The metrics routes -- the channel list, one channel, its members -- are reads of who is talking to whom, and a read with no policy behind it is a tenant boundary that exists everywhere except the dashboard.

Implementations are safe for concurrent use.

func NewMemoryBroker added in v0.2.0 

func NewMemoryBroker() Broker

NewMemoryBroker is the Broker a server runs on: the channels this process holds, in a map.

Memory is in the name because it is the thing to know before running a second instance, not a hedge against a stored one arriving later. A channel is a set of open sockets and a socket is held by one process, so the channels of an instance cannot be anywhere but in it. What crosses instances is the events, over Relay, and the metrics routes ask the fleet rather than a shared store.

It takes no argument. There is nothing to configure about a map, and an option on it would be a second way to build the one registry this package has (RULE 9).

func RelayedBroker added in v0.2.0 

func RelayedBroker(base Broker, relay *Relay) Broker

relayedBroker is the Broker a server holds when it was built with a Relay, and it is the inbound half of horizontal scaling: what makes this instance receive from the bus the channels it is actually holding, and only those.

The two methods it wraps are the two ends of a channel's life on one instance, which is what Broker already says they are: Broker.FindOrCreate is what a subscription calls, because the first subscriber is what brings a channel into existence, and Broker.Remove is what the last subscriber leaving calls. So the fleet's topic is joined where a channel appears here and left where it goes, and neither a Protocol nor an application's Broker has to remember either call.

Not having to remember is the reason this wraps instead of being asked of every implementation. Broker is an interface an application writes, and a relay that had to be joined by hand would be joined by hand wrongly in exactly one deployment -- where the symptom is a client hearing nothing, which from inside a browser is indistinguishable from a quiet channel (RULE 9).

Broker.Find and Broker.All are the embedded ones and are untouched, because they read. A route that looks a channel up is not a socket listening on one, and an instance that opened a pipe from the fleet for every channel a dashboard mentioned would be receiving traffic it has nobody to deliver to. RelayedBroker is the Broker an instance uses when it is one of several.

It wraps the application's Broker so that a channel coming into existence here subscribes this instance to the fleet's topic for it, and the last subscriber leaving unsubscribes. Without it an instance relays what its own API publishes and never hears what the others publish.

Wire it once and hand it to both

The same value has to reach the Protocol -- NewPusher takes a Broker -- and ServerConfig.Broker. A Protocol holding the raw Broker never joins a topic, however well the server is configured, because a socket's subscription passes through the Protocol and not through the server. 

relay, err := joaju.NewRelay(ctx, id, bus, log)
broker := joaju.RelayedBroker(joaju.NewMemoryBroker(), relay)
protocol := joaju.NewPusher(broker, subscribe, joaju.PusherConfig{})
server, err := joaju.NewServer(joaju.ServerConfig{
	Broker: broker, Protocol: protocol, Relay: relay,
})

NewServer refuses a Relay whose Broker did not come from here, so the half-wired arrangement is a startup error and not a message that never arrives.

A nil relay returns base unchanged, and wrapping twice returns the wrapper rather than a second layer.

type Bus 

type Bus interface {
	// Publish sends one message to one channel and answers how many subscribers
	// received it. It is the PUBLISH command.
	Publish(ctx context.Context, channel string, message any) (int64, error)
	// Subscribe listens on a set of channels and calls callback for each
	// message, with the message first and the channel it arrived on second. It
	// blocks until ctx is cancelled or the connection is lost.
	Subscribe(ctx context.Context, channels []string, callback func(message, channel string)) error
}

Bus is the little of a Redis connection that horizontal scaling needs.

The two methods are github.com/arandu-io/hesape/redis/connections.Connection's Publish and Subscribe, signature for signature, so that a *connections.Connection satisfies this by having been written -- there is no adapter to keep in step.

It is declared here rather than imported for the reason the same shape is declared in hesape's broadcasters package: github.com/arandu-io/hesape/redis is a separate module, because the driver beneath it is a third-party dependency and Go has no optional dependency (ADR 0048). Importing it would put that driver in this repository's go.mod, and a graph with one entry in it is what writing the ws subpackage bought (ADR 0052). Stating the contract and letting the application pass its connection in costs nothing and keeps the graph.

Publish and Subscribe both prefix the channel name with the connection's own key prefix, on both sides, which is why nothing here compensates for it and why the name handed to the callback is not the name passed to Subscribe.

type Channel 

type Channel interface {
	// Name is the channel's name, with its tenant.
	Name() ChannelName

	// Connections is every current subscriber.
	//
	// Reverb returns a map keyed by connection id; this returns a slice,
	// because every caller of it iterates. [Channel.Find] is the keyed lookup.
	Connections() []Subscriber

	// Find is the subscriber on this channel with the given socket id, if it
	// is subscribed.
	Find(id SocketID) (Subscriber, bool)

	// Subscribe adds a connection to the channel.
	//
	// g must be a Grant issued for broadcasting.ChannelJoin by a
	// [SubscriptionPolicy] asked about this exact [Subscription], and its
	// tenant must be the tenant in [Channel.Name] -- otherwise
	// [ErrWrongTenant]. An implementation checks the Grant before it touches
	// its state, on every kind of channel and not only the guarded ones:
	// [ChannelType.Guarded] says whether a policy may allow a subscription
	// freely, not whether one is consulted.
	//
	// member is the presence data the client offered, and is ignored off a
	// presence channel. On a presence channel, subscribing announces the new
	// member to the others with [EventMemberAdded].
	Subscribe(ctx context.Context, g auth.Grant, conn *Connection, member Member) error

	// Unsubscribe removes a connection from the channel, and on a presence
	// channel announces its departure with [EventMemberRemoved].
	//
	// It takes no Grant. Leaving is not a read: nothing is disclosed by it, and
	// a socket that drops has no one left to ask.
	Unsubscribe(ctx context.Context, conn *Connection) error

	// Subscribed reports whether this connection is on this channel.
	Subscribed(conn *Connection) bool

	// Broadcast delivers the event to every subscriber except the one named by
	// [Event.Socket].
	Broadcast(ctx context.Context, e Event) error

	// BroadcastToAll delivers the event to every subscriber, [Event.Socket]
	// included. It is what the server itself sends on -- a member list, a
	// refusal -- where there is no sender to spare.
	BroadcastToAll(ctx context.Context, e Event) error

	// Data is what [EventSubscriptionSucceeded] carries to a new subscriber.
	//
	// It is empty on every kind of channel but a presence one, where it is
	// Reverb's presence block: {"presence": {"count": n, "ids": [...],
	// "hash": {...}}}. It stays a map because it is a payload on its way to
	// JSON and nothing in this repository reads a field out of it.
	Data() map[string]any
}

Channel is a group of sockets that receive the same events.

The method names are Reverb's Channel class, so that the six kinds of channel read as what they are there. The differences are three:

  • Subscribe takes a context and an auth.Grant. Reverb's takes a connection and two optional strings, one of which is a signature it verifies itself. A subscription that a Policy did not decide is the leak this repository exists not to have, and a Grant in the signature is how the compiler asks for the decision.
  • Find takes a SocketID, and so covers Reverb's find() and findById() both.
  • The reads carry no context and return no error, because the sockets in a channel are the ones this process holds and looking at them is not I/O. Everything that writes to a socket does both.

Implementations are safe for concurrent use. One connection is one goroutine and a channel is shared between all of them, so the alternative is a lock every caller has to remember.

func NewChannel 

func NewChannel(name ChannelName) (Channel, error)

NewChannel is the only way to a Channel, and it makes the kind ChannelName.Type names.

It is Reverb's ChannelBroker::create, which switches on the same prefixes to pick a subclass. There is no subclass to pick here and no argument that says which kind to build: the kind is in the name, the name came from a Grant, and a caller who could ask for a public channel under a private name would have found the way around ChannelType.Guarded.

The zero ChannelName is refused. It is what a failed NewChannelName hands back, and a channel built from one would be a channel with no tenant.

type ChannelName 

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

ChannelName is a channel name with its tenant already in it.

It has only unexported fields, so it cannot be written as a struct literal with a tenant somebody chose. NewChannelName is the way to one, and it reads the tenant off an auth.Grant -- which is RULE 14 expressed as a type, the same shape auth.Grant itself uses to make an authorization decision unforgeable.

It has two string forms and they are not interchangeable: 

[ChannelName.String]    "acme:private-orders.17"  the key, and what Redis publishes on
[ChannelName.Requested] "private-orders.17"       what the client sent, and what goes back to it

The client never sees the first. It asked for the second, it is answered about the second, and the tenant it was scoped to is not its to know or to choose.

The zero value is not a channel. ChannelName.IsZero reports it and ChannelName.String answers the empty string rather than a bare separator, so a zero value cannot be mistaken for a channel named "".

func NewChannelName 

func NewChannelName(g auth.Grant, requested string) (ChannelName, error)

NewChannelName builds the name a channel is held under, out of the Grant and the raw name the client asked for.

requested is exactly what came off the wire, prefix and all: "private-orders.17". It is refused if it is empty or contains broadcasting.TenantSeparator, because that is where the tenant goes and a client that names one is a client choosing whose events it hears. The Grant is refused if it carries no tenant, or one that cannot be a namespace.

Both refusals are broadcasting.RequestedChannel and broadcasting.TenantChannel doing the work, because the SSE fanout already authenticates channel names and two definitions of a valid one are how the looser of the two gets found (RULE 9).

func (ChannelName) IsZero 

func (n ChannelName) IsZero() bool

IsZero reports whether this is the zero value, which is not a channel.

func (ChannelName) Requested 

func (n ChannelName) Requested() string

Requested is the name the client asked for, with no tenant in it.

This is the value that goes back on the wire, in the "channel" field of every frame, because it is the name the client used and the only one it knows.

func (ChannelName) String 

func (n ChannelName) String() string

String is the published name, "<tenant>:<channel>".

It is the key a Broker holds the channel under and the name a message is published on for the other servers in the fleet, and it is never sent to a client.

func (ChannelName) Tenant 

func (n ChannelName) Tenant() string

Tenant is whose channel this is. It came from the Grant and from nowhere else.

func (ChannelName) Type 

func (n ChannelName) Type() ChannelType

Type reads the kind off the prefix, which is Reverb's ChannelBroker::create.

The order matters and is the reason this is one function: "private-cache-" also begins with "private-", so the compound prefixes are tested first. Two copies of this switch would be two copies of that ordering, and the second one is where a private cache channel quietly becomes a plain private one.

It reads ChannelName.Requested and not ChannelName.String, because the prefix is at the front of the name the client sent and the published name has the tenant in front of it. That exact mistake is on record in hesape/broadcasting: IsGuardedChannel asked the published name whether it began with "private-", so it answered false for every private channel it existed to protect.

type ChannelType 

type ChannelType uint8

ChannelType is which of the six kinds of channel a name denotes.

The kind is read off the name's prefix rather than stored, because the client chooses it by what it asks for: a client that sends "private-orders.17" is asking for a private channel by naming one. ChannelName.Type is the rule, and it is the only place the prefixes are compared. 

const (
	// PublicChannel is Reverb's Channel: no prefix, no authorization, anyone
	// connected may listen.
	//
	// It still cannot cross a tenant. Every name carries one (RULE 14), so
	// "public" means public within one customer's namespace and never wider.
	PublicChannel ChannelType = iota
	// PrivateChannel is "private-": a subscription a [SubscriptionPolicy] has
	// to allow.
	PrivateChannel
	// PresenceChannel is "presence-": private, and its subscribers are
	// published to each other as [Member] values.
	PresenceChannel
	// CacheChannel is "cache-": public, and it replays its last event to
	// whoever subscribes next.
	CacheChannel
	// PrivateCacheChannel is "private-cache-": [PrivateChannel] and
	// [CacheChannel] together.
	PrivateCacheChannel
	// PresenceCacheChannel is "presence-cache-": [PresenceChannel] and
	// [CacheChannel] together.
	PresenceCacheChannel
)

The six channel kinds, which are Reverb's six Channel classes.

Reverb's Channels directory holds a seventh file, ChannelBroker, which is not a kind of channel: it is the rule that picks one, and here that rule is ChannelName.Type and the thing that hands channels out is Broker.

func (ChannelType) Cache 

func (t ChannelType) Cache() bool

Cache reports whether this kind of channel replays its last event to whoever subscribes next.

func (ChannelType) Guarded 

func (t ChannelType) Guarded() bool

Guarded reports whether a subscription to this kind of channel has to be authorized by a SubscriptionPolicy.

It is Reverb's UsePusherChannelConventions::isGuardedChannel, and it answers no for the public kinds -- which does not make them reachable across a tenant, because the tenant is in every name and comes from the Grant that built it.

func (ChannelType) Presence 

func (t ChannelType) Presence() bool

Presence reports whether this kind of channel publishes its subscribers to each other, which is what makes Member and Channel.Data mean anything.

func (ChannelType) String 

func (t ChannelType) String() string

String names the kind, for a log line and for the metrics routes.

type ClientEvents 

type ClientEvents bool

ClientEvents is whether this server relays the events a client publishes, and its zero value is off.

A client event is a frame one browser sends that every other browser on the channel receives, and the server does not look at it. It is the only path in this repository where a client is the source of what other clients are told, and it is the half of the protocol worth being afraid of:

  • the payload is the sender's, so whatever the receivers render from it is rendered from a stranger's bytes;
  • the sender is a browser, so its identity is whatever a SubscriptionPolicy settled at subscription time and nothing more;
  • nothing on the server sees the message go past, so there is no audit trail, no validation and no rate limit that is not built here.

So it is off unless something turns it on, which is Pusher's own default and Reverb's. When it is on, ClientEvents.Accept still holds two lines that do not move: the channel has to be one a policy guarded, and the sender has to be subscribed to it.

Why there are two states and not three

Reverb spells the same switch as accept_client_events_from, and it reads three values: "members", "all", and anything else for off. The two here are its off and its "members". The third was measured against this file rather than declined on principle, and what it changes there is three things -- two of which cannot exist here, and the third of which is the authorization:

  • under "all" the sender need not be on the channel. There is no frame a client can send that reaches a Channel it did not subscribe to: the caller of Accept is handed the seat the socket already holds, and nothing on that path asks a Broker for a channel. Adding the lookup would be a socket publishing into a private channel no SubscriptionPolicy ever saw (RULE 17), which is the leak this repository exists not to have;
  • under "members" Reverb rebuilds the relayed payload out of three named fields, because under "all" it forwards the sender's frame verbatim -- every extra top-level key it carried included, a user_id the sender wrote for itself among them. Here a relayed frame is built field by field out of an Event, and there is no verbatim path for a setting to pick;
  • under "all" no user_id is stamped. Here it comes off the channel's record of the seat, so the only way not to have one is not to have a seat, which is the first point again.

So the setting would be a name for the membership check with a value that turns it off, and one more way to relay a client event (RULE 9) -- the looser of two paths being the one that ends up in production. Reverb's own two defaults already disagree about which value it is: config/reverb.php:90 publishes "members" and ConfigApplicationProvider falls back to "all" when the key is absent. A switch that is one thing in the config file and another in the code is a switch nobody knows the state of. 

const (
	// ClientEventsOff refuses every client event. It is the zero value.
	ClientEventsOff ClientEvents = false
	// ClientEventsOn relays a client event from a subscriber of a guarded
	// channel, and only from one.
	ClientEventsOn ClientEvents = true
)

The two states of ClientEvents, named so a call site reads as a decision rather than as a bare true.

func (ClientEvents) Accept 

func (c ClientEvents) Accept(f Frame, channel ChannelName, from SocketID, sender Member, subscribed bool) (Event, error)

Accept decides whether a client-published frame may be relayed, and turns it into the Event that would be relayed.

channel is the ChannelName the caller resolved from the frame's channel field with NewChannelName and the socket's Grant -- so the tenant is already settled and is the Grant's, never the name the sender typed. Its ChannelName.Requested has to match what the frame said, which catches the caller resolving one name and relaying another. from is the sender, and it ends up in Event.Socket so the relay does not send them their own message back. sender and subscribed are the seat the channel is holding for them, and Channel.Find answers both at once.

sender becomes Event.UserID, and is the whole of who the receivers are told this came from. It is the Member the channel seated, so it is the identity a SubscriptionPolicy was asked about; it is the zero Member off a presence channel, and the relayed frame then carries no user_id at all. The frame's own Frame.UserID is never read here, and that is the point of taking the sender as an argument rather than off f: a client that writes a user_id into a client event is a client naming somebody else as the sender, and the channel's record is the one place that answer is not the sender's to write.

The refusals, in the order they are made: not a client event at all is ErrInvalidMessage; the switch being off is ErrClientEventsDisabled; a public channel is ErrClientEventChannel; not being subscribed is ErrNotSubscribed. Each is a ProtocolError, so ErrorFrame carries it to the client with its code intact.

type ConnectPolicy 

type ConnectPolicy = auth.Policy[Handshake]

ConnectPolicy decides whether a client may open a socket. Its Grant is issued for Connect.

It is an alias and not a new interface because there is one way to authorize in this ecosystem, and it is auth.Policy (RULE 9). The alias exists so the resource type is named at the point the decision is described.

type Connection 

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

Connection is one client's open socket.

It has only unexported fields and one constructor, so it cannot be written as a struct literal: NewConnection takes an auth.Grant and refuses a Grant that authorized nothing. That is the same shape auth.Grant uses on itself, and it is here for the same reason -- a Connection carries the tenant every channel name it reaches is built from, and a Connection assembled by hand would be a tenant somebody typed.

It is a struct and not an interface because there is one kind of connection. What varies is where the bytes go, and that is Sink.

func NewConnection 

func NewConnection(g auth.Grant, id SocketID, sink Sink) (*Connection, error)

NewConnection is the only way to a Connection.

g must be a Grant issued for Connect -- that is, one an auth.Policy answered about this handshake. A zero Grant, or one issued for another action, is refused here rather than at the first channel the client names.

func (*Connection) Grant 

func (c *Connection) Grant() auth.Grant

Grant is the proof the handshake was authorized.

It is issued for Connect and it is not a subscription: a caller that has it still has to run a SubscriptionPolicy to reach a channel, because auth.Grant.Check refuses a Grant issued for another action.

func (*Connection) ID 

func (c *Connection) ID() SocketID

ID is the socket id, which is Reverb's Connection::id.

func (*Connection) Send 

func (c *Connection) Send(ctx context.Context, message []byte) error

Send writes one encoded frame to this client.

func (*Connection) Subject 

func (c *Connection) Subject() auth.Subject

Subject is who this socket was authenticated as.

func (*Connection) Tenant 

func (c *Connection) Tenant() string

Tenant is whose socket this is, read off the Grant and from nowhere else.

func (*Connection) Terminate 

func (c *Connection) Terminate(ctx context.Context) error

Terminate closes this client's socket.

type Counter 

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

Counter is an Observer that counts, and the one an application usually wants.

It answers what GET /apps/{appId}/connections and the Pulse cards in Reverb answer: how many are connected, how many messages crossed, how many channels exist. It holds numbers and nothing else -- no history, no per-connection record -- because a server that remembers every connection is a server whose memory grows with churn.

Safe for concurrent use. Every method may be called from any connection's goroutine at any time, which is the normal case rather than the exception.

func NewCounter 

func NewCounter() *Counter

NewCounter has no Reverb counterpart: Laravel's recorders are registered into Pulse, and this is the value that plays their part.

func (*Counter) ChannelCreated 

func (c *Counter) ChannelCreated(_ context.Context, name ChannelName)

func (*Counter) ChannelRemoved 

func (c *Counter) ChannelRemoved(_ context.Context, name ChannelName)

func (*Counter) ConnectionClosed 

func (c *Counter) ConnectionClosed(_ context.Context, _ SocketID, tenant, _ string)

func (*Counter) ConnectionOpened 

func (c *Counter) ConnectionOpened(_ context.Context, _ SocketID, tenant string)

func (*Counter) MessageReceived 

func (c *Counter) MessageReceived(_ context.Context, _ SocketID, _ []byte)

MessageReceived and MessageSent count frames and never keep them.

The bytes are on the Observer interface because a diagnosis wants them; a counter that stored them would be a full traffic log nobody asked for, of customer data, growing without bound.

func (*Counter) MessageSent 

func (c *Counter) MessageSent(_ context.Context, _ SocketID, _ []byte)

func (*Counter) Read 

func (c *Counter) Read(tenant string) TenantCount

Read is one tenant's numbers, copied.

A copy rather than a pointer, so that a caller reading a dashboard cannot hold a reference that changes while it renders -- or write through it.

func (*Counter) Tenants 

func (c *Counter) Tenants() []string

Tenants is every tenant this counter has seen, for an operator's dashboard.

It is NOT reachable from a request handler: the HTTP routes answer for the Grant's tenant and one tenant only. Anything calling this is running with the operator's own authority, outside the request path.

type ErrorCode 

type ErrorCode int

ErrorCode is a code from Pusher's error table, sent in the data of EventError.

The number is the contract, not the message: an existing client branches on the code and only prints the text. The ranges are Pusher's, and a client reads them without a table -- 4000 to 4099 means do not come back, 4100 to 4199 means come back after a backoff, 4200 to 4299 means come back now. 

const (
	// CodeOverQuota is 4004: too many sockets are open.
	CodeOverQuota ErrorCode = 4004
	// CodeUnauthorized is 4009: the client may not do that. Every refusal a
	// Policy produces arrives here.
	CodeUnauthorized ErrorCode = 4009
	// CodeOverCapacity is 4100: come back later, with a backoff.
	CodeOverCapacity ErrorCode = 4100
	// CodeInvalidMessage is 4200: the frame was not something this server could
	// read. It is the code for everything that is not one of the others.
	CodeInvalidMessage ErrorCode = 4200
	// CodeRateLimited is 4301: the client is sending too fast, or asked for
	// something this server does not offer it.
	CodeRateLimited ErrorCode = 4301
)

The codes this server sends. They are the ones laravel/reverb spells, kept at the same numbers because a client that already handles Reverb handles these.

type Event 

type Event struct {
	// Name is the event name, and goes out in the frame's "event" field.
	// [ProtocolPrefix] and [InternalPrefix] are reserved: an event a client
	// published may use neither.
	Name string
	// Channel is where it goes.
	Channel ChannelName
	// Data is the payload, left encoded because this package never reads it.
	Data json.RawMessage
	// Socket is who sent it, and it exists so that they do not receive it.
	//
	// A client that publishes a message has already drawn its own result on
	// screen; delivering it back makes the message appear twice. The client
	// learns its own id from [EventConnectionEstablished] and puts it in the
	// publish. [Channel.Broadcast] skips it; [Channel.BroadcastToAll] ignores
	// this field, which is the only difference between the two.
	//
	// Empty means the event came from the server and goes to everyone.
	Socket SocketID

	// UserID is who published a relayed client event, and is empty on
	// everything else.
	//
	// It is the sender's [Member.UserID] as the channel holds it, which is the
	// identity a [SubscriptionPolicy] settled when it seated them -- never the
	// one the sending frame claimed. A receiver drawing "somebody is typing"
	// has no other way to name them: the payload is a stranger's bytes, so a
	// user_id inside it says whatever that stranger wanted it to say.
	//
	// It is empty off a presence channel, because a channel that publishes
	// nothing about who is listening has no member to name. The field is then
	// absent from the frame rather than empty in it -- see [EventFrame].
	UserID string
}

Event is what travels: one message, on one channel.

It has no JSON tags, and that is deliberate -- it is the value this server passes around, not the frame. The protocol layer builds the frame, and two fields do not go out as they are held:

  • Channel goes out as ChannelName.Requested, never as ChannelName.String. The client asked for "private-orders.17" and that is the name it is answered about; the tenant is not its to see.
  • Data goes out as a JSON string containing JSON, which is what the Pusher protocol specifies and what Reverb's json_encode of the data field produces. It is held decoded here so that nothing double-encodes twice.

type Frame 

type Frame struct {
	// Event is the event name, and is required. It is one of the [ProtocolPrefix]
	// names, one of the [InternalPrefix] names, an application event, or a
	// [ClientEventPrefix] one.
	Event string
	// Channel is the channel the frame is about, with no tenant in it. It is
	// absent on the frames that are about the socket rather than a channel:
	// [EventConnectionEstablished], [EventPing], [EventPong] and most
	// [EventError] frames.
	Channel string
	// Data is the payload, decoded. It is encoded as a JSON string on the way
	// out and decoded from one on the way in.
	Data json.RawMessage
	// UserID is the sender of a relayed client event, and is empty on every
	// other frame. It is Pusher's user_id, top level and beside the channel
	// rather than inside the data.
	UserID string
}

Frame is one Pusher message, in both directions.

The envelope is three fields, and the protocol's oddity is in the middle one: 

{"event":"pusher:connection_established","data":"{\"socket_id\":\"7.1\"}"}

data is a JSON string that contains JSON. It is not a nesting anybody would choose, and it is not ours to change -- every Pusher client parses the string a second time, so a frame that put an object there is a frame they cannot read. Frame.Data holds the inner value decoded, and the encoding happens once, here, so that no caller has to remember which side of it they are on.

The frame's Channel is a plain string and not a ChannelName, because that is what the field is: on the way in it is what the client asked for, and on the way out it is ChannelName.Requested. The constructors below do the conversion, and they are the reason no code path can write ChannelName.String -- with the tenant in front of it -- into a frame.

The zero value is not a frame: encoding one without an event name fails rather than emitting {}.

func CacheMiss 

func CacheMiss(name ChannelName) (Frame, error)

CacheMiss tells a new subscriber of a cache channel that there is nothing to replay, so it stops waiting for an event that is not coming.

func ConnectionEstablished 

func ConnectionEstablished(id SocketID, activityTimeout time.Duration) Frame

ConnectionEstablished is the first frame the server sends, and it hands the client the socket id it will quote back when it publishes.

activityTimeout is how long the client may stay silent before it should ping, and it goes out in seconds because the protocol counts in seconds. A duration under a second is sent as one: zero would tell the client to ping immediately. A zero or negative duration omits the field, and the client falls back to its own default.

func Decode 

func Decode(message []byte) (Frame, error)

Decode reads one frame off a socket.

It refuses a frame with no event name, which json.Unmarshal alone would accept, and it refuses it as ErrInvalidMessage -- so the caller can hand the error straight to ErrorFrame and the client is told 4200 rather than the parser's complaint about byte 31.

func ErrorFrame 

func ErrorFrame(err error) Frame

ErrorFrame is the frame a client is sent about any error at all.

It is the only translation from an internal error to something a client reads, and it exists so there is one: a caller that built the frame itself would be a caller choosing how much of the cause to disclose, and it takes one such caller to send a policy's refusal -- which names the subject, the action and the resource -- to the browser that was refused.

A ProtocolError keeps its own code. An auth.ErrForbidden from any depth, which is what a ConnectPolicy or a SubscriptionPolicy produces, becomes ErrUnauthorized. Everything else becomes ErrInvalidMessage, and its text is dropped. The caller logs err.

func EventFrame 

func EventFrame(e Event) (Frame, error)

EventFrame is the frame that carries an Event to a subscriber.

It is the one place Event becomes bytes, and it is where the channel loses its tenant: ChannelName.Requested goes out, ChannelName.String never does. The client asked for "private-orders.17" and is answered about "private-orders.17"; that it was "acme:private-orders.17" here is not its to see, and a frame naming the tenant would tell every subscriber the name of the namespace their neighbours are in.

Event.Socket is not in the frame. It says who not to send this to, which is the sender's business and the channel's, and it is answered before this is called.

Event.UserID is, and only when there is one. It is set on a client event relayed from a presence channel and on nothing else, and Frame.MarshalJSON leaves the field out when it is empty -- so a private channel's frames keep the shape they had. A user_id of "" is a key a client validating a schema has to be taught to expect, in exchange for a value that never names anybody.

func MemberAdded 

func MemberAdded(name ChannelName, m Member) (Frame, error)

MemberAdded tells the rest of a presence channel that somebody joined. It carries the whole Member, which is user_id and user_info.

func MemberRemoved 

func MemberRemoved(name ChannelName, m Member) (Frame, error)

MemberRemoved tells the rest of a presence channel that somebody left.

It carries the user_id alone, and not Member.Info: the departure identifies a member the others already have, and sending their information again on the way out would put a copy of it in a frame that nobody reads it from.

func Ping 

func Ping() Frame

Ping is the server asking a silent socket whether it is still there.

func Pong 

func Pong() Frame

Pong answers a client's EventPing. It carries no data, which is the frame Pusher's clients expect: {"event":"pusher:pong"}.

func SubscriptionSucceeded 

func SubscriptionSucceeded(name ChannelName, data map[string]any) (Frame, error)

SubscriptionSucceeded confirms a subscription and carries Channel.Data.

data is empty on every kind of channel but a presence one, and it still goes out as an object: an internal frame always carries {} rather than no data field, because Reverb's json_encode((object) $data) always produces one and a client that finds the field missing has nothing to parse.

func (Frame) ClientMaySend 

func (f Frame) ClientMaySend() error

ClientMaySend is the refusal, if any, of a frame that arrived from a client.

It is the whole list of what a socket may say, and it is a closed one:

  • pusher:subscribe, pusher:unsubscribe, pusher:ping and pusher:pong;
  • anything beginning with client-, which ClientEvents then decides about.

Everything else is ErrInvalidMessage. That includes the "pusher_internal:" events, and refusing them is the reason this function exists rather than the server switching on the name: pusher_internal:member_added is the frame that tells a presence channel who joined, and a client able to send one is a client able to invent members of a channel it is on. It also includes an application event with no prefix at all, which is published over the HTTP route by something holding the credentials, never over a socket by a browser.

func (Frame) IsClientEvent 

func (f Frame) IsClientEvent() bool

IsClientEvent reports whether a client published this, rather than the server. See ClientEvents.

func (Frame) IsInternal 

func (f Frame) IsInternal() bool

IsInternal reports whether this is one of the events only the server may send, the "pusher_internal:" ones.

func (Frame) IsProtocol 

func (f Frame) IsProtocol() bool

IsProtocol reports whether this is one of the protocol's own events, the "pusher:" ones.

func (Frame) MarshalJSON 

func (f Frame) MarshalJSON() ([]byte, error)

MarshalJSON writes the envelope, with Frame.Data encoded as the protocol's string-containing-JSON.

It is a method rather than a function so that the encoding cannot be sidestepped: json.Marshal of a Frame, anywhere, produces the same bytes Encode does.

func (Frame) Subscribe 

func (f Frame) Subscribe() (SubscribeRequest, error)

Subscribe decodes the data of a EventSubscribe frame.

func (*Frame) UnmarshalJSON 

func (f *Frame) UnmarshalJSON(b []byte) error

UnmarshalJSON reads the envelope, undoing the double encoding of the data field when it finds it.

The field arrives either way and both are legal: pusher-js sends the data of pusher:subscribe as an object, and sends the data of a client event as a string containing JSON. A string whose contents are not JSON stays the string it is -- which is the same rule Reverb applies, and it means a client that sends the data "123" is heard as the number 123. That ambiguity is the protocol's, and matching it is the point.

func (Frame) Unsubscribe 

func (f Frame) Unsubscribe() (string, error)

Unsubscribe decodes the data of a EventUnsubscribe frame, and is the channel name the client asked to leave -- raw, with no tenant in it, like SubscribeRequest.Channel.

type Handshake 

type Handshake struct {
	// Socket is the id minted for this client.
	Socket SocketID
	// Origin is the Origin header of the upgrade request, verbatim. It is what
	// the browser claims, so it is evidence for a policy and never a tenant.
	Origin string
}

Handshake is what a ConnectPolicy decides about: one client asking to open a socket.

The origin is here rather than in configuration because refusing an origin is a decision about who may connect, and this repository has one place where those are made. Reverb keeps an allowed-origins list per application; a policy answers the same question and answers it beside every other one.

type InstanceID 

type InstanceID string

InstanceID identifies one joaju process within a fleet of them.

It is the whole of the deduplication. Redis pub/sub has no notion of a publisher, so an instance that publishes on a topic it is subscribed to hears its own message come back -- and it has already delivered that message to its own connections, which is what publishing was in aid of. Every relayed message carries the id of the instance that sent it, and an instance drops the ones carrying its own: without that, every message would reach half the fleet once and the instance it came from twice.

It has to differ between instances and nothing here can check that. A hostname, a pod name or a random string minted at start-up are all fine; a constant compiled into the binary is not, because then no instance can tell its own traffic from its neighbour's and the deduplication silently drops every message in the fleet.

type Member 

type Member struct {
	// UserID is user_id: who this subscriber is, within the tenant.
	UserID string `json:"user_id"`
	// Info is user_info: whatever else the application publishes about them,
	// left encoded because this package never looks inside it.
	Info json.RawMessage `json:"user_info,omitempty"`
}

Member is what a presence channel publishes about one subscriber.

The field names are Pusher's, because they are the JSON keys a browser client reads out of EventMemberAdded and out of the presence block of Channel.Data.

UserID is not the tenant and is not authority. It says which person a subscriber is inside one customer's channel; whose channel it is was settled by the Grant that built the ChannelName before this value was looked at.

type NopObserver 

type NopObserver struct{}

NopObserver ignores everything, and is what a server without one uses.

It exists so that the server never checks for nil on a path it runs per message. The check would be free; the branch in six places would not be, and a nil that reaches one of them is a panic on a live connection.

func (NopObserver) ChannelCreated 

func (NopObserver) ChannelCreated(context.Context, ChannelName)

func (NopObserver) ChannelRemoved 

func (NopObserver) ChannelRemoved(context.Context, ChannelName)

func (NopObserver) ConnectionClosed 

func (NopObserver) ConnectionClosed(context.Context, SocketID, string, string)

func (NopObserver) ConnectionOpened 

func (NopObserver) ConnectionOpened(context.Context, SocketID, string)

func (NopObserver) MessageReceived 

func (NopObserver) MessageReceived(context.Context, SocketID, []byte)

func (NopObserver) MessageSent 

func (NopObserver) MessageSent(context.Context, SocketID, []byte)

type Observer 

type Observer interface {
	// ChannelCreated is the first subscription to a channel that did not exist.
	ChannelCreated(ctx context.Context, name ChannelName)

	// ChannelRemoved is the last subscriber leaving.
	ChannelRemoved(ctx context.Context, name ChannelName)

	// ConnectionOpened is a socket that finished the upgrade and was
	// authorised. A refused upgrade never reaches here.
	ConnectionOpened(ctx context.Context, id SocketID, tenant string)

	// ConnectionClosed is a socket that went away, for any reason: the client
	// closed it, the pong did not arrive in time, or the server terminated it.
	//
	// Reverb splits this into a close and a ConnectionPruned, because it prunes
	// on a schedule and has to say which happened. Here the read deadline does
	// the pruning, so there is one event and a reason string.
	ConnectionClosed(ctx context.Context, id SocketID, tenant, reason string)

	// MessageReceived is a frame that arrived from a client, before it is acted
	// on. The raw bytes, because that is what a diagnosis wants.
	MessageReceived(ctx context.Context, id SocketID, message []byte)

	// MessageSent is a frame written to a client.
	MessageSent(ctx context.Context, id SocketID, message []byte)
}

Observer is told what the server did, after it did it.

It exists so that an application can count, log or audit without the server knowing about counting, logging or auditing. Reverb fires five events for the same purpose, through Laravel's dispatcher; there is no dispatcher here (ADR 0001), so the destination is a value somebody passes in.

Every method runs after the fact, and its return value is ignored

None of these can refuse anything. A channel is created because somebody authorised a subscription; a message is sent because it was already broadcast. An observer that could veto would be a second authorisation path, and RULE 17 allows one -- the SubscriptionPolicy is it.

It must not block

The server calls these on the path that serves connections. An observer that talks to a database on MessageSent turns every broadcast into a round trip, and a thousand messages a second into a thousand of them. Count in memory, and let something else read the counter.

Nothing here is called with a lock held, so an observer may call back into the server without deadlocking. That is deliberate and it is tested.

type Protocol 

type Protocol interface {
	// Open is called once, after the handshake was authorized and the socket is
	// writable, and before any client frame is read. It is where
	// [EventConnectionEstablished] goes out.
	//
	// An error closes the socket.
	Open(ctx context.Context, conn *Connection) error

	// Message is called for each frame the client sends.
	//
	// An error does not close the socket: a client that asks for a channel it
	// may not have is told so with [EventError] and goes on using the ones it
	// has. The server logs it and reads the next frame.
	Message(ctx context.Context, conn *Connection, message []byte) error

	// Close is called once, after the socket is gone and the connection has
	// been dropped from the server's registry. It is where a channel's
	// subscribers are cleaned up and [EventMemberRemoved] goes out.
	//
	// It gets a context that is already cancelled if the process is shutting
	// down, so an implementation that has to reach Redis derives one with its
	// own deadline.
	Close(ctx context.Context, conn *Connection)
}

Protocol is the frame layer: what the Server hands a socket's traffic to.

The server owns the socket -- the upgrade, the two goroutines, the deadlines, the registry -- and owns no part of the Pusher protocol. It sends no frame of its own, not even EventConnectionEstablished, because a second place that builds a frame is a second answer to what a frame looks like (RULE 9). The one it does put on the wire is a refusal this package already declared and it only encodes -- ErrRateLimited, when ServerConfig.MaxMessagesPerSecond is what dropped the frame -- because that limit is the socket's and a Protocol cannot answer for a frame it was never handed.

The three methods are Reverb's Protocols\Pusher\Server: open(), message() and close(). Its fourth, error(), is not here -- in Go the error comes back from the call that failed.

An implementation is called from the one goroutine that reads a given socket, so calls concerning one Connection are ordered and never concurrent with each other. Calls concerning different connections are concurrent.

func NewPusher added in v0.2.0 

func NewPusher(broker Broker, subscribe SubscriptionPolicy, cfg PusherConfig) Protocol

NewPusher is the Protocol that speaks the Pusher protocol, and the one a Server is built with.

broker is where a channel is reached and where one begins and ends: the first subscriber to a name brings the channel into existence and the last one to leave drops it, which is what Broker.FindOrCreate and Broker.Remove exist for. subscribe is asked about every subscription, on every kind of channel.

It takes neither the app id, nor the socket's limits, nor a logger, and that is the line these arguments are drawn on: what the Server holds it holds because it owns the socket, a second copy here would be a second answer the first time a deployment changed one of them, and every method of Protocol is handed the Connection that carries the rest. What is left is PusherConfig, which is what nothing else could tell this type.

It answers a Protocol and no error, because there is nothing here that can fail. The one exception is a nil Broker or a nil SubscriptionPolicy, and it panics: that is a wiring mistake, it happens in the same breath as the NewServer call that would have refused the same omission in its config, and the alternative is a nil interface reached from a live socket's goroutine at the first subscription -- with a stack that names hesape/auth rather than whoever built this.

type ProtocolError 

type ProtocolError struct {
	// Code is the number the client branches on.
	Code ErrorCode
	// Message is the sentence the client may print. It is fixed per code, and
	// is Reverb's wording where Reverb has one.
	Message string
}

ProtocolError is a refusal expressed in the protocol's own terms.

It carries only a code and a fixed sentence, and that is deliberate: it is the one error in this package whose text reaches a client, so it never contains the reason. A Policy refuses a subscription with a sentence naming the subject and the resource, and that sentence belongs in the log; what the client is told is 4009.

The values are comparable, so the ones declared below can be compared with errors.Is even after being wrapped.

func (ProtocolError) Error 

func (e ProtocolError) Error() string

Error makes a ProtocolError an error.

func (ProtocolError) Frame 

func (e ProtocolError) Frame() Frame

Frame is the EventError frame that carries this refusal to the client.

type PusherConfig added in v0.2.0 

type PusherConfig struct {
	// ActivityTimeout is how long a client may stay silent before it should send
	// [EventPing], and it is what [EventConnectionEstablished] carries. Zero
	// leaves the field out of the frame and the client falls back to its own
	// default.
	//
	// It is the protocol's number and not the socket's, which is why it is here
	// rather than read off [ServerConfig.PongTimeout]. That one is a read
	// deadline, and any frame at all resets it -- including the WebSocket ping
	// the writer sends on its own every [ServerConfig.PingInterval], which the
	// client's stack answers without the client knowing. This one is an
	// instruction a browser obeys, and it belongs below PongTimeout: a client
	// that pings after the deadline pings a socket that has already been hung up
	// on.
	ActivityTimeout time.Duration

	// ClientEvents is whether one browser's frame may be relayed to the others on
	// the channel. It is off in its zero value, which is Pusher's default and
	// Reverb's; [ClientEvents] is where the reason is written down.
	ClientEvents ClientEvents

	// Observer is told when a channel came into existence and when it went. Nil
	// means [NopObserver].
	//
	// It is the interface [ServerConfig.Observer] takes and an application hands
	// one value to both, because the two halves cannot be announced from one
	// place: a socket is opened and closed by the server, and a channel is
	// created and dropped by nothing but this.
	//
	// [Observer.MessageReceived] and [Observer.MessageSent] are deliberately not
	// announced from here. Only one of the two could be -- a frame a client sends
	// arrives at [Protocol.Message] unless the socket's rate limit dropped it,
	// while a frame a client receives is usually written by a [Channel]
	// delivering a broadcast, which this type never sees. Counting one side of a
	// conversation would show a server that receives a thousand messages and
	// sends none, and a number that is wrong in a knowable direction is worse
	// than the zero it replaces.
	Observer Observer
}

PusherConfig is the rest of what NewPusher takes: the settings whose zero value is a decision rather than an omission.

The two things the protocol cannot work without are arguments to the constructor and not fields here, because a field nobody filled in is a server that starts and a Grant nobody asked for.

type Relay 

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

Relay is horizontal scaling: it carries what this instance received to the other instances, and delivers what they received to this instance's connections.

It is the shape ADR 0052 settles on, and it is Reverb's: every server publishes on Redis pub/sub, every server subscribes, and no server has to know which of the others holds a given socket. One topic per channel per tenant -- see Topic -- so an instance receives only the traffic of channels it is actually holding.

The flow of one message, and it is worth reading in order, because the local delivery is not this type's job:

  1. a client publishes on a channel this instance holds;
  2. the caller delivers it locally with Channel.Broadcast, which skips the socket that sent it;
  3. the caller hands the same Event to Relay.Publish;
  4. every instance holding that channel receives it, this one included;
  5. this one drops it, on the InstanceID; the others deliver it.

Step three is called by the code that received the event from outside, and never by the code in step four. A relay that republished what it was relayed would be a fleet talking to itself forever.

The other direction: the metrics routes

The same bus carries the four metrics routes' questions, and it carries them because those routes are wrong without it. A count of open sockets served out of one process is a count of one process's sockets, and two instances serving one application answer two different numbers of which neither is the answer.

So a route asks: it publishes a [fleetQuestion] on MetricsTopic, every other instance answers with what it holds on MetricsReplyTopic of the asker, and the asker adds those to what it answered from its own Broker. An instance that does not answer in time is left out and the route replies anyway -- see [Relay.ask] for why a partial answer is the answer.

A Relay is safe for concurrent use, and it is meant to be: one connection is one goroutine, and each of them can publish.

func NewRelay 

func NewRelay(ctx context.Context, id InstanceID, bus Bus, log *slog.Logger) (*Relay, error)

NewRelay is the only way to a Relay.

ctx is the server's lifetime and not a request's: closing it stops every subscription, which is what Relay.Close does through it.

bus may be nil, and that is a supported deployment rather than a mistake: one instance has nobody to relay to. It starts degraded and says so once, because an operator who meant to configure Redis and did not needs to read that in the log rather than infer it from clients on different instances not seeing each other.

log may be nil, in which case slog.Default is used.

func (*Relay) Close 

func (r *Relay) Close() error

Close stops every subscription and waits for them.

It is idempotent, and after it every Relay.Join and Relay.Publish answers ErrRelayClosed rather than working on a relay whose goroutines are gone.

func (*Relay) Degraded 

func (r *Relay) Degraded() bool

Degraded reports whether this instance is currently unable to reach the bus.

A degraded instance still works: it accepts connections, authorizes subscriptions and delivers to the sockets it holds. What it has lost is the other instances, so a client on this one and a client on another stop hearing each other. That is worth answering GET /up with and worth putting on the metrics route, because from inside one browser it is indistinguishable from a quiet channel.

func (*Relay) ID 

func (r *Relay) ID() InstanceID

ID is which instance this is.

func (*Relay) Join 

func (r *Relay) Join(g auth.Grant, ch Channel) error

Join starts relaying a channel: from now on, what the other instances publish on it is delivered to ch.

g must be a Grant issued for broadcasting.ChannelJoin, and its tenant must be the tenant of ch's name, or ErrWrongTenant. That is the same Grant the SubscriptionPolicy answered for the first subscriber, and asking for it here is not ceremony: this is the call that opens a pipe from every other instance into this channel, and a pipe nobody authorized is RULE 17 with a network in the middle of it.

It takes no context because it starts I/O rather than doing any -- the subscription runs on the lifetime the relay was built with. It is idempotent: a channel already joined is left as it is, since the first subscriber is what brings a channel into existence and every later one finds it already relayed.

A relay with no bus, or one that cannot reach the bus, joins successfully. Refusing here would turn a Redis outage into every subscription failing, when what an outage actually costs is the other instances.

func (*Relay) Joined 

func (r *Relay) Joined() int

Joined reports how many channels this instance is relaying. It answers the metrics route, and it is what a test asserts on.

func (*Relay) Leave 

func (r *Relay) Leave(name ChannelName) error

Leave stops relaying a channel.

The caller is whatever removed the channel -- Broker.Remove, when the last subscriber left. It takes no Grant for the reason Channel.Unsubscribe takes none: leaving discloses nothing.

A channel that was never joined is not an error. It is what a broker that removes a channel unconditionally will do, and there is nothing to report.

func (*Relay) Publish 

func (r *Relay) Publish(ctx context.Context, e Event) error

Publish sends an event to the other instances.

It is called once per event, by whatever received that event from outside -- a client frame, or POST /apps/{appId}/events -- and after the local delivery, not instead of it. This instance's own connections are served by Channel.Broadcast; this call is only about the other instances.

A bus that cannot be reached is not an error and does not come back as one. The event has already been delivered to every socket this instance holds, so there is nothing to fail: what was lost is reach, and reach is reported by Relay.Degraded and by one log line at the moment it goes. Returning an error would make a client whose message was delivered read that it was not, and would make every publish in the fleet fail for as long as Redis was down -- which is exactly the outage the caller is being protected from.

What does come back as an error is a malformed event: no channel, no name, or data that will not encode. Those are bugs in the caller and no amount of Redis fixes them.

type Server 

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

Server answers the nine routes: the socket and the eight of the HTTP API.

It is an http.Handler and not a net/http.Server, so that it is mounted the way any other handler is -- behind hesape/auth's Authenticate middleware, which is where the auth.Subject on the request context comes from. That middleware is the only thing in the ecosystem that calls auth.WithSubject, and this server does not authenticate anybody: it reads the subject the middleware put there and asks a Policy about it.

That is the one real difference from Reverb's API routes, which verify an app_secret HMAC on every call. Reverb has to, because it is a separate process with no session to read. Here the request has already been through the framework's front door, and a second credential of its own would be a second way to prove who is calling (RULE 9).

The authorization shape, which every route follows 

[ConnectPolicy]       once per request  may this subject be here at all
[SubscriptionPolicy]  once per channel  may this subject reach this channel

Both run on the API routes and not only on the socket, because listing channels, counting subscribers and reading a presence channel's members are reads of who is talking to whom, and RULE 17 opens no exception for reads. A dashboard that lists channels without a policy is a tenant boundary that holds everywhere except the dashboard.

A Handshake is what the ConnectPolicy is asked about on an API route as well, with Socket empty -- an API caller is asking the same question a browser asks, minus the socket it wants opened. Inventing a third auth.Action for it would mean a third policy an application has to remember to write, and the question it would answer is the one Connect already answers.

The Connect Grant is also what a ChannelName on an API route is built from, because a name needs a tenant and RULE 14 says a tenant comes off a Grant. The channel a caller named in the path is never trusted for that: it supplies the name after the tenant, and nothing else.

func NewServer 

func NewServer(cfg ServerConfig) (*Server, error)

NewServer builds the server, or says which part of the config was missing.

func (*Server) Close 

func (s *Server) Close(ctx context.Context)

Close terminates every socket this server holds. It is what a shutdown calls, and it takes no Grant because it crosses every tenant on purpose.

func (*Server) Connections 

func (s *Server) Connections(g auth.Grant) (int, error)

Connections is how many sockets this process holds for the Grant's tenant.

It takes a Grant for the reason Broker.All does: a count of who is connected is a read, and a count that spans tenants tells one customer how many of another's people are online. The Grant has to be one a ConnectPolicy issued -- auth.Grant.Check is what says so -- and the tenant it carries is the only filter, because it is the only one that did not come in with the request (RULE 14).

func (*Server) ServeHTTP 

func (s *Server) ServeHTTP(w http.ResponseWriter, r *http.Request)

ServeHTTP routes the nine routes.

func (*Server) Terminate 

func (s *Server) Terminate(ctx context.Context, g auth.Grant, subject string) (int, error)

Terminate closes every socket the Grant's tenant holds for one subject, and answers how many it closed.

It is the route Reverb calls terminate_connections, and it is what a sign-out or a revoked membership calls: the socket was authorized once, at the handshake, and nothing about it expires on its own.

type ServerConfig 

type ServerConfig struct {
	// AppID is the {appId} the API routes carry, and AppKey is the {appKey} the
	// socket route carries. A request naming another app is answered 404.
	//
	// One server is one application (ADR 0052): Reverb serves several from one
	// instance because hosting PHP is expensive, and a Go binary is a process.
	// These are here so the route table is Reverb's and the names in a client's
	// configuration mean what they mean everywhere else.
	AppID  string
	AppKey string

	// Broker holds the channels. Every route that reads or writes channel state
	// goes through it, and there is no method on it that reaches a channel
	// without a Grant.
	Broker Broker

	// Connect decides whether a subject may be on this server at all. It runs
	// on the socket route and on every API route, before anything else.
	Connect ConnectPolicy

	// Subscribe decides whether a subject may reach one channel. It runs once
	// per channel touched, on every route that touches one.
	Subscribe SubscriptionPolicy

	// Protocol is the frame layer. See [Protocol].
	Protocol Protocol

	// Relay is the other instances, and a deployment of more than one needs it.
	// With one, an event published here reaches the sockets the other instances
	// hold, what they publish reaches the sockets held here, and the four
	// metrics routes answer for the whole fleet. Without one, each of those
	// three is answered by this process alone -- which is the true answer for a
	// deployment of one and the wrong answer for any other: two instances
	// serving one application each hold half the sockets, and neither can see
	// the other half.
	//
	// It is optional because a relay is a Redis, and a server that refused to
	// start without one would make the single-instance deployment the harder of
	// the two. See [Relay].
	Relay *Relay

	// MetricsTimeout is how long a metrics route waits for the other instances
	// before it answers with what has arrived. Zero means
	// [DefaultMetricsTimeout].
	//
	// It is a bound and not a promise: an instance that answers after it is
	// left out of that reply, because a dashboard served late by one instance
	// being replaced is a dashboard that is down. See [Relay.ask].
	MetricsTimeout time.Duration

	// Log is where a refusal and a dropped socket are recorded. nil means
	// slog.Default.
	Log *slog.Logger

	// MaxMessageSize, MaxBodySize, OutboundQueue, WriteTimeout, PingInterval
	// and PongTimeout are the socket's limits. Zero means the Default of the
	// same name.
	MaxMessageSize int64
	MaxBodySize    int64

	// MaxConnections is how many sockets ONE TENANT may hold. Zero means
	// [DefaultMaxConnections]; a negative number means no limit, and saying so
	// takes writing -1 rather than leaving a field out.
	//
	// Per tenant and not per server, for the reason RULE 14 gives: a global
	// limit lets one customer's traffic refuse another customer's connections,
	// which is a denial of service one of them did not cause and cannot see.
	MaxConnections int

	// MaxMessagesPerSecond is how many frames ONE SOCKET may send in a second.
	// Zero, which is the default, is no limit.
	//
	// Per socket and not per tenant, which is the opposite of what
	// MaxConnections does and is the same reasoning taken one layer in: a socket
	// is the smallest thing a noisy client owns, so metering it spends nothing
	// of what the tenant's other sockets have. An allowance shared across a
	// tenant would let one runaway browser tab refuse the frames of every other
	// tab that customer has open, which is the denial of service the connection
	// limit exists to keep out.
	//
	// Zero is no limit rather than a Default of its own, because there is no
	// rate that is right for traffic this server has not seen: a chat client and
	// one that reports a cursor position differ by two orders of magnitude, and
	// a limit that refuses the frames of a correct client is worse than no limit
	// at all. It is turned on by a deployment that measured its own traffic.
	//
	// A frame past the limit is answered with [ErrRateLimited] and dropped, and
	// the socket stays open. There is no second setting that closes it instead:
	// two ways to answer one refusal is two behaviours to explain (RULE 9), and
	// the client that a limit is aimed at is the one worth keeping addressable.
	MaxMessagesPerSecond int

	// Observer is told what happened, after it happened. Nil means
	// [NopObserver]. See [Observer] for why nothing there can refuse anything.
	Observer      Observer
	OutboundQueue int
	WriteTimeout  time.Duration
	PingInterval  time.Duration
	PongTimeout   time.Duration
}

ServerConfig is what a Server is built from.

Everything without a default is required, and NewServer refuses a config missing one rather than filling in something safe-looking: a server with no ConnectPolicy would accept every socket, and a nil policy is exactly the mistake RULE 17 exists to make impossible.

type Sink 

type Sink interface {
	// Send writes one message. The message is a complete protocol frame,
	// already encoded.
	Send(ctx context.Context, message []byte) error
	// Terminate closes the socket. It is Reverb's Connection::terminate.
	Terminate(ctx context.Context) error
}

Sink is the write half of a socket: where a Connection's bytes go.

It is an interface so that this package declares no transport. The ws subpackage implements it over a real socket, and a test implements it over a slice.

It is named for what it is in Go terms rather than after Reverb's names for the same two operations, which are methods on the Connection class itself -- send() and terminate(). Splitting them out is what lets Connection be a struct with a constructor that cannot be bypassed: an interface cannot have one.

type SocketID 

type SocketID string

SocketID identifies one open socket, and is the id the protocol calls socket_id.

It is handed to the client in EventConnectionEstablished and comes back in a publish as the sender, so Channel.Broadcast can skip it. Pusher's clients expect the shape "<digits>.<digits>"; nothing here reads it, and the package that mints one keeps to that shape because the clients print it.

type SubscribeRequest 

type SubscribeRequest struct {
	// Channel is the name the client asked for, exactly as it sent it and with
	// no tenant in it. The caller turns it into a [ChannelName] with
	// [NewChannelName], which is where the tenant comes from the Grant.
	Channel string `json:"channel"`

	// Auth is the client's Pusher authentication signature, and this package
	// does not verify it. It is carried to [Subscription.Auth], where a
	// [SubscriptionPolicy] may.
	//
	// In Pusher, and so in Reverb, this string is the whole authorization: the
	// application signs "socket_id:channel" with the shared secret over a
	// separate HTTP round trip, and the socket server checks the HMAC. That is
	// what nothing here does, and the reason still stands where it was written.
	// In a mounted application the subject on the Grant came through the
	// framework's front door, and a signature that could also allow a
	// subscription would be exactly the second mechanism RULE 9 forbids --
	// particularly this one, which allows a channel without ever naming a
	// tenant.
	//
	// What is different is who is asked. The signature travels as evidence
	// rather than as authority: it allows nothing by itself, it builds no
	// tenant and no Grant, and a policy that ignores it refuses precisely what
	// it refused before. In a process that authenticates nobody there is no
	// first mechanism for it to be a second one to, and there it is the only
	// evidence about a browser there is -- which is why cmd/joaju could serve
	// no private or presence channel while [Subscription] had no field for it.
	Auth string `json:"auth,omitempty"`

	// ChannelData is the presence information the client offered, and it is a
	// claim rather than a fact -- see [Subscription.Member]. Read it with
	// [SubscribeRequest.Member].
	//
	// It reaches a policy undecoded as well, as [Subscription.ChannelData],
	// because it is the third part of what [SubscribeRequest.Auth] was computed
	// over and re-encoding it would change the bytes and so the hash.
	ChannelData json.RawMessage `json:"channel_data,omitempty"`
}

SubscribeRequest is the data of a EventSubscribe frame: what a client asked to listen to, and what it offered in support.

func (SubscribeRequest) Member 

func (r SubscribeRequest) Member() (Member, error)

Member reads the presence data the client offered, and is the zero Member when it offered none.

It accepts a user_id that arrived as a number as well as one that arrived as a string, because the clients send both and Reverb casts. The number is kept as it was written rather than parsed and reprinted, so 007 stays 007.

What it returns is a claim. A client sends its own channel_data, so the user_id in it is the one the client typed: a SubscriptionPolicy compares it against the subject on the Grant, and one that does not is a policy that lets a subscriber join a presence channel as somebody else.

type Subscriber 

type Subscriber struct {
	// Conn is the socket.
	Conn *Connection
	// Member is the presence data, and is the zero value on a channel that is
	// not a presence channel.
	Member Member
}

Subscriber is one Connection's membership of one Channel.

It is Reverb's ChannelConnection, which exists there for the same reason: presence data belongs to the pair, not to the socket. One socket may be in a presence channel as one member and in another as a different one.

type Subscription 

type Subscription struct {
	// Channel is the channel asked for, tenant already in it.
	Channel ChannelName
	// Member is the presence data the client offered, and is the zero value
	// off a presence channel. It came from the client, so a policy that lets a
	// subscriber name their own Member.UserID is a policy that lets them
	// impersonate one -- compare it against the subject.
	Member Member
	// Socket is which of the subject's connections is asking.
	Socket SocketID
	// Auth is the Pusher subscription signature the client offered --
	// "<app key>:<hex HMAC-SHA256>" -- verbatim, or empty when it offered none.
	//
	// It is evidence and not authority, which is the whole of why it may be
	// here. Nothing in this package reads it, nothing derives a tenant from it,
	// and holding it allows no channel: the decision is still the
	// [SubscriptionPolicy]'s and the Grant it issues is still the only way to a
	// [Channel]. It is carried for the reason [Handshake.Origin] is carried --
	// a policy cannot weigh evidence it is never shown, and this is the evidence
	// the Pusher protocol puts on the wire about a browser.
	//
	// A policy that checks it recomputes the HMAC of
	// "<socket id>:<channel>:<channel data>" -- the last part only where there
	// is one -- under the app secret and compares in constant time. The channel
	// in that string is [ChannelName.Requested] and never [ChannelName.String]:
	// the client signed the name it sent, and the tenant was never its to see.
	// The socket id is what makes the signature one connection's; a policy that
	// leaves it out accepts a signature anybody who saw it can replay.
	//
	// A policy for a mounted application ignores it, and should. There the
	// subject on the Grant arrived through the framework's front door, so a
	// signature that could also allow a subscription would be the second
	// mechanism RULE 9 forbids. Where there is no front door -- cmd/joaju is
	// that process -- there is no first mechanism for it to be a second one to.
	Auth string
	// ChannelData is the presence data exactly as it arrived, and it is what
	// the third part of the signed string above is.
	//
	// It is [Subscription.Member] before it was read: the same bytes, undecoded.
	// Both are here because a policy compares fields and a signature covers
	// bytes -- re-encoding Member would produce a different JSON text, and so a
	// different hash, for the same claim.
	ChannelData json.RawMessage
}

Subscription is what a SubscriptionPolicy decides about: one client asking to listen on one channel.

The Channel is a ChannelName, which means the tenant was settled before the policy ran. A policy is asked whether this subject may hear this channel; it is never asked whose channel it is, because that was not the client's to say.

type SubscriptionPolicy 

type SubscriptionPolicy = auth.Policy[Subscription]

SubscriptionPolicy decides whether a client may listen on a channel. Its Grant is issued for broadcasting.ChannelJoin, which hesape/broadcasting already declares for the SSE fanout's half of the same question.

It runs for every channel, on every subscription, including a resubscription after a reconnect. Subscribing is a read, and RULE 17 has no exception for reads: a channel a policy never saw is a channel whose contents nobody decided anyone could have.

type TenantCount 

type TenantCount struct {
	// Connections is how many are open right now. It goes down.
	Connections int64
	// Channels is how many exist right now. It goes down.
	Channels int64
	// Received and Sent are totals since the process started. They only go up,
	// so a reader can subtract two readings and get a rate.
	Received int64
	Sent     int64
}

TenantCount is one tenant's numbers.

Per tenant and never global, for the reason RULE 14 gives everywhere else: a single number tells one customer how busy another one is. The rate of a competitor's traffic is a business fact, and an operator reading a dashboard is not a reason to publish it.

Source Files

View all Source files

Directories

Path Synopsis
cmd
joaju command
Command joaju runs the WebSocket server as a process.
 Command joaju runs the WebSocket server as a process.
redis module
ws
Package ws is the WebSocket protocol of RFC 6455, written for this server.
 Package ws is the WebSocket protocol of RFC 6455, written for this server.
internal/autobahn/echo command
Command echo is the server the Autobahn TestSuite fuzzing client attacks.
 Command echo is the server the Autobahn TestSuite fuzzing client attacks.
internal/autobahn/report command
Command report turns an Autobahn TestSuite run into a verdict.
 Command report turns an Autobahn TestSuite run into a verdict.

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.