protocol

package
v0.0.0-...-668d940 Latest Latest
Warning

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

 Go to latest Published: Aug 19, 2026 License: Apache-2.0 Imports: 15 Imported by: 0

Documentation

Index

Constants

View Source 
const (
	// FlagHasPresence is set on ATTACHED when the channel has a
	// non-empty presence set, signalling the SDK that a SYNC will
	// follow (DESIGN.md §12.4).
	FlagHasPresence int64 = 1 << 0

	// FlagHasBacklog is set on ATTACHED when the attach replayed any
	// backlog before live delivery — a rewind or a resume gap-fill
	// (DESIGN.md §4.2, §4.3). SDKs surface it as
	// ChannelStateChange.hasBacklog (RTL2i). Cleared on a fresh attach
	// that replayed nothing.
	FlagHasBacklog int64 = 1 << 1

	// FlagResumed indicates the channel state was resumed from the
	// client's supplied channelSerial: the gap between the client's
	// cursor and the live tail was replayed in full. Cleared when the
	// server could not satisfy the resume in full (e.g. cap exceeded,
	// retention aged-out) — clients should treat the absence of this
	// flag as a discontinuity.
	FlagResumed int64 = 1 << 2

	// FlagAttachResume is set by the SDK on an ATTACH that continues an
	// existing attachment (a non-clean attach, RTL4j) — e.g. re-attaching
	// after a connection resume. Like a supplied channelSerial it marks the
	// attach as a resume, which suppresses rewind (DESIGN.md §4.3): a
	// continuation must not replay history the client has already seen.
	FlagAttachResume int64 = 1 << 5

	// Channel-mode flags (DESIGN.md §4.2). ATTACH.flags selects the
	// requested modes; ATTACHED.flags carries the effective set.
	FlagPresence          int64 = 1 << 16 // enter/update/leave presence
	FlagPublish           int64 = 1 << 17 // publish MESSAGE
	FlagSubscribe         int64 = 1 << 18 // receive MESSAGE
	FlagPresenceSubscribe int64 = 1 << 19 // receive PRESENCE + presence sync
	// Annotation modes (DESIGN.md §14.3). ANNOTATION_PUBLISH is part of the
	// no-mode-bits ATTACH default set (§4.2, matching the reference's
	// MODE_DEFAULT); ANNOTATION_SUBSCRIBE is opt-in — raw annotation
	// delivery must be requested explicitly. Bits match Ably's wire
	// constants (1<<20 MAY_HAVE_PRESENCE is internal-only and unused here).
	FlagAnnotationPublish   int64 = 1 << 21 // publish ANNOTATION
	FlagAnnotationSubscribe int64 = 1 << 22 // receive raw ANNOTATION frames
)

Flags carried on ATTACH / ATTACHED. The low bits are server-set status flags; the high bits (1<<16 and up) are the channel-mode bitfield, matching Ably's wire constants (DESIGN.md §4.2, §12).

View Source 
const DeltaAppend = "delta-append"

DeltaAppend is the Alt key under which an append's incremental delta message rides on the full aggregated version (DESIGN.md §13.3). Matches Ably's reference constant.

Variables

View Source 
var AnnotationAggregations = map[string]bool{
	"distinct.v1": true,
	"unique.v1":   true,
	"multiple.v1": true,
	"flag.v1":     true,
	"total.v1":    true,
}

AnnotationAggregations is the set of v1 summarisation methods an annotation type's `<aggregation>` suffix must name (DESIGN.md §14.1), pinned to Ably's reference list.

View Source 
var AnnotationAnonymousAggregations = map[string]bool{
	"multiple.v1": true,
	"total.v1":    true,
}

AnnotationAnonymousAggregations is the subset of aggregation methods an unidentified (anonymous) client may publish (DESIGN.md §14.1), mirroring Ably's AllowedAnonymousAggregationMethods: multiple.v1 (reactions that mix identified and anonymous) and total.v1 (anonymous tally).

Functions

func DataFromExternal 

func DataFromExternal(v any, inEncoding string) (data any, outEncoding string, err error)

DataFromExternal converts an external value (an `any` decoded from JSON or msgpack) into the server's canonical Data representation — a Go string or a []byte — together with the canonical encoding.

This is the `Data any` equivalent of the reference's ablyrpc.DataFromExternal, itself the Go equivalent of node's Data.normalise / Data.fromEncoded. The rules are identical:

  • a string whose outer encoding is base64 is decoded to bytes with the base64 suffix stripped;
  • raw bytes are kept as bytes;
  • objects and arrays are JSON-encoded to a string and the encoding is deliberately overridden to "json" (avoiding a redundant "json/json" if the caller already specified it on an unenveloped publish);
  • anything else (a bool or a number) is stringified — node uses String(data); "%v" is the close Go equivalent.

A nil value returns nil data (no payload) with the encoding unchanged.

func Marshal 

func Marshal(m *ProtocolMessage, f Format) ([]byte, error)

Marshal encodes a ProtocolMessage in the given format.

func MsgpackToJSON 

func MsgpackToJSON(dec *msgpack.Decoder) (string, error)

