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

centrifuge

package module
v0.40.1 Latest Latest
Warning

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

 Go to latest Published: Oct 20, 2026 License: MIT Imports: 54 Imported by: 128

README

codecov.io GoDoc

This library has no v1 release, API may change. Before v1 release patch version updates only have backwards compatible changes and fixes, minor version updates can have backwards-incompatible API changes. Master branch can have unreleased code. Only two last Go minor versions are officially supported by this library.

The Centrifuge library is a general purpose real-time messaging library for Go programming language. Real-time messaging can help create interactive applications where events are delivered to online users with milliseconds delay. Chats apps, live comments, multiplayer games, real-time data visualizations, telemetry, collaborative tools, etc. can all be built on top of Centrifuge library.

The library is built on top of efficient client-server protocol schema and exposes various real-time oriented primitives for a developer. Centrifuge solves problems developers may come across when building complex real-time applications – like scalability to many server nodes, proper persistent connection management and invalidation, subscription multiplexing, fast reconnect with message recovery, WebSocket fallback options (without sticky sessions requirement in distributed scenario). And it all comes with ready to use client SDKs for both web and mobile development. See the full list of highlighs below.

Centrifuge library is used by:

  • Centrifugo - the main product of Centrifugal Labs. Centrifuge was decoupled into separate library from Centrifugo at some point.
  • Grafana - the most popular observability platform. Centrifuge library powers Grafana Live subsystem to stream data to panels. See cool demo of WebSocket telemetry from the Assetto Corsa racing simulator to the Grafana dashboard.

Why using Centrifuge

Centrifuge library provides a lot of top of raw WebSocket transport. Important library highlights:

  • Fast and optimized for low-latency communication with millions of client connections. See test stand with 1 million connections in Kubernetes
  • WebSocket bidirectional transport using JSON or binary Protobuf formats, both based on a strict Protobuf schema. Code generation is used to push both JSON and Protobuf serialization performance to the limits
  • Our own WebSocket emulation layer over HTTP-streaming (JSON + Protobuf) and Eventsource (JSON) without sticky sessions requirement for distributed setup
  • Possibility to use unidirectional transports without using custom Centrifuge SDK library: see examples for GRPC, EventSource(SSE), HTTP-streaming, Unidirectional WebSocket
  • Experimental support for WebSocket over HTTP/2 (RFC 8441), see the examples
  • Built-in horizontal scalability with Redis PUB/SUB, consistent Redis sharding, Redis Sentinel and Redis Cluster support, super-optimized Redis communication layer
  • Effective non-blocking broadcasts towards client connections using individual queues
  • Native authentication over HTTP middleware or custom token-based (ex. JWT)
  • Channel concept to broadcast message to all active subscribers
  • Client-side and server-side channel subscriptions
  • Bidirectional asynchronous message communication, RPC calls, builtin PING/PONG
  • Presence information for channels (show all active clients in a channel)
  • History information for channels (ephemeral streams with size and TTL retention)
  • Join/leave events for channels (aka client goes online/offline)
  • Possibility to register a custom PUB/SUB Broker and PresenceManager implementations
  • Option to register custom Transport, like Centrifugo does with WebTransport
  • Message recovery mechanism for channels to survive PUB/SUB delivery problems, short network disconnects or node restart
  • Cache channels – a way to quickly deliver latest publication from channel history to the client upon subscription
  • Delta compression using Fossil algorithm for publications inside a channel to reduce bandwidth usage
  • Support for client-supplied publication filters to drop unnecessary channel publications on the server side (see the example)
  • Per-client and per-channel batching controls for reduced system calls and better CPU utilization
  • Out-of-the-box observability using Prometheus instrumentation
  • Client SDKs for main application environments all following single behaviour spec (see list of SDKs below).

Real-time SDK

For bidirectional communication between a client and a Centrifuge-based server we have a set of official client real-time SDKs:

These SDKs abstract asynchronous communication complexity from the developer: handle framing, reconnect with backoff, timeouts, multiplex channel subscriptions over single connection, etc.

If you opt for a unidirectional communication then you may leverage Centrifuge possibilities without any specific SDK on client-side - simply by using native browser API or GRPC-generated code. See examples of unidirectional communication over GRPC, EventSource(SSE), HTTP-streaming, WebSocket.

Explore Centrifuge

Installation

go get github.com/centrifugal/centrifuge

Tutorial

Let's take a look on how to build the simplest real-time chat with Centrifuge library. Clients will be able to connect to a server over Websocket, send a message into a channel and this message will be instantly delivered to all active channel subscribers. On a server side we will accept all connections and will work as a simple PUB/SUB proxy without worrying too much about permissions. In this example we will use Centrifuge Javascript client (centrifuge-js) on a frontend.

Start a new Go project and create main.go:

package main

import (
	"log"
	"net/http"

	// Import this library.
	"github.com/centrifugal/centrifuge"
)

// Authentication middleware example. Centrifuge expects Credentials
// with current user ID set. Without provided Credentials client
// connection won't be accepted. Another way to authenticate connection
// is reacting to node.OnConnecting event where you may authenticate
// connection based on a custom token sent by a client in first protocol
// frame. See _examples folder in repo to find real-life auth samples
// (OAuth2, Gin sessions, JWT etc).
func auth(h http.Handler) http.Handler {
	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		ctx := r.Context()
		// Put authentication Credentials into request Context.
		// Since we don't have any session backend here we simply
		// set user ID as empty string. Users with empty ID called
		// anonymous users, in real app you should decide whether
		// anonymous users allowed to connect to your server or not.
		cred := &centrifuge.Credentials{
			UserID: "",
		}
		newCtx := centrifuge.SetCredentials(ctx, cred)
		r = r.WithContext(newCtx)
		h.ServeHTTP(w, r)
	})
}

func main() {
	// Node is the core object in Centrifuge library responsible for
	// many useful things. For example Node allows publishing messages
	// into channels with its Publish method. Here we initialize Node
	// with Config which has reasonable defaults for zero values.
	node, err := centrifuge.New(centrifuge.Config{})
	if err != nil {
		log.Fatal(err)
	}

	// Set ConnectHandler called when client successfully connected to Node.
	// Your code inside a handler must be synchronized since it will be called
	// concurrently from different goroutines (belonging to different client
	// connections). See information about connection life cycle in library readme.
	// This handler should not block – so do minimal work here, set required
	// connection event handlers and return.
	node.OnConnect(func(client *centrifuge.Client) {
		// In our example transport will always be Websocket but it can be different.
		transportName := client.Transport().Name()
		// In our example clients connect with JSON protocol but it can also be Protobuf.
		transportProto := client.Transport().Protocol()
		log.Printf("client connected via %s (%s)", transportName, transportProto)

		// Set SubscribeHandler to react on every channel subscription attempt
		// initiated by a client. Here you can theoretically return an error or
		// disconnect a client from a server if needed. But here we just accept
		// all subscriptions to all channels. In real life you may use a more
		// complex permission check here. The reason why we use callback style
		// inside client event handlers is that it gives a possibility to control
		// operation concurrency to developer and still control order of events.
		client.OnSubscribe(func(e centrifuge.SubscribeEvent, cb centrifuge.SubscribeCallback) {
			log.Printf("client subscribes on channel %s", e.Channel)
			cb(centrifuge.SubscribeReply{}, nil)
		})

		// By default, clients can not publish messages into channels. By setting
		// PublishHandler we tell Centrifuge that publish from a client-side is
		// possible. Now each time client calls publish method this handler will be
		// called and you have a possibility to validate publication request. After
		// returning from this handler Publication will be published to a channel and
		// reach active subscribers with at most once delivery guarantee. In our simple
		// chat app we allow everyone to publish into any channel but in real case
		// you may have more validation.
		client.OnPublish(func(e centrifuge.PublishEvent, cb centrifuge.PublishCallback) {
			log.Printf("client publishes into channel %s: %s", e.Channel, string(e.Data))
			cb(centrifuge.PublishReply{}, nil)
		})

		// Set Disconnect handler to react on client disconnect events.
		client.OnDisconnect(func(e centrifuge.DisconnectEvent) {
			log.Printf("client disconnected")
		})
	})

	// Run node. This method does not block. See also node.Shutdown method
	// to finish application gracefully.
	if err := node.Run(); err != nil {
		log.Fatal(err)
	}

	// Now configure HTTP routes.

	// Serve Websocket connections using WebsocketHandler.
	wsHandler := centrifuge.NewWebsocketHandler(node, centrifuge.WebsocketConfig{})
	http.Handle("/connection/websocket", auth(wsHandler))

	// The second route is for serving index.html file.
	http.Handle("/", http.FileServer(http.Dir("./")))

	log.Printf("Starting server, visit http://localhost:8000")
	if err := http.ListenAndServe(":8000", nil); err != nil {
		log.Fatal(err)
	}
}

Also create file index.html near main.go with content:

<!DOCTYPE html>
<html lang="en">
    <head>
        <meta charset="utf-8">
        <script type="text/javascript" src="https://unpkg.com/centrifuge@5.8.0/dist/centrifuge.js"></script>
        <title>Centrifuge chat example</title>
    </head>
    <body>
        <input type="text" id="input" />
        <script type="text/javascript">
            function drawText(text) {
                const div = document.createElement('div');
                div.innerHTML = text + '<br>';
                document.body.appendChild(div);
            }
            
            // Create Centrifuge object with Websocket endpoint address set in main.go
            const centrifuge = new Centrifuge('ws://localhost:8000/connection/websocket');

            centrifuge.on('connected', function(ctx){
                drawText('Connected over ' + ctx.transport);
            });
            
            const sub = centrifuge.newSubscription("chat");
            sub.on('publication', function(ctx) {
                drawText(JSON.stringify(ctx.data));
            });
            // Move subscription to subscribing state.
            sub.subscribe();
            
            const input = document.getElementById("input");
            input.addEventListener('keyup', function(e) {
                if (e.key === "Enter") {
                    e.preventDefault();
                    sub.publish(this.value);
                    input.value = '';
                }
            });
            // After setting event handlers – initiate actual connection with server.
            centrifuge.connect();
        </script>
    </body>
</html>

Then run Go app as usual:

go run main.go

Open several browser tabs with http://localhost:8000 and see chat in action.

While this example is only the top of an iceberg, it should give you a good insight on library API. Check out examples folder for more. We recommend to start looking from chat_json example, which extends the basics shown here and demonstrates more possibilities of Centrifuge protocol:

Chat example

[!IMPORTANT]
Keep in mind that Centrifuge library is not a framework to build chat applications. It's a general purpose real-time transport for your messages with some helpful primitives. You can build many kinds of real-time apps on top of this library including chats but depending on application you may need to write business logic yourself.

Tips and tricks

Some useful advices about library here.

Connection life cycle

Let's describe some aspects related to connection life cycle and event handling in Centrifuge:

  • If you set middleware for transport handlers (like WebsocketHandler) – then it will be called first before a client sent any command to a server and handler had a chance to start working. Just like a regular HTTP middleware. You can put Credentials to Context to authenticate connection.
  • node.OnConnecting called as soon as client sent Connect command to server. At this point no Client instance exists. You have incoming Context and Transport information. You still can authenticate Client at this point (based on string token sent from client side or any other way). Also, you can add extra data to context and return modified context to Centrifuge. Context cancelled as soon as client connection closes. This handler is synchronous and connection read loop can't proceed until you return ConnectReply.
  • node.OnConnect then called (after a reply to Connect command already written to connection). Inside OnConnect closure you have a possibility to define per-connection event handlers. If particular handler not set then client will get ErrorNotAvailable errors requesting it. Remember that none of event handlers available in Centrifuge should block forever – do minimal work, start separate goroutines if you need blocking code.
  • Client initiated request handlers called one by one from connection reading goroutine. This includes OnSubscribe, OnPublish, OnPresence, OnPresenceStats, OnHistory, client-side OnRefresh, client-side OnSubRefresh.
  • Other handlers like OnAlive, OnDisconnect, server-side OnSubRefresh, server-side OnRefresh called from separate internal goroutines.
  • OnAlive handler must not be called after OnDisconnect.
  • Client initiated request handlers can be processed asynchronously in goroutines to manage operation concurrency. This is achieved using callback functions. See concurrency example for more details.
Channel history stream

Centrifuge Broker interface supports saving Publication to history stream on publish. Depending on Broker implementation this feature can be missing though. Builtin Memory and Redis brokers support keeping Publication stream.

When using default MemoryBroker Publication stream kept in process memory and lost as soon as process restarts. RedisBroker keeps Publication stream in Redis LIST or STREAM data structures – reliability inherited from Redis configuration in this case.

Centrifuge library publication stream not meant to be used as the only source of missed Publications for a client. It mostly exists to help many clients reconnect at once (load balancer reload, application deploy) without creating a massive spike in load on your main application database. So application database still required in idiomatic use case.

Centrifuge message recovery protocol feature designed to be used together with reasonably small Publication stream size as all missed publications sent towards client in one protocol frame on resubscribe to channel.

Logging

Centrifuge library exposes logs with different log level. In your app you can set special function to handle these log entries in a way you want.

// Function to handle Centrifuge internal logs.
func handleLog(e centrifuge.LogEntry) {
	log.Printf("%s: %v", e.Message, e.Fields)
}

cfg := centrifuge.DefaultConfig
cfg.LogLevel = centrifuge.LogLevelDebug
cfg.LogHandler = handleLog
Allowed origin for WebSocket

When connecting to Centrifuge WebSocket endpoint from web browsers you need to configure allowed Origin. This is important to prevent CSRF-like/WebSocket hijacking attacks. See this post for example.

By default, CheckOrigin function of WebSocket handler will ensure that connection request originates from same host as your service. To override this behaviour you can provide your own implementation of CheckOrigin function to allow origins you trust. For example, your Centrifuge runs on http://localhost:8000 but you want it to allow WebSocket connections from http://localhost:3000:

centrifuge.NewWebsocketHandler(node, centrifuge.WebsocketConfig{
	CheckOrigin: func(r *http.Request) bool {
		originHeader := r.Header.Get("Origin")
		if originHeader == "" {
			return true
		}
		return originHeader == "http://localhost:3000"
	},
})

Note, that if WebSocket Upgrade does not contain Origin header – it means it does not come from web browser and security concerns outlined above are not applied in that case. So we can safely return true in this case in the example above.

CORS for HTTP-based transports

Centrifuge has two HTTP-based fallback transports for WebSocket – see HTTPStreamHandler and SSEHandler. To connect to those from web browser from the domain which is different from your transport endpoint domain you may need to wrap handlers with CORS middleware:

func CORS(h http.Handler) http.Handler {
	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		header := w.Header()
		if originAllowed(r) {
			header.Set("Access-Control-Allow-Origin", r.Header.Get("origin"))
			if allowHeaders := r.Header.Get("Access-Control-Request-Headers"); allowHeaders != "" && allowHeaders != "null" {
				header.Add("Access-Control-Allow-Headers", allowHeaders)
			}
			header.Set("Access-Control-Allow-Credentials", "true")
		}
		h.ServeHTTP(w, r)
	})
}

http.Handle("/connection/http_stream", CORS(centrifuge.NewHTTPStreamHandler(node, centrifuge.HTTPStreamHandlerConfig{})))

You can also configure CORS on load-balancer/reverse-proxy level.

Server timeouts and HTTP-based real-time transports

Centrifuge uses http.ResponseController when working with timeouts in HTTP-streaming and Server-Sent Events (SSE) handlers. This allows having custom timeouts for HTTP server. But if you are using HTTP middlewares which provide a custom implementation of http.ResponseWriter – then make sure they implement Unwrap method to access original http.ResponseWriter for ResponseController to work correctly. As per http package documentation:

The ResponseWriter should be the original value passed to the [Handler.ServeHTTP] method, or have an Unwrap method returning the original ResponseWriter.

If handlers can't access original http.ResponseWriter – then you will observe connection closing corresponding to your HTTP server's ReadTimeout setting.

For contributors

Running integration tests locally

To run integration tests over Redis, Redis + Sentinel, Redis Cluster:

docker compose up
go test -tags integration ./...

To clean up container state:

docker compose down -v

Documentation

Overview

Package centrifuge is a real-time messaging library that abstracts several bidirectional transports (Websocket and its emulation over HTTP-Streaming, SSE/EventSource) and provides primitives to build scalable real-time applications with Go. It's also used as a core of Centrifugo server (https://github.com/centrifugal/centrifugo).

Centrifuge library provides several features on top of plain Websocket implementation and comes with several client SDKs – see more details in the library README on GitHub – https://github.com/centrifugal/centrifuge.

The API of this library is almost all goroutine-safe except cases where one-time operations like setting callback handlers performed, also your code inside event handlers should be synchronized since event handlers can be called concurrently. Library expects that code inside event handlers will not block. See more information about client connection lifetime and event handler order/concurrency in README on GitHub.

Also check out examples in repo to see main library concepts in action.

Index

Constants

View Source 
const (
	MapPhaseLive   int32 = 0 // Join live pub/sub, switch to real-time streaming (default)
	MapPhaseStream int32 = 1 // Paginating over stream (history catch-up)
	MapPhaseState  int32 = 2 // Paginating over state (map state)
)

Map subscription phase constants.

View Source 
const (
	// SharedPollModeVersionless is the versionless mode where the backend
	// returns {key, data} without versions. Centrifugo detects changes via content hash
	// and generates internal synthetic versions. This is the default mode (also "").
	SharedPollModeVersionless = "versionless"
	// SharedPollModeVersioned is the versioned mode where the backend
	// returns items with {key, data, version}. Versions are included in requests,
	// allowing the backend to skip unchanged items. Enables direct publish and
	// cached initial data.
	SharedPollModeVersioned = "versioned"
)
View Source 
const (
	PushFlagConnect uint64 = 1 << iota
	PushFlagDisconnect
	PushFlagSubscribe
	PushFlagJoin
	PushFlagLeave
	PushFlagUnsubscribe
	PushFlagPublication
	PushFlagMessage
)

It's possible to disable certain types of pushes to be sent to a client connection by using ClientConfig.DisabledPushFlags.

View Source 
const (
	// UnsubscribeCodeClient set when unsubscribe event was initiated
	// by an explicit client-side unsubscribe call.
	// Code is less than 2000 since it's never sent to a client connection.
	UnsubscribeCodeClient uint32 = 0
	// UnsubscribeCodeDisconnect set when unsubscribe event was initiated
	// by a client disconnect process.
	// Code is less than 2000 since it's never sent to a client connection.
	UnsubscribeCodeDisconnect uint32 = 1
	// UnsubscribeCodeServer set when unsubscribe event was initiated
	// by an explicit server-side unsubscribe call.
	UnsubscribeCodeServer uint32 = 2000
	// UnsubscribeCodeInsufficient set when client unsubscribed from
	// a channel due to insufficient state in a stream. We expect client to
	// resubscribe after receiving this since it's still may be possible to
	// recover a state since known StreamPosition.
	UnsubscribeCodeInsufficient uint32 = 2500
	// UnsubscribeCodeExpired set when client subscription expired.
	UnsubscribeCodeExpired uint32 = 2501
	// UnsubscribeCodeStateInvalidated set when server determines that the
	// subscription's cached state and/or token are no longer valid. Upon
	// receiving this code, the client SDK should clear the subscription
	// token (to force obtaining a new one via getToken) and clear any cached
	// map subscription state (to force full state re-sync instead of stream
	// recovery on resubscribe).
	UnsubscribeCodeStateInvalidated uint32 = 2502
)

Known unsubscribe codes.

View Source 
const ConnectionFlagDictionaryCompression int64 = 1 << 0

ConnectionFlagDictionaryCompression is advertised by a client in ConnectRequest.flag to say it can decode dictionary-compressed frames, and echoed in ConnectResult.flag when the server enabled it.

A client must not assume a feature it advertised was accepted: the server may have it switched off, or may decline for this particular connection.

View Source 
const NoLimit = -1

NoLimit defines that limit should not be applied.

Variables

View Source 
var (
	// DisconnectShutdown issued when node is going to shut down.
	DisconnectShutdown = Disconnect{
		Code:   3001,
		Reason: "shutdown",
	}
	// DisconnectServerError issued when internal error occurred on server.
	DisconnectServerError = Disconnect{
		Code:   3004,
		Reason: "internal server error",
	}
	// DisconnectExpired issued when client connection expired.
	DisconnectExpired = Disconnect{
		Code:   3005,
		Reason: "connection expired",
	}
	// DisconnectSubExpired issued when client subscription expired.
	DisconnectSubExpired = Disconnect{
		Code:   3006,
		Reason: "subscription expired",
	}
	// DisconnectSlow issued when client can't read messages fast enough.
	DisconnectSlow = Disconnect{
		Code:   3008,
		Reason: "slow",
	}
	// DisconnectWriteError issued when an error occurred while writing to
	// client connection.
	DisconnectWriteError = Disconnect{
		Code:   3009,
		Reason: "write error",
	}
	// DisconnectInsufficientState issued when server detects wrong client
	// position in channel Publication stream. Disconnect allows client
	// to restore missed publications on reconnect.
	DisconnectInsufficientState = Disconnect{
		Code:   3010,
		Reason: "insufficient state",
	}
	// DisconnectForceReconnect issued when server disconnects connection.
	DisconnectForceReconnect = Disconnect{
		Code:   3011,
		Reason: "force reconnect",
	}
	// DisconnectNoPong may be issued when server disconnects bidirectional
	// connection due to no pong received to application-level server-to-client
	// pings in a configured time.
	DisconnectNoPong = Disconnect{
		Code:   3012,
		Reason: "no pong",
	}
	// DisconnectTooManyRequests may be issued when client sends too many commands to a server.
	DisconnectTooManyRequests = Disconnect{
		Code:   3013,
		Reason: "too many requests",
	}
	// DisconnectStateInvalidated may be issued when server determines that
	// client's cached state and/or token are no longer valid. Upon receiving
	// this code, the client SDK should clear its connection token (to force
	// obtaining a new one via getToken) and clear all map subscription state
	// (to force full state re-sync instead of stream recovery on reconnect).
	DisconnectStateInvalidated = Disconnect{
		Code:   3014,
		Reason: "state invalidated",
	}
)

Some predefined non-terminal disconnect structures used by the library internally.

View Source 
var (
	// DisconnectInvalidToken issued when client came with invalid token.
	DisconnectInvalidToken = Disconnect{
		Code:   3500,
		Reason: "invalid token",
	}
	// DisconnectBadRequest issued when client uses malformed protocol frames.
	DisconnectBadRequest = Disconnect{
		Code:   3501,
		Reason: "bad request",
	}
	// DisconnectStale issued to close connection that did not become
	// authenticated in configured interval after dialing.
	DisconnectStale = Disconnect{
		Code:   3502,
		Reason: "stale",
	}
	// DisconnectForceNoReconnect issued when server disconnects connection
	// and asks it to not reconnect again.
	DisconnectForceNoReconnect = Disconnect{
		Code:   3503,
		Reason: "force disconnect",
	}
	// DisconnectConnectionLimit can be issued when client connection exceeds a
	// configured connection limit (per user ID or due to other rule).
	DisconnectConnectionLimit = Disconnect{
		Code:   3504,
		Reason: "connection limit",
	}
	// DisconnectChannelLimit can be issued when client connection exceeds a
	// configured channel limit.
	DisconnectChannelLimit = Disconnect{
		Code:   3505,
		Reason: "channel limit",
	}
	// DisconnectInappropriateProtocol can be issued when client connection format can not
	// handle incoming data. For example, this happens when JSON-based clients receive
	// binary data in a channel. This is usually an indicator of programmer error, JSON
	// clients can not handle binary.
	DisconnectInappropriateProtocol = Disconnect{
		Code:   3506,
		Reason: "inappropriate protocol",
	}
	// DisconnectPermissionDenied may be issued when client attempts accessing a server without
	// enough permissions.
	DisconnectPermissionDenied = Disconnect{
		Code:   3507,
		Reason: "permission denied",
	}
	// DisconnectNotAvailable may be issued when ErrorNotAvailable does not fit message type, for example
	// we issue DisconnectNotAvailable when client sends asynchronous message without MessageHandler set
	// on server side.
	DisconnectNotAvailable = Disconnect{
		Code:   3508,
		Reason: "not available",
	}
	// DisconnectTooManyErrors may be issued when client generates too many errors.
	DisconnectTooManyErrors = Disconnect{
		Code:   3509,
		Reason: "too many errors",
	}
)

The codes below are built-in terminal codes.

View Source 
var (
	// ErrorInternal means server error, if returned this is a signal
	// that something went wrong with server itself and client most probably
	// not guilty.
	ErrorInternal = &Error{
		Code:      100,
		Message:   "internal server error",
		Temporary: true,
	}
	// ErrorUnauthorized says that request is unauthorized.
	ErrorUnauthorized = &Error{
		Code:    101,
		Message: "unauthorized",
	}
	// ErrorUnknownChannel means that channel name does not exist.
	ErrorUnknownChannel = &Error{
		Code:    102,
		Message: "unknown channel",
	}
	// ErrorPermissionDenied means that access to resource not allowed.
	ErrorPermissionDenied = &Error{
		Code:    103,
		Message: "permission denied",
	}
	// ErrorMethodNotFound means that method sent in command does not exist.
	ErrorMethodNotFound = &Error{
		Code:    104,
		Message: "method not found",
	}
	// ErrorAlreadySubscribed returned when client wants to subscribe on channel
	// it already subscribed to.
	ErrorAlreadySubscribed = &Error{
		Code:    105,
		Message: "already subscribed",
	}
	// ErrorLimitExceeded says that some sort of limit exceeded, server logs should
	// give more detailed information. See also ErrorTooManyRequests which is more
	// specific for rate limiting purposes.
	ErrorLimitExceeded = &Error{
		Code:    106,
		Message: "limit exceeded",
	}
	// ErrorBadRequest says that server can not process received
	// data because it is malformed. Retrying request does not make sense.
	ErrorBadRequest = &Error{
		Code:    107,
		Message: "bad request",
	}
	// ErrorNotAvailable means that resource is not enabled.
	ErrorNotAvailable = &Error{
		Code:    108,
		Message: "not available",
	}
	// ErrorTokenExpired indicates that connection or subscription token expired.
	// In this case client should drop the current token and try to ask a new one.
	ErrorTokenExpired = &Error{
		Code:    109,
		Message: "token expired",
	}
	// ErrorExpired indicates that connection expired (no token involved).
	ErrorExpired = &Error{
		Code:    110,
		Message: "expired",
	}
	// ErrorTooManyRequests means that server rejected request due to
	// its rate limiting strategies.
	ErrorTooManyRequests = &Error{
		Code:      111,
		Message:   "too many requests",
		Temporary: true,
	}
	// ErrorUnrecoverablePosition means that stream does not contain required
	// range of publications to fulfill a history query. This can happen due to
	// expiration, size limitation or due to wrong epoch.
	ErrorUnrecoverablePosition = &Error{
		Code:    112,
		Message: "unrecoverable position",
	}
	// ErrorConcurrentPagination means a pagination request was sent while
	// another pagination request for the same channel is in progress.
	ErrorConcurrentPagination = &Error{
		Code:      113,
		Message:   "concurrent pagination",
		Temporary: true,
	}
)