MsgpackToJSON returns the JSON encoding of the next msgpack value from the given decoder using as few allocations as possible.

func ParseAnnotationType 

func ParseAnnotationType(typ string) (name, aggregation string, ok bool)

ParseAnnotationType splits an annotation type into its `<name>` and `<aggregation>` parts (DESIGN.md §14.1). The aggregation is empty for an unaggregated type. ok is false for a malformed type (empty name, or more than one ':' separator).

func Unmarshal 

func Unmarshal(data []byte, f Format, m *ProtocolMessage) error

Unmarshal decodes data into m using the given format.

Types

type Action 

type Action int8
const (
	ActionHeartbeat    Action = 0
	ActionAck          Action = 1
	ActionNack         Action = 2
	ActionConnect      Action = 3
	ActionConnected    Action = 4
	ActionDisconnect   Action = 5
	ActionDisconnected Action = 6
	ActionClose        Action = 7
	ActionClosed       Action = 8
	ActionError        Action = 9
	ActionAttach       Action = 10
	ActionAttached     Action = 11
	ActionDetach       Action = 12
	ActionDetached     Action = 13
	ActionPresence     Action = 14
	ActionMessage      Action = 15
	ActionSync         Action = 16
	ActionAuth         Action = 17
	// ActionAnnotation carries annotation publishes and deliveries
	// (DESIGN.md §14). Pinned to Ably's wire value 21.
	ActionAnnotation Action = 21
)

func (Action) String 

func (a Action) String() string

type Aggregation 

type Aggregation struct {
	// Method is the aggregation suffix (distinct.v1 / unique.v1 /
	// multiple.v1 / flag.v1 / total.v1) that selects the populated field.
	Method string
	// Values backs distinct.v1 and unique.v1: per-value sets of the
	// distinct client ids that annotated with that value.
	Values map[string]*ClientIDList
	// Counts backs multiple.v1: per-value tallies of client id counts.
	Counts map[string]*ClientIDCounts
	// Flag backs flag.v1: the set of clients that set the flag.
	Flag *ClientIDList
	// Total backs total.v1: an anonymous tally.
	Total *TotalAggregation
}

Aggregation is one entry in a Summary: the rolled-up annotations of a single type, shaped by that type's aggregation method. Exactly one of the typed fields is populated, selected by Method (the aggregation suffix of the annotation type, e.g. "distinct.v1").

type Annotation 

type Annotation struct {
	ID            string           `json:"id,omitempty"            msgpack:"id,omitempty"`
	Serial        string           `json:"serial,omitempty"        msgpack:"serial,omitempty"`
	Action        AnnotationAction `json:"action"                  msgpack:"action"`
	ClientID      string           `json:"clientId,omitempty"      msgpack:"clientId,omitempty"`
	ConnectionID  string           `json:"connectionId,omitempty"  msgpack:"connectionId,omitempty"`
	Type          string           `json:"type,omitempty"          msgpack:"type,omitempty"`
	Name          string           `json:"name,omitempty"          msgpack:"name,omitempty"`
	MessageSerial string           `json:"messageSerial,omitempty" msgpack:"messageSerial,omitempty"`
	Count         int              `json:"count,omitempty"         msgpack:"count,omitempty"`
	Data          any              `json:"data,omitempty"          msgpack:"data,omitempty"`
	Encoding      string           `json:"encoding,omitempty"      msgpack:"encoding,omitempty"`
	// Extras is a free-form JSON object attached to the annotation,
	// preserved verbatim through publish, fan-out and storage (DESIGN.md
	// §8, §14.1). Carried as a map for the same round-trip reason as
	// Message.Extras; the reference's ablyrpc.Annotation carries it too.
	Extras    map[string]any `json:"extras,omitempty"        msgpack:"extras,omitempty"`
	Timestamp int64          `json:"timestamp,omitempty"     msgpack:"timestamp,omitempty"`
	// Summary is the post-fold summary snapshot of this annotation's target
	// message, stamped by StoreAnnotation for the MESSAGE/summary (action 4)
	// delivery frame (DESIGN.md §14.2, §14.3). It is server-internal: never
	// encoded to a client and never part of the persisted annotation
	// payload (excluded from both json and msgpack). On the single-process
	// backends it rides the in-memory cm to the appender; on the Postgres
	// cluster path it is persisted in a dedicated channel_messages.summary
	// column and reconstructed here on the LISTEN load, so a node that never
	// witnessed earlier annotations emits the identical summary.
	Summary Summary `json:"-" msgpack:"-"`
}

Annotation is one annotation attached to an existing message (DESIGN.md §14.1). It rides the channel stream as a third cm kind (kind = annotation) carried in ChannelMessage.Annotations. Field names and enum values are pinned to Ably's wire shape.

ID and Serial carry the same split as on Message (DESIGN.md §8):

  • ID is client-supplied and optional; it carries idempotency intent.
  • Serial is server-assigned on publish in the form `<channelSerial>:<idx>`.

MessageSerial is the TARGET message's identity serial — the message the annotation is attached to (client-supplied). Type has the form `<name>:<aggregation>` (see AnnotationAggregations). Action carries no omitempty: an annotation.create must emit action=0 on the wire to match Ably's SDKs.

func (*Annotation) DecodeMsgpack 

func (a *Annotation) DecodeMsgpack(dec *msgpack.Decoder) error

DecodeMsgpack decodes an Annotation, normalising the Data field via the same two-pass scheme as Message.DecodeMsgpack.

func (*Annotation) EncodeMsgpack 

func (a *Annotation) EncodeMsgpack(enc *msgpack.Encoder) error

EncodeMsgpack encodes the Annotation as a map mirroring its struct tags, forcing Data whenever non-nil (see the file comment). Summary is server-internal (msgpack:"-") and is not emitted.

func (*Annotation) MarshalJSON 

func (a *Annotation) MarshalJSON() ([]byte, error)

MarshalJSON encodes an Annotation for a JSON transport, applying the binary→base64 egress rule to its payload (see Message.MarshalJSON).

func (*Annotation) UnmarshalJSON 

func (a *Annotation) UnmarshalJSON(b []byte) error

UnmarshalJSON decodes a JSON Annotation, normalising its payload to the canonical Data representation via DataFromExternal (see data.go).

func (*Annotation) Validate 

func (a *Annotation) Validate() *AnnotationError

Validate applies the §14.1 annotation validation, mirroring Ably's ablyrpc.Annotation.Validate: a missing MessageSerial or Type is rejected (40000), a malformed type or unknown aggregation is rejected, and an anonymous (empty ClientID) publish is confined to the anonymous-allowed aggregation methods. For multiple.v1 a zero Count defaults to 1. It returns nil when the annotation is valid, mutating Count where defaulted.

type AnnotationAction 

type AnnotationAction int8

AnnotationAction is the action carried by an Annotation — the annotation-stream analogue of MessageAction/PresenceAction. Values are pinned to Ably's Annotation.Action wire enum (DESIGN.md §14.1). 

const (
	// AnnotationCreate attaches a new annotation to an existing message.
	AnnotationCreate AnnotationAction = 0
	// AnnotationDelete removes the caller's own contribution(s) for the
	// annotation type, per the aggregation method's fold.
	AnnotationDelete AnnotationAction = 1
)

func (AnnotationAction) String 

func (a AnnotationAction) String() string

type AnnotationError 

type AnnotationError struct {
	Message    string
	Code       int
	StatusCode int
}

AnnotationError is a validation failure of an inbound annotation. Its Code/StatusCode mirror Ably's ablyrpc.Annotation.Validate error shapes (all 40000 / 400); callers surface it as a NACK (WS) or 400 (REST).

func (*AnnotationError) Error 

func (e *AnnotationError) Error() string

func (*AnnotationError) ErrorInfo 

func (e *AnnotationError) ErrorInfo() *ErrorInfo

ErrorInfo renders the AnnotationError as the wire ErrorInfo carried on a NACK / REST error body.

type AuthDetails 

type AuthDetails struct {
	AccessToken string `json:"accessToken,omitempty" msgpack:"accessToken,omitempty"`
}

AuthDetails carries the token supplied on an inband AUTH ProtocolMessage (Ably AD2). AccessToken is the JWT the client presents to re-authenticate an established connection (DESIGN.md §3).

type ChannelMessage 

type ChannelMessage struct {
	ID            string             `json:"id,omitempty"            msgpack:"id,omitempty"`
	ChannelSerial string             `json:"channelSerial,omitempty" msgpack:"channelSerial,omitempty"`
	Messages      []*Message         `json:"messages,omitempty"      msgpack:"messages,omitempty"`
	Presence      []*PresenceMessage `json:"presence,omitempty"      msgpack:"presence,omitempty"`
	Annotations   []*Annotation      `json:"annotations,omitempty"   msgpack:"annotations,omitempty"`
}