Here we define well-known errors that can be used in client protocol replies.

View Source 
var DisconnectConnectionClosed = Disconnect{
	Code:   3000,
	Reason: "connection closed",
}

DisconnectConnectionClosed is a special Disconnect object used when client connection was closed without any advice from a server side. This can be a clean disconnect, or temporary disconnect of the client due to internet connection loss. Server can not distinguish the actual reason of disconnect.

Functions

func HandleReadFrame added in v0.28.0 

func HandleReadFrame(c *Client, r io.Reader, messageSizeLimit int64) bool

HandleReadFrame is a helper to read Centrifuge commands from frame-based io.Reader and process them. Frame-based means that EOF treated as the end of the frame, not the entire connection close.

messageSizeLimit bounds the size of a single command and must be positive: the underlying stream decoder rejects a non-positive limit, since the length prefix is attacker-controlled and used as an allocation size. Callers pass the transport's configured limit (coerced to a positive default).

func LogLevelToString 

func LogLevelToString(l LogLevel) string

LogLevelToString transforms Level to its string representation.

func MakeOrderedCursor added in v0.39.0 

func MakeOrderedCursor(score, key string) string

MakeOrderedCursor creates a cursor for ordered state: "score\x00key".

func NewClient added in v0.1.0 

func NewClient(ctx context.Context, n *Node, t Transport) (*Client, ClientCloseFunc, error)

NewClient initializes new Client.

func SetCredentials 

func SetCredentials(ctx context.Context, cred *Credentials) context.Context

SetCredentials allows setting connection Credentials to Context. Credentials set to Context in authentication middleware will be used by Centrifuge library to authenticate user.

Types

type AliveHandler added in v0.10.0 

type AliveHandler func()

AliveHandler called periodically while connection alive. This is a helper to do periodic things which can tolerate some approximation in time. This callback will run every ClientPresenceUpdateInterval and can save you a timer.

type Broker 

type Broker interface {
	// RegisterBrokerEventHandler called once on start when Broker already set to Node. At
	// this moment node is ready to process broker events.
	RegisterBrokerEventHandler(BrokerEventHandler) error

	// Subscribe node on channels to listen all messages coming from them.
	// Subscribe and Unsubscribe of the same channel are never called
	// concurrently: callers serialize them per channel.
	Subscribe(channels ...string) error
	// Unsubscribe node from channels to stop listening messages from them.
	Unsubscribe(channels ...string) error

	// Publish allows sending data into channel. Data should be
	// delivered to all clients subscribed to this channel at moment on any
	// Centrifuge node (with at most once delivery guarantee).
	//
	// Broker can optionally maintain publication history inside channel according
	// to PublishOptions provided. See History method for rules that should be implemented
	// for accessing publications from history stream.
	//
	// Saving message to a history stream and publish to PUB/SUB should be an atomic
	// operation per channel. If this is not true – then publication to one channel
	// must be serialized on the caller side, i.e. publish requests must be issued one
	// after another. Otherwise, the order of publications and stable behaviour of
	// subscribers with positioning/recovery enabled can't be guaranteed.
	//
	// PublishResult.StreamPosition describes stream epoch and offset assigned to
	// the publication. For channels without history this should be zero value.
	// PublishResult.Suppressed and PublishResult.SuppressReason indicate whether
	// publication was suppressed and why (e.g. idempotency deduplication, version skip).
	Publish(ch string, data []byte, opts PublishOptions) (PublishResult, error)
	// PublishJoin publishes Join Push message into channel.
	PublishJoin(ch string, info *ClientInfo) error
	// PublishLeave publishes Leave Push message into channel.
	PublishLeave(ch string, info *ClientInfo) error

	// History used to extract Publications from history stream.
	// Publications returned according to HistoryFilter which allows to set several
	// filtering options. StreamPosition returned describes current history stream
	// top offset and epoch.
	History(ch string, opts HistoryOptions) ([]*Publication, StreamPosition, error)
	// RemoveHistory removes history from channel. This is in general not
	// needed as history expires automatically (based on history_lifetime)
	// but sometimes can be useful for application logic.
	RemoveHistory(ch string) error
}

Broker is responsible for PUB/SUB mechanics.

type BrokerEventHandler 

type BrokerEventHandler interface {
	// HandlePublication to handle received Publications.
	HandlePublication(ch string, pub *Publication, sp StreamPosition, useDelta bool, prevPub *Publication) error
	// HandleJoin to handle received Join messages.
	HandleJoin(ch string, info *ClientInfo) error
	// HandleLeave to handle received Leave messages.
	HandleLeave(ch string, info *ClientInfo) error
}

BrokerEventHandler can handle messages received from PUB/SUB system.

type CacheEmptyEvent added in v0.33.0 

type CacheEmptyEvent struct {
	Channel string
}

CacheEmptyEvent is issued when recovery mode is used but Centrifuge can't find Publication in history to recover from. This event allows application to decide what to do in this case – it's possible to populate the cache by sending actual data to a channel.

type CacheEmptyHandler added in v0.33.0 

type CacheEmptyHandler func(CacheEmptyEvent) (CacheEmptyReply, error)

CacheEmptyHandler allows setting cache empty handler function.

type CacheEmptyReply added in v0.33.0 

type CacheEmptyReply struct {
	// Populated when set to true tells Centrifuge that cache was populated and
	// in that case Centrifuge will try to recover missed Publication from history
	// one more time.
	Populated bool
}

CacheEmptyReply contains fields determining the reaction on cache empty event.

type ChannelBatchConfig added in v0.35.0 

type ChannelBatchConfig struct {
	// MaxSize is the maximum number of messages to batch before flushing.
	MaxSize int64
	// MaxDelay is the maximum time to wait before flushing.
	MaxDelay time.Duration
	// FlushLatestPublication if true, then Centrifuge flushes only the latest publication
	// in the batch upon reaching the MaxSize or MaxDelay. Skipping on this level does
	// not work with delta compression.
	FlushLatestPublication bool
}

ChannelBatchConfig allows configuring how to write push messages to a channel during broadcasts (applied for publication, join and leave pushes). This API is EXPERIMENTAL and may be changed/removed. If MaxSize is set to 0 then no batching by size will be performed. If MaxDelay is set to 0 then no batching by time will be performed. If both MaxSize and MaxDelay are set to 0 then no batching will be performed.

type ChannelContext 

type ChannelContext struct {

	// Source is a source of subscription application can set in SubscribeHandler.
	Source uint8
	// contains filtered or unexported fields
}

ChannelContext contains extra context for channel connection subscribed to. Note: this struct is aligned to consume less memory.

type ChannelMediumOptions added in v0.33.0 

type ChannelMediumOptions struct {
	// KeepLatestPublication enables keeping latest publication which was broadcasted to channel subscribers on
	// this Node in the channel medium layer. This is helpful for supporting deltas in at most once scenario.
	KeepLatestPublication bool

	// SharedPositionSync when true delegates connection position checks to the channel medium. In that case
	// check is only performed no more often than once in Config.ClientChannelPositionCheckDelay thus reducing
	// the load on broker in cases when channel has many subscribers. When message loss is detected medium layer
	// tells caller about this and also marks all channel subscribers with insufficient state flag. By default,
	// medium is not used for sync – in that case each individual connection syncs position independently.
	SharedPositionSync bool
	// contains filtered or unexported fields
}

ChannelMediumOptions is an EXPERIMENTAL way to enable using a channel medium layer in Centrifuge. Note, channel medium layer is very unstable at the moment – do not use it in production! Channel medium layer is an optional per-channel intermediary between Broker PUB/SUB and Client connections. This intermediary layer may be used for various per-channel tweaks and optimizations. Channel medium comes with memory overhead depending on ChannelMediumOptions. At the same time, it can provide significant benefits in terms of overall system efficiency and flexibility.

type Client 

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

Client represents client connection to server.

func (*Client) AcquireStorage added in v0.29.2 

func (c *Client) AcquireStorage() (map[string]any, func(map[string]any))

AcquireStorage returns an attached connection storage (a map) and a function to be called when the application finished working with the storage map. Be accurate when using this API – avoid acquiring storage for a long time - i.e. on the time of IO operations. Do the work fast and release with the updated map. The API designed this way to allow reading, modifying or fully overriding storage map and avoid making deep copies each time. Note, that if storage map has not been initialized yet - i.e. if it's nil - then it will be initialized to an empty map and then returned – so you never receive nil map when acquiring. The purpose of this map is to simplify handling user-defined state during the lifetime of connection. Try to keep this map reasonably small. This API is EXPERIMENTAL and may be changed/removed.

func (*Client) Channels 

func (c *Client) Channels() []string

Channels returns a slice of channels client connection currently subscribed to.

func (*Client) ChannelsWithContext added in v0.26.0 

func (c *Client) ChannelsWithContext() map[string]ChannelContext

ChannelsWithContext returns a map of channels client connection currently subscribed to with a ChannelContext.

func (*Client) Connect added in v0.16.0 

func (c *Client) Connect(req ConnectRequest)

Connect supposed to be called only from a unidirectional transport layer to pass initial information about connection and thus initiate Node.OnConnecting event. Bidirectional transport initiate connecting workflow automatically since client passes Connect command upon successful connection establishment with a server. If there is an error during connect method processing Centrifuge extracts Disconnect from it and closes the connection with that Disconnect message.

func (*Client) ConnectNoErrorToDisconnect added in v0.33.5 

func (c *Client) ConnectNoErrorToDisconnect(req ConnectRequest) error

ConnectNoErrorToDisconnect is the same as Client.Connect but does not try to extract Disconnect code from the error returned by the connect logic, instead it just returns the error to the caller. This error must be handled by the caller on the Transport level, and the connection must be closed on Transport level upon receiving an error.

func (*Client) ConnectedAtMS added in v0.37.1 

func (c *Client) ConnectedAtMS() int64

ConnectedAtMS returns timestamp in milliseconds when client connected.

func (*Client) Context added in v0.10.0 

func (c *Client) Context() context.Context

Context returns client Context. This context will be canceled as soon as client connection closes.

func (*Client) Disconnect added in v0.10.0 

func (c *Client) Disconnect(disconnect ...Disconnect)

Disconnect client connection with specific disconnect code and reason. If zero args or nil passed then DisconnectForceNoReconnect is used.

This method internally creates a new goroutine at the moment to do closing stuff. An extra goroutine is required to solve disconnect and alive callback ordering/sync problems. Will be a noop if client already closed. As this method runs a separate goroutine client connection will be closed eventually (i.e. not immediately).

func (*Client) HandleCommand added in v0.24.0 

func (c *Client) HandleCommand(cmd *protocol.Command, cmdProtocolSize int) bool

HandleCommand processes a single protocol.Command. Supposed to be called only from a transport connection reader.

func (*Client) ID 

func (c *Client) ID() string

ID returns unique client connection id.

func (*Client) Info added in v0.18.0 

func (c *Client) Info() []byte

Info returns connection info.

func (*Client) IsSubscribed added in v0.11.0 

func (c *Client) IsSubscribed(ch string) bool

IsSubscribed returns true if client subscribed to a channel.

func (*Client) Labels added in v0.39.0 

func (c *Client) Labels() map[string]string

Labels returns custom labels attached to the client connection. These labels are set via ConnectReply.Labels. The returned map must not be modified by the caller - use it for reading only. Labels are set once during connect and never change.

func (*Client) LatestPingPongLatency added in v0.37.1 

func (c *Client) LatestPingPongLatency() (time.Duration, bool)

LatestPingPongLatency returns latest ping-pong latency duration. It may be not available if no ping-pong messages were exchanged yet, or in case of unidirectional transport. In that case second return value will be false.

func (*Client) OnAlive added in v0.13.0 

func (c *Client) OnAlive(h AliveHandler)

OnAlive allows setting AliveHandler. AliveHandler called periodically for active client connection.

func (*Client) OnDisconnect added in v0.13.0 

func (c *Client) OnDisconnect(h DisconnectHandler)

OnDisconnect allows setting DisconnectHandler. DisconnectHandler called when client disconnected.

func (*Client) OnHistory added in v0.13.0 

func (c *Client) OnHistory(h HistoryHandler)

OnHistory allows settings HistoryHandler. HistoryHandler called when History request from client received. At this moment you can only return a custom error or disconnect client.

func (*Client) OnMapPublish added in v0.39.0 

func (c *Client) OnMapPublish(h MapPublishHandler)

OnMapPublish allows setting MapPublishHandler. MapPublishHandler called when client publishes into map channel.

func (*Client) OnMapRemove added in v0.39.0 

func (c *Client) OnMapRemove(h MapRemoveHandler)

OnMapRemove allows setting MapRemoveHandler. MapRemoveHandler called when client wants to remove a key from map channel.

func (*Client) OnMessage added in v0.13.0 

func (c *Client) OnMessage(h MessageHandler)

OnMessage allows setting MessageHandler. MessageHandler called when client sent asynchronous message.

func (*Client) OnPresence added in v0.13.0 

func (c *Client) OnPresence(h PresenceHandler)

OnPresence allows setting PresenceHandler. PresenceHandler called when Presence request from client received. At this moment you can only return a custom error or disconnect client.

func (*Client) OnPresenceStats added in v0.13.0 

func (c *Client) OnPresenceStats(h PresenceStatsHandler)

OnPresenceStats allows settings PresenceStatsHandler. PresenceStatsHandler called when Presence Stats request from client received. At this moment you can only return a custom error or disconnect client.

func (*Client) OnPublish added in v0.13.0 

func (c *Client) OnPublish(h PublishHandler)

OnPublish allows setting PublishHandler. PublishHandler called when client publishes message into channel.

func (*Client) OnRPC added in v0.13.0 

func (c *Client) OnRPC(h RPCHandler)

OnRPC allows setting RPCHandler. RPCHandler will be executed on every incoming RPC call.

func (*Client) OnRefresh added in v0.13.0 

func (c *Client) OnRefresh(h RefreshHandler)

OnRefresh allows setting RefreshHandler. RefreshHandler called when it's time to refresh expiring client connection.

func (*Client) OnStateSnapshot added in v0.23.0 

func (c *Client) OnStateSnapshot(h StateSnapshotHandler)

OnStateSnapshot allows settings StateSnapshotHandler. This API is EXPERIMENTAL and may be changed/removed.

func (*Client) OnSubRefresh added in v0.13.0 

func (c *Client) OnSubRefresh(h SubRefreshHandler)

OnSubRefresh allows setting SubRefreshHandler. SubRefreshHandler called when it's time to refresh client subscription.

func (*Client) OnSubscribe added in v0.13.0 

func (c *Client) OnSubscribe(h SubscribeHandler)

OnSubscribe allows setting SubscribeHandler. SubscribeHandler called when client subscribes on a channel.

func (*Client) OnTrack added in v0.39.0 

func (c *Client) OnTrack(h TrackHandler)

OnTrack allows setting TrackHandler. TrackHandler called when client sends a track request on a keyed channel.

func (*Client) OnUnsubscribe added in v0.13.0 

func (c *Client) OnUnsubscribe(h UnsubscribeHandler)

OnUnsubscribe allows setting UnsubscribeHandler. UnsubscribeHandler called when client unsubscribes from channel, and once for every subscribe attempt allowed by SubscribeHandler which ended without becoming a subscription (UnsubscribeEvent.Subscribed is false then).

func (*Client) OnUntrack added in v0.39.0 

func (c *Client) OnUntrack(h UntrackHandler)

OnUntrack allows setting UntrackHandler. UntrackHandler called when client untracks keys on a keyed channel.

func (*Client) Profile added in v0.39.0 

func (c *Client) Profile() string

Profile returns the application context this connection belongs to, as resolved at connect time - the value an OnConnecting handler set, or failing that the one the client declared in its connect command. Empty means the connection is unclassified.

Nothing in this package interprets it. It exists so features layered on top can group connections whose traffic looks alike, and so operators can filter and measure by it.

func (*Client) ProtocolConnect added in v0.34.3 

func (c *Client) ProtocolConnect(req *protocol.ConnectRequest)

ProtocolConnect accepts protocol.ConnectRequest directly. It adds dependency to protocol package, so prefer using Connect or ConnectNoErrorToDisconnect methods until necessary.

func (*Client) ProtocolConnectNoErrorToDisconnect added in v0.34.3 

func (c *Client) ProtocolConnectNoErrorToDisconnect(req *protocol.ConnectRequest) error

ProtocolConnectNoErrorToDisconnect accepts protocol.ConnectRequest directly. It adds dependency to protocol package, so prefer ConnectNoErrorToDisconnect methods until necessary.

func (*Client) Refresh added in v0.18.0 

func (c *Client) Refresh(opts ...RefreshOption) error

func (*Client) Send 

func (c *Client) Send(data []byte) error

Send data to client. This sends an asynchronous message – data will be just written to connection. on client side this message can be handled with Message handler.

func (*Client) StateSnapshot added in v0.23.0 

func (c *Client) StateSnapshot() (any, error)

StateSnapshot allows collecting current state copy. Mostly useful for connection introspection from the outside. This API is EXPERIMENTAL and may be changed/removed.

func (*Client) Subscribe added in v0.5.0 

func (c *Client) Subscribe(channel string, opts ...SubscribeOption) error

Subscribe client to a channel.

func (*Client) Transport 

func (c *Client) Transport() TransportInfo

Transport returns client connection transport information.

func (*Client) Unsubscribe 

func (c *Client) Unsubscribe(ch string, unsubscribe ...Unsubscribe)

Unsubscribe allows unsubscribing client from channel. Called before ConnectHandler returned (e.g. Node.Unsubscribe of a subscription made on connect, or from inside ConnectHandler), it takes effect right after ConnectHandler returns: so UnsubscribeHandler set in ConnectHandler gets the call, and not while ConnectHandler runs.

func (*Client) UserID 

func (c *Client) UserID() string

UserID returns user id associated with client connection.

func (*Client) WritePublication added in v0.30.3 

func (c *Client) WritePublication(channel string, publication *Publication, sp StreamPosition) error

WritePublication allows sending publications to Client subscription directly without HUB and Broker semantics. The possible use case is to turn subscription to a channel into an individual data stream. This API is EXPERIMENTAL and may be changed/removed.

type ClientCloseFunc added in v0.10.0 

type ClientCloseFunc func() error

ClientCloseFunc must be called on Transport handler close to clean up Client.

type ClientInfo 

type ClientInfo struct {
	// ClientID is a client unique id.
	ClientID string
	// UserID is an ID of authenticated user. Zero value means anonymous user.
	UserID string
	// ConnInfo is additional information about connection.
	ConnInfo []byte
	// ChanInfo is additional information about connection in context of
	// channel subscription.
	ChanInfo []byte
}

ClientInfo contains information about client connection.

type Closer 

type Closer interface {
	// Close when called should clean up used resources.
	Close(ctx context.Context) error
}

Closer is an interface that Broker and PresenceManager can optionally implement if they need to close any resources on Centrifuge Node graceful shutdown.

type CommandProcessedEvent added in v0.28.0 

type CommandProcessedEvent struct {
	// Command which was processed. May be pooled - see comment of CommandProcessedEvent.
	Command *protocol.Command
	// Error may be set to non-nil if Command processing resulted into error.
	Error error
	// Reply to the command. Reply may be pooled - see comment of CommandProcessedEvent.
	// This Reply may be nil, for example in the following cases:
	// 1. For Send command since send commands do not have replies
	// 2. When command processing resulted into disconnection of the client without sending a reply.
	// 3. When unidirectional transport connects (Centrifuge creates Connect Command artificially
	//    with id: 1 and never sends replies to the unidirectional transport, only pushes).
	Reply *protocol.Reply
	// Started is a time command was passed to Client for processing.
	Started time.Time
}

CommandProcessedEvent contains protocol.Command processed by Client. Command and Reply types and their fields in the event MAY BE POOLED by Centrifuge, so code which wants to use them AFTER CommandProcessedHandler handler returns MUST MAKE A COPY.

type CommandProcessedHandler added in v0.28.0 

type CommandProcessedHandler func(*Client, CommandProcessedEvent)

CommandProcessedHandler allows setting a callback which will be called after Client processed a protocol.Command. This exists mostly for real-time connection tracing purposes. CommandProcessedHandler may be called after the corresponding Reply written to the connection and TransportWriteHandler called. But for tracing purposes this seems tolerable as commands and replies may be matched by id. Also, carefully read docs for CommandProcessedEvent to avoid possible bugs.

type CommandReadEvent added in v0.27.0 

type CommandReadEvent struct {
	// Command which was read from the connection. May be pooled - see comment of CommandReadEvent.
	Command *protocol.Command
	// CommandSize is a size of command in bytes in its protocol representation.
	CommandSize int
}

CommandReadEvent contains protocol.Command processed by Client. Command type and its fields in the event MAY BE POOLED by Centrifuge, so code which wants to use Command AFTER CommandReadHandler handler returns MUST MAKE A COPY.

type CommandReadHandler added in v0.27.0 

type CommandReadHandler func(*Client, CommandReadEvent) error

CommandReadHandler allows setting a callback which will be called before Client processed a protocol.Command read from the connection. Return an error if you want to prevent command execution. Also, carefully read docs for CommandReadEvent to avoid possible bugs.

type Config 

type Config struct {
	// Version of server – if set will be sent to a client on connection
	// establishment phase in reply to connect command from a client.
	Version string
	// Name is a unique name of the current server Node. Name used as human-readable
	// and meaningful node identifier. If not set then os.Hostname will be used.
	Name string
	// LogLevel is a log level. By default, nothing will be logged by Centrifuge.
	LogLevel LogLevel
	// LogHandler is a handler function Node will send logs to.
	LogHandler LogHandler
	// NodeInfoMetricsAggregateInterval sets interval for automatic metrics
	// aggregation. It's not reasonable to have it less than one second.
	// Zero value means 60 * time.Second.
	NodeInfoMetricsAggregateInterval time.Duration
	// ClientConnectIncludeServerTime tells Centrifuge to append `time` field to Connect result of client protocol.
	// This field contains Unix timestamp in milliseconds and represents current server time. By default, server time
	// is not included.
	ClientConnectIncludeServerTime bool
	// ClientPresenceUpdateInterval sets an interval how often connected
	// clients update presence information.
	// Zero value means 25 * time.Second.
	ClientPresenceUpdateInterval time.Duration

	// ClientExpiredCloseDelay is an extra time given to client to refresh
	// its connection in the end of connection TTL. At moment only used for
	// a client-side refresh workflow.
	// Zero value means 25 * time.Second.
	ClientExpiredCloseDelay time.Duration
	// ClientExpiredSubCloseDelay is an extra time given to client to
	// refresh its expiring subscription in the end of subscription TTL.
	// At the moment only used for a client-side subscription refresh workflow.
	// Zero value means 25 * time.Second.
	ClientExpiredSubCloseDelay time.Duration
	// ClientStaleCloseDelay is a timeout after which connection will be
	// closed if still not authenticated (i.e. no valid connect command
	// received yet).
	// Zero value means 15 * time.Second.
	ClientStaleCloseDelay time.Duration
	// ClientChannelPositionCheckDelay defines minimal time from previous
	// client position check in channel. If client does not pass check it
	// will be disconnected with DisconnectInsufficientState.
	// Zero value means 40 * time.Second.
	ClientChannelPositionCheckDelay time.Duration
	// Maximum allowed time lag for publications for subscribers with positioning on.
	// When exceeded we mark connection with insufficient state. By default, not used - i.e.
	// Centrifuge does not take lag into account for positioning.
	// See also pub_sub_time_lag_seconds as a helpful metric.
	ClientChannelPositionMaxTimeLag time.Duration
	// ClientQueueMaxSize is a maximum size of client's message queue in
	// bytes. After this queue size exceeded Centrifuge closes client's connection.
	// Zero value means 1048576 bytes (1MB).
	ClientQueueMaxSize int
	// ClientChannelLimit sets upper limit of client-side channels each client
	// can subscribe to. Client-side subscriptions attempts will get an ErrorLimitExceeded
	// in subscribe reply. Server-side subscriptions above limit will result into
	// DisconnectChannelLimit.
	// Zero value means 128.
	ClientChannelLimit int
	// UserConnectionLimit limits number of client connections to single Node
	// from user with the same ID. Zero value means unlimited. Anonymous users
	// can't be tracked.
	UserConnectionLimit int
	// ChannelMaxLength is the maximum length of a channel name. This is only checked
	// for client-side subscription requests.
	// Zero value means 255.
	ChannelMaxLength int
	// HistoryMaxPublicationLimit allows limiting the maximum number of publications to be
	// asked over client API history call. This is useful when you have large streams and
	// want to prevent a massive number of missed messages to be sent to a client when
	// calling history without any limit explicitly set. By default, no limit used.
	// This option does not affect Node.History method. See also RecoveryMaxPublicationLimit.
	HistoryMaxPublicationLimit int
	// RecoveryMaxPublicationLimit allows limiting the number of Publications that could be
	// restored during the automatic recovery process. See also HistoryMaxPublicationLimit.
	// By default, no limit used.
	RecoveryMaxPublicationLimit int
	// UseSingleFlight allows turning on mode where singleflight will be automatically used
	// for Node.History (including recovery), Node.Presence/Node.PresenceStats,
	// and Node.MapStateRead/Node.MapStreamRead/Node.MapStats calls.
	UseSingleFlight bool
	// HistoryMetaTTL sets a time of stream meta key expiration in Redis. Stream
	// meta key is a Redis HASH that contains top offset in channel and epoch value.
	// In some cases – when channels created for а short time and then
	// not used anymore – created stream meta keys can stay in memory while
	// not actually useful. For example, you can have a personal user channel but
	// after using your app for a while user left it forever. In long-term
	// perspective this can be an unwanted memory leak. Setting a reasonable
	// value to this option (usually much bigger than history retention period)
	// can help. In this case unused channel stream metadata will eventually expire.
	//
	// Keep this value much larger than history stream TTL used when publishing.
	// When zero Centrifuge uses default 30 days which we believe is more than enough
	// for most use cases.
	HistoryMetaTTL time.Duration
	// Metrics is MetricsConfig to configure Prometheus metrics provided by Centrifuge.
	Metrics MetricsConfig
	// GetChannelMediumOptions is a way to provide ChannelMediumOptions for specific channel.
	// This function is called each time new channel appears on the Node.
	// See the doc comment for ChannelMediumOptions for more details about channel medium concept.
	GetChannelMediumOptions func(channel string) ChannelMediumOptions
	// GetBroker when set allows returning a custom Broker to use for a specific channel. If not set
	// then the default Node's Broker is always used for all channels. Also, Node's default Broker is
	// always used for control channels. It's the responsibility of an application to call Broker.Run
	// method of all brokers except the default one (called automatically inside Node.Run). Also, a
	// proper Broker shutdown is the responsibility of application because Node does not know about
	// custom Broker instances. When GetBroker returns false as the second argument then Node will
	// use the default Broker for the channel.
	GetBroker func(channel string) (Broker, bool)
	// GetPresenceManager when set allows returning a custom PresenceManager to use for a specific
	// channel. If not set then the default Node's PresenceManager is always used for all channels.
	// A proper PresenceManager shutdown is the responsibility of application because Node does not
	// know about custom PresenceManager instances. When GetPresenceManager returns false as the second
	// argument then Node will use the default PresenceManager for the channel.
	GetPresenceManager func(channel string) (PresenceManager, bool)
	// Tell Centrifuge how to transform connect error codes to disconnect objects for unidirectional
	// transports. If not set or code not found in the mapping then Centrifuge falls back to the default
	// mapping defined internally.
	UnidirectionalCodeToDisconnect map[uint32]Disconnect
	// GetChannelBatchConfig allows configuring per-channel write batching. Batching config if
	// returned is applied for publications and join/leave channel pushes for all channel subscribers.
	// The cost of batching are extra goroutines, buffers and extra timers for each channel used in
	// batching, so you can expect memory overhead. But batching may be useful for reducing CPU usage
	// coming from write system calls in channels with high publication rate. If GetChannelBatchConfig
	// not set then no batching is used on per-channel level. This function may be called in the hot
	// broadcast path, so must be fast. This is an EXPERIMENTAL feature.
	GetChannelBatchConfig func(channel string) ChannelBatchConfig
	// DictionaryCompression enables connection-level frame compression against a
	// shared dictionary. Nil - the default - disables it entirely: nothing is
	// allocated and no compression code path is entered.
	//
	// This package deliberately ships no implementation. Deciding which
	// dictionary a connection gets, where dictionaries come from and what may go
	// into one are product decisions with real consequences, so they belong to
	// whoever supplies the engine rather than here.
	//
	// It only applies to connections whose client advertised support, so it is
	// safe to enable for a mixed fleet of old and new clients. This is an
	// EXPERIMENTAL feature.
	DictionaryCompression DictionaryCompression
	// ClientTimerScheduler if set will be used for scheduling client timers.
	// This is an EXPERIMENTAL API.
	ClientTimerScheduler TimerScheduler

	// Map contains options for map subscriptions.
	// This is an EXPERIMENTAL API.
	Map MapConfig
	// SharedPoll contains options for shared poll subscriptions.
	// This is an EXPERIMENTAL API.
	SharedPoll SharedPollConfig
	// contains filtered or unexported fields
}