ChannelMessage is one atomic publish on a channel: a server-assigned channelSerial (the discrete attach/resume point in the channel's stream) plus the one or more Messages published in that batch.

One publish (REST request or inbound MESSAGE frame) maps to exactly one ChannelMessage; subscribers receive ChannelMessages as the atomic delivery unit (one outbound MESSAGE frame per ChannelMessage). Storage persists ChannelMessages keyed by ChannelSerial. A ChannelMessage carries exactly one of Messages (a data publish), Presence (a presence publish), or Annotations (an annotation publish, DESIGN.md §14.1) — the three ride one ordered stream distinguished by which slice is populated (DESIGN.md §12.1).

ID is the batch identifier for a message publish (DESIGN.md §8): the client-supplied idempotency key, or a server-generated 8-char base64 string when none is supplied. Each contained Message.ID is stamped "<ID>:<idx>" (storage.StampMessageIDs), so the batch id is the idempotency key indexed by storage.

Empty for a presence cm: presence has no batch key because presence identity is per-member, not per-batch. Each contained PresenceMessage.ID is instead stamped "<connectionId>:<msgSerial>: <index>" at realtime publish, and storage-level idempotency keys off those contained ids — so no cm-level id is ever minted for presence.

type ChannelParams 

type ChannelParams map[string]string

ChannelParams is the ATTACH/ATTACHED channel-params map. On the wire it is a string→string map, but SDKs may send a value as a non-string — ably-js's `channels.get(name, {params: {rewind: 1}})` puts the number 1 in the map, so the wire carries `{"rewind": 1}` (an integer value). A plain map[string]string decode rejects that and drops the whole frame, so the server would silently never answer the ATTACH and the channel would hang in ATTACHING. To match the reference server's tolerant paramsFromMap, the custom decoders below coerce every value to its string form on ingress. Encoding stays the default map[string]string marshalling (values are always strings by the time we echo them).

func (*ChannelParams) DecodeMsgpack 

func (p *ChannelParams) DecodeMsgpack(dec *msgpack.Decoder) error

DecodeMsgpack mirrors UnmarshalJSON for the msgpack transport.

func (*ChannelParams) UnmarshalJSON 

func (p *ChannelParams) UnmarshalJSON(b []byte) error

UnmarshalJSON decodes the params object leniently, coercing non-string values to strings (see ChannelParams).

type ClientIDCounts 

type ClientIDCounts struct {
	Total             int            `json:"total"                    msgpack:"total"`
	ClientIDs         map[string]int `json:"clientIds,omitempty"      msgpack:"clientIds,omitempty"`
	TotalUnidentified int            `json:"totalUnidentified"        msgpack:"totalUnidentified"`
	Clipped           bool           `json:"clipped,omitempty"        msgpack:"clipped,omitempty"`
	TotalClientIDs    int            `json:"totalClientIds,omitempty" msgpack:"totalClientIds,omitempty"`
}

ClientIDCounts is the multiple.v1 per-value tally: a sum of counts from all annotations (Total), the per-identified-client sums (ClientIDs), the sum from anonymous clients (TotalUnidentified), and the number of distinct identified clients (TotalClientIDs). Total and TotalUnidentified carry no omitempty so a value with only anonymous contributions still reports totalUnidentified, matching Ably.

type ClientIDList 

type ClientIDList struct {
	Total     int      `json:"total"               msgpack:"total"`
	ClientIDs []string `json:"clientIds,omitempty" msgpack:"clientIds,omitempty"`
	Clipped   bool     `json:"clipped,omitempty"   msgpack:"clipped,omitempty"`
}

ClientIDList is a de-duplicated, insertion-ordered set of the client ids that contributed to a distinct/unique value or set a flag, plus the total count (DESIGN.md §14.2). Total equals len(ClientIDs) for the folds implemented here (no unidentified contributions reach a ClientIDList).

type ConnectionDetails 

type ConnectionDetails struct {
	// ClientID is the connection's resolved clientId (§3.2): a concrete
	// value, "*" for a wildcard bearer, or omitted for an anonymous
	// connection.
	ClientID string `json:"clientId,omitempty" msgpack:"clientId,omitempty"`
	// ConnectionKey is the opaque key an SDK would resume with. Since
	// connection-state resume is a non-goal (§1, §11), it is the
	// process-local connectionId and is not recoverable.
	ConnectionKey string `json:"connectionKey,omitempty" msgpack:"connectionKey,omitempty"`
	// MaxMessageSize is the largest permitted payload of a single publish
	// (bytes). SDKs reject oversize publishes client-side.
	MaxMessageSize int64 `json:"maxMessageSize,omitempty" msgpack:"maxMessageSize,omitempty"`
	// MaxFrameSize is the largest permitted WebSocket frame / POST body
	// (bytes).
	MaxFrameSize int64 `json:"maxFrameSize,omitempty" msgpack:"maxFrameSize,omitempty"`
	// MaxInboundRate is the advisory ceiling on messages per second from
	// this connection.
	MaxInboundRate int64 `json:"maxInboundRate,omitempty" msgpack:"maxInboundRate,omitempty"`
	// ConnectionStateTTLMs is how long (ms) an SDK should treat the
	// connection state as recoverable after an abrupt disconnect (DF1a).
	ConnectionStateTTLMs int64 `json:"connectionStateTtl,omitempty" msgpack:"connectionStateTtl,omitempty"`
	// MaxIdleIntervalMs is the maximum time (ms) the server will leave the
	// server→client direction idle before sending a HEARTBEAT; it equals
	// the server heartbeat cadence (CD2h).
	MaxIdleIntervalMs int64 `json:"maxIdleInterval,omitempty" msgpack:"maxIdleInterval,omitempty"`
}

ConnectionDetails is sent inside the CONNECTED ProtocolMessage and tells the SDK its resolved identity plus the limits/params it should adopt for this connection (Ably's CD2* / DESIGN.md §2.1, §8). Field names and wire tags match the SDKs' connectionDetails shape so they decode it unchanged. The two duration fields are whole milliseconds on the wire, as SDKs encode durations as integer milliseconds.

type ErrorInfo 

type ErrorInfo struct {
	Message    string `json:"message,omitempty"    msgpack:"message,omitempty"`
	Code       int    `json:"code,omitempty"       msgpack:"code,omitempty"`
	StatusCode int    `json:"statusCode,omitempty" msgpack:"statusCode,omitempty"`
	HRef       string `json:"href,omitempty"       msgpack:"href,omitempty"`
}

ErrorInfo describes an error in Ably's standard wire form, attached to a ProtocolMessage when the server needs to convey a non-fatal problem to the client (e.g. a resume that could not fully replay).

type Format 

type Format int

Format is the wire encoding of a ProtocolMessage frame. 

const (
	FormatJSON Format = iota
	FormatMsgpack
)

func FormatFromQuery 

func FormatFromQuery(v string) (Format, error)

FormatFromQuery returns the format selected by the `format` query parameter on a WebSocket upgrade request. Empty defaults to JSON.

func (Format) String 

func (f Format) String() string

String returns the format name for diagnostics.

type Message 

type Message struct {
	ID           string        `json:"id,omitempty"           msgpack:"id,omitempty"`
	Serial       string        `json:"serial,omitempty"       msgpack:"serial,omitempty"`
	Action       MessageAction `json:"action"                 msgpack:"action"`
	ClientID     string        `json:"clientId,omitempty"     msgpack:"clientId,omitempty"`
	ConnectionID string        `json:"connectionId,omitempty" msgpack:"connectionId,omitempty"`
	// ConnectionKey is an inbound-only field on a REST publish: it names a
	// live realtime connection to publish on behalf of (DESIGN.md §8, §13).
	// The REST publish path resolves it to that connection's connectionId
	// (and clientId), stamps those, then clears it — connectionKey is never
	// persisted or delivered (mirrors the reference, which resolves it to a
	// derived connectionId and delivers only that).
	ConnectionKey string `json:"connectionKey,omitempty" msgpack:"connectionKey,omitempty"`
	Name          string `json:"name,omitempty"         msgpack:"name,omitempty"`
	Data          any    `json:"data,omitempty"         msgpack:"data,omitempty"`
	Encoding      string `json:"encoding,omitempty"     msgpack:"encoding,omitempty"`
	// Extras is a free-form JSON object the client attaches to a message
	// (headers, push metadata, and — for the AI Transport SDK — extras.ai).
	// The server treats it as opaque and preserves it verbatim through
	// publish, fan-out, storage, history and mutations (DESIGN.md §8, §13.2).
	// It is a map rather than a raw-JSON carrier so it round-trips both the
	// JSON and msgpack wire/storage encodings — json.RawMessage would not
	// survive the msgpack storage payload. This mirrors the reference's
	// extras object (ablyrpc holds it as structpb.Struct, wire-encoded as a
	// plain object under either format).
	Extras    map[string]any  `json:"extras,omitempty"       msgpack:"extras,omitempty"`
	Timestamp int64           `json:"timestamp,omitempty"    msgpack:"timestamp,omitempty"`
	Version   *MessageVersion `json:"version,omitempty"      msgpack:"version,omitempty"`
	// Summary is the fold of this message's annotations (DESIGN.md §14.2),
	// keyed by annotation type. It rides the latest-version projection so
	// message reads (GET .../messages, .../messages/{serial}, history)
	// carry the current summary, and it is the payload of the outbound
	// MESSAGE/summary (action 4) delivery frame.
	Summary Summary `json:"summary,omitempty" msgpack:"summary,omitempty"`
	// Alt carries alternative in-band representations of this message,
	// keyed by role (DESIGN.md §13.3). Its sole current use is the
	// append delta: an append is persisted and fanned out as a full
	// action=update Message whose Data is the rolled-up aggregate, with
	// Alt[DeltaAppend] holding the incremental append (action=append,
	// just the new data) the server hands a caught-up subscriber instead
	// of the full version. It is a server-internal carrier — persisted so
	// resume and cross-node fan-out can still choose delta vs full — and
	// is resolved away before a frame reaches a client, so it is excluded
	// from the client-facing JSON encoding.
	Alt map[string]*Message `json:"-" msgpack:"alt,omitempty"`
}

Message is a published message payload — one Message within a ChannelMessage atomic publish (see DESIGN.md §8).

ID and Serial are distinct identifiers:

  • ID is client-supplied and optional; it carries idempotency intent so the server can reject duplicate publishes within the retention window.
  • Serial is the stable message IDENTITY (DESIGN.md §8, §13.1): it names the message, unchanged across every version. For a create it is server-assigned in the form `<channelSerial>:<idx>` (the containing ChannelMessage's serial plus this Message's position in the batch); an update/delete/append repeats the target's Serial and carries a fresh Version.

Action and Version split message-vs-version identity (DESIGN.md §13.1). Action distinguishes a create from a mutation; Version names a single version of the message. For a create, Version.Serial == Serial; each subsequent mutation lands at a new channelSerial and gets a fresh, strictly-greater Version.Serial. Action carries no omitempty: a create must emit action=0 on the wire to match Ably's SDKs.

func (*Message) DecodeMsgpack 

func (m *Message) DecodeMsgpack(dec *msgpack.Decoder) error

DecodeMsgpack decodes a Message. It captures the raw msgpack value and decodes it twice: once via the method-less shadow type (sidestepping the custom-encoder recursion) for every field with the library's full type fidelity, and once into a lone msgpackData field to normalise the payload. (The two passes avoid a struct with two `data` keys, which msgpack — unlike encoding/json — rejects rather than shadowing by depth.) The payload is then normalised via DataFromExternal to the canonical Data representation: an outer-base64 string is decoded to raw bytes with the suffix stripped, and a msgpack object/array is transcoded to a JSON string (via MsgpackToJSON) with the encoding set to "json" after the fact. Mirrors the reference's ablyrpc.Message.DecodeMsgpack.

func (*Message) EncodeMsgpack 

func (m *Message) EncodeMsgpack(enc *msgpack.Encoder) error

EncodeMsgpack encodes the Message as a map mirroring its struct tags, forcing Data whenever non-nil (see the file comment).

func (*Message) HasAppendDelta 

func (m *Message) HasAppendDelta() bool

HasAppendDelta reports whether m is an append aggregate — a full version carrying an incremental append in Alt (DESIGN.md §13.3). It is how the delivery path and the version-history collapse distinguish an append from an ordinary update, which are otherwise both action=update.

func (*Message) MarshalJSON 

func (m *Message) MarshalJSON() ([]byte, error)

MarshalJSON encodes a Message for a JSON (text) transport, applying the binary→base64 egress rule (DESIGN.md §8; see data.go): the Data field is emitted via encoding/json's default handling — a []byte payload becomes a base64 string — and "base64" is appended to the emitted encoding so a JSON subscriber decodes the payload back to bytes. String payloads and the canonical (stored) encoding are emitted unchanged. Mirrors the reference's ablyrpc.Message.MarshalJSON, where the base64 suffix is appended at JSON emission and never stored.

func (*Message) UnmarshalJSON 

func (m *Message) UnmarshalJSON(b []byte) error

UnmarshalJSON decodes a JSON Message, normalising its payload to the server's canonical Data representation via DataFromExternal (see data.go): an outer-base64 string is decoded to raw bytes with the base64 suffix stripped, and objects/arrays are canonicalised to a JSON string. Mirrors the reference's ablyrpc.Message.UnmarshalJSON.

type MessageAction 

type MessageAction int8

MessageAction is the action carried by a Message — the data-stream analogue of PresenceAction. Values are pinned to Ably's MessageAction wire enum (the SDKs' constants): a create is the default original publish; update/delete/append are mutations of an existing message (DESIGN.md §13.1); summary (4) is the server-generated annotation summary delivery (DESIGN.md §14.3). The value meta (3) and others Ably defines for object messages are intentionally omitted — this server only models the mutable-message and summary subset. 

const (
	// MessageCreate is an original publish (the default). Every message
	// the publish path produces carries this; it must be emitted on the
	// wire even at its zero value, so Message.Action has no omitempty.
	MessageCreate MessageAction = 0
	// MessageUpdate replaces fields of an existing message with a new
	// version (DESIGN.md §13.2).
	MessageUpdate MessageAction = 1
	// MessageDelete soft-deletes an existing message — a tombstone
	// version (DESIGN.md §13.2).
	MessageDelete MessageAction = 2
	// MessageSummary is a server-generated annotation summary delivered to
	// ordinary SUBSCRIBE attachments (DESIGN.md §14.3): a MESSAGE carrying
	// the target message's unchanged serial and the fold of its
	// annotations. Pinned to Ably's value 4.
	MessageSummary MessageAction = 4
	// MessageAppend concatenates onto an existing message's data
	// (DESIGN.md §13.3). Pinned to Ably's value 5.
	MessageAppend MessageAction = 5
)

func (MessageAction) IsMutation 

func (a MessageAction) IsMutation() bool

IsMutation reports whether a is an update, delete or append — i.e. a mutation of an existing message rather than an original create.

func (MessageAction) String 

func (a MessageAction) String() string

type MessageVersion 

type MessageVersion struct {
	Serial      string         `json:"serial,omitempty"      msgpack:"serial,omitempty"`
	Timestamp   int64          `json:"timestamp,omitempty"   msgpack:"timestamp,omitempty"`
	ClientID    string         `json:"clientId,omitempty"    msgpack:"clientId,omitempty"`
	Description string         `json:"description,omitempty" msgpack:"description,omitempty"`
	Metadata    map[string]any `json:"metadata,omitempty"    msgpack:"metadata,omitempty"`
}

MessageVersion names a single version of a message (DESIGN.md §13.1). Its wire shape matches Ably's version object. Serial is the `<channelSerial>:<idx>` of the publish that produced this version (for a create it equals the message's own Serial; for a mutation it is the mutation publish's position). Timestamp, ClientID, Description and Metadata stamp the operation: ClientID is the operating client (which may differ from the message's creator), and Description/Metadata are optional operator-supplied annotations.