Config contains Node configuration options.

type ConnectEvent 

type ConnectEvent struct {
	// ClientID that was generated by library for client connection.
	ClientID string
	// Token received from client as part of Connect Command.
	Token string
	// Data received from client as part of Connect Command.
	Data []byte
	// Name can contain client name if provided on connect.
	Name string
	// Version can contain client version if provided on connect.
	Version string
	// Profile can contain the profile name the client declared on connect.
	// This is an unvalidated client assertion, exactly like Name and Version:
	// treat it as a request, not a fact. Return ConnectReply.Profile to set the
	// profile the server will actually use - whatever is returned there wins,
	// and returning an empty one discards the client's claim entirely.
	Profile string
	// Transport contains information about transport used by client.
	Transport TransportInfo
	// Channels is a list of channels a client wants to subscribe to
	// (server-side). It's just a way for a client to provide this list.
	// Server should use ConnectReply.Subscriptions to tell Centrifuge
	// the final list of server-side subscriptions for a connection which
	// can differ from the Channels list.
	Channels []string
	// Headers can be set by SDK to pass custom key-value headers from
	// client to server. It's up to the application to decide what to do
	// with these headers.
	Headers map[string]string
}

ConnectEvent contains fields related to connecting event (when a server received Connect protocol command from client).

type ConnectHandler 

type ConnectHandler func(*Client)

ConnectHandler called when client connected to server and ready to communicate.

type ConnectReply 

type ConnectReply struct {
	// Context allows returning a modified context.
	Context context.Context
	// Credentials should be set if app wants to authenticate connection.
	// This field is optional since auth Credentials could be set through
	// HTTP middleware.
	Credentials *Credentials
	// Data allows setting custom data in connect reply.
	Data []byte
	// Subscriptions map contains channels to subscribe connection to on server-side.
	Subscriptions map[string]SubscribeOptions
	// ClientSideRefresh tells library to use client-side refresh logic:
	// i.e. send refresh commands with new connection token. If not set
	// then server-side refresh mechanism will be used.
	ClientSideRefresh bool
	// Storage can be used to fill initial connection storage during connecting.
	// This data may be then accessed/modified/replaced later during Client's lifetime
	// over Client.AcquireStorage() call. This API is EXPERIMENTAL.
	Storage map[string]any

	// Profile is the application context this connection belongs to - which view,
	// screen or client kind it is. Connections sharing a profile see traffic of a
	// similar shape.
	//
	// Setting it here overrides whatever the client declared in its connect
	// command, and is the recommended way to classify connections: a claim in the
	// token this handler already reads is one change to the service issuing
	// tokens, rather than a release of every client.
	//
	// It is carried on the Client and can be read with Client.Profile(). Nothing
	// in this package interprets it; it exists so features layered on top - the
	// obvious one being compression dictionaries - can group connections that
	// look alike.
	Profile string

	// MaxMessagesInFrame is the maximum number of messages (replies and pushes) which
	// Centrifuge Client message writer will collect from the client's queue before sending
	// to the connection. By default, it's 16. Use -1 to disable the limit.
	MaxMessagesInFrame int
	// WriteDelay is a time Centrifuge will try to collect messages inside message writer loop
	// before sending them towards this connection. Enabling WriteDelay may reduce CPU usage of
	// both server and client in case of high message rate inside individual connections. The
	// reduction happens due to the lesser number of system calls to execute. Enabling WriteDelay
	// limits the maximum throughput of messages towards the connection which may be achieved.
	// For example, if WriteDelay is 100ms then the max throughput per second will be
	// (1000 / 100) * MaxMessagesInFrame (16 by default), i.e. 160 messages per second. This
	// should be more than enough for target Centrifuge use cases (frontend apps) though.
	WriteDelay time.Duration
	// WriteWithTimer enables using write timer for this client connection. This mode is only
	// active when WriteDelay is set to a non-zero value. In this mode Centrifuge uses a timer
	// to schedule batched writes to the connection instead of checking for messages in a
	// dedicated goroutine loop.
	WriteWithTimer bool
	// ReplyWithoutQueue when enabled will force Centrifuge to avoid using Client write
	// queue for sending replies to commands for this connection. Replies sent directly to
	// the Client's transport thus avoiding possible delays caused by writer loop, but replies
	// lose a chance to be batched.
	ReplyWithoutQueue bool
	// QueueInitialCap set an initial capacity for client's message queue, the size of queue
	// can grow further, but won't be reduced below QueueInitialCap. By default, it's 2.
	QueueInitialCap int
	// QueueShrinkDelay is a time Centrifuge will wait, after the client's message
	// queue has been drained, before shrinking it back down. This option only
	// applies when WriteDelay is set to a non-zero value.
	//
	// It matters because shrinking is evaluated immediately after a drain, when
	// the queue is empty by construction. Acting on that would discard the ring
	// the writer had just filled, and the next frame would rebuild it one
	// doubling at a time - allocating on every write cycle for the lifetime of
	// the connection. Waiting instead lets the queue settle: under sustained load
	// the timer keeps resetting and capacity is simply retained, and once the
	// connection goes quiet the timer fires and the memory is released in full.
	//
	// Zero value (the default) means Centrifuge picks a sensible delay itself
	// (currently one second). Pass a negative value to shrink immediately after
	// every drain - available for completeness, but it reintroduces the per-cycle
	// allocations described above and is not recommended.
	//
	// Broadcasting to 200 subscribers with WriteDelay set, per publication:
	//
	//	shrink immediately   19.1us   84 allocs   34 KB
	//	default delay        12.7us    2 allocs  376 B
	//
	// A longer delay holds capacity through longer quiet gaps, which suits
	// connections that burst repeatedly; a shorter one returns memory sooner.
	QueueShrinkDelay time.Duration
	// PingPongConfig if set, will override Transport's PingPongConfig to enable setting ping/pong interval
	// for individual client.
	PingPongConfig *PingPongConfig
	// Labels is a map of custom label key-value pairs that will be attached to the client connection.
	// Labels can be used for any application purpose. Additionally, labels defined in MetricsConfig.ClientLabels
	// will be exported to Prometheus metrics for all client metrics.
	// For exported metric labels: if a defined label is not found in this map, an empty string will be used.
	Labels map[string]string
}

ConnectReply contains reaction to ConnectEvent.

type ConnectRequest added in v0.16.0 

type ConnectRequest struct {
	// Token is an optional token from a client.
	Token string
	// Data is an optional custom data from a client.
	Data []byte
	// Name of a client.
	Name string
	// Version of a client.
	Version string
	// Subs is a map with channel subscription state (for recovery on connect).
	Subs map[string]SubscribeRequest
	// Headers represent headers which may be used for headers emulation feature.
	Headers map[string]string
}

ConnectRequest can be used in a unidirectional connection case to pass initial connection information from a client-side.

type ConnectingHandler added in v0.0.2 

type ConnectingHandler func(context.Context, ConnectEvent) (ConnectReply, error)

ConnectingHandler called when new client authenticates on server.

type ControlEventHandler added in v0.36.0 

type ControlEventHandler interface {
	// HandleControl to handle received control data.
	HandleControl(data []byte) error
}

ControlEventHandler can handle messages received from Controller.

type Controller added in v0.36.0 

type Controller interface {
	// RegisterControlEventHandler called once on start when controller already set to node. At
	// this moment node is ready to process controller events.
	RegisterControlEventHandler(ControlEventHandler) error
	// PublishControl allows sending control command data. If nodeID is empty string
	// then message should be delivered to all running nodes, if nodeID is set then
	// message should be delivered only to node with specified ID. If shardKey is set
	// then it can be used for ordering control (now not used).
	PublishControl(data []byte, nodeID, shardKey string) error
}

type Credentials 

type Credentials struct {
	// UserID tells library an ID of current user. Leave this empty string
	// if you need access from anonymous user.
	UserID string
	// ExpireAt allows setting time in future when connection must be validated.
	// In this case Client.OnRefresh callback must be set by application. Zero
	// value means no expiration.
	ExpireAt int64
	// Info contains additional information about connection. This data will be
	// included untouched into Join/Leave messages, into Presence information,
	// also info can become a part of published message as part of ClientInfo.
	// In some cases having additional info can be an undesired overhead – but
	// you are simply free to not use this field at all.
	Info []byte
}

Credentials allow authenticating connection when set into context.

func GetCredentials added in v0.5.0 

func GetCredentials(ctx context.Context) (*Credentials, bool)

GetCredentials allows extracting Credentials from Context (if set previously).

type DeltaType added in v0.33.0 

type DeltaType string
const (

	// DeltaTypeFossil is Fossil delta encoding. See https://fossil-scm.org/home/doc/tip/www/delta_encoder_algorithm.wiki.
	DeltaTypeFossil DeltaType = "fossil"
)

type DictionaryAwareTransport added in v0.39.0 

type DictionaryAwareTransport interface {
	// SetDictionaryCompression installs a codec that must not encode the next
	// frame written, because that frame is the connect reply carrying the
	// dictionary this codec uses.
	SetDictionaryCompression(cc DictionaryConnection)
	// CloseDictionaryCompression is called once the writer has stopped, so an
	// implementation's Close cannot race the last Encode.
	CloseDictionaryCompression()
}

DictionaryAwareTransport is implemented by transports that can carry dictionary-compressed frames. A transport that does not implement it is left alone: no dictionary is offered on it, and the connect reply never claims compression that was not installed. That is what makes it safe to enable the feature for a mixed fleet, where a client may arrive over a fallback transport that cannot carry a compressed frame.

It is exported so transports defined outside this package can opt in, and only for that. Centrifuge calls these methods itself, as part of the connect handshake and connection teardown; application code should not. Installing a codec outside the handshake compresses frames against a dictionary the client was never sent, and it has no way to report that it cannot read them. Whether a transport can is a question about framing rather than effort: a compressed frame is arbitrary bytes, so a transport that delimits messages with a newline cannot carry one without re-encoding it and giving the saving straight back.

type DictionaryCompression added in v0.39.0 

type DictionaryCompression interface {
	// NewDictionaryConnection is called once per connection, after the client has been
	// authenticated and before the connect reply is written.
	//
	// Returning nil leaves the connection uncompressed, which is the right answer
	// for a client that cannot decode, a profile with nothing to offer, or any
	// case the implementation would rather sit out.
	NewDictionaryConnection(params DictionaryConnectionParams) DictionaryConnection
}

DictionaryCompression compresses outgoing frames against a shared dictionary.

This package provides no implementation. It defines the shape of one and the points where a connection hands work to it, so compression can be supplied from outside - which is where the decisions live: which dictionary a connection gets, where dictionaries come from, and what may go in one.

A nil DictionaryCompression means the feature is absent. Nothing is allocated and no code path here is entered.

type DictionaryConnection added in v0.39.0 

type DictionaryConnection interface {
	// Dictionary returns what to put in ConnectResult.dict, or nil to send
	// nothing.
	//
	// It is called once, before the connect reply is written.
	//
	// Compression always begins on the frame AFTER the connect reply, whether
	// this returns bytes or only an id. The reply itself goes out in the plain
	// protocol - text JSON, length-prefixed Protobuf - so a client never has to
	// work out whether the frame that establishes compression was itself
	// compressed.
	//
	// A reply could be compressed when the client already holds the dictionary,
	// and briefly was. Doing so requires the client to decide how to decode the
	// reply before it can read it, and the two encodings are not reliably
	// distinguishable: a Protobuf ping is an empty reply, a single 0x00 byte,
	// which is exactly what a raw frame marker looks like. Every scheme that
	// resolves that costs either a byte on every frame or an invariant about
	// minimum frame sizes that nobody would think to preserve - to save one
	// small frame, once per connection.
	//
	// Returning only an id means the client presented that id and already holds
	// the bytes, so nothing is transferred - which is where the saving on a
	// returning connection actually is, and it is unaffected by any of this.
	//
	// Set only the id, with no bytes, when this engine RECOGNISES the id the
	// client advertised - meaning it holds that dictionary and will compress
	// against it. Matching the advertised id is necessary but not sufficient:
	// echoing back an id merely because the client sent it is a security bug,
	// not just a correctness one.
	//
	// A client's stored dictionary can be tampered with - browser storage is
	// writable by anything running on the origin - and a client cannot fully
	// detect it, since bytes and id rewritten together verify locally. What
	// stops that is the server refusing to recognise an id it never issued, so
	// the tampered bytes are replaced rather than used. An engine that echoes
	// the advertised id removes exactly that defence: the client then decodes
	// real server frames against attacker-chosen content, because DEFLATE back
	// references resolve into whatever dictionary is installed.
	//
	// Centrifuge cannot check this for you. It does not know which dictionaries
	// an engine holds, so it can only reject an id the client never mentioned.
	Dictionary() *protocol.Dictionary

	// Encode is called for every frame this connection compresses, on the
	// connection's write goroutine - including the connect reply when the
	// client already held the dictionary.
	//
	// It returns the bytes to write and whether they must go out as a binary
	// message, which compressed payloads need even on a JSON connection.
	//
	// One publication reaches every subscriber of a channel as the same bytes,
	// so implementations are called with an identical frame once per
	// subscriber, from that many goroutines, at nearly the same instant. What
	// an implementation does about that decides what the feature costs: a
	// dictionary compression is dominated by loading the dictionary rather than
	// by the frame, so compressing a fan-out once instead of once per
	// subscriber is the difference between the feature being viable and not.
	//
	// A cache alone does not achieve it. The subscribers arrive together and
	// all miss, because the first has not finished compressing yet - measured
	// on a four-subscriber channel, the same frame was compressed three or four
	// times over while the cache reported a plausible hit rate. Collapsing the
	// duplicates as well removed half the compressions and a third of the CPU.
	//
	// This package deliberately provides neither, so that engines keep control
	// of their own memory and lifetime. It is what an implementation should
	// build first.
	Encode(frame []byte) (out []byte, binary bool)

	// Close is called once when the connection goes away, on the same goroutine
	// as the final Encode and never concurrently with one.
	//
	// Implementations that batch their accounting need this, and need it to be
	// exact rather than best-effort. Without it, everything a connection did
	// since its last flush is lost - and what is lost is not a random sample:
	// short connections are the ones that end before a flush, and they are also
	// the ones that most often pay to be sent a dictionary. Dropping them makes
	// compression look better than it is, in a number an operator uses to decide
	// whether the feature is worth keeping on.
	Close()
}

DictionaryConnection compresses one connection's outgoing frames against a dictionary. It is named for the dictionary rather than for compression in general: a connection may also be using permessage-deflate, delta compression or channel compaction, and none of those go through here.

type DictionaryConnectionParams added in v0.39.0 

type DictionaryConnectionParams struct {
	// ProtocolType is the connection's protocol. A dictionary built from JSON
	// frames is useless on Protobuf, so implementations must keep them apart.
	ProtocolType ProtocolType

	// ClientFlags is the capability bitmask the client advertised.
	ClientFlags int64

	// Profile is the application context this connection belongs to, resolved by
	// the server. Empty means unclassified.
	Profile string

	// UserID is the authenticated user, or empty for an anonymous connection.
	//
	// It is here because it is the only identity that survives a reconnect, and
	// a staged rollout needs one: a cohort re-drawn on every reconnect is not a
	// cohort, and a client that fell out of one would pay to be sent a
	// dictionary it already had. Implementations that split traffic should
	// derive the split from this rather than from anything per-connection.
	UserID string

	// HeldDictionaryID is the dictionary the client says it already has from an
	// earlier connection. An id identifies dictionary content, so a match means
	// both sides hold the same bytes and the dictionary can be named rather than
	// sent. An unrecognised id is simply ignored.
	HeldDictionaryID string
}

DictionaryConnectionParams describes the client a DictionaryConnection is being made for. It is a struct rather than an argument list so more can be negotiated later without breaking implementations outside this package.

type Disconnect 

type Disconnect struct {
	// Code is a disconnect code.
	Code uint32 `json:"code,omitempty"`
	// Reason is a short description of disconnect code for humans.
	Reason string `json:"reason"`
}

Disconnect allows configuring how client will be disconnected from a server. A server can provide a Disconnect.Code and Disconnect.Reason to a client. Clients can execute some custom logic based on a certain Disconnect.Code. Code is also used for metric collection. Disconnect.Reason is optional and exists mostly for human-readable description of returned code – i.e. for logging, debugging etc.

The important note is that Disconnect.Reason must be less than 127 bytes due to WebSocket protocol limitations.

Codes have some rules which should be followed by a client connector implementation. These rules described below.

Codes in range 0-2999 should not be used by a Centrifuge library user. Those are reserved for the client-side and transport specific needs. Codes in range >=5000 should not be used also. Those are reserved by Centrifuge.

Client should reconnect upon receiving code in range 3000-3499, 4000-4499, >=5000. For codes <3000 reconnect behavior can be adjusted for specific transport.

Codes in range 3500-3999 and 4500-4999 are application terminal codes, no automatic reconnect should be made by a client implementation.

Library users supposed to use codes in range 4000-4999 for creating custom disconnects.

func (Disconnect) Error added in v0.10.0 

func (d Disconnect) Error() string

Error to use Disconnect as a callback handler error to signal Centrifuge that client must be disconnected with corresponding Code and Reason.

func (Disconnect) String added in v0.8.2 

func (d Disconnect) String() string

String representation.

type DisconnectEvent 

type DisconnectEvent struct {
	// Disconnect contains a Disconnect object which identifies the code and reason
	// of disconnect process. When disconnect was not initiated by a server this
	// is always DisconnectConnectionClosed.
	Disconnect
}

DisconnectEvent contains fields related to disconnect event.

type DisconnectHandler 

type DisconnectHandler func(DisconnectEvent)

DisconnectHandler called when client disconnects from server. The important thing to remember is that you should not rely entirely on this handler to clean up non-expiring resources (in your database for example). Why? Because in case of any non-graceful node shutdown (kill -9, process crash, machine lost) disconnect handler will never be called (obviously) so you can have stale data.

type DisconnectOption added in v0.8.0 

type DisconnectOption func(options *DisconnectOptions)

DisconnectOption is a type to represent various Disconnect options.

func WithCustomDisconnect added in v0.23.0 

func WithCustomDisconnect(disconnect Disconnect) DisconnectOption

WithCustomDisconnect allows setting custom Disconnect.

func WithDisconnectAllUsers added in v0.39.0 

func WithDisconnectAllUsers(enabled bool) DisconnectOption

WithDisconnectAllUsers changes the meaning of an empty userID passed to Node.Disconnect from "anonymous bucket only" to "all users including anonymous." It is a no-op when userID is non-empty.

func WithDisconnectClient added in v0.18.0 

func WithDisconnectClient(clientID string) DisconnectOption

WithDisconnectClient allows setting Client.

func WithDisconnectClientWhitelist added in v0.18.0 

func WithDisconnectClientWhitelist(whitelist []string) DisconnectOption

WithDisconnectClientWhitelist allows setting ClientWhitelist.

func WithDisconnectLabelFilter added in v0.39.0 

func WithDisconnectLabelFilter(f *FilterNode) DisconnectOption

WithDisconnectLabelFilter restricts the disconnect to connections whose Client.Labels match the filter. See DisconnectOptions.LabelFilter for the detailed contract.

func WithDisconnectSession added in v0.21.1 

func WithDisconnectSession(sessionID string) DisconnectOption

WithDisconnectSession allows setting session ID to disconnect.

type DisconnectOptions added in v0.8.0 

type DisconnectOptions struct {
	// Disconnect represents custom disconnect to use.
	// By default, DisconnectForceNoReconnect will be used.
	Disconnect *Disconnect
	// ClientWhitelist contains client IDs to keep.
	ClientWhitelist []string
	// contains filtered or unexported fields
}

DisconnectOptions define some fields to alter behaviour of Disconnect operation.

type EmulationConfig added in v0.23.0 

type EmulationConfig struct {
	// MaxRequestBodySize limits request body size (in bytes). By default we accept 64kb max.
	MaxRequestBodySize int
}

EmulationConfig is a config for EmulationHandler.

type EmulationHandler added in v0.23.0 

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

EmulationHandler allows receiving client protocol commands from client and proxy them to the right node (where client session lives). This makes it possible to use unidirectional transports for server-to-clients data flow but still emulate bidirectional connection - thanks to this handler. Redirection to the correct node works over Survey.

func NewEmulationHandler added in v0.23.0 

func NewEmulationHandler(node *Node, config EmulationConfig) *EmulationHandler

NewEmulationHandler creates new EmulationHandler.

func (*EmulationHandler) ServeHTTP added in v0.23.0 

func (s *EmulationHandler) ServeHTTP(rw http.ResponseWriter, r *http.Request)

type Error 

type Error struct {
	Code      uint32
	Message   string
	Temporary bool
}

Error represents client reply error. Library user can define own application specific errors. When defining new custom errors use error codes in range [400, 1999] assuming that codes in interval 0-399 are reserved by Centrifuge.

func (*Error) Error added in v0.8.0 

func (e *Error) Error() string

type FilterNode added in v0.39.0 

type FilterNode = protocol.FilterNode

FilterNode is a filter expression tree for matching against key-value tags. Used for server-side publication filtering (ServerTagsFilter in SubscribeOptions).

type HTTPStreamConfig added in v0.23.0 

type HTTPStreamConfig struct {
	PingPongConfig
	// MaxRequestBodySize limits request body size.
	MaxRequestBodySize int
}

HTTPStreamConfig represents config for HTTPStreamHandler.

type HTTPStreamHandler added in v0.23.0 

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

HTTPStreamHandler handles WebSocket client connections. WebSocket protocol is a bidirectional connection between a client and a server for low-latency communication.

func NewHTTPStreamHandler added in v0.23.0 

func NewHTTPStreamHandler(node *Node, config HTTPStreamConfig) *HTTPStreamHandler

NewHTTPStreamHandler creates new HTTPStreamHandler.

func (*HTTPStreamHandler) ServeHTTP added in v0.23.0 

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

type HistoryCallback added in v0.13.0 

type HistoryCallback func(HistoryReply, error)

HistoryCallback should be called with HistoryReply or error.

type HistoryEvent added in v0.10.0 

type HistoryEvent struct {
	Channel string
	Filter  HistoryFilter
}

HistoryEvent has channel operation called for.

type HistoryFilter 

type HistoryFilter struct {
	// Since used to extract publications from stream since provided StreamPosition.
	Since *StreamPosition
	// Limit number of publications to return.
	// -1 means no limit - i.e. return all publications currently in stream.
	// 0 means that caller only interested in current stream top position so
	// Broker should not return any publications.
	Limit int
	// Reverse direction.
	Reverse bool
}

HistoryFilter allows filtering history according to fields set.

type HistoryHandler added in v0.10.0 

type HistoryHandler func(HistoryEvent, HistoryCallback)

HistoryHandler must handle incoming command from client.

type HistoryOption added in v0.8.0 

type HistoryOption func(options *HistoryOptions)

HistoryOption is a type to represent various History options.

func WithHistoryFilter added in v0.29.0 

func WithHistoryFilter(filter HistoryFilter) HistoryOption

func WithHistoryMetaTTL added in v0.29.0 

func WithHistoryMetaTTL(metaTTL time.Duration) HistoryOption

func WithLimit added in v0.8.0 

func WithLimit(limit int) HistoryOption

WithLimit allows setting HistoryOptions.Limit.

func WithReverse added in v0.18.0 

func WithReverse(reverse bool) HistoryOption

WithReverse allows setting HistoryOptions.Reverse option.

func WithSince added in v0.17.0 

func WithSince(sp *StreamPosition) HistoryOption

WithSince allows setting HistoryOptions.Since option.

type HistoryOptions added in v0.8.0 

type HistoryOptions struct {
	// Filter for history publications.
	Filter HistoryFilter
	// MetaTTL allows overriding default (set in Config.HistoryMetaTTL) history
	// meta information expiration time.
	MetaTTL time.Duration
}

HistoryOptions define some fields to alter History method behaviour.

type HistoryReply added in v0.10.0 

type HistoryReply struct {
	Result *HistoryResult
}

HistoryReply contains fields determining the reaction on history request.

type HistoryResult added in v0.8.0 

type HistoryResult struct {
	// StreamPosition embedded here describes current stream top offset and epoch.
	StreamPosition
	// Publications extracted from history storage according to HistoryFilter.
	Publications []*Publication
}

HistoryResult contains Publications and current stream top StreamPosition.

type Hub 

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

Hub tracks Client connections on the current Node.

func (*Hub) BroadcastPublication added in v0.17.1 

func (h *Hub) BroadcastPublication(ch string, pub *Publication, sp StreamPosition) error

BroadcastPublication sends message to all clients subscribed on a channel on the current Node. Usually this is NOT what you need since in most cases you should use Node.Publish method which uses a Broker to deliver publications to all Nodes in a cluster and maintains publication history in a channel with incremental offset. By calling BroadcastPublication messages will only be sent to the current node subscribers without any defined offset semantics, without delta support.

func (*Hub) BroadcastPublicationDelta added in v0.39.0 

func (h *Hub) BroadcastPublicationDelta(ch string, pub *Publication, prevPub *Publication, sp StreamPosition) error

BroadcastPublicationDelta is like BroadcastPublication but supports delta compression. When prevPub is non-nil, subscribers with delta enabled receive a computed delta instead of the full publication data. Only sent to the current node subscribers.

func (*Hub) Channels 

func (h *Hub) Channels() []string

Channels returns a slice of all active channels.

func (*Hub) Connections added in v0.23.0 

func (h *Hub) Connections() map[string]*Client

Connections returns all user connections to the current Node.

func (*Hub) NumChannels 

func (h *Hub) NumChannels() int

NumChannels returns a total number of different channels.

func (*Hub) NumClients 

func (h *Hub) NumClients() int

NumClients returns total number of client connections.

func (*Hub) NumSubscribers 

func (h *Hub) NumSubscribers(ch string) int

NumSubscribers returns number of current subscribers for a given channel.

func (*Hub) NumSubscriptions added in v0.18.0 

func (h *Hub) NumSubscriptions() int

NumSubscriptions returns a total number of subscriptions.

func (*Hub) NumUsers 

func (h *Hub) NumUsers() int

NumUsers returns a number of unique users connected.

func (*Hub) UserConnections added in v0.18.0 

func (h *Hub) UserConnections(userID string) map[string]*Client

UserConnections returns all user connections to the current Node.

type Info 

type Info struct {
	Nodes []NodeInfo
}

Info contains information about all known server nodes.

type KeyMode added in v0.39.0 

type KeyMode string

KeyMode controls conditional publishing based on key existence in the state. 

const (
	// KeyModeReplace always writes the key (default behavior, backward compatible).
	KeyModeReplace KeyMode = ""

	// KeyModeIfNew only writes if the key does NOT exist in the state.
	// Use cases: lobby slots, distributed locks, first-claim resources.
	// When key already exists, Suppressed=true with SuppressReason="key_exists".
	KeyModeIfNew KeyMode = "if_new"

	// KeyModeIfExists only writes if the key ALREADY exists in the state.
	// Use cases: heartbeats, update-only operations, presence renewal.
	// When key does not exist, Suppressed=true with SuppressReason="key_not_found".
	KeyModeIfExists KeyMode = "if_exists"
)

type LogEntry 

type LogEntry struct {
	Level   LogLevel
	Message string
	Fields  map[string]any
	Error   error
}

LogEntry represents log entry.

type LogHandler 

type LogHandler func(LogEntry)

LogHandler handles log entries - i.e. writes into correct destination if necessary.

type LogLevel 

type LogLevel int

LogLevel describes the chosen log level. 

const (
	// LogLevelNone means no logging.
	LogLevelNone LogLevel = iota
	// LogLevelTrace turns on trace logs - should only be used during development. This
	// log level shows all client-server communication.
	LogLevelTrace
	// LogLevelDebug turns on debug logs - it's generally too much for production in normal
	// conditions but can help when developing and investigating problems in production.
	LogLevelDebug
	// LogLevelInfo logs useful server information. This includes various information
	// about problems with client connections which is not Centrifuge errors but
	// in most situations malformed client behaviour.
	LogLevelInfo
	// LogLevelWarn logs server warnings. This may contain tips for a developer about things
	// which should be addressed but usually not immediately.
	LogLevelWarn
	// LogLevelError level logs only server errors. This is logging that means non-working
	// Centrifuge and may require effort from developers/administrators to make things
	// work again.
	LogLevelError
)

type MapBroker added in v0.39.0 

type MapBroker interface {
	// RegisterEventHandler registers a BrokerEventHandler to handle publications
	// received from pub/sub. Called by Node.Run() during startup.
	RegisterEventHandler(BrokerEventHandler) error

	// Subscribe registers this server node to receive pub/sub messages for the channels.
	// Called when a client subscribes to a map channel on this node. Subscribe
	// and Unsubscribe of the same channel are never called concurrently:
	// callers serialize them per channel.
	Subscribe(channels ...string) error

	// Unsubscribe removes this server node from receiving pub/sub messages for the channels.
	// Called when the last client on this node unsubscribes from the channel.
	Unsubscribe(channels ...string) error

	// Publish updates the state and broadcasts the change to subscribers.
	//
	// The operation:
	//  1. Updates the key in the state (subject to KeyMode conditions)
	//  2. Appends the publication to the stream (if StreamSize > 0)
	//  3. Broadcasts to pub/sub subscribers
	//
	// Returns MapUpdateResult with Suppressed=false when the state was changed.
	// When Suppressed=true (idempotency, version conflict, or KeyMode condition),
	// no stream entry is created and no pub/sub broadcast occurs.
	Publish(ctx context.Context, ch string, key string, opts MapPublishOptions) (MapUpdateResult, error)

	// Remove removes a key from the state and notifies subscribers.
	//
	// The operation:
	//  1. Removes the key from the state
	//  2. Appends a removal publication to the stream (if StreamSize > 0)
	//  3. Broadcasts removal to pub/sub subscribers
	//
	// Returns MapUpdateResult with Suppressed=false when the key was removed.
	// When Suppressed=true (key not found or idempotency), no stream entry or broadcast occurs.
	Remove(ctx context.Context, ch string, key string, opts MapRemoveOptions) (MapUpdateResult, error)

	// ReadStream retrieves publications from the channel's history stream.
	//
	// Used by clients to catch up on changes after receiving the state.
	// Supports filtering by position (Since) and pagination (Limit).
	// Returns publications, current stream position, and error.
	//
	// Special Limit values:
	//  - Limit = 0: Only return current stream position (no publications)
	//  - Limit = -1: Return all publications (no limit)
	ReadStream(ctx context.Context, ch string, opts MapReadStreamOptions) (MapStreamResult, error)

	// ReadState retrieves the current key-value state for a channel.
	//
	// Returns all keys in the state as publications (each with Key field set).
	// Supports cursor-based pagination for large states.
	//
	// If opts.Revision is provided, validates that the epoch hasn't changed.
	// Returns ErrorUnrecoverablePosition if epoch changed (client must restart).
	ReadState(ctx context.Context, ch string, opts MapReadStateOptions) (MapStateResult, error)

	// Stats returns statistics about the channel in storage.
	Stats(ctx context.Context, ch string) (MapStats, error)

	// Clear deletes all data for a channel (state and stream).
	// Use for cleanup when a channel is no longer needed.
	Clear(ctx context.Context, ch string, opts MapClearOptions) error
}

MapBroker is the interface for managing map subscription state. See "MapBroker Overview" above for detailed documentation.

type MapChannelOptions added in v0.39.0 

type MapChannelOptions struct {
	// Mode controls synchronization, recovery, and data lifecycle.
	// Required. Zero value = not configured = error.
	Mode MapMode
	// KeyTTL sets automatic expiration for entries in this channel.
	// Required when Mode.HasExpiry() (must be > 0).
	// Must be 0 when Mode is Persistent.
	KeyTTL time.Duration
	// StreamSize sets the maximum number of entries in the recovery stream.
	// Zero = auto-derived (100) for Recoverable/Persistent. Must be 0 for Ephemeral.
	StreamSize int
	// StreamTTL sets how long stream entries are retained.
	// Zero = auto-derived (1 minute) for Recoverable/Persistent. Must be 0 for Ephemeral.
	StreamTTL time.Duration
	// MetaTTL sets how long stream metadata (epoch, offset) is retained.
	// Zero = auto-derived: Recoverable: StreamTTL*10, Persistent: permanent (no expiry).
	// Must be 0 for Ephemeral.
	MetaTTL time.Duration
	// DefaultPageSize sets the default number of items per page when
	// the client does not specify a page size. Zero means default (100).
	DefaultPageSize int
	// MinPageSize sets the minimum number of items per page in map
	// subscription pagination requests. This prevents excessive round trips when
	// clients send very small page sizes. Zero means default (100).
	MinPageSize int
	// MaxPageSize sets the maximum number of items a client can request
	// per page in map subscription pagination requests. Zero means default (1000).
	MaxPageSize int
	// LiveTransitionMaxPublicationLimit sets the maximum number of stream publications
	// to recover during map subscription live transition. If limit is reached,
	// ErrorUnrecoverablePosition is returned. Zero means MaxPageSize.
	LiveTransitionMaxPublicationLimit int
	// SubscribeCatchUpTimeout sets the maximum time a client can spend paginating
	// through state and stream phases before going live. If exceeded, the client is
	// disconnected with DisconnectSlow. Zero means 5 seconds (default). Negative
	// means no timeout.
	SubscribeCatchUpTimeout time.Duration
	// contains filtered or unexported fields
}

MapChannelOptions contains configuration for map channels. Every map channel must have Mode explicitly set — zero value is an error.

Note on Ephemeral mode: TTL removal events are delivered via pub/sub but NOT logged to a stream (there is no stream). If a client misses the removal pub/sub message (e.g., during a brief disconnect), it will retain stale entries until the next full state resync. This is expected behavior for ephemeral use cases like cursor tracking — use Recoverable mode for data requiring consistency.

func ResolveAndValidateMapChannelOptions added in v0.39.0 

func ResolveAndValidateMapChannelOptions(resolver func(channel string) MapChannelOptions, channel string) (MapChannelOptions, error)

ResolveAndValidateMapChannelOptions resolves and validates map channel options for a channel. Returns an error if the resolver is nil, Mode is not set, or the configuration is invalid. Auto-derives stream defaults for stream-backed modes.

type MapClearOptions added in v0.39.0 

type MapClearOptions struct{}

MapClearOptions defines options for removing all channel data. Currently empty but reserved for future options (e.g., selective removal).

type MapConfig added in v0.39.0 

type MapConfig struct {
	// GetMapBroker when set allows returning a custom MapBroker to use for a specific channel.
	// This enables sharding map state across multiple databases based on channel. If not set
	// then the default Node's MapBroker (set via SetMapBroker) is always used for all channels.
	// It's the responsibility of an application to properly initialize and shutdown all MapBroker
	// instances. When GetMapBroker returns false as the second argument then Node will use the
	// default MapBroker for the channel.
	GetMapBroker func(channel string) (MapBroker, bool)
	// GetMapChannelOptions returns channel options for map channels.
	// Required for map channel operations. If nil, any map operation returns an error.
	// Each channel must have SyncMode and RetentionMode explicitly set.
	// NOTE: this callback is invoked on every broker operation (Publish, Remove,
	// ReadState, ReadStream, etc.) — it must be fast and should not perform I/O.
	// This is an EXPERIMENTAL API.
	GetMapChannelOptions func(channel string) MapChannelOptions
}

type MapCurrentEntry added in v0.39.0 

type MapCurrentEntry struct {
	// Offset is the current offset of the entry — use as ExpectedPosition.Offset
	// when retrying CAS.
	Offset uint64
	// Data is the current value of the entry — typically used to compute the
	// next value in a read-modify-write CAS pattern.
	Data []byte
}

MapCurrentEntry contains the current state of a map entry when an operation is suppressed due to a CAS position mismatch (SuppressReasonPositionMismatch). It lets the caller retry the CAS without an extra ReadState round-trip: 

newData := compute(result.CurrentEntry.Data)
retry := MapPublishOptions{
    Data:             newData,
    ExpectedPosition: &StreamPosition{
        Offset: result.CurrentEntry.Offset,
        Epoch:  result.Position.Epoch,  // channel epoch lives on the outer Position
    },
}

Set only when MapUpdateResult.SuppressReason == SuppressReasonPositionMismatch. Treat as read-only — implementations may point Data at internal storage.

type MapMode added in v0.39.0 

type MapMode int

MapMode controls the synchronization, recovery, and data lifecycle behavior of a map channel. 

const (
	// MapModeEphemeral – PUB/SUB only, no stream. Missed updates during disconnect
	// are lost — reconnect triggers full state resync. Entries expire after KeyTTL.
	MapModeEphemeral MapMode = iota + 1
	// MapModeRecoverable – stream-backed with offset-based recovery. Entries expire after
	// KeyTTL. TTL removal events are logged to the stream (recoverable delivery).
	MapModeRecoverable
	// MapModePersistent – stream-backed with offset-based recovery. Entries live
	// forever (until explicitly removed). No TTL, no expiry.
	MapModePersistent
)

func (MapMode) HasExpiry added in v0.39.0 

func (m MapMode) HasExpiry() bool

HasExpiry returns true for modes where entries expire via TTL (Ephemeral, Recoverable).

func (MapMode) HasStream added in v0.39.0 

func (m MapMode) HasStream() bool

HasStream returns true for modes that maintain a recovery stream (Recoverable, Persistent).

func (MapMode) IsEphemeral added in v0.39.0 

func (m MapMode) IsEphemeral() bool

IsEphemeral returns true for MapModeEphemeral.

type MapPublishCallback added in v0.39.0 

type MapPublishCallback func(MapPublishReply, error)

MapPublishCallback should be called with MapPublishReply or error.

type MapPublishEvent added in v0.39.0 

type MapPublishEvent struct {
	// Channel client wants to publish data to.
	Channel string
	// Key sent by the client (may be empty if server assigns keys).
	Key string
	// Data client wants to publish.
	Data []byte
	// ClientInfo about client connection.
	ClientInfo *ClientInfo
}

MapPublishEvent contains fields related to map publish event.

type MapPublishHandler added in v0.39.0 

type MapPublishHandler func(MapPublishEvent, MapPublishCallback)

MapPublishHandler called when client publishes into map channel.

type MapPublishOptions added in v0.39.0 

type MapPublishOptions struct {
	// IdempotencyKey enables duplicate detection. If the same key is seen within
	// IdempotentResultTTL, the operation is suppressed (Suppressed=true).
	// Use for safe client retries without duplicating state changes.
	IdempotencyKey      string
	IdempotentResultTTL time.Duration

	// Data is the publication payload (usually JSON, but can be binary also).
	// Stored in state and published to stream/pub-sub.
	Data []byte
	// Tags are key-value metadata for filtering. Subscribers can filter by tags
	// to receive only relevant publications.
	Tags map[string]string
	// ClientInfo attaches client information to the publication.
	ClientInfo *ClientInfo
	// UseDelta enables delta compression. When true, only the difference from
	// the previous value is stored/sent (requires client support).
	UseDelta bool

	// Version enables per-key version-based ordering. If set, the publish only
	// succeeds if the key's current version is lower. Each key tracks its own
	// version independently. Requires a non-empty key and StreamSize > 0.
	Version uint64
	// VersionEpoch scopes the Version per key. Different epochs reset version
	// comparison. Empty string is valid — use when only incremental version matters.
	VersionEpoch string

	// KeyMode controls conditional publishing based on key existence.
	// Default (empty/KeyModeReplace) always writes.
	// KeyModeIfNew only writes if key doesn't exist (for claiming resources).
	// KeyModeIfExists only writes if key exists (for updates/heartbeats).
	KeyMode KeyMode

	// RefreshTTLOnSuppress when true will refresh the TTL of existing key even when
	// publish is suppressed due to KeyMode (e.g., KeyModeIfNew when key exists).
	// Useful for presence keep-alive without generating new stream entries.
	RefreshTTLOnSuppress bool

	// ExpectedPosition enables Compare-And-Swap semantics.
	// If set, publish only succeeds if the key's current position (offset+epoch) matches.
	// Returns Suppressed=true with SuppressReasonPositionMismatch on mismatch.
	// Use with ReadState Key filter for atomic read-modify-write.
	ExpectedPosition *StreamPosition
	// contains filtered or unexported fields
}

MapPublishOptions defines options for publishing to a map channel.

type MapPublishReply added in v0.39.0 

type MapPublishReply struct {
	// Key overrides the key from the client request. If set, this key is used
	// instead of the one sent by the client. This allows the server to control
	// key assignment (e.g., set key to client ID for cursor channels).
	Key string
	// Options to control map publication.
	Options MapPublishOptions
	// Result if set will tell Centrifuge that message already published to
	// channel by handler code. In this case Centrifuge won't try to publish
	// into channel again after handler returned MapPublishReply.
	Result *MapUpdateResult
}

MapPublishReply contains fields determining the result on map publish.

type MapReadStateOptions added in v0.39.0 

type MapReadStateOptions struct {
	// Revision is the stream position from a previous state read.
	// If provided, the server validates that the epoch hasn't changed.
	// Returns ErrorUnrecoverablePosition if epoch changed (state invalidated).
	Revision *StreamPosition

	// Cursor is the pagination cursor from a previous ReadState call.
	// Empty string starts from the beginning.
	Cursor string

	// Limit controls maximum entries returned per page:
	//  -1 = no limit (return all entries in state)
	//   0 = return only current stream position (no entries)
	//  >0 = return at most this many entries
	//
	// Note: for unordered state in Redis, pagination uses HSCAN with COUNT hint.
	// Redis may return more entries than Limit on some pages (especially for small
	// hashes in listpack encoding). Callers should not rely on exact Limit enforcement
	// for unordered reads. ordered state uses ZRANGEBYSCORE with LIMIT — exact page sizes.
	Limit int

	// Key filters to a single entry by exact key match.
	// When set, returns at most one publication (or empty if key not found).
	// Cursor/Limit are ignored when Key is set.
	Key string

	// Asc requests ascending sort direction for ordered channels.
	// When false (default), ordered state is sorted by score DESC.
	// When true, ordered state is sorted by score ASC.
	// Ignored for unordered channels.
	Asc bool

	// AllowCached allows reading from cache instead of primary storage if
	// available. This option is always set by the subscriptions flow for
	// optimized delivery.
	// Application code should NOT set this if consistent results are important
	// (for example, when making CAS operations).
	// Broker can theoretically not provide this functionality.
	AllowCached bool
}

MapReadStateOptions defines options for reading a channel's state.

type MapReadStreamOptions added in v0.39.0 

type MapReadStreamOptions struct {
	// Filter controls which publications are returned (position, limit, direction).
	Filter StreamFilter
	// AllowCached allows reading from a read replica or cache instead of primary
	// storage. When false (default), reads always go to the primary for strong
	// consistency. Position checks must NOT set this flag.
	AllowCached bool
}

MapReadStreamOptions defines options for reading a channel's history stream.

type MapRemoveCallback added in v0.39.0 

type MapRemoveCallback func(MapRemoveReply, error)

MapRemoveCallback should be called with MapRemoveReply or error.

type MapRemoveEvent added in v0.39.0 

type MapRemoveEvent struct {
	// Channel client wants to remove a key from.
	Channel string
	// Key sent by the client (may be empty if server assigns keys).
	Key string
	// ClientInfo about client connection.
	ClientInfo *ClientInfo
}

MapRemoveEvent contains fields related to map remove event.

type MapRemoveHandler added in v0.39.0 

type MapRemoveHandler func(MapRemoveEvent, MapRemoveCallback)

MapRemoveHandler called when client wants to remove a key from map channel.

type MapRemoveOptions added in v0.39.0 

type MapRemoveOptions struct {
	// IdempotencyKey enables duplicate detection for removal operations.
	IdempotencyKey      string
	IdempotentResultTTL time.Duration

	// ExpectedPosition enables Compare-And-Swap semantics for removal.
	// If set, remove only succeeds if the key's current position (offset+epoch) matches.
	// Returns Suppressed=true with SuppressReasonPositionMismatch on mismatch.
	ExpectedPosition *StreamPosition

	// Tags to include in the removal publication. When not set, tags are automatically
	// read from the existing state entry (if available).
	Tags map[string]string
}

MapRemoveOptions defines options for removing a key from a map channel.

type MapRemoveReply added in v0.39.0 

type MapRemoveReply struct {
	// Key overrides the key from the client request.
	Key string
	// Options to control map removal.
	Options MapRemoveOptions
	// Result if set will tell Centrifuge that removal already performed
	// by handler code. In this case Centrifuge won't try to remove again.
	Result *MapUpdateResult
}

MapRemoveReply contains fields determining the result on map remove.

type MapStateResult added in v0.39.0 

type MapStateResult struct {
	// Publications contains the state entries (each with Key field set).
	Publications []*Publication
	// Position is the current stream position at the time of reading.
	Position StreamPosition
	// Cursor is the pagination cursor for the next page ("" if done).
	Cursor string
}

MapStateResult contains the result of a ReadState operation.

type MapStats added in v0.39.0 

type MapStats struct {
	// NumKeys is the current number of keys in the state.
	NumKeys int
}

MapStats provides statistics about a channel's state.

type MapStatsResult added in v0.39.0 

type MapStatsResult struct {
	MapStats
}

MapStatsResult wraps keyed stats result.

type MapStreamResult added in v0.39.0 

type MapStreamResult struct {
	// Publications contains the stream entries.
	Publications []*Publication
	// Position is the current stream position.
	Position StreamPosition
}

MapStreamResult contains the result of a ReadStream operation.

type MapUpdateResult added in v0.39.0 

type MapUpdateResult struct {
	// Position is the current stream position after the operation.
	Position StreamPosition
	// Suppressed is true when the operation did NOT change the state.
	// When Suppressed is true, no message is appended to the stream and no
	// publication is sent to PUB/SUB.
	Suppressed bool
	// SuppressReason explains why the operation was suppressed (empty when Suppressed is false).
	SuppressReason SuppressReason
	// CurrentEntry — set only when SuppressReason is SuppressReasonPositionMismatch.
	// See the MapCurrentEntry doc for usage in CAS-retry patterns.
	CurrentEntry *MapCurrentEntry
}

MapUpdateResult contains the result of Publish operation.

type MemoryBroker added in v0.16.0 

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

MemoryBroker is builtin default Broker which allows running Centrifuge-based server without any external broker. All data managed inside process memory.

With this Broker you can only run single Centrifuge node. If you need to scale you should consider using another Broker implementation instead – for example RedisBroker.

Running single node can be sufficient for many use cases especially when you need maximum performance and not too many online clients. Consider configuring your load balancer to have one backup Centrifuge node for HA in this case.

func NewMemoryBroker added in v0.16.0 

func NewMemoryBroker(n *Node, _ MemoryBrokerConfig) (*MemoryBroker, error)

NewMemoryBroker initializes MemoryBroker.

func (*MemoryBroker) Close added in v0.17.1 

func (b *MemoryBroker) Close(_ context.Context) error

Close is noop for now.

func (*MemoryBroker) History added in v0.16.0 

func (b *MemoryBroker) History(ch string, opts HistoryOptions) ([]*Publication, StreamPosition, error)

History - see Broker interface description.

func (*MemoryBroker) Publish added in v0.16.0 

func (b *MemoryBroker) Publish(ch string, data []byte, opts PublishOptions) (PublishResult, error)

Publish adds message into history hub and calls node method to handle message. We don't have any PUB/SUB here as MemoryBroker is single node only.

func (*MemoryBroker) PublishJoin added in v0.16.0 

func (b *MemoryBroker) PublishJoin(ch string, info *ClientInfo) error

PublishJoin - see Broker interface description.

func (*MemoryBroker) PublishLeave added in v0.16.0 

func (b *MemoryBroker) PublishLeave(ch string, info *ClientInfo) error

PublishLeave - see Broker interface description.

func (*MemoryBroker) RegisterBrokerEventHandler added in v0.36.0 

func (b *MemoryBroker) RegisterBrokerEventHandler(h BrokerEventHandler) error

RegisterBrokerEventHandler runs memory broker.

func (*MemoryBroker) RemoveHistory added in v0.16.0 

func (b *MemoryBroker) RemoveHistory(ch string) error

RemoveHistory - see Broker interface description.

func (*MemoryBroker) Subscribe added in v0.16.0 

func (b *MemoryBroker) Subscribe(_ ...string) error

Subscribe is noop here.

func (*MemoryBroker) Unsubscribe added in v0.16.0 

func (b *MemoryBroker) Unsubscribe(_ ...string) error

Unsubscribe node from channel. Noop here.

type MemoryBrokerConfig added in v0.16.0 

type MemoryBrokerConfig struct{}

MemoryBrokerConfig is a memory broker config.

type MemoryMapBroker added in v0.39.0 

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

MemoryMapBroker is builtin default MapBroker which allows running Centrifuge-based server without any external storage. All data managed inside process memory.

With this MapBroker you can only run single Centrifuge node. If you need to scale you should consider using another MapBroker implementation instead – for example RedisMapBroker.

func NewMemoryMapBroker added in v0.39.0 

func NewMemoryMapBroker(n *Node, _ MemoryMapBrokerConfig) (*MemoryMapBroker, error)

NewMemoryMapBroker initializes MemoryMapBroker.

func (*MemoryMapBroker) Clear added in v0.39.0 

func (e *MemoryMapBroker) Clear(_ context.Context, ch string, _ MapClearOptions) error

func (*MemoryMapBroker) Close added in v0.39.0 

func (e *MemoryMapBroker) Close(_ context.Context) error

Close shuts down the broker.

func (*MemoryMapBroker) Publish added in v0.39.0 

func (e *MemoryMapBroker) Publish(ctx context.Context, ch string, key string, opts MapPublishOptions) (MapUpdateResult, error)

Publish publishes data to channel with optional key for keyed state.

func (*MemoryMapBroker) ReadState added in v0.39.0 

func (e *MemoryMapBroker) ReadState(ctx context.Context, ch string, opts MapReadStateOptions) (MapStateResult, error)

ReadState retrieves keyed state with revisions.

func (*MemoryMapBroker) ReadStream added in v0.39.0 

func (e *MemoryMapBroker) ReadStream(ctx context.Context, ch string, opts MapReadStreamOptions) (MapStreamResult, error)

ReadStream retrieves publications from stream.

func (*MemoryMapBroker) RegisterEventHandler added in v0.39.0 

func (e *MemoryMapBroker) RegisterEventHandler(h BrokerEventHandler) error

RegisterEventHandler registers event handler and runs memory map broker.

func (*MemoryMapBroker) Remove added in v0.39.0 

func (e *MemoryMapBroker) Remove(ctx context.Context, ch string, key string, opts MapRemoveOptions) (MapUpdateResult, error)

Remove removes a key from keyed state.

func (*MemoryMapBroker) Stats added in v0.39.0 

func (e *MemoryMapBroker) Stats(ctx context.Context, ch string) (MapStats, error)

Stats returns state statistics.

func (*MemoryMapBroker) Subscribe added in v0.39.0 

func (e *MemoryMapBroker) Subscribe(_ ...string) error

Subscribe is noop here.

func (*MemoryMapBroker) Unsubscribe added in v0.39.0 

func (e *MemoryMapBroker) Unsubscribe(_ ...string) error

Unsubscribe is noop here.

type MemoryMapBrokerConfig added in v0.39.0 