type PresenceAction 

type PresenceAction int8

PresenceAction is the action carried by a PresenceMessage — the presence-stream analogue of a message action. Values match Ably's wire enum (DESIGN.md §12.1). 

const (
	PresenceAbsent  PresenceAction = 0
	PresencePresent PresenceAction = 1
	PresenceEnter   PresenceAction = 2
	PresenceLeave   PresenceAction = 3
	PresenceUpdate  PresenceAction = 4
)

func (PresenceAction) String 

func (a PresenceAction) String() string

type PresenceMessage 

type PresenceMessage struct {
	ID           string         `json:"id,omitempty"           msgpack:"id,omitempty"`
	Serial       string         `json:"serial,omitempty"       msgpack:"serial,omitempty"`
	Action       PresenceAction `json:"action"                 msgpack:"action"`
	ClientID     string         `json:"clientId,omitempty"     msgpack:"clientId,omitempty"`
	ConnectionID string         `json:"connectionId,omitempty" msgpack:"connectionId,omitempty"`
	Data         any            `json:"data,omitempty"         msgpack:"data,omitempty"`
	Encoding     string         `json:"encoding,omitempty"     msgpack:"encoding,omitempty"`
	// Extras is a free-form JSON object attached to the presence message,
	// preserved verbatim through publish, presence sync/fan-out and storage
	// (DESIGN.md §8, §12.1). Carried as a map for the same round-trip reason
	// as Message.Extras.
	Extras    map[string]any `json:"extras,omitempty"       msgpack:"extras,omitempty"`
	Timestamp int64          `json:"timestamp,omitempty"    msgpack:"timestamp,omitempty"`
}

PresenceMessage is a single presence operation — the presence-stream analogue of Message (DESIGN.md §12.1). A member's identity in the presence set is the pair (ConnectionID, ClientID).

ID and Serial (DESIGN.md §8, §12.1):

  • ID is the presence-newness key. A genuine op (published by a live connection) is stamped "<connectionId>:<msgSerial>:<index>" at realtime publish, so SDKs order it by (msgSerial, index); a client may instead supply its own id (idempotency intent). A synthesized event (fixture member, teardown/detach LEAVE) stays id-less and is ordered by Timestamp.
  • Serial is server-assigned on publish in the form `<channelSerial>:<idx>`.

func (*PresenceMessage) DecodeMsgpack 

func (p *PresenceMessage) DecodeMsgpack(dec *msgpack.Decoder) error

DecodeMsgpack decodes a PresenceMessage, normalising the Data field via the same two-pass scheme as Message.DecodeMsgpack.

func (*PresenceMessage) EncodeMsgpack 

func (p *PresenceMessage) EncodeMsgpack(enc *msgpack.Encoder) error

EncodeMsgpack encodes the PresenceMessage as a map mirroring its struct tags, forcing Data whenever non-nil (see the file comment).

func (*PresenceMessage) MarshalJSON 

func (p *PresenceMessage) MarshalJSON() ([]byte, error)