type MemoryMapBrokerConfig struct{}

MemoryMapBrokerConfig is a memory map broker config.

type MemoryPresenceManager added in v0.16.0 

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

MemoryPresenceManager is builtin default PresenceManager which allows running Centrifuge-based server without any external storage. All data managed inside process memory.

With this PresenceManager you can only run single Centrifuge node. If you need to scale you should consider using another PresenceManager implementation instead – for example RedisPresenceManager.

Running single node can be sufficient for many use cases especially when you need maximum performance and not too many online clients. Consider configuring your load balancer to have one backup Centrifuge node for HA in this case.

func NewMemoryPresenceManager added in v0.16.0 

func NewMemoryPresenceManager(n *Node, c MemoryPresenceManagerConfig) (*MemoryPresenceManager, error)

NewMemoryPresenceManager initializes MemoryPresenceManager.

func (*MemoryPresenceManager) AddPresence added in v0.16.0 

func (m *MemoryPresenceManager) AddPresence(ch string, uid string, info *ClientInfo) error

AddPresence - see PresenceManager interface description.

func (*MemoryPresenceManager) Close added in v0.17.1 

func (m *MemoryPresenceManager) Close(_ context.Context) error

Close is noop for now.

func (*MemoryPresenceManager) Presence added in v0.16.0 

func (m *MemoryPresenceManager) Presence(ch string) (map[string]*ClientInfo, error)

Presence - see PresenceManager interface description.

func (*MemoryPresenceManager) PresenceStats added in v0.16.0 

func (m *MemoryPresenceManager) PresenceStats(ch string) (PresenceStats, error)

PresenceStats - see PresenceManager interface description.

func (*MemoryPresenceManager) RemovePresence added in v0.16.0 

func (m *MemoryPresenceManager) RemovePresence(ch string, clientID string, _ string) error

RemovePresence - see PresenceManager interface description.

type MemoryPresenceManagerConfig added in v0.16.0 

type MemoryPresenceManagerConfig struct{}

MemoryPresenceManagerConfig is a MemoryPresenceManager config.

type MessageEvent 

type MessageEvent struct {
	// Data contains message untouched payload.
	Data []byte
}

MessageEvent contains fields related to message request.

type MessageHandler 

type MessageHandler func(MessageEvent)

MessageHandler must handle incoming async message from client. So Centrifuge feels similar to pure WebSocket API. Though in general, we recommend using RPC where possible to send data to a server from a client as it provides a better flow control.

type Metrics 

type Metrics struct {
	Interval float64
	Items    map[string]float64
}

Metrics aggregation over time interval for node.

type MetricsConfig added in v0.34.0 

type MetricsConfig struct {
	// MetricsNamespace is a Prometheus metrics namespace to use for Centrifuge metrics.
	// If not set then the default metrics namespace name "centrifuge" will be used.
	MetricsNamespace string
	// RegistererGatherer is a Prometheus registerer and gatherer. If not set then a
	// prometheus.DefaultRegisterer and prometheus.DefaultGatherer will be used.
	RegistererGatherer RegistererGatherer

	// GetChannelNamespaceLabel if set will be used by Centrifuge to extract channel_namespace
	// label for channel related metrics. Make sure to maintain low cardinality of returned values
	// to avoid issues with Prometheus performance. This function may introduce sufficient overhead
	// since it's called in hot paths - so it should be fast. By default, Centrifuge uses cache
	// of resolved channel namespace labels to avoid calling this function too often. See below
	// ChannelNamespaceCacheSize and ChannelNamespaceCacheTTL options to tweak the cache behavior.
	GetChannelNamespaceLabel func(channel string) string
	// ChannelNamespaceCacheSize sets the size of the cache for channel namespace label resolution.
	// Zero value will use cache size equal to 4096. Set -1 to disable cache (in that case make sure
	// your GetChannelNamespaceLabel is fast and ideally does not allocate because it's called in hot
	// paths).
	ChannelNamespaceCacheSize int
	// ChannelNamespaceCacheTTL sets the time after which resolved channel namespace for a channel
	// will expire in the cache. If zero – default TTL 10 seconds is used.
	ChannelNamespaceCacheTTL time.Duration

	// RegisteredClientNames is an optional list of known client names which will be allowed to be
	// attached as labels to metrics. If client passed a name which is not in the list – then Centrifuge
	// will use string "unregistered" as a client_name label. If a client passed no name at all Centrifuge
	// will use string "unnamed". We need to be strict here to avoid Prometheus cardinality issues.
	RegisteredClientNames []string
	// CheckRegisteredClientVersion is a function to check whether the version passed by a client with a
	// particular name is valid and can be used in metric values. When function is not set or returns
	// false Centrifuge will use "unregistered" value for a client version. If a client passed no version
	// at all Centrifuge will use string "unversioned". Note, the name argument here is an original name of
	// client passed to Centrifuge.
	CheckRegisteredClientVersion func(clientName string, clientVersion string) bool
	// EnableRecoveredPublicationsHistogram enables histogram tracking of number of publications
	// recovered during subscription successful recovery operations.
	EnableRecoveredPublicationsHistogram bool
	// ExposeTransportAcceptProtocol enables exposing in labels the accept protocol used by client's transport.
	// If not enabled - empty string will be used as a label value.
	ExposeTransportAcceptProtocol bool
	// ClientLabels defines which Client.Labels keys are exported as additional
	// Prometheus dimensions on per-connection metrics. Label values are read
	// from ConnectReply.Labels set in the OnConnecting handler. If a configured
	// key is missing on a particular client, an empty string is used.
	//
	// Exported dimension names are prefixed with "app_" to guarantee no
	// collision with built-in metric labels (transport, method, code, …).
	// For example, ClientLabels: []string{"region", "tier"} appears in
	// Prometheus as the dimensions "app_region" and "app_tier". Your
	// application code reads the unprefixed keys from Client.Labels()
	// unchanged — the prefix is applied only on the metric export path.
	//
	// Warning: each unique combination of values creates a new Prometheus
	// time series. Keep cardinality low by using controlled finite value
	// sets (region, tier, app version) — do not export user IDs, session
	// IDs, or any unbounded input.
	//
	// If empty (the default), client labels are disabled and there is no
	// overhead on the metric emission path.
	ClientLabels []string
	// EnableNativeHistograms switches every Histogram instrument in the
	// package to Prometheus native (sparse, exponential) schema with no
	// explicit buckets exposed, and stops exposing the legacy Summary
	// counterparts of dual-instrument metrics. Designed for OpenTelemetry
	// export via the client_golang Prometheus bridge — native histograms
	// map to OTel ExponentialHistogram, and dropping Summaries keeps OTel
	// Summary (which most backends treat as second-class) out of the
	// pipeline.
	//
	// Centrifuge exposes both a Summary and a Histogram for the two
	// distribution metrics that have historically been Summaries:
	//   - {ns}_client_command_duration_seconds (Summary) +
	//     {ns}_client_command_duration_seconds_histogram (Histogram)
	//   - {ns}_node_survey_duration_seconds (Summary) +
	//     {ns}_node_survey_duration_seconds_histogram (Histogram)
	// Both are constructed and observed unconditionally.
	//
	// Default is false: today's behavior is preserved (Summaries still
	// exposed; companion Histograms additionally exposed with classic
	// explicit buckets — see Histograms section in metrics docs for the
	// exact bucket lists).
	//
	// When set to true:
	//   - Summaries are no longer exposed (the underlying instrument
	//     becomes a no-op so cached observers continue to satisfy
	//     prometheus.Observer without nil-checks).
	//   - All Histograms switch to native (sparse, exponential) schema
	//     with no explicit buckets.
	//   - Text-format Prometheus scrapes lose _bucket series on every
	//     Histogram; only _count and _sum remain visible. Use the protobuf
	//     scrape format (Prom 2.40+) to receive the native histogram data.
	//
	// Native histograms are experimental in client_golang.
	EnableNativeHistograms bool
}

type Node 

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

Node is a heart of Centrifuge library – it keeps and manages client connections, maintains information about other Centrifuge nodes in cluster, keeps references to common things (like Broker and PresenceManager, Hub) etc. By default, Node uses in-memory implementations of Broker and PresenceManager - MemoryBroker and MemoryPresenceManager which allow running a single Node only. To scale use other implementations of Broker and PresenceManager like builtin RedisBroker and RedisPresenceManager.

func New 

func New(c Config) (*Node, error)

New creates Node with provided Config.

func (*Node) AddMapBrokerCleanupRemoved added in v0.39.0 

func (n *Node) AddMapBrokerCleanupRemoved(name string, count int64)

AddMapBrokerCleanupRemoved adds to the map broker cleanup removed-entries counter for observability.

func (*Node) Config 

func (n *Node) Config() Config

Config returns Node's Config.

func (*Node) Disconnect 

func (n *Node) Disconnect(userID string, opts ...DisconnectOption) error

Disconnect allows closing all user connections on all nodes.

func (*Node) HandleControl added in v0.33.0 

func (n *Node) HandleControl(data []byte) error

HandleControl coming from Broker.

func (*Node) HandleJoin added in v0.33.0 

func (n *Node) HandleJoin(ch string, info *ClientInfo) error

HandleJoin coming from Broker.

func (*Node) HandleLeave added in v0.33.0 

func (n *Node) HandleLeave(ch string, info *ClientInfo) error

HandleLeave coming from Broker.

func (*Node) HandlePublication added in v0.33.0 

func (n *Node) HandlePublication(ch string, pub *Publication, sp StreamPosition, delta bool, prevPub *Publication) error

HandlePublication coming from Broker.

func (*Node) History 

func (n *Node) History(ch string, opts ...HistoryOption) (HistoryResult, error)

History allows extracting Publications in channel. The channel must belong to namespace where history is on.

func (*Node) Hub 

func (n *Node) Hub() *Hub

Hub returns node's Hub.

func (*Node) ID added in v0.15.0 

func (n *Node) ID() string

ID returns unique Node identifier. This is a UUID v4 value.

func (*Node) IncMapBrokerCleanupErrors added in v0.39.0 

func (n *Node) IncMapBrokerCleanupErrors(name string)

IncMapBrokerCleanupErrors increments the map broker cleanup error counter for observability.

func (*Node) IncTransportOutgoingClose added in v0.39.0 

func (n *Node) IncTransportOutgoingClose(transport string, code int)

IncTransportOutgoingClose increments the counter of close frames sent to clients, labeled by transport and close code. It is exported so that custom transports implemented outside this package (e.g. unidirectional WebSocket) can record their outgoing close codes too. Only server-sent close codes should be passed - client-supplied close codes must not be recorded as their label cardinality is not bounded.

func (*Node) Info 

func (n *Node) Info() (Info, error)

Info returns aggregated stats from all nodes.

func (*Node) MapClear added in v0.39.0 

func (n *Node) MapClear(ctx context.Context, ch string, opts MapClearOptions) error

MapClear deletes all data for a map channel (state and stream). Use for cleanup when a channel data is no longer needed.

func (*Node) MapPublish added in v0.39.0 

func (n *Node) MapPublish(ctx context.Context, ch string, key string, opts MapPublishOptions) (MapUpdateResult, error)

MapPublish publishes data to a map channel. This updates the snapshot and optionally broadcasts to subscribers.

func (*Node) MapRemove added in v0.39.0 

func (n *Node) MapRemove(ctx context.Context, ch string, key string, opts MapRemoveOptions) (MapUpdateResult, error)

MapRemove removes a key from a map channel. This removes the key from snapshot and optionally broadcasts removal to subscribers.

func (*Node) MapStateRead added in v0.39.0 

func (n *Node) MapStateRead(ctx context.Context, ch string, opts MapReadStateOptions) (MapStateResult, error)

MapStateRead retrieves keyed snapshot for a channel.

func (*Node) MapStats added in v0.39.0 

func (n *Node) MapStats(ctx context.Context, ch string) (MapStatsResult, error)

MapStats retrieves stats for a map channel.

func (*Node) MapStreamRead added in v0.39.0 

func (n *Node) MapStreamRead(ctx context.Context, ch string, opts MapReadStreamOptions) (MapStreamResult, error)

MapStreamRead retrieves keyed stream for a channel.

func (*Node) Notify added in v0.17.0 

func (n *Node) Notify(op string, data []byte, toNodeID string) error

Notify allows sending an asynchronous notification to all other nodes (or to a single specific node). Unlike Survey, it does not wait for any response. If toNodeID is not an empty string then a notification will be sent to a concrete node in cluster, otherwise a notification sent to all running nodes. See a corresponding Node.OnNotification method to handle received notifications.

func (*Node) NotifyShutdown 

func (n *Node) NotifyShutdown() chan struct{}

NotifyShutdown returns a channel which will be closed on node shutdown.

func (*Node) ObserveTransportFrameSize added in v0.39.1 

func (n *Node) ObserveTransportFrameSize(transport string, size int, c *Client)

ObserveTransportFrameSize records the size in bytes of one protocol frame read from a client connection. It is exported so that custom transports implemented outside this package can record frame sizes too.

A frame is not a command: the protocol batches, so one frame may carry many commands. Transports that read a continuous stream rather than discrete frames have nothing frame-shaped to record here.

func (*Node) OnCacheEmpty added in v0.33.0 

func (n *Node) OnCacheEmpty(h CacheEmptyHandler)

OnCacheEmpty allows setting CacheEmptyHandler. CacheEmptyHandler called when client subscribes on a channel with RecoveryModeCache but there is no cached value in channel. In response to this handler it's possible to tell Centrifuge what to do with subscribe request – keep it, or return error.

func (*Node) OnCommandProcessed added in v0.28.0 

func (n *Node) OnCommandProcessed(handler CommandProcessedHandler)

OnCommandProcessed allows setting CommandProcessedHandler. This should be done before Node.Run called.

func (*Node) OnCommandRead added in v0.27.0 

func (n *Node) OnCommandRead(handler CommandReadHandler)

OnCommandRead allows setting CommandReadHandler. This should be done before Node.Run called.

func (*Node) OnConnect added in v0.10.0 

func (n *Node) OnConnect(handler ConnectHandler)

OnConnect allows setting ConnectHandler. ConnectHandler called after client connection successfully established, authenticated and Connect Reply already sent to client. This is a place where application can start communicating with client.

func (*Node) OnConnecting added in v0.10.0 

func (n *Node) OnConnecting(handler ConnectingHandler)

OnConnecting allows setting ConnectingHandler. ConnectingHandler will be called when client sends Connect command to server. In this handler server can reject connection or provide Credentials for it.

func (*Node) OnNodeInfoSend added in v0.18.0 

func (n *Node) OnNodeInfoSend(handler NodeInfoSendHandler)

OnNodeInfoSend allows setting NodeInfoSendHandler. This should be done before Node.Run called.

func (*Node) OnNotification added in v0.17.0 

func (n *Node) OnNotification(handler NotificationHandler)

OnNotification allows setting NotificationHandler. This should be done before Node.Run called.

func (*Node) OnSharedPoll added in v0.39.0 

func (n *Node) OnSharedPoll(handler SharedPollHandler)

OnSharedPoll allows setting SharedPollHandler. SharedPollHandler is called by the refresh worker to fetch current item data from the backend. Called per-channel, not per-client.

func (*Node) OnSurvey added in v0.15.0 

func (n *Node) OnSurvey(handler SurveyHandler)

OnSurvey allows setting SurveyHandler. This should be done before Node.Run called.

func (*Node) OnTransportWrite added in v0.18.0 

func (n *Node) OnTransportWrite(handler TransportWriteHandler)

OnTransportWrite allows setting TransportWriteHandler. This should be done before Node.Run called.

func (*Node) Presence 

func (n *Node) Presence(ch string) (PresenceResult, error)

Presence returns a map with information about active clients in channel.

func (*Node) PresenceStats 

func (n *Node) PresenceStats(ch string) (PresenceStatsResult, error)

PresenceStats returns presence stats from PresenceManager.

func (*Node) Publish 

func (n *Node) Publish(channel string, data []byte, opts ...PublishOption) (PublishResult, error)

Publish sends data to all clients subscribed on channel at this moment. All running nodes will receive Publication and send it to all local channel subscribers.

Data expected to be valid marshaled JSON or any binary payload. Connections that work over JSON protocol can not handle binary payloads. Connections that work over Protobuf protocol can work both with JSON and binary payloads.

So the rule here: if you have channel subscribers that work using JSON protocol then you can not publish binary data to these channel.

Channels in Centrifuge are ephemeral and its settings not persisted over different publish operations. So if you want to have a channel with history stream behind you need to provide WithHistory option on every publish. To simplify working with different channels you can make some type of publish wrapper in your own code.

The returned PublishResult contains embedded StreamPosition that describes position inside stream Publication was added too. For channels without history enabled (i.e. when Publications only sent to PUB/SUB system) StreamPosition will be an empty struct (i.e. PublishResult.Offset will be zero).

func (*Node) Refresh added in v0.18.0 

func (n *Node) Refresh(userID string, opts ...RefreshOption) error

Refresh user connection. Without any options will make user connections non-expiring. Note, that OnRefresh event won't be called in this case since this is a server-side refresh.

func (*Node) RemoveHistory 

func (n *Node) RemoveHistory(ch string) error

RemoveHistory removes channel history.

func (*Node) Run 

func (n *Node) Run() error

Run performs node startup actions. At moment must be called once on start after Controller and Broker set to Node.

func (*Node) SetBroker 

func (n *Node) SetBroker(b Broker)

SetBroker allows setting Broker implementation to use. For historical reasons and to keep existing API, we also check if Broker implements Controller and if so we set it as Node's Controller (but only if Controller not explicitly set).

func (*Node) SetController added in v0.36.0 

func (n *Node) SetController(c Controller)

SetController allows setting Controller implementation to use.

func (*Node) SetMapBroker added in v0.39.0 

func (n *Node) SetMapBroker(e MapBroker)

SetMapBroker allows setting MapBroker to use.

func (*Node) SetMapBrokerCleanupLag added in v0.39.0 

func (n *Node) SetMapBrokerCleanupLag(name string, seconds float64)

SetMapBrokerCleanupLag sets the map broker cleanup lag gauge for observability.

func (*Node) SetPresenceManager 

func (n *Node) SetPresenceManager(m PresenceManager)

SetPresenceManager allows setting PresenceManager to use.

func (*Node) SharedPollNotify added in v0.39.0 

func (n *Node) SharedPollNotify(notifications []SharedPollNotificationItem)

SharedPollNotify submits notifications that trigger immediate backend polls for the specified keys. Notifications are batched per channel according to SharedPollChannelOptions before triggering polls. Safe for concurrent use. Notifications for unknown channels are silently dropped.

func (*Node) SharedPollPublish added in v0.39.0 

func (n *Node) SharedPollPublish(ctx context.Context, channel string, key string, version uint64, epoch string, data []byte) error

SharedPollPublish pushes data directly to a SharedPoll channel for a specific key. The version must be in the same space as versions returned by the OnSharedPoll handler — stale versions (≤ current) are ignored within a given epoch. When PublishEnabled is set in channel options, the publication is distributed to all nodes via Broker PUB/SUB. Otherwise, local-only.

epoch is an optional channel-level string identifying the publisher's epoch. If it differs from the channel's stored epoch, all current subscribers are unsubscribed with insufficient-state code so they re-track from version 0 on resubscribe — this lets a publisher that resets its in-memory version counter (e.g., after a process restart) deliver fresh state without freezing connected clients. Use empty epoch to skip this check (pure version comparison).

func (*Node) Shutdown 

func (n *Node) Shutdown(ctx context.Context) error

Shutdown sets shutdown flag to Node so handlers could stop accepting new requests and disconnects clients with shutdown reason.

func (*Node) Subscribe added in v0.16.0 

func (n *Node) Subscribe(userID string, channel string, opts ...SubscribeOption) error

Subscribe subscribes user to a channel. Note, that OnSubscribe event won't be called in this case since this is a server-side subscription. If user have been already subscribed to a channel then its subscription will be updated and subscribe notification will be sent to a client-side.

func (*Node) Survey added in v0.15.0 

func (n *Node) Survey(ctx context.Context, op string, data []byte, toNodeID string) (map[string]SurveyResult, error)

Survey allows collecting data from all running Centrifuge nodes. This method publishes control messages, then waits for replies from all running nodes. The maximum time to wait can be controlled over context timeout. If provided context does not have a deadline for survey then this method uses default 10 seconds timeout. Keep in mind that Survey does not scale very well as number of Centrifuge Node grows. Though it has reasonably good performance to perform rare tasks even with relatively large number of nodes. If toNodeID is not an empty string then a survey will be sent only to the concrete node in a cluster, otherwise a survey sent to all running nodes. See a corresponding Node.OnSurvey method to handle received surveys. Survey ops starting with `centrifuge_` are reserved by Centrifuge library.

func (*Node) Unsubscribe 

func (n *Node) Unsubscribe(userID string, channel string, opts ...UnsubscribeOption) error

Unsubscribe unsubscribes user from a channel. If a channel is empty string then user will be unsubscribed from all channels.

type NodeInfo 

type NodeInfo struct {
	UID         string
	Name        string
	Version     string
	NumClients  uint32
	NumUsers    uint32
	NumSubs     uint32
	NumChannels uint32
	Uptime      uint32
	Metrics     *Metrics
	Data        []byte
}

NodeInfo contains information about node.

type NodeInfoSendHandler added in v0.18.0 

type NodeInfoSendHandler func() NodeInfoSendReply

NodeInfoSendHandler called every time the control node frame is published and allows modifying Node control frame sending. Currently, attaching an arbitrary data to it. See NodeInfoSendReply.

type NodeInfoSendReply added in v0.18.0 

type NodeInfoSendReply struct {
	// Data allows setting an arbitrary data to the control node frame which is
	// published by each Node periodically, so it will be available in the
	// result of Node.Info call for the current Node description. Keep this
	// data reasonably small.
	Data []byte
}

NodeInfoSendReply can modify sending Node control frame in some ways.

type NotificationEvent added in v0.17.0 

type NotificationEvent struct {
	FromNodeID string
	Op         string
	Data       []byte
}

NotificationEvent with Op and Data.

type NotificationHandler added in v0.17.0 

type NotificationHandler func(NotificationEvent)

NotificationHandler allows handling notifications.

type PingPongConfig added in v0.26.0 

type PingPongConfig struct {
	// PingInterval tells how often to issue server-to-client pings.
	// For zero value 25 secs will be used. To disable sending app-level pings use -1.
	PingInterval time.Duration
	// PongTimeout sets time for pong check after issuing a ping.
	// For zero value 10 seconds will be used. To disable pong checks use -1.
	// PongTimeout must be less than PingInterval in current implementation.
	PongTimeout time.Duration
}

PingPongConfig allows configuring application level ping-pong behavior. Note that in current implementation PingPongConfig.PingInterval must be greater than PingPongConfig.PongTimeout.

type PresenceCallback added in v0.13.0 

type PresenceCallback func(PresenceReply, error)

PresenceCallback should be called with PresenceReply or error.

type PresenceEvent added in v0.9.0 

type PresenceEvent struct {
	Channel string
}

PresenceEvent has channel operation called for.

type PresenceHandler added in v0.9.0 

type PresenceHandler func(PresenceEvent, PresenceCallback)

PresenceHandler called when presence request received from client.

type PresenceManager 

type PresenceManager interface {
	// Presence returns actual presence information for channel.
	Presence(ch string) (map[string]*ClientInfo, error)
	// PresenceStats returns short stats of current presence data
	// suitable for scenarios when caller does not need full client
	// info returned by presence method.
	PresenceStats(ch string) (PresenceStats, error)
	// AddPresence sets or updates presence information in channel
	// for connection with specified identifier. PresenceManager should
	// have a property to expire client information that was not updated
	// (touched) after some configured time interval.
	AddPresence(ch string, clientID string, info *ClientInfo) error
	// RemovePresence removes presence information for connection
	// with specified client and user identifiers.
	RemovePresence(ch string, clientID string, userID string) error
}

PresenceManager is responsible for channel presence management.

type PresenceReply added in v0.9.0 

type PresenceReply struct {
	Result *PresenceResult
}

PresenceReply contains fields determining the reaction on presence request.

type PresenceResult added in v0.10.0 

type PresenceResult struct {
	Presence map[string]*ClientInfo
}

PresenceResult wraps presence.

type PresenceStats 

type PresenceStats struct {
	// NumClients is a number of client connections in channel.
	NumClients int
	// NumUsers is a number of unique users in channel.
	NumUsers int
}

PresenceStats represents a short presence information for channel.

type PresenceStatsCallback added in v0.13.0 

type PresenceStatsCallback func(PresenceStatsReply, error)

PresenceStatsCallback should be called with PresenceStatsReply or error.

type PresenceStatsEvent added in v0.10.0 

type PresenceStatsEvent struct {
	Channel string
}

PresenceStatsEvent has channel operation called for.

type PresenceStatsHandler added in v0.10.0 

type PresenceStatsHandler func(PresenceStatsEvent, PresenceStatsCallback)

PresenceStatsHandler must handle incoming command from client.

type PresenceStatsReply added in v0.10.0 

type PresenceStatsReply struct {
	Result *PresenceStatsResult
}

PresenceStatsReply contains fields determining the reaction on presence request.

type PresenceStatsResult added in v0.10.0 

type PresenceStatsResult struct {
	PresenceStats
}

PresenceStatsResult wraps presence stats.

type ProtocolType added in v0.1.0 

type ProtocolType string

ProtocolType represents client connection transport encoding format. 

const (
	// ProtocolTypeJSON means JSON-based protocol.
	ProtocolTypeJSON ProtocolType = "json"
	// ProtocolTypeProtobuf means Protobuf protocol.
	ProtocolTypeProtobuf ProtocolType = "protobuf"
)

type ProtocolVersion added in v0.20.0 

type ProtocolVersion uint8

ProtocolVersion defines protocol behavior. 

const (
	// ProtocolVersion2 is the current stable client protocol.
	ProtocolVersion2 ProtocolVersion = 2
)

type Publication 

type Publication struct {
	// Offset is an incremental position number inside a history stream.
	// Zero value means that channel does not maintain Publication stream.
	Offset uint64
	// Epoch is a multi-purpose channel-level epoch string. For stream
	// subscriptions it pairs with Offset for stream-position tracking
	// (history loss / stream recreation detection). For shared-poll
	// versioned subscriptions it carries the publisher's per-channel
	// epoch — a change in epoch resets per-key versions and unsubscribes
	// current subscribers with insufficient-state code so they re-track
	// from version 0 on resubscribe.
	Epoch string
	// Data to be published to a channel (to be delivered to subscribers).
	Data []byte
	// Info is optional information about client connection published this data.
	Info *ClientInfo
	// Tags contains a map with custom key-values attached to a Publication. Tags map
	// will be delivered to a client.
	Tags map[string]string
	// Optional time of publication as Unix timestamp milliseconds. At this point
	// we use it for calculating PUB/SUB time lag, it's not exposed to the client
	// protocol.
	Time int64
	// Channel is only set when subscription channel does not match channel in Publication.
	// This is a case for wildcard subscriptions. Client SDK then should use this channel
	// in PublicationContext.
	Channel string
	// Key is used for map subscriptions (state and presence/membership).
	// For presence: Key = ClientID.
	// For map state: Key = arbitrary string key.
	Key string
	// Removed indicates this is a remove/leave event (true) vs add/join event (false).
	// Used in map subscriptions to distinguish event types.
	Removed bool
	// Score is used for ordered map subscriptions (leaderboards, priority queues).
	Score int64
	// Version of the publication. Used by SharedPoll for stale-check.
	Version uint64
}