MarshalJSON encodes a PresenceMessage for a JSON transport, applying the binary→base64 egress rule to its payload (see Message.MarshalJSON).

func (*PresenceMessage) UnmarshalJSON 

func (p *PresenceMessage) UnmarshalJSON(b []byte) error

UnmarshalJSON decodes a JSON PresenceMessage, normalising its payload to the canonical Data representation via DataFromExternal (see data.go).

type ProtocolMessage 

type ProtocolMessage struct {
	Action       Action `json:"action"                  msgpack:"action"`
	ID           string `json:"id,omitempty"            msgpack:"id,omitempty"`
	ConnectionID string `json:"connectionId,omitempty"  msgpack:"connectionId,omitempty"`
	// Channel is a pointer so the server can emit an explicit empty channel
	// (channel:"") while frames that carry no channel omit the field. An
	// ERROR frame with the channel field absent is classified connection-
	// fatal by SDKs, but one carrying the field (even "") is routed to the
	// channel (ably-js transport.ts onProtocolMessage) — so an empty-name
	// ATTACH rejection must set Channel to a pointer-to-"" to reach the wire
	// and fail the channel rather than the connection. pointer+omitempty
	// emits the value whenever it is non-nil (including "") and omits it when
	// nil. Read an inbound frame's channel via GetChannel (absent means "").
	Channel       *string `json:"channel,omitempty"       msgpack:"channel,omitempty"`
	ChannelSerial string  `json:"channelSerial,omitempty" msgpack:"channelSerial,omitempty"`
	// MsgSerial is the per-connection publish counter (§8), a pointer so
	// the server can emit msgSerial:0 while every non-publish frame omits
	// the field. pointer+omitempty emits the value whenever it is non-nil
	// (including 0) and omits it when nil, so an ACK/NACK always carries an
	// explicit msgSerial — SDKs read it positionally and compute NaN
	// if it is absent — while HEARTBEAT/CONNECTED/MESSAGE
	// deliveries, which never set it, stay free of a spurious msgSerial:0.
	// Read an inbound frame's serial via GetMsgSerial (absent means 0).
	MsgSerial *int64 `json:"msgSerial,omitempty"     msgpack:"msgSerial,omitempty"`
	Timestamp int64  `json:"timestamp,omitempty"     msgpack:"timestamp,omitempty"`
	Count     int    `json:"count,omitempty"         msgpack:"count,omitempty"`
	// Res carries the per-message publish results back to the publisher
	// on an ACK (Ably's TR4s shape): one entry per message in the ack
	// window, each holding the server-assigned serials. SDKs read it to
	// populate publish/update results (DESIGN.md §8, §13.1).
	Res         []*PublishResult   `json:"res,omitempty"           msgpack:"res,omitempty"`
	Flags       int64              `json:"flags,omitempty"         msgpack:"flags,omitempty"`
	Messages    []*Message         `json:"messages,omitempty"      msgpack:"messages,omitempty"`
	Presence    []*PresenceMessage `json:"presence,omitempty"      msgpack:"presence,omitempty"`
	Annotations []*Annotation      `json:"annotations,omitempty"   msgpack:"annotations,omitempty"`
	Error       *ErrorInfo         `json:"error,omitempty"         msgpack:"error,omitempty"`
	Params      ChannelParams      `json:"params,omitempty"        msgpack:"params,omitempty"`
	// ConnectionDetails carries the resolved identity and connection
	// limits on the CONNECTED frame (DESIGN.md §2.1, §8).
	ConnectionDetails *ConnectionDetails `json:"connectionDetails,omitempty" msgpack:"connectionDetails,omitempty"`
	// Auth carries a fresh token on an inbound AUTH frame for inband
	// re-authentication (DESIGN.md §2.1, §3); field name/tags match
	// the SDKs' authDetails shape so they encode it unchanged.
	Auth *AuthDetails `json:"auth,omitempty" msgpack:"auth,omitempty"`
}