Publication is a data sent to a channel.

type PublishCallback added in v0.13.0 

type PublishCallback func(PublishReply, error)

PublishCallback should be called with PublishReply or error.

type PublishEvent 

type PublishEvent struct {
	// Channel client wants to publish data to.
	Channel string
	// Data client wants to publish.
	Data []byte
	// ClientInfo about client connection.
	ClientInfo *ClientInfo
}

PublishEvent contains fields related to publish event. Note that this event called before actual publish to Broker so handler has an option to reject this publication returning an error.

type PublishHandler 

type PublishHandler func(PublishEvent, PublishCallback)

PublishHandler called when client publishes into channel.

type PublishOption 

type PublishOption func(*PublishOptions)

PublishOption is a type to represent various Publish options.

func WithClientInfo added in v0.13.0 

func WithClientInfo(info *ClientInfo) PublishOption

WithClientInfo adds ClientInfo to Publication.

func WithDelta added in v0.33.0 

func WithDelta(enabled bool) PublishOption

WithDelta tells Broker to use delta streaming.

func WithHistory added in v0.12.0 

func WithHistory(size int, ttl time.Duration, metaTTL ...time.Duration) PublishOption

WithHistory tells Broker to save message to history stream with provided size and ttl.

func WithIdempotencyKey added in v0.31.0 

func WithIdempotencyKey(key string) PublishOption

WithIdempotencyKey tells Broker the idempotency key for the publication. See PublishOptions.IdempotencyKey.

func WithIdempotentResultTTL added in v0.31.0 

func WithIdempotentResultTTL(ttl time.Duration) PublishOption

WithIdempotentResultTTL sets the time of expiration for results of idempotent publications. See PublishOptions.IdempotentResultTTL for more description and defaults.

func WithKey added in v0.39.0 

func WithKey(key string) PublishOption

WithKey sets a key for the publication. When set, the publication is associated with a specific key within the channel. This may enable per-key debouncing or channel level per-key batching. The key is delivered to subscribers in the Publication.

func WithTags added in v0.20.0 

func WithTags(tags map[string]string) PublishOption

WithTags allows setting Publication.Tags.

func WithVersion added in v0.36.0 

func WithVersion(version uint64, versionEpoch string) PublishOption

WithVersion allows application to provide a tip for Centrifuge about internal application version of Publication. This is helpful to drop non-actual publications on Centrifuge Broker level. Publications may be non-actual in case of unordered publish calls from application side. In some cases application may want Centrifuge to avoid delivery of these publications. This is mostly useful for scenarios when channel publications contain the entire state, so skipping intermediary messages is safe and beneficial. This also means that Centrifuge history will contain the most recent Publication, so the recovery will return the proper state (and state should be eventually consistent in case of at least one delivery). This option must be used only with channels that have history enabled. Note, these version and versionEpoch are not used as Centrifuge channel stream position. Centrifuge still generates its own independent StreamPosition for each Publication in channel streams with history, but it additionally starts keeping version and versionEpoch provided here. If versionEpoch is an empty string, then Centrifuge does not look at it when comparing versions.

type PublishOptions 

type PublishOptions struct {
	// HistoryTTL sets history ttl to expire inactive history streams.
	// Current Broker implementations only work with seconds resolution for TTL.
	HistoryTTL time.Duration
	// HistorySize sets history size limit to prevent infinite stream growth.
	HistorySize int
	// HistoryMetaTTL allows overriding default (set in Config.HistoryMetaTTL)
	// history meta information expiration time upon publish.
	HistoryMetaTTL time.Duration
	// ClientInfo to include into Publication. By default, no ClientInfo will be appended.
	ClientInfo *ClientInfo
	// Tags to set Publication.Tags.
	Tags map[string]string
	// IdempotencyKey is an optional key for idempotent publish. Broker implementation
	// may cache these keys for some time to prevent duplicate publications. In this case
	// the returned result is the same as from the previous publication with the same key.
	IdempotencyKey string
	// IdempotentResultTTL sets the time of expiration for results of idempotent publications
	// (publications with idempotency key provided). Memory and Redis brokers implement this TTL
	// with second precision, so don't set something less than one second here. By default,
	// Centrifuge uses 5 minutes as idempotent result TTL.
	IdempotentResultTTL time.Duration
	// UseDelta enables using delta encoding for the publication.
	UseDelta bool
	// Version of Publication. This is a tip to Centrifuge to skip non-actual
	// publications. Mostly useful for cases when Publication contains the entire
	// state. Version only used when history is configured.
	Version uint64
	// VersionEpoch is a string that is used to identify the epoch of version of the
	// publication. Use it if version may be reused in the future. For example, if
	// version comes from in-memory system which can lose data, or due to eviction, etc.
	VersionEpoch string
	// Key associates the publication with a specific key within the channel.
	// For stream subscriptions: enables per-key debouncing via DebouncingBroker
	// and is delivered to subscribers on the Publication for client-side use.
	// For map subscriptions: plays central role – state is compacted by key,
	// delta compression per key, channel batching latest publication is per key.
	Key string

	// Removed indicates this is a removal event in map subscriptions.
	Removed bool

	// Offset is the stream offset for map publications. When set, this offset
	// is used instead of broker-assigned offset (used for map broker fan-out).
	Offset uint64
	// Epoch is the stream epoch for map publications. When set along with Offset,
	// the broker will include position information in the message for fan-out.
	Epoch string
	// PrevData carries previous publication data for delta computation.
	// Used when map broker fans out through a standard Broker (e.g. Redis).
	// The receiving node uses this as prevPub in HandlePublication.
	PrevData []byte
	// contains filtered or unexported fields
}

PublishOptions define some fields to alter behaviour of Publish operation.

type PublishReply 

type PublishReply struct {
	// Options to control publication.
	Options PublishOptions

	// Result if set will tell Centrifuge that message already published to
	// channel by handler code. In this case Centrifuge won't try to publish
	// into channel again after handler returned PublishReply. This can be
	// useful if you need to know new Publication offset in your code, or you
	// want to make sure message successfully published to Broker on server
	// side (otherwise only client will get an error).
	Result *PublishResult
}

PublishReply contains fields determining the result on publish.

type PublishResult added in v0.8.0 

type PublishResult struct {
	StreamPosition
	// Suppressed is true when the operation was suppressed (e.g. due to idempotency key deduplication).
	Suppressed bool
	// SuppressReason explains why the operation was suppressed (empty when Suppressed is false).
	SuppressReason SuppressReason
}

PublishResult returned from Publish operation.

type RPCCallback added in v0.13.0 

type RPCCallback func(RPCReply, error)

RPCCallback should be called as soon as handler decides what to do with connection RPCEvent.

type RPCEvent 

type RPCEvent struct {
	// Method is an optional string that contains RPC method name client wants to call.
	// This is an optional field, by default clients send RPC without any method set.
	Method string
	// Data contains RPC untouched payload.
	Data []byte
}

RPCEvent contains fields related to rpc request.

type RPCHandler 

type RPCHandler func(RPCEvent, RPCCallback)

RPCHandler must handle incoming command from client.

type RPCReply 

type RPCReply struct {
	// Data to return in RPC reply to client.
	Data []byte
}

RPCReply contains fields determining the reaction on rpc request.

type RecoveryMode added in v0.33.0 

type RecoveryMode uint8
const (
	RecoveryModeStream RecoveryMode = 0
	RecoveryModeCache  RecoveryMode = 1
)

type RedisAuthCredentials added in v0.39.0 

type RedisAuthCredentials struct {
	Username string
	Password string
	// RefreshAfter, when non-zero, schedules AuthCredentialsFn to be called again at this
	// time to re-authenticate an already established connection with fresh credentials
	// (sends AUTH). Useful when the server limits the lifetime of a connection authenticated
	// with a short-lived token (e.g. AWS ElastiCache IAM auth disconnects after 12 hours).
	RefreshAfter time.Time
}

RedisAuthCredentials contains the credentials returned by AuthCredentialsFn.

type RedisAuthCredentialsContext added in v0.39.0 

type RedisAuthCredentialsContext struct {
	// Address is the address of the Redis server being connected to.
	Address net.Addr
}

RedisAuthCredentialsContext is passed to AuthCredentialsFn.

type RedisBroker added in v0.16.0 

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

RedisBroker uses Redis to implement Broker functionality. This broker allows scaling Centrifuge-based server to many instances and load balance client connections between them. Centrifuge nodes will be connected over Redis PUB/SUB. RedisBroker supports standalone Redis, Redis in master-replica setup with Sentinel, Redis Cluster. Also, it supports client-side consistent sharding between isolated Redis setups. By default, Redis >= 5 required (due to the fact RedisBroker uses STREAM data structure to keep publication history for a channel).

func NewRedisBroker added in v0.16.0 

func NewRedisBroker(n *Node, config RedisBrokerConfig) (*RedisBroker, error)

NewRedisBroker initializes Redis Broker.

func (*RedisBroker) Close added in v0.26.0 

func (b *RedisBroker) Close(_ context.Context) error

func (*RedisBroker) History added in v0.16.0 

func (b *RedisBroker) History(ch string, opts HistoryOptions) ([]*Publication, StreamPosition, error)

History - see Broker.History.

func (*RedisBroker) Publish added in v0.16.0 

func (b *RedisBroker) Publish(ch string, data []byte, opts PublishOptions) (PublishResult, error)

Publish - see Broker.Publish.

func (*RedisBroker) PublishControl added in v0.16.0 

func (b *RedisBroker) PublishControl(data []byte, nodeID string, _ string) error

PublishControl - see Broker.PublishControl.

func (*RedisBroker) PublishJoin added in v0.16.0 

func (b *RedisBroker) PublishJoin(ch string, info *ClientInfo) error

PublishJoin - see Broker.PublishJoin.

func (*RedisBroker) PublishLeave added in v0.16.0 

func (b *RedisBroker) PublishLeave(ch string, info *ClientInfo) error

PublishLeave - see Broker.PublishLeave.

func (*RedisBroker) RegisterBrokerEventHandler added in v0.36.0 

func (b *RedisBroker) RegisterBrokerEventHandler(h BrokerEventHandler) error

RegisterBrokerEventHandler – see Broker.RegisterBrokerEventHandler.

func (*RedisBroker) RegisterControlEventHandler added in v0.36.0 

func (b *RedisBroker) RegisterControlEventHandler(h ControlEventHandler) error

func (*RedisBroker) RemoveHistory added in v0.16.0 

func (b *RedisBroker) RemoveHistory(ch string) error

RemoveHistory - see Broker.RemoveHistory.

func (*RedisBroker) Subscribe added in v0.16.0 

func (b *RedisBroker) Subscribe(channels ...string) error

Subscribe - see Broker.Subscribe.

func (*RedisBroker) Unsubscribe added in v0.16.0 

func (b *RedisBroker) Unsubscribe(channels ...string) error

Unsubscribe - see Broker.Unsubscribe.

type RedisBrokerConfig added in v0.16.0 

type RedisBrokerConfig struct {
	// Prefix to use before every channel name and key in Redis. By default,
	// RedisBroker will use prefix "centrifuge".
	Prefix string

	// Shards is a slice of RedisShard to use. At least one shard must be provided.
	// Data will be consistently sharded by channel over provided Redis shards.
	Shards []*RedisShard

	// UseLists allows enabling usage of Redis LIST instead of STREAM data
	// structure to keep history. LIST support exist mostly for backward
	// compatibility since STREAM seems superior. If you have a use case
	// where you need to turn on this option in new setup - please share,
	// otherwise LIST support can be removed at some point in the future.
	// Iteration over history in reversed order not supported with lists.
	UseLists bool

	// Subscribe on replica Redis nodes. This only works for Redis Cluster
	// and Sentinel setups and requires replica client to be initialized in
	// each RedisShard using RedisShardConfig.ReplicaClientEnabled.
	SubscribeOnReplica bool

	// SkipPubSub enables mode when Redis broker only saves history, without
	// publishing to channels and using PUB/SUB.
	SkipPubSub bool

	// Name of broker, for observability purposes – i.e. becomes part of metrics/logs.
	// By default, empty string is used.
	Name string

	// NumShardedPubSubPartitions when greater than zero allows turning on a mode in which
	// broker will use Redis Cluster with sharded PUB/SUB feature available in
	// Redis >= 7: https://redis.io/docs/manual/pubsub/#sharded-pubsub
	//
	// To achieve sharded PUB/SUB efficiency RedisBroker reduces 16384 Redis Cluster
	// slots to the NumShardedPubSubPartitions value and starts a separate PUB/SUB for each
	// partition. This is necessary because in Centrifuge case one node can work with
	// thousands of different channels – and we can't afford running a separate
	// PUB/SUB connection for each of 16384 possible slots. We re-use partition
	// connection for many channels and make sure that all channels in the partition
	// point to the same Redis Cluster slot.
	//
	// By default, sharded PUB/SUB is not used in Redis Cluster case - Centrifuge uses
	// globally distributed PUBLISH commands in Redis Cluster where each publish is
	// distributed to all nodes in Redis Cluster.
	//
	// Note (!), that turning on NumShardedPubSubPartitions will cause Centrifuge to generate
	// different key names for history and different Redis channel names than in the base
	// Redis Cluster mode due to reasons outlined above.
	NumShardedPubSubPartitions int

	// UsePrecomputedPartitionTags switches sharded PUB/SUB partition hash tags
	// from the bare partition index ("0", "1", ...) to a precomputed table
	// chosen so CRC16 hash slots distribute evenly across any cluster size.
	// The bare-index scheme can collide badly on larger clusters — see
	// https://github.com/centrifugal/centrifuge/issues/554. Enabling this
	// option spreads partitions across slots so cluster nodes get balanced
	// PUB/SUB load even at higher shard counts.
	//
	// When enabled, NumShardedPubSubPartitions must equal one of the sizes
	// returned by redispartition.PrecomputedSizes() (16, 32, 64, 128, 256,
	// 512, 1024, 2048, 4096). Construction returns an error otherwise. The
	// exact tag table is bundled in internal/redispartition/precomputed.go;
	// the tags are stable and will not change.
	//
	// Toggling this option changes the Redis key/channel naming scheme, so
	// flipping it on a running deployment requires a coordinated restart
	// with state cleared (existing channels and history will appear under
	// different keys).
	//
	// Default: false (backward-compatible bare-index tags).
	UsePrecomputedPartitionTags bool

	// LoadSHA1 enables loading SHA1 from Redis via SCRIPT LOAD instead of calculating
	// it on the client side. This is useful for FIPS compliance.
	LoadSHA1 bool
	// contains filtered or unexported fields
}

RedisBrokerConfig is a config for Broker.

type RedisMapBroker added in v0.39.0 

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

RedisMapBroker is a Redis-based MapBroker. Note – it does not work properly with Redis eviction, use with disabled eviction to avoid undefined state.

Message Formats ===============

This broker uses simplified message formats published by Lua scripts:

1. No prefix:

  • Raw protobuf bytes (Publication)
  • Used by direct PUBLISH calls without Lua scripts when stream is not used.

2. Non-delta publications:

  • Format: "offset:epoch:Publication"
  • Where Publication is in protobuf format.

3. Delta publications:

  • Format: "d:offset:epoch:prev_len:prev_publication:curr_len:curr_publication"
  • Where prev_publication and curr_publication are protocol.Publication in protobuf.
  • Enables atomic publishing of current + previous publication for delta compression of publication data.
  • Prev is atomically fetched from stream.

Storage: - Streams (XADD): keeps protocol.Publication - States (HSET): For keyed state - may keep latest protocol.Publication or custom state.

Pagination:

  • ordered state use ZRANGEBYSCORE/ZRANGEBYLEX with LIMIT — exact page sizes.
  • Unordered state use HSCAN with COUNT — COUNT is only a hint, Redis may return more entries than requested (especially for small hashes in listpack encoding). Callers should not rely on exact Limit enforcement for unordered reads.

func NewRedisMapBroker added in v0.39.0 

func NewRedisMapBroker(n *Node, conf RedisMapBrokerConfig) (*RedisMapBroker, error)

NewRedisMapBroker initializes RedisMapBroker.

func (*RedisMapBroker) Clear added in v0.39.0 

func (e *RedisMapBroker) Clear(ctx context.Context, ch string, _ MapClearOptions) error

func (*RedisMapBroker) Close added in v0.39.0 

func (e *RedisMapBroker) Close(_ context.Context) error

Close closes the broker.

func (*RedisMapBroker) Publish added in v0.39.0 

func (e *RedisMapBroker) Publish(ctx context.Context, ch string, key string, opts MapPublishOptions) (MapUpdateResult, error)

Publish publishes data to a stateful channel with optional keyed state.

func (*RedisMapBroker) ReadPresenceState added in v0.39.0 

func (e *RedisMapBroker) ReadPresenceState(ctx context.Context, ch string, opts MapReadStateOptions) ([]*Publication, StreamPosition, error)

ReadPresenceState retrieves presence state with per-entry revisions for converged membership. Each entry includes its revision so client can filter: entry.Revision <= state_revision. If opts.Revision is provided and epoch changed, returns empty entries. Returns Publications with Key=ClientID, Info=ClientInfo, Offset/Epoch for revision.

func (*RedisMapBroker) ReadPresenceStream added in v0.39.0 

func (e *RedisMapBroker) ReadPresenceStream(ctx context.Context, ch string, opts MapReadStreamOptions) (MapStreamResult, error)

ReadPresenceStream retrieves presence event stream (joins/leaves) for a channel. Returns Publications with Info=ClientInfo and Removed flag (true for leave, false for join).

func (*RedisMapBroker) ReadState added in v0.39.0 

func (e *RedisMapBroker) ReadState(ctx context.Context, ch string, opts MapReadStateOptions) (MapStateResult, error)

ReadState retrieves state entries with per-entry revisions for a channel. Each entry includes its revision so client can filter: entry.Revision <= state_revision. If opts.Revision is provided and epoch changed, returns empty entries. Returns entries, stream position, next cursor for pagination, and error. Cursor "0" or "" means end of iteration.

func (*RedisMapBroker) ReadStream added in v0.39.0 

func (e *RedisMapBroker) ReadStream(ctx context.Context, ch string, opts MapReadStreamOptions) (MapStreamResult, error)

ReadStream retrieves publication stream for a channel.

func (*RedisMapBroker) ReadStream2 added in v0.39.0 

func (e *RedisMapBroker) ReadStream2(ctx context.Context, ch string, opts MapReadStreamOptions) (MapStreamResult, error)

ReadStream2 is a 2-call version of ReadStream that splits metadata and publication reads. Call 1: Get metadata (epoch, top_offset) using simpler Lua script Call 2: Read publications using native XRANGE/XREVRANGE Key difference: Must filter out publications with offset > top_offset (non-atomic).

func (*RedisMapBroker) RegisterEventHandler added in v0.39.0 

func (e *RedisMapBroker) RegisterEventHandler(h BrokerEventHandler) error

RegisterEventHandler registers a BrokerEventHandler to handle messages from Pub/Sub.

func (*RedisMapBroker) Remove added in v0.39.0 

func (e *RedisMapBroker) Remove(ctx context.Context, ch string, key string, opts MapRemoveOptions) (MapUpdateResult, error)

Remove removes a key from keyed state state.

func (*RedisMapBroker) Stats added in v0.39.0 

func (e *RedisMapBroker) Stats(ctx context.Context, ch string) (MapStats, error)

Stats returns short stats of current presence data. This is a read-only operation - cleanup is handled by the cleanup worker.

func (*RedisMapBroker) Subscribe added in v0.39.0 

func (e *RedisMapBroker) Subscribe(channels ...string) error

Subscribe to channels.

func (*RedisMapBroker) Unsubscribe added in v0.39.0 

func (e *RedisMapBroker) Unsubscribe(channels ...string) error

Unsubscribe from channels.

type RedisMapBrokerConfig added in v0.39.0 

type RedisMapBrokerConfig struct {
	// Shards is a slice of RedisShard to use. At least one shard must be provided.
	Shards []*RedisShard
	// Prefix to use before every channel name and key in Redis.
	Prefix string
	// Name of broker, for observability purposes – i.e. becomes part of metrics/logs labels.
	// By default, empty string is used.
	Name string
	// LoadSHA1 enables loading SHA1 from Redis via SCRIPT LOAD instead of calculating
	// it on the client side. This is useful for FIPS compliance.
	LoadSHA1 bool
	// IdempotentResultTTL is a time-to-live for idempotent result.
	IdempotentResultTTL time.Duration
	// SubscribeOnReplica allows subscribing on replica Redis nodes.
	SubscribeOnReplica bool
	// SkipPubSub enables mode when broker only works with data structures, without
	// publishing to channels and using PUB/SUB.
	SkipPubSub bool
	// NumShardedPubSubPartitions when greater than zero allows turning on a mode in which
	// broker will use Redis Cluster with sharded PUB/SUB feature available in
	// Redis >= 7: https://redis.io/docs/manual/pubsub/#sharded-pubsub
	NumShardedPubSubPartitions int

	// UsePrecomputedPartitionTags switches sharded PUB/SUB partition hash tags
	// from the bare partition index ("0", "1", ...) to a precomputed table
	// chosen so CRC16 hash slots distribute evenly across any cluster size.
	// See RedisBrokerConfig.UsePrecomputedPartitionTags for details and
	// migration constraints.
	//
	// When enabled, NumShardedPubSubPartitions must equal one of the sizes
	// returned by redispartition.PrecomputedSizes() (16, 32, 64, 128, 256,
	// 512, 1024, 2048, 4096). Construction returns an error otherwise. The
	// exact tag table is bundled in internal/redispartition/precomputed.go;
	// the tags are stable and will not change.
	//
	// Default: false (backward-compatible bare-index tags).
	UsePrecomputedPartitionTags bool

	// CleanupInterval defines how often to run the cleanup worker that
	// generates remove events for expired keyed state entries (presence and state).
	// Default is 1 second. Set to -1 to disable cleanup (make sure you understand the consequences).
	// Applies to all channels using TTL-based state.
	CleanupInterval time.Duration
	// CleanupBatchSize defines max entries to process per channel per cleanup cycle.
	// Default is 100. Applies to all keyed state (presence and state).
	CleanupBatchSize int
	// contains filtered or unexported fields
}

RedisMapBrokerConfig is a config for RedisMapBroker.

type RedisPresenceManager added in v0.16.0 

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

RedisPresenceManager keeps presence in Redis thus allows scaling nodes.

func NewRedisPresenceManager added in v0.16.0 

func NewRedisPresenceManager(n *Node, config RedisPresenceManagerConfig) (*RedisPresenceManager, error)

NewRedisPresenceManager creates new RedisPresenceManager.

func (*RedisPresenceManager) AddPresence added in v0.16.0 

func (m *RedisPresenceManager) AddPresence(ch string, uid string, info *ClientInfo) error

AddPresence - see PresenceManager interface description.

func (*RedisPresenceManager) Close added in v0.26.0 

func (m *RedisPresenceManager) Close(_ context.Context) error

func (*RedisPresenceManager) Presence added in v0.16.0 

func (m *RedisPresenceManager) Presence(ch string) (map[string]*ClientInfo, error)

Presence - see PresenceManager interface description.

func (*RedisPresenceManager) PresenceStats added in v0.16.0 

func (m *RedisPresenceManager) PresenceStats(ch string) (PresenceStats, error)

PresenceStats - see PresenceManager interface description.

func (*RedisPresenceManager) RemovePresence added in v0.16.0 

func (m *RedisPresenceManager) RemovePresence(ch string, clientID string, userID string) error

RemovePresence - see PresenceManager interface description.

type RedisPresenceManagerConfig added in v0.16.0 

type RedisPresenceManagerConfig struct {
	// Prefix to use before every channel name and key in Redis. By default,
	// "centrifuge" prefix will be used.
	Prefix string

	// PresenceTTL is an interval how long to consider presence info
	// valid after receiving presence update. This allows to automatically
	// clean up unnecessary presence entries after TTL passed. Zero value
	// means 60 seconds.
	PresenceTTL time.Duration

	// Shards is a slice of RedisShard to use. At least one shard must be provided.
	// Data will be consistently sharded by channel over provided Redis shards.
	Shards []*RedisShard

	// EnableUserMapping when returns true tells RedisPresenceManager to additionally store
	// user to num client connections hash map and sorted set with unique users in Redis.
	// This increases Redis memory usage since additional structures are used, but provides
	// a way to optimize presence stats retrieving as we can calculate stats quickly on
	// Redis side instead of loading the entire presence information. By default, user mapping
	// is not maintained.
	EnableUserMapping func(channel string) bool

	// UseHashFieldTTL allows using HEXPIRE command to set TTL for hash field. It's only available
	// since Redis 7.4.0 thus disabled by default. Using hash field TTL can be useful to avoid
	// maintaining expiration index in ZSET – so both useful from the throughput and memory usage
	// perspective.
	UseHashFieldTTL bool

	// ReadFromReplica enables reading presence information from replica Redis servers.
	// This only works in Redis Cluster and Sentinel setups and requires replica client
	// to be initialized in each RedisShard using RedisShardConfig.ReplicaClientEnabled.
	ReadFromReplica bool

	// LoadSHA1 enables loading SHA1 from Redis via SCRIPT LOAD instead of calculating
	// it on the client side. This is useful for FIPS compliance.
	LoadSHA1 bool
}

RedisPresenceManagerConfig is a config for RedisPresenceManager.

type RedisShard added in v0.16.0 

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

func NewRedisShard added in v0.16.0 

func NewRedisShard(_ *Node, conf RedisShardConfig) (*RedisShard, error)

NewRedisShard initializes new Redis shard.

func (*RedisShard) Close added in v0.26.0 

func (s *RedisShard) Close()

func (*RedisShard) Mode added in v0.37.0 

func (s *RedisShard) Mode() RedisShardMode

type RedisShardConfig 

type RedisShardConfig struct {
	// Address is a Redis server connection address. Address can be:
	// - host:port
	// - tcp://[[[user]:password]@]host:port[/db][?option1=value1&optionN=valueN]
	// - redis://[[[user]:password]@]host:port[/db][?option1=value1&optionN=valueN]
	// - rediss://[[[user]:password]@]host:port[/db][?option1=value1&optionN=valueN]
	// - unix://[[[user]:password]@]path[?option1=value1&optionN=valueN]
	// It's also possible to use Address with redis+sentinel:// scheme to connect to Redis Sentinel:
	// - redis+sentinel://[[[user]:password]@]host:port?sentinel_master_name=mymaster?addr=host2:port2&addr=host3:port3
	// In case of using redis+sentinel://, sentinel_master_name is required and host:port points to Sentinel instance.
	// It's also possible to connect to Redis Cluster by providing ClusterAddresses instead of Address.
	// It's also possible to connect to Redis Sentinel by providing SentinelAddresses instead of Address.
	Address string
	// ClusterAddresses is a slice of seed cluster addresses to connect to.
	// Each address should be in form of host:port. If ClusterAddresses set then
	// RedisShardConfig.Address not used at all.
	ClusterAddresses []string
	// SentinelAddresses is a slice of Sentinel addresses. Each address should
	// be in form of host:port. If set then Redis address will be automatically
	// discovered from Sentinel. For Sentinel the name of the master instance
	// Sentinel monitors (SentinelMasterName) must be provided. If SentinelAddresses
	// set then RedisShardConfig.Address not used at all.
	SentinelAddresses []string

	// SentinelMasterName is a name of Redis instance master Sentinel monitors.
	SentinelMasterName string
	// SentinelUser is a user for Sentinel ACL-based auth.
	SentinelUser string
	// SentinelPassword is a password for Sentinel. Works with Sentinel >= 5.0.1.
	SentinelPassword string
	// SentinelClientName is a client name for established connections to Sentinel.
	SentinelClientName string
	// SentinelTLSConfig is a TLS configuration for Sentinel connections.
	SentinelTLSConfig *tls.Config

	// DB is Redis database number. If not set then database 0 used.
	// Does not make sense in Redis Cluster case.
	DB int
	// User is a username for Redis ACL-based auth.
	User string
	// Password is password to use when connecting to Redis. If zero then password not used.
	Password string
	// ClientName for established connections with Redis. See https://redis.io/commands/client-setname/
	ClientName string
	// TLSConfig contains connection TLS configuration.
	TLSConfig *tls.Config

	// ConnectTimeout is a timeout on connect operation.
	// By default, 1 second is used.
	ConnectTimeout time.Duration
	// IOTimeout is a timeout on Redis connection operations. This is used as a write deadline
	// for connection, also Redis client we use internally periodically (once in a second) PINGs
	// Redis with this timeout for PING operation to find out stale/broken/blocked connections.
	// By default, 4 seconds is used.
	IOTimeout time.Duration

	// ForceRESP2 if set to true forces using RESP2 protocol for communicating with Redis.
	// By default, Redis client tries to detect supported Redis protocol automatically
	// trying RESP3 first.
	ForceRESP2 bool

	// ReplicaClientEnabled once set to true will initialize replica client for this shard.
	// Replica client can then be used for read-only operations from replica nodes in Redis
	// Cluster or Redis Sentinel setups (single Redis is not allowed). Replica client will
	// be initialized with the same options as the main client but with ReplicaOnly option
	// set to true.
	ReplicaClientEnabled bool

	// AuthCredentialsFn is an optional function to dynamically provide auth credentials.
	// When set, it is called by the Redis client to obtain credentials for each new connection,
	// enabling short-lived token-based authentication (e.g. GCP IAM, AWS IAM).
	AuthCredentialsFn func(RedisAuthCredentialsContext) (RedisAuthCredentials, error)
}

RedisShardConfig contains Redis connection options.

type RedisShardMode added in v0.37.0 

type RedisShardMode string
const (
	RedisShardModeStandalone RedisShardMode = "standalone"
	RedisShardModeCluster    RedisShardMode = "cluster"
	RedisShardModeSentinel   RedisShardMode = "sentinel"
)

type RefreshCallback added in v0.13.0 

type RefreshCallback func(RefreshReply, error)

RefreshCallback should be called as soon as handler decides what to do with connection refresh event.

type RefreshEvent 

type RefreshEvent struct {
	// ClientSideRefresh is true for refresh initiated by client-side refresh workflow.
	ClientSideRefresh bool
	// Token will only be set in case of using client-side refresh mechanism.
	Token string
}

RefreshEvent contains fields related to refresh event.

type RefreshHandler 

type RefreshHandler func(RefreshEvent, RefreshCallback)

RefreshHandler called when it's time to validate client connection and update its expiration time if it's still actual.

Centrifuge library supports two ways of refreshing connection: client-side and server-side.

The default mechanism is server-side, this means that as soon refresh handler set and connection expiration time happens (by timer) – refresh handler will be called.

If ClientSideRefresh in ConnectReply inside ConnectingHandler set to true then library uses client-side refresh mechanism. In this case library relies on Refresh commands sent from client periodically to refresh connection. Refresh command contains updated connection token.

type RefreshOption added in v0.18.0 

type RefreshOption func(options *RefreshOptions)

RefreshOption is a type to represent various Refresh options.

func WithRefreshAllUsers added in v0.39.0 

func WithRefreshAllUsers(enabled bool) RefreshOption

WithRefreshAllUsers changes the meaning of an empty userID passed to Node.Refresh from "anonymous bucket only" to "all users including anonymous." It is a no-op when userID is non-empty.

func WithRefreshClient added in v0.18.0 

func WithRefreshClient(clientID string) RefreshOption

WithRefreshClient to limit refresh only for specified client ID.

func WithRefreshExpireAt added in v0.18.0 

func WithRefreshExpireAt(expireAt int64) RefreshOption

WithRefreshExpireAt to set unix seconds in the future when connection should expire. Zero value means no expiration.

func WithRefreshExpired added in v0.18.0 

func WithRefreshExpired(expired bool) RefreshOption

WithRefreshExpired to set expired flag - connection will be closed with DisconnectExpired.

func WithRefreshInfo added in v0.18.0 

func WithRefreshInfo(info []byte) RefreshOption

WithRefreshInfo to override connection info.

func WithRefreshLabelFilter added in v0.39.0 

func WithRefreshLabelFilter(f *FilterNode) RefreshOption

WithRefreshLabelFilter restricts the refresh to connections whose Client.Labels match the filter. See RefreshOptions.LabelFilter for the detailed contract.

func WithRefreshSession added in v0.21.1 

func WithRefreshSession(sessionID string) RefreshOption

WithRefreshSession to limit refresh only for specified session ID.

type RefreshOptions added in v0.18.0 

type RefreshOptions struct {
	// Expired can close connection with expired reason.
	Expired bool
	// ExpireAt defines time in future when subscription should expire,
	// zero value means no expiration.
	ExpireAt int64
	// Info defines custom channel information, zero value means no channel information.
	Info []byte
	// contains filtered or unexported fields
}

RefreshOptions ...

type RefreshReply 

type RefreshReply struct {
	// Expired tells Centrifuge that connection expired. In this case connection will be
	// closed with DisconnectExpired.
	Expired bool
	// ExpireAt defines time in future when connection should expire,
	// zero value means no expiration.
	ExpireAt int64
	// Info allows modifying connection information,
	// zero value means no modification of current connection Info.
	Info []byte
}

RefreshReply contains fields determining the reaction on refresh event.

type RegistererGatherer added in v0.34.0 

type RegistererGatherer interface {
	prometheus.Registerer
	prometheus.Gatherer
}

RegistererGatherer defines an interface that combines Registerer and Gatherer from Prometheus. Prometheus Registry implements both interfaces.

type SSEConfig added in v0.23.0 

type SSEConfig struct {
	PingPongConfig
	// MaxRequestBodySize limits initial request body size (when SSE starts with POST).
	MaxRequestBodySize int
}

SSEConfig represents config for SSEHandler.

type SSEHandler added in v0.23.0 

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

SSEHandler handles WebSocket client connections. WebSocket protocol is a bidirectional connection between a client and a server for low-latency communication.

func NewSSEHandler added in v0.23.0 

func NewSSEHandler(node *Node, config SSEConfig) *SSEHandler

NewSSEHandler creates new SSEHandler.

func (*SSEHandler) ServeHTTP added in v0.23.0 

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

type SharedPollChannelOptions added in v0.39.0 

type SharedPollChannelOptions struct {
	// MaxKeysPerConnection limits how many keys a single connection
	// can track in this channel. Zero value means 5000.
	MaxKeysPerConnection int
	// RefreshInterval sets how often the refresh worker calls OnSharedPoll.
	// Zero value means 10 * time.Second.
	RefreshInterval time.Duration
	// RefreshBatchSize sets the maximum number of item keys per OnSharedPoll call.
	// Zero value means 1000.
	RefreshBatchSize int
	// Mode controls request format.
	// "versionless" (default, also ""): backend returns {key, data} without versions.
	//   Centrifugo detects changes via content hash and generates internal versions.
	// "versioned": backend returns {key, data, version}. Versions are included in
	//   requests, allowing the backend to skip unchanged items.
	Mode string
	// KeepLatestData controls whether TrackedEntry stores the last data payload
	// per item. Default false.
	KeepLatestData bool
	// RefreshIntervalFn, if set, is called after each refresh cycle with the
	// cycle's wall time, and returns the interval before the next cycle. Overrides
	// static RefreshInterval. Used by Centrifugo PRO for adaptive backpressure.
	RefreshIntervalFn func(cycleDuration time.Duration) time.Duration
	// CallTimeout sets the maximum duration for each OnSharedPoll callback invocation.
	// Zero value means 30 * time.Second.
	CallTimeout time.Duration
	// ChannelShutdownDelay is the delay before shutting down a channel
	// state after the last item is untracked. Zero means default (1s).
	// Set to -1 for immediate shutdown with no delay.
	ChannelShutdownDelay time.Duration
	// NotificationBatchMaxSize sets the maximum number of notified keys to
	// accumulate before triggering an immediate backend poll. The batch fires
	// when either this size or NotificationBatchMaxDelay is reached first.
	// Zero disables size-based batching (batching still fires on delay if
	// NotificationBatchMaxDelay is set).
	NotificationBatchMaxSize int
	// NotificationBatchMaxDelay sets the maximum time to wait while
	// accumulating notified keys before triggering an immediate backend poll.
	// Zero disables delay-based batching (batching still fires on size if
	// NotificationBatchMaxSize is set). When neither is set, notifications
	// are dispatched immediately (no batching).
	NotificationBatchMaxDelay time.Duration
	// TrackExpiredExtraDelay is extra time given to client to refresh track signature
	// after it expires. Keys not refreshed within this delay are silently removed
	// from server state. Zero means 25 * time.Second.
	TrackExpiredExtraDelay time.Duration
	// PublishEnabled enables cross-node distribution for SharedPollPublish.
	// When true, SharedPollManager subscribes to the Broker (via node.getBroker)
	// for this channel, allowing SharedPollPublish to distribute publications
	// to all nodes. When false, SharedPollPublish is local-only.
	PublishEnabled bool
}

SharedPollChannelOptions configures a shared poll channel.

type SharedPollConfig added in v0.39.0 

type SharedPollConfig struct {
	// GetSharedPollChannelOptions returns per-channel options for shared poll
	// channels. Required for shared poll operations — if nil, any shared poll
	// subscribe returns an error. Called when a new channel first appears in
	// SharedPollManager. Must be fast, no I/O.
	// This is an EXPERIMENTAL API.
	GetSharedPollChannelOptions func(channel string) (SharedPollChannelOptions, bool)
	// ConcurrencyLimit sets the maximum number of concurrent backend
	// calls across all shared poll channels. All channels share this pool.
	// Zero means 64.
	ConcurrencyLimit int
}

type SharedPollEvent added in v0.39.0 

type SharedPollEvent struct {
	Channel string
	Items   []SharedPollItem
}

SharedPollEvent contains fields for a shared poll refresh call.

type SharedPollHandler added in v0.39.0 

type SharedPollHandler func(ctx context.Context, event SharedPollEvent) (SharedPollResult, error)

SharedPollHandler is called by the refresh worker to fetch current item data from the backend. Registered on Node, not per-client.

type SharedPollItem added in v0.39.0 

type SharedPollItem struct {
	Key     string
	Version uint64 // 0 when Mode is "versionless"
}

SharedPollItem represents an item being refreshed.

type SharedPollManager added in v0.39.0 

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

SharedPollManager manages all shared poll channels for a Node. One instance per Node. Thread-safe.

func (*SharedPollManager) Epoch added in v0.39.0 

func (m *SharedPollManager) Epoch(channel string, isVersionless bool) string

Epoch returns the epoch string for subscribe replies.

Versionless channels: server-generated, stable for the lifetime of channel state. Changes when state is recreated after shutdown delay expires.

Versioned channels: publisher-supplied, set on first publish/refresh. Changes (epoch flip) trigger unsubscribe of all current subscribers with insufficient-state code so they re-track from version 0 on resubscribe.

func (*SharedPollManager) SharedPollRevokeKeys added in v0.39.0 

func (m *SharedPollManager) SharedPollRevokeKeys(channel string, keys []string, users []string, excludeUsers []string)

SharedPollRevokeKeys removes items for matching connections. Sends Publication{Removed: true} to affected subscribers and cleans up hub + itemIndex.

type SharedPollNotification added in v0.39.0 

type SharedPollNotification struct {
	Items []SharedPollNotificationItem `json:"items"`
}

SharedPollNotification is the top-level JSON envelope for notifications published to the notification pub/sub channel.

type SharedPollNotificationItem added in v0.39.0 

type SharedPollNotificationItem struct {
	Channel string `json:"channel"`
	Key     string `json:"key"`
}

SharedPollNotificationItem represents a lightweight notification that a specific key in a channel has changed. Used to trigger immediate backend polls for just the affected keys instead of waiting for the next timer-based cycle.

type SharedPollRefreshItem added in v0.39.0 

type SharedPollRefreshItem struct {
	Key      string
	Data     []byte
	Version  uint64
	Removed  bool
	PrevData []byte
}

SharedPollRefreshItem represents a single item in a shared poll refresh response.

type SharedPollResult added in v0.39.0 

type SharedPollResult struct {
	Items []SharedPollRefreshItem
	// Epoch is the publisher's current epoch for this channel. When set, an
	// epoch change vs the channel's stored epoch resets all per-key versions
	// and unsubscribes current subscribers with insufficient-state code,
	// forcing a fresh subscribe handshake. Empty epoch means no epoch
	// invalidation logic — pure version comparison.
	//
	// Only meaningful for versioned shared-poll channels. Versionless
	// channels use server-generated channel-level epoch and ignore this.
	Epoch string
}

SharedPollResult contains the response from a shared poll refresh.

type StateSnapshotHandler added in v0.23.0 

type StateSnapshotHandler func() (any, error)

StateSnapshotHandler must return a copy of current client's internal state. Returning a copy is important to avoid data races.

type StreamFilter added in v0.39.0 

type StreamFilter struct {
	// Since filters to publications after this position. Only publications
	// with Offset > Since.Offset are returned. The epoch must match.
	Since *StreamPosition

	// Limit controls maximum publications returned:
	//  -1 = no limit (return all publications in stream)
	//   0 = return only current stream position (no publications)
	//  >0 = return at most this many publications
	Limit int

	// Reverse returns publications in reverse order (newest first).
	// Default is oldest first (ascending offset order).
	Reverse bool
}

StreamFilter controls which publications are returned from ReadStream.

type StreamPosition added in v0.8.0 

type StreamPosition struct {
	// Offset defines publication incremental offset inside a stream.
	Offset uint64
	// Epoch allows handling situations when storage
	// lost stream entirely for some reason (expired or lost after restart) and we
	// want to track this fact to prevent successful recovery from another stream.
	// I.e. for example we have a stream [1, 2, 3], then it's lost and new stream
	// contains [1, 2, 3, 4], client that recovers from position 3 will only receive
	// publication 4 missing 1, 2, 3 from new stream. With epoch, we can tell client
	// that correct recovery is not possible.
	Epoch string
}

StreamPosition contains fields to describe position in stream. At moment this is used for automatic recovery mechanics. More info about stream recovery in Centrifugo docs: https://centrifugal.dev/docs/server/history_and_recovery.

type SubRefreshCallback added in v0.13.0 

type SubRefreshCallback func(SubRefreshReply, error)

SubRefreshCallback should be called as soon as handler decides what to do with connection SubRefreshEvent.

type SubRefreshEvent 

type SubRefreshEvent struct {
	// ClientSideRefresh is true for refresh initiated by client-side subscription
	// refresh workflow.
	ClientSideRefresh bool
	// Channel to which SubRefreshEvent belongs to.
	Channel string
	// Token will only be set in case of using client-side subscription refresh mechanism.
	Token string
}

SubRefreshEvent contains fields related to subscription refresh event.

type SubRefreshHandler 

type SubRefreshHandler func(SubRefreshEvent, SubRefreshCallback)

SubRefreshHandler called when it's time to validate client subscription to channel and update it's state if needed.

If ClientSideRefresh in SubscribeReply inside SubscribeHandler set to true then library uses client-side subscription refresh mechanism. In this case library relies on SubRefresh commands sent from client periodically to refresh subscription. SubRefresh command contains updated subscription token.

type SubRefreshReply 

type SubRefreshReply struct {
	// Expired tells Centrifuge that subscription expired. In this case connection will be
	// closed with DisconnectExpired.
	Expired bool
	// ExpireAt is a new Unix time of expiration. Zero value means no expiration.
	ExpireAt int64
	// Info is a new channel-scope info. Zero value means do not change previous one.
	Info []byte
	// ServerTagsFilter is an optional updated server-side tags filter, see
	// SubscribeOptions.ServerTagsFilter. It is applied on both client-side and
	// server-side subscription refresh. When nil, the existing filter is left
	// unchanged. When set, Centrifuge compares it with the current filter, and
	// if different, starts delivering by the new one. A subscription which
	// can't just switch filters must start over:
	//   - a map subscription is unsubscribed with UnsubscribeCodeStateInvalidated
	//     to re-sync its state;
	//   - a subscription using delta compression is unsubscribed with
	//     UnsubscribeCodeInsufficient to resubscribe without delta, as delta is
	//     not negotiated together with a server tags filter;
	//   - a server-side subscription which needs either is disconnected with
	//     DisconnectInsufficientState, since a client can't resubscribe to it.
	// Once a subscription has a server tags filter, it is refused history, as
	// subscribing with one is. Return the same filter in subscribe or connect
	// options as on refresh: otherwise every subscription starts without it,
	// and a server-side delta subscription reconnects on every expiration.
	ServerTagsFilter *FilterNode
}

SubRefreshReply contains fields determining the reaction on subscription refresh event.

type SubscribeCallback added in v0.13.0 

type SubscribeCallback func(SubscribeReply, error)

SubscribeCallback should be called as soon as handler decides what to do with connection subscribe event, exactly once.

type SubscribeEvent 

type SubscribeEvent struct {
	// Channel client wants to subscribe to.
	Channel string
	// Token will only be set for token channels. This is a task of application
	// to check that subscription to a channel has valid token.
	Token string
	// Data received from client as part of Subscribe Command.
	Data []byte
	// Positioned is true when Client wants to create subscription with positioned property.
	Positioned bool
	// Recoverable is true when Client wants to create subscription with recoverable property.
	Recoverable bool
	// JoinLeave is true when Client wants to receive join/leave messages.
	JoinLeave bool
	// Type is the subscription type requested by the client (map, presence, etc.).
	// For regular subscriptions this is SubscriptionTypeStream (zero value).
	Type SubscriptionType
}

SubscribeEvent contains fields related to subscribe event.

type SubscribeHandler 

type SubscribeHandler func(SubscribeEvent, SubscribeCallback)

SubscribeHandler called when client wants to subscribe on channel. If the SubscribeCallback is invoked with a nil error, exactly one UnsubscribeHandler call follows for the channel: when the subscription ends, or when the attempt ends without becoming a subscription (UnsubscribeEvent.Subscribed is false then). Note that for map subscriptions SubscribeHandler is called on the first subscribe request only, not on further pagination requests.

An unsubscribe of the client from the channel (Client.Unsubscribe, Node.Unsubscribe) waits for the subscribe in progress, up to 5 seconds, and disconnects the client if it does not finish: don't unsubscribe the client from the channel inside SubscribeHandler before invoking the callback.

type SubscribeOption added in v0.16.0 

type SubscribeOption func(*SubscribeOptions)

SubscribeOption is a type to represent various Subscribe options.

func WithAutoCacheRecover added in v0.39.0 

func WithAutoCacheRecover(enabled bool) SubscribeOption

WithAutoCacheRecover allows setting SubscribeOptions.AutoCacheRecover. It makes Centrifuge initiate cache recovery for a subscription even if the client did not request it. See SubscribeOptions.AutoCacheRecover for details.

func WithChannelInfo added in v0.16.0 

func WithChannelInfo(chanInfo []byte) SubscribeOption

WithChannelInfo ...

func WithEmitJoinLeave added in v0.24.0 

func WithEmitJoinLeave(enabled bool) SubscribeOption

WithEmitJoinLeave ...

func WithEmitPresence added in v0.24.0 

func WithEmitPresence(enabled bool) SubscribeOption

WithEmitPresence ...

func WithExpireAt added in v0.16.0 

func WithExpireAt(expireAt int64) SubscribeOption

WithExpireAt allows setting ExpireAt field.

func WithPositioning added in v0.24.0 

func WithPositioning(enabled bool) SubscribeOption

WithPositioning ...

func WithPushJoinLeave added in v0.24.0 

func WithPushJoinLeave(enabled bool) SubscribeOption

WithPushJoinLeave ...

func WithRecoverSince added in v0.18.0 

func WithRecoverSince(since *StreamPosition) SubscribeOption

WithRecoverSince allows setting SubscribeOptions.RecoverFrom.

func WithRecovery added in v0.24.0 

func WithRecovery(enabled bool) SubscribeOption

WithRecovery ...

func WithRecoveryMode added in v0.33.0 

func WithRecoveryMode(mode RecoveryMode) SubscribeOption

WithRecoveryMode ...

func WithSubscribeAllUsers added in v0.39.0 

func WithSubscribeAllUsers(enabled bool) SubscribeOption

WithSubscribeAllUsers changes the meaning of an empty userID passed to Node.Subscribe from "anonymous bucket only" to "all users including anonymous." It is a no-op when userID is non-empty — the per-user path is always taken in that case. Typically combined with WithSubscribeLabelFilter to narrow the fleet-wide dispatch.

func WithSubscribeClient added in v0.16.0 

func WithSubscribeClient(clientID string) SubscribeOption

WithSubscribeClient allows setting client ID that should be subscribed. This option not used when Client.Subscribe called.

func WithSubscribeData added in v0.16.0 

func WithSubscribeData(data []byte) SubscribeOption

WithSubscribeData allows setting custom data to send with subscribe push.

func WithSubscribeHistoryMetaTTL added in v0.29.0 

func WithSubscribeHistoryMetaTTL(metaTTL time.Duration) SubscribeOption

WithSubscribeHistoryMetaTTL allows setting SubscribeOptions.HistoryMetaTTL.

func WithSubscribeLabelFilter added in v0.39.0 

func WithSubscribeLabelFilter(f *FilterNode) SubscribeOption

WithSubscribeLabelFilter restricts the subscribe to connections whose Client.Labels match the filter. See SubscribeOptions.LabelFilter for the detailed contract.

func WithSubscribeSession added in v0.21.1 

func WithSubscribeSession(sessionID string) SubscribeOption

WithSubscribeSession allows setting session ID that should be subscribed. This option not used when Client.Subscribe called.

func WithSubscribeSource added in v0.26.0 

func WithSubscribeSource(source uint8) SubscribeOption

WithSubscribeSource allows setting SubscribeOptions.Source.

type SubscribeOptions added in v0.13.0 

type SubscribeOptions struct {

	// ExpireAt defines time in future when subscription should expire,
	// zero value means no expiration.
	ExpireAt int64
	// ChannelInfo defines custom channel information, zero value means no channel information.
	ChannelInfo []byte
	// EmitPresence turns on participating in channel presence - i.e. client
	// subscription will emit presence updates to PresenceManager and will be visible
	// in a channel presence result.
	EmitPresence bool
	// EmitJoinLeave turns on emitting Join and Leave events from the subscribing client.
	// See also PushJoinLeave if you want current client to receive join/leave messages.
	EmitJoinLeave bool
	// PushJoinLeave turns on receiving channel Join and Leave events by the client.
	// Subscriptions which emit join/leave events should have EmitJoinLeave on.
	PushJoinLeave bool
	// When position is on client will additionally sync its position inside a stream
	// to prevent publication loss. The loss can happen due to at most once guarantees
	// of PUB/SUB model. Make sure you are enabling EnablePositioning in channels that
	// maintain Publication history stream. When EnablePositioning is on Centrifuge will
	// include StreamPosition information to subscribe response - for a client to be
	// able to manually track its position inside a stream.
	EnablePositioning bool
	// EnableRecovery turns on automatic recovery for a channel. In this case
	// client will try to recover missed messages upon resubscribe to a channel
	// after reconnect to a server. This option also enables client position
	// tracking inside a stream (i.e. enabling EnableRecovery will automatically
	// enable EnablePositioning option) to prevent occasional publication loss.
	// Make sure you are using EnableRecovery in channels that maintain Publication
	// history stream.
	EnableRecovery bool
	// RecoveryMode is by default RecoveryModeStream, but can be also RecoveryModeCache.
	RecoveryMode RecoveryMode
	// Data to send to a client with Subscribe Push.
	Data []byte
	// RecoverSince will try to subscribe a client and recover from a certain StreamPosition.
	RecoverSince *StreamPosition
	// AutoCacheRecover, when set, makes Centrifuge initiate cache recovery for a subscription
	// even if the client did not request it (i.e. without a recover flag/position in the
	// subscribe request). It applies to both client-initiated and server-side subscriptions and
	// requires EnableRecovery to be set.
	//
	// AutoCacheRecover only takes effect in cache recovery mode (RecoveryModeCache): there it
	// delivers the latest publication on each (re)subscribe regardless of position, which is
	// especially useful for unidirectional clients with server-side subscriptions that can't
	// request recovery themselves, or to avoid requiring an empty "since" on the client.
	//
	// In stream recovery mode AutoCacheRecover is ignored: forcing recovery without a client
	// position can't preserve continuity (there is no baseline to compare against), and the
	// cache-style "deliver latest" semantics don't map onto a stream. To recover a stream
	// subscription use the client recover flag/position, or SubscribeOptions.RecoverSince for a
	// specific position; to deliver an initial backlog use SubscribeReply.Publications.
	// AutoCacheRecover is also ignored for map subscriptions.
	AutoCacheRecover bool

	// HistoryMetaTTL allows to override default (set in Config.HistoryMetaTTL) history
	// meta information expiration time.
	HistoryMetaTTL time.Duration

	// AllowedDeltaTypes is a whitelist of DeltaType subscribers can negotiate. At this point Centrifuge
	// only supports DeltaTypeFossil. If zero value – clients won't be able to negotiate delta encoding
	// within a channel and will receive full data in publications.
	// Delta encoding is an EXPERIMENTAL feature and may be changed.
	AllowedDeltaTypes []DeltaType
	// Source is a way to mark the source of Subscription - i.e. where it comes from. May be useful
	// for inspection of a connection during its lifetime.
	Source uint8
	// AllowChannelCompaction if true allows client to negotiate channel ID compaction –
	// Centrifuge will replace channel names with shorter IDs in subscription pushes.
	// If disabled, clients receive the full channel name in all pushes. Requires support
	// in client SDK.
	AllowChannelCompaction bool
	// AllowTagsFilter if set to true allows client to use publication filter by tags. If not allowed
	// and client provided a filter – the BadRequest error will be returned.
	// Important note here, since channel permissions are managed on channel level, tags filtering
	// must be used as a bandwidth optimization, not an access control mechanism.
	AllowTagsFilter bool
	// ServerTagsFilter is a server-controlled tags filter applied to publications before delivery.
	// Unlike AllowTagsFilter (which enables client-side filtering), this filter is set by the server
	// (via subscribe proxy or JWT) and cannot be overridden by the client. When both server and
	// client filters are set, they are applied independently (AND semantics). ServerTagsFilter
	// can not be used together with Delta Compression in subscription. History is not filtered
	// by it, so a subscription with ServerTagsFilter is refused history requests; recovery on
	// subscribe applies the filter and stays the way to get missed publications.
	ServerTagsFilter *FilterNode

	// Type defines the subscription type. Use SubscriptionTypeMap for map subscriptions.
	// For regular subscriptions this can be left as zero value (SubscriptionTypeStream).
	Type SubscriptionType
	// MapClientPresenceChannel is the full channel name for client presence.
	// When set, client presence will be published to this channel on subscribe.
	// Empty string means no client presence publishing.
	MapClientPresenceChannel string
	// MapUserPresenceChannel is the full channel name for user presence.
	// When set, user presence will be published to this channel on subscribe.
	// Empty string means no user presence publishing.
	MapUserPresenceChannel string
	// MapRemoveClientOnUnsubscribe enables automatic cleanup of map state when the
	// subscription ends – the key matching current client ID will be removed.
	// This is useful for ephemeral state like cursor positions or temporary resources
	// that should not persist after the client leaves.
	MapRemoveClientOnUnsubscribe bool
	// ClientPublishDebounceInterval when > 0, included in the subscribe result
	// as publish_debounce (milliseconds). The SDK debounces client-initiated publishes
	// to this channel locally.
	ClientPublishDebounceInterval time.Duration
	// contains filtered or unexported fields
}