ProtocolMessage is one frame on the realtime WebSocket connection.

func (*ProtocolMessage) GetChannel 

func (m *ProtocolMessage) GetChannel() string

GetChannel returns the frame's channel, treating an absent (nil) Channel as "". Inbound frames omit the field when there is no channel; a pointer-to-"" is a deliberately channel-scoped empty name.

func (*ProtocolMessage) GetMsgSerial 

func (m *ProtocolMessage) GetMsgSerial() int64

GetMsgSerial returns the frame's msgSerial, treating an absent (nil) MsgSerial as 0. Inbound publish-like frames address their ACK by this per-connection counter; SDKs — and this server — omit msgSerial:0 on the first publish, so absent-means-0 preserves inbound compatibility (DESIGN.md §8).

type PublishResult 

type PublishResult struct {
	Serials []string `json:"serials,omitempty" msgpack:"serials,omitempty"`
}

PublishResult is one entry in an ACK's Res array (Ably's TR4s): the serials the server assigned to one acknowledged message. For a create it carries the message's Serial; for a mutation, the new version serial.

type Summary 

type Summary map[string]*Aggregation

A Summary is the per-message fold of that message's annotations (DESIGN.md §14.2), keyed by the full annotation type (`<name>:<aggregation>`). Each Aggregation is shaped by the type's aggregation method. It rides the message projection (so reads carry the current summary) and the MESSAGE/summary (action 4) delivery frame.

The wire shape is method-specific and un-wrapped, pinned to Ably's: distinct.v1 / unique.v1 encode as `{value: {total, clientIds:[…]}}`, multiple.v1 as `{value: {total, clientIds:{id:count}, totalUnidentified}}`, flag.v1 as a single `{total, clientIds:[…]}`, total.v1 as `{total}`. The Summary map itself carries the codec (Summary.MarshalJSON / EncodeMsgpack and their inverses) because only it holds the type key that names the method — an Aggregation cannot decode itself.

func (Summary) Apply 

func (s Summary) Apply(a *Annotation) Summary

Apply folds one annotation into the summary and returns the result. It treats the receiver as immutable — the fold runs against a clone — so the pre-fold summary (already stamped on an earlier annotation cm) is never disturbed. The annotation type names the method; an unaggregated or unknown-method type is ignored (returns the summary unchanged). An aggregation that folds down to nothing is dropped from the map.

func (Summary) Clone 

func (s Summary) Clone() Summary

Clone returns a deep copy of the summary. StoreAnnotation stamps a clone onto each annotation cm so its delivered snapshot is independent of later folds in the same batch (DESIGN.md §14.2).

func (*Summary) DecodeMsgpack 

func (s *Summary) DecodeMsgpack(dec *msgpack.Decoder) error

DecodeMsgpack decodes the msgpack summary map, dispatching per entry on the type's aggregation-method suffix.

func (Summary) EncodeMsgpack 

func (s Summary) EncodeMsgpack(enc *msgpack.Encoder) error

EncodeMsgpack encodes the summary as a msgpack map of the same shape.

func (Summary) MarshalJSON 

func (s Summary) MarshalJSON() ([]byte, error)

MarshalJSON encodes the summary as {type: <method-shaped object>}.

func (*Summary) UnmarshalJSON 

func (s *Summary) UnmarshalJSON(b []byte) error

UnmarshalJSON decodes a {type: <method-shaped object>} map, dispatching each entry on the type's aggregation-method suffix. Unknown methods are skipped (matching Ably's summary decode).

type TotalAggregation 

type TotalAggregation struct {
	Total int `json:"total" msgpack:"total"`
}

TotalAggregation is the total.v1 anonymous tally.

Source Files

View all Source files

Jump to

Keyboard shortcuts

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