SubscribeOptions define per-subscription options.

type SubscribeReply 

type SubscribeReply struct {
	// Options to control subscription.
	Options SubscribeOptions

	// ClientSideRefresh tells library to use client-side refresh logic: i.e. send
	// SubRefresh commands with new Subscription Token. If not set then server-side
	// SubRefresh handler will be used.
	ClientSideRefresh bool

	// Publications to include in the subscribe result. These are delivered to the
	// client as publication events before live publications begin — same mechanism
	// as recovery publications. Useful for delivering initial state on subscribe.
	Publications []*Publication

	// SubscriptionReady channel if provided will be closed as soon as Centrifuge
	// written subscribe reply to the connection, so it's possible to start writing
	// publications into a channel using experimental Client.WritePublication method.
	// In usual flow you don't need to provide this channel at all.
	// This is EXPERIMENTAL and may be removed in the future.
	SubscriptionReady chan struct{}
}

SubscribeReply contains fields determining the reaction on subscribe event.

type SubscribeRequest added in v0.16.0 

type SubscribeRequest struct {
	// Recover enables publication recovery for a channel.
	Recover bool
	// Epoch last seen by a client.
	Epoch string
	// Offset last seen by a client.
	Offset uint64
}

SubscribeRequest contains state of subscription to a channel.

type SubscriptionType added in v0.39.0 

type SubscriptionType int32

SubscriptionType defines the type of subscription. 

const (
	// SubscriptionTypeStream is a regular PUB/SUB subscription (default).
	SubscriptionTypeStream SubscriptionType = 0
	// SubscriptionTypeMap is a map subscription with keyed state.
	SubscriptionTypeMap SubscriptionType = 1
	// SubscriptionTypeMapClients is a client presence subscription on a map channel.
	SubscriptionTypeMapClients SubscriptionType = 2
	// SubscriptionTypeMapUsers is a user presence subscription on a map channel.
	SubscriptionTypeMapUsers SubscriptionType = 3
	// SubscriptionTypeSharedPoll is a shared poll subscription.
	SubscriptionTypeSharedPoll SubscriptionType = 4
)

func (SubscriptionType) IsMapPresence added in v0.39.0 

func (t SubscriptionType) IsMapPresence() bool

IsMapPresence reports whether t is a map presence subscription type (SubscriptionTypeMapClients or SubscriptionTypeMapUsers).

func (SubscriptionType) String added in v0.39.0 

func (t SubscriptionType) String() string

type SuppressReason added in v0.39.0 

type SuppressReason string

SuppressReason explains why a Publish or Remove operation was suppressed. 

const (
	// SuppressReasonNone indicates the operation was not suppressed.
	SuppressReasonNone SuppressReason = ""

	// SuppressReasonIdempotency indicates duplicate detected via idempotency key.
	SuppressReasonIdempotency SuppressReason = "idempotency"

	// SuppressReasonVersion indicates skipped due to out-of-order version.
	SuppressReasonVersion SuppressReason = "version"

	// SuppressReasonKeyExists indicates KeyModeIfNew was used but key already exists.
	SuppressReasonKeyExists SuppressReason = "key_exists"

	// SuppressReasonKeyNotFound indicates KeyModeIfExists was used but key doesn't exist,
	// or Remove was called on a non-existent key.
	SuppressReasonKeyNotFound SuppressReason = "key_not_found"

	// SuppressReasonPositionMismatch indicates CAS failed - key's position changed.
	SuppressReasonPositionMismatch SuppressReason = "position_mismatch"
)

type SurveyCallback added in v0.15.0 

type SurveyCallback func(SurveyReply)

SurveyCallback should be called with SurveyReply as soon as survey completed.

type SurveyEvent added in v0.15.0 

type SurveyEvent struct {
	Op   string
	Data []byte
}

SurveyEvent with Op and Data of survey.

type SurveyHandler added in v0.15.0 

type SurveyHandler func(SurveyEvent, SurveyCallback)

SurveyHandler allows setting survey handler function.

type SurveyReply added in v0.15.0 

type SurveyReply struct {
	Code uint32
	Data []byte
}

SurveyReply contains survey reply fields.

type SurveyResult added in v0.15.0 

type SurveyResult struct {
	Code uint32
	Data []byte
}

SurveyResult from node.

type TimerCanceler added in v0.36.0 

type TimerCanceler interface {
	// Cancel the timer.
	Cancel()
}

TimerCanceler is the interface returned from ScheduleTimer which allows the task to be cancelled. EXPERIMENTAL API.

type TimerScheduler added in v0.36.0 

type TimerScheduler interface {
	// ScheduleTimer adds a callback for later execution. The TimerCanceler is returned.
	ScheduleTimer(duration time.Duration, callback func()) TimerCanceler
}

TimerScheduler is the interface for scheduling timers.

Callbacks may block: some client periodic operations perform network calls (subscription refresh via RefreshHandler, and — for schedulers that run callbacks on a goroutine shared between connections — anything the application does in AliveHandler). An implementation that runs several connections' callbacks on one goroutine therefore lets a single slow connection delay the others' pings, so it should bound how many callbacks share a goroutine.

Centrifuge does not rely on this for presence updates specifically: when a TimerScheduler is set it runs the presence tick on its own goroutine, since that tick calls into PresenceManager/MapBroker on every subscribed channel.

EXPERIMENTAL API.

type TrackBatch added in v0.39.0 

type TrackBatch struct {
	Items []TrackItem
	// Signature is an opaque string sent by the client to prove authorization for
	// the requested channel and keys in this batch. In Centrifugo this is an HMAC
	// token generated by the application backend. The handler MUST verify Signature
	// before allowing the batch to proceed — returning nil error without validation
	// lets any client subscribe to arbitrary keys, bypassing access control.
	Signature string
}

TrackBatch is one signed group of items in a TrackEvent. Each batch carries the FULL set of items its Signature was computed over.

type TrackBatchReply added in v0.39.0 

type TrackBatchReply struct {
	// ExpireAt is the Unix timestamp (seconds) after which the keys in this batch
	// are considered expired and stop receiving updates. The client must re-track
	// with a fresh signature to continue.
	//
	// Setting a reasonable ExpireAt is important for proper permission control: it
	// limits how long a client can receive data after a single authorization check.
	// Without it (ExpireAt=0), a successfully tracked client receives updates
	// indefinitely — even if the user's access has since been revoked.
	// A typical value matches the signature/token TTL issued by the backend
	// (e.g. 5–15 minutes).
	ExpireAt int64
}

TrackBatchReply contains the per-batch reaction inside a TrackReply.

type TrackCallback added in v0.39.0 

type TrackCallback func(TrackReply, error)

TrackCallback should be called with TrackReply or error.

type TrackEvent added in v0.39.0 

type TrackEvent struct {
	Channel string
	Batches []TrackBatch
}

TrackEvent contains fields related to a track request on a keyed channel. A single request may carry multiple signed batches (e.g. when the SDK replays cached signatures after reconnect) — the handler MUST validate every batch's signature before allowing the track to proceed.

type TrackHandler added in v0.39.0 

type TrackHandler func(TrackEvent, TrackCallback)

TrackHandler is called when a client sends a track request on a shared poll channel. The handler MUST validate every TrackBatch.Signature in TrackEvent.Batches and return an error if any is invalid — this is the sole authorization gate for which keys a client may receive data for. On success, return a TrackReply with one TrackBatchReply per batch (parallel to event.Batches), each with ExpireAt set to bound how long the client may receive updates for that batch without re-authorization.

type TrackItem added in v0.39.0 

type TrackItem struct {
	Key     string
	Version uint64
}

TrackItem represents a single item being tracked (key + version).

type TrackReply added in v0.39.0 

type TrackReply struct {
	Batches []TrackBatchReply
}

TrackReply contains the reaction to a keyed track event. Batches is parallel to TrackEvent.Batches — len(Batches) MUST equal len(event.Batches).

type Transport 

type Transport interface {
	TransportInfo
	// Write should write single push data into a connection. Every byte slice
	// here is a single Reply (or Push for unidirectional transport) encoded
	// according transport ProtocolType.
	Write([]byte) error
	// WriteMany should write data into a connection. Every byte slice here is a
	// single Reply (or Push for unidirectional transport) encoded according
	// transport ProtocolType.
	// The reason why we have both Write and WriteMany here is to have a path
	// without extra allocations for massive broadcasts (since variadic args cause
	// allocation).
	WriteMany(...[]byte) error
	// Close must close transport. Transport implementation can optionally
	// handle Disconnect passed here. For example builtin WebSocket transport
	// sends Disconnect as part of websocket.CloseMessage.
	Close(Disconnect) error
}

Transport abstracts a connection transport between server and client. It does not contain Read method as reading can be handled by connection handler code (for example by WebsocketHandler.ServeHTTP).

type TransportAcceptedLabels added in v0.38.0 

type TransportAcceptedLabels struct {
	// Transport is the transport type (e.g., "websocket", "http_stream", "sse").
	Transport string
	// AcceptProtocol is the transport protocol used to accept connection (can be "h1", "h2", "h3").
	AcceptProtocol string
}

TransportAcceptedLabels contains labels for transport connection metrics. This struct is designed to be extensible - additional label fields can be added in the future without breaking compatibility.

type TransportInfo 

type TransportInfo interface {
	// Name returns a name of transport.
	Name() string
	// AcceptProtocol protocol should return information about network protocol used to
	// create transport. This is useful to distinguish between different HTTP protocols
	// used to create transport. For example, WebSocket transport can be created over
	// HTTP/1.1 and HTTP/2 protocols. For HTTP, it's recommended to use "h1", "h2", "h3"
	// values. This must be low-cardinality set of values. It is used only for observability
	// purposes, custom transport implementations can choose to return an empty string here.
	AcceptProtocol() string
	// Protocol returns an underlying transport protocol type used by transport.
	// JSON or Protobuf protocol types are supported by Centrifuge. Message encoding
	// happens of Client level.
	Protocol() ProtocolType
	// ProtocolVersion returns protocol version used by transport.
	ProtocolVersion() ProtocolVersion
	// Unidirectional returns whether transport is unidirectional. For
	// unidirectional transports Client writes Push protobuf messages
	// without additional wrapping pushes into Reply type.
	Unidirectional() bool
	// Emulation must return true for transport that uses Centrifuge emulation layer.
	// See EmulationHandler for more details.
	Emulation() bool
	// DisabledPushFlags returns a disabled push flags for specific transport.
	// For example this allows to disable sending Disconnect push in case of
	// bidirectional WebSocket implementation since disconnect data sent inside
	// Close frame.
	DisabledPushFlags() uint64
	// PingPongConfig returns application-level server-to-client ping
	// configuration.
	PingPongConfig() PingPongConfig
}

TransportInfo has read-only transport description methods. Some of these methods can modify the behaviour of Client.

type TransportWriteEvent added in v0.18.0 

type TransportWriteEvent struct {
	// Data represents single Centrifuge protocol message which is going to be sent
	// into the connection. For unidirectional transports this is an encoded protocol.Push
	// type, for bidirectional transports this is an encoded protocol.Reply type.
	Data []byte
	// Channel will be set if TransportWriteEvent relates to some channel.
	Channel string
	// Key of Publication (optional).
	Key string
	// FrameType tells what is being sent inside Data.
	FrameType protocol.FrameType
}

TransportWriteEvent called just before sending data into the client connection. The event is triggered from inside each client's message queue consumer – so it should not directly affect Hub broadcast latencies.

type TransportWriteHandler added in v0.18.0 

type TransportWriteHandler func(*Client, TransportWriteEvent) bool

TransportWriteHandler called just before writing data to the Transport. At this moment application can skip sending data to a client returning false from a handler. The main purpose of this handler is not a message filtering based on data content but rather tracing stuff.

type Unsubscribe added in v0.23.0 

type Unsubscribe struct {
	// Code is unsubscribe code. Several unsubscribe codes already used by
	// a library, see for example UnsubscribeCodeClient, UnsubscribeCodeDisconnect,
	// UnsubscribeCodeServer, UnsubscribeCodeInsufficient. In theory, we can also
	// allow applications to set their custom unsubscribe codes in the future.
	Code uint32 `json:"code"`
	// Reason is a short description of unsubscribe code for humans. Suitable for
	// logs for better connection behavior observability.
	Reason string `json:"reason,omitempty"`
}

Unsubscribe describes how client must be unsubscribed (or was unsubscribed) from a channel. Codes for unsubscribe advices going to client connections must be in range [2000, 2999]. Unsubscribe codes >= 2500 coming from server to client result into resubscribe attempt. Codes [0, 2099] and [2500, 2599] are reserved for Centrifuge library internal use and must not be used by applications to create custom Unsubscribe structs.

func (Unsubscribe) String added in v0.23.0 

func (d Unsubscribe) String() string

String representation.

type UnsubscribeEvent 

type UnsubscribeEvent struct {
	// Channel client unsubscribed from.
	Channel string
	// ServerSide set to true for server-side subscription unsubscribe events.
	ServerSide bool
	// Unsubscribe identifies the source of unsubscribe (i.e. why unsubscribed event happened).
	Unsubscribe
	// Disconnect can be additionally set to identify the reason of disconnect when Unsubscribe.Code
	// is UnsubscribeCodeDisconnect - i.e. when unsubscribe caused by a client disconnection process.
	// Otherwise, it's nil.
	Disconnect *Disconnect
	// Subscribed is true when the client was subscribed to the channel. It is
	// false when the event ends a subscribe attempt which SubscribeHandler allowed
	// but which never became a subscription: Centrifuge refused it after the
	// handler (for example, an invalid expiration or a failed history or map state
	// read, reported with UnsubscribeCodeServer), or the attempt was abandoned by an
	// unsubscribe or by a disconnect while it was in progress (reported with that
	// unsubscribe's or disconnect's code). Map subscriptions are subscribed only
	// once they reach the live phase, so an unsubscribe during state or stream
	// pagination also has Subscribed set to false.
	Subscribed bool
}

UnsubscribeEvent contains fields related to unsubscribe event.

type UnsubscribeHandler 

type UnsubscribeHandler func(UnsubscribeEvent)

UnsubscribeHandler called when client unsubscribed from channel, and also when a subscribe attempt allowed by SubscribeHandler ended without becoming a subscription (see UnsubscribeEvent.Subscribed). Every SubscribeHandler call whose SubscribeCallback is invoked with a nil error is followed by exactly one UnsubscribeHandler call for that channel, so per-subscription state created in SubscribeHandler can be released in UnsubscribeHandler. A SubscribeCallback invoked with an error is not followed by UnsubscribeHandler.

For an attempt which ended without becoming a subscription, the handler is called from a separate goroutine (never from inside the SubscribeCallback call), possibly after the error reply was sent to the client. A subscribe to the same channel calls SubscribeHandler only after that call was made, so it comes before the SubscribeHandler call of the next attempt on the channel (unless it takes more than 5 seconds: a warning is logged then and SubscribeHandler is called; until all such calls of the client are done, its subscribes don't wait for them). Subscribes to other channels don't wait, unless the client has many such calls in progress. The same holds for the UnsubscribeHandler call of a subscription the server unsubscribes (Client.Unsubscribe, Node.Unsubscribe, expiration, insufficient state); for a client unsubscribe it is made before the reply. This does not apply to server-side subscriptions (Client.Subscribe), which don't call SubscribeHandler.

On disconnect, UnsubscribeHandler calls for the connection's subscriptions and the attempts in progress come before DisconnectHandler: Centrifuge waits for subscribes in progress (up to 5 seconds for each, 10 seconds in total), then up to 30 seconds for UnsubscribeHandler calls in progress (a warning is logged then), before calling DisconnectHandler. An attempt whose SubscribeCallback is invoked after that ends after DisconnectHandler. Release per-connection state in DisconnectHandler and per-subscription state in UnsubscribeHandler, in a way which works in either order. If a handler allocates something for an allowed subscription, it should do so before invoking the SubscribeCallback.

type UnsubscribeOption added in v0.5.0 

type UnsubscribeOption func(options *UnsubscribeOptions)

UnsubscribeOption is a type to represent various Unsubscribe options.

func WithCustomUnsubscribe added in v0.23.0 

func WithCustomUnsubscribe(unsubscribe Unsubscribe) UnsubscribeOption

WithCustomUnsubscribe allows setting custom Unsubscribe.

func WithUnsubscribeAllUsers added in v0.39.0 

func WithUnsubscribeAllUsers(enabled bool) UnsubscribeOption

WithUnsubscribeAllUsers changes the meaning of an empty userID passed to Node.Unsubscribe from "anonymous bucket only" to "all users including anonymous." It is a no-op when userID is non-empty.

func WithUnsubscribeClient added in v0.16.0 

func WithUnsubscribeClient(clientID string) UnsubscribeOption

WithUnsubscribeClient allows setting client ID that should be unsubscribed. This option not used when Client.Unsubscribe called.

func WithUnsubscribeLabelFilter added in v0.39.0 

func WithUnsubscribeLabelFilter(f *FilterNode) UnsubscribeOption

WithUnsubscribeLabelFilter restricts the unsubscribe to connections whose Client.Labels match the filter. See UnsubscribeOptions.LabelFilter for the detailed contract.

func WithUnsubscribeSession added in v0.21.1 

func WithUnsubscribeSession(sessionID string) UnsubscribeOption

WithUnsubscribeSession allows setting session ID that should be unsubscribed. This option not used when Client.Unsubscribe called.

type UnsubscribeOptions added in v0.5.0 

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

UnsubscribeOptions ...

type UntrackEvent added in v0.39.0 

type UntrackEvent struct {
	// Channel client untracked keys from.
	Channel string
	// Keys are the keys being untracked.
	Keys []string
}

UntrackEvent contains fields related to an untrack request on a keyed channel.

type UntrackHandler added in v0.39.0 

type UntrackHandler func(UntrackEvent)

UntrackHandler called when client untracks keys on a keyed channel.

type WebsocketConfig 

type WebsocketConfig struct {
	// CheckOrigin func to provide custom origin check logic.
	// nil means that sameHostOriginCheck function will be used which
	// expects Origin host to match request Host.
	CheckOrigin func(r *http.Request) bool

	// ReadBufferSize is a parameter that is used for raw websocket Upgrader.
	// If set to zero reasonable default value will be used.
	ReadBufferSize int

	// WriteBufferSize is a parameter that is used for raw websocket Upgrader.
	// If set to zero reasonable default value will be used.
	WriteBufferSize int

	// UseWriteBufferPool enables using buffer pool for writes.
	UseWriteBufferPool bool

	// MessageSizeLimit sets the maximum size in bytes of allowed message from client.
	// By default, 65536 bytes (64KB) will be used.
	// Note that with Compression enabled this only bounds the compressed bytes
	// received on the wire - see DecompressedMessageSizeLimit which bounds the
	// size after decompression.
	MessageSizeLimit int

	// DecompressedMessageSizeLimit sets the maximum size in bytes of a single
	// message after permessage-deflate decompression. It only has an effect when
	// Compression is enabled. MessageSizeLimit alone bounds only the compressed
	// bytes received on the wire, so without this limit a small compressed frame
	// could be inflated into a much larger amount of memory (a "decompression
	// bomb"). When set to zero (the default) the limit is derived from
	// MessageSizeLimit multiplied by 10 (const multiplier).
	DecompressedMessageSizeLimit int

	// WriteTimeout is maximum time of write message operation.
	// Slow client will be disconnected.
	// By default, 1 * time.Second will be used.
	WriteTimeout time.Duration

	// Compression allows enabling websocket permessage-deflate
	// compression support for raw websocket connections. It does
	// not guarantee that compression will be used - i.e. it only
	// says that server will try to negotiate it with client.
	// Note: enabling compression may lead to performance degradation.
	Compression bool

	// CompressionLevel sets a level for websocket compression.
	// See possible value description at https://golang.org/pkg/compress/flate/#NewWriter
	CompressionLevel int

	// CompressionMinSize allows setting minimal limit in bytes for
	// message to use compression when writing it into client connection.
	// By default, it's 0 - i.e. all messages will be compressed when
	// WebsocketCompression enabled and compression negotiated with client.
	CompressionMinSize int

	// CompressionPreparedMessageCacheSize when greater than zero tells Centrifuge to use
	// prepared WebSocket messages for connections with compression. This generally introduces
	// overhead but at the same time may drastically reduce compression memory and CPU spikes
	// during broadcasts. See also BenchmarkWsBroadcastCompressionCache.
	// This option is EXPERIMENTAL, do not use in production. Contact maintainers if it
	// works well for your use case, and you want to enable it in production.
	CompressionPreparedMessageCacheSize int64

	// DisableHTTP1Upgrade disables support for HTTP/1.1 Upgrade WebSocket handshakes.
	// When true, only HTTP/2 Extended CONNECT is accepted (EnableHTTP2ExtendedConnect
	// must be true, otherwise no connections will be accepted).
	DisableHTTP1Upgrade bool

	// ProcessCommandsOffReadLoop makes each connection's read loop hand every
	// frame to a short-lived goroutine and wait for it, instead of processing the
	// frame itself.
	//
	// It exists to keep the read loop's goroutine stack small. Command
	// processing runs the whole dispatch chain - handlers, publish, broker,
	// broadcast fan-out - and a goroutine's stack only ever grows to the deepest
	// call it makes and is not given back while the goroutine lives. A read loop
	// that has processed one command therefore keeps that peak stack for as long
	// as the connection is open, even while completely idle. Moving the work to a
	// goroutine that exits afterwards returns the deep stack to the runtime's
	// stack cache, where connections reuse it instead of each owning one.
	//
	// The trade is one goroutine start and two scheduler handoffs per frame.
	// That is a latency cost, not a throughput one: it is visible on a
	// connection that sends a command and waits for the reply with nothing else
	// going on, and it disappears once many connections are busy at the same
	// time, where the handoff overlaps with other work. Under that kind of load
	// per-connection memory is lower as well, because only the connections
	// actually mid-frame hold a deep stack rather than every one of them.
	//
	// Ordering is unaffected: the read loop waits for each frame to finish
	// before reading the next.
	//
	// Off by default. It suits large pools of mostly-idle connections, where it
	// roughly halves the read loop's stack.
	ProcessCommandsOffReadLoop bool

	PingPongConfig
}

WebsocketConfig represents config for WebsocketHandler.

type WebsocketHandler 

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

WebsocketHandler handles WebSocket client connections. WebSocket protocol is a bidirectional connection between a client and a server for low-latency communication.

func NewWebsocketHandler 

func NewWebsocketHandler(node *Node, config WebsocketConfig) *WebsocketHandler

NewWebsocketHandler creates new WebsocketHandler.

func (*WebsocketHandler) ServeHTTP 

func (s *WebsocketHandler) ServeHTTP(rw http.ResponseWriter, r *http.Request)

Source Files

View all Source files

Directories

Path Synopsis
_examples module
internal
bpool
cancelctx
controlpb
controlproto
convert
dissolve
epoch
Package epoch provides epoch string generation for stream position tracking.
 Package epoch provides epoch string generation for stream position tracking.
filter
fossilutf8
Package fossilutf8 aligns fossil deltas to UTF-8 character boundaries.
 Package fossilutf8 aligns fossil deltas to UTF-8 character boundaries.
lazyutf8
Package lazyutf8 answers whether a payload is valid UTF-8, scanning it at most once.
 Package lazyutf8 answers whether a payload is valid UTF-8, scanning it at most once.
memstream
metricsnap
Package metricsnap takes snapshots of Prometheus metrics for node info: what each metric did since the previous snapshot, flattened into key-value items, e.g.
 Package metricsnap takes snapshots of Prometheus metrics for node info: what each metric did since the previous snapshot, flattened into key-value items, e.g.
nowtime
priority
Package priority provides priority queue.
 Package priority provides priority queue.
queue
readerpool
recovery
redispartition
Package redispartition exposes precomputed string tags whose CRC16 hash slots distribute evenly across Redis Cluster nodes for any cluster size up to the configured number of partitions.
 Package redispartition exposes precomputed string tags whose CRC16 hash slots distribute evenly across Redis Cluster nodes for any cluster size up to the configured number of partitions.
redispartition/cmd/precompute command
Command precompute regenerates ../../precomputed.go with partition tags for a configurable list of partition counts.
 Command precompute regenerates ../../precomputed.go with partition tags for a configurable list of partition counts.
saferand
timers
websocket
Package websocket implements the WebSocket protocol defined in RFC 6455.
 Package websocket implements the WebSocket protocol defined in RFC 6455.
websocket/examples/autobahn command
Command server is a test server for the Autobahn WebSockets Test Suite.
 Command server is a test server for the Autobahn WebSockets Test Suite.

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